velesdb_memory/context/segment.rs
1//! Deterministic transcript segmentation for the `compile_transcript` MCP
2//! tool (V2b-2, see the crate's `PLAN.md`, section V2b).
3//!
4//! [`segment_transcript`] turns a raw agent-session transcript — plain text
5//! with role markers, or JSONL — into an ordered list of
6//! [`TranscriptSegment`]s, each wrapping an ordinary [`super::ContextFragment`]
7//! plus the audit metadata (`turn`, `role`, `kind`, byte range) the
8//! `compile_transcript` tool reports alongside the compiled context. The
9//! resulting fragments feed the existing, unmodified [`super::ContextCompiler`]
10//! pipeline — this module only decides *how to cut the transcript up*, never
11//! what to keep or drop.
12//!
13//! **Zero regex, zero clock, single linear scan per stage** — same
14//! determinism contract as [`super::chunk`]: the same transcript + the same
15//! [`SegmentationPolicy`] always segment byte-identically (see
16//! `segmentation_twice_is_byte_identical` in the test suite).
17//!
18//! # Pipeline
19//!
20//! 1. **Format detection** ([`detect_and_segment`]): `jsonl` when every
21//! non-empty line parses as a `{role, content}` JSON object, `plain`
22//! otherwise. A caller-forced format that does not parse is a hard error —
23//! never a silent fallback to the other format.
24//! 2. **Turns**: `jsonl` — one line, one turn, `role` taken directly from the
25//! parsed JSON. `plain` — a CLOSED table of markers (`"System:"`,
26//! `"User:"`, `"Human:"`, `"Assistant:"`, `"AI:"`, `"Tool:"`,
27//! `"### User"`, `"### Assistant"`), first match at the start of a line
28//! opens a new turn; a transcript with no marker at all is one turn with
29//! `role: None`.
30//! 3. **Sub-segmentation** (`plain` turns only — a `jsonl` turn's `content` is
31//! a JSON-decoded string, not a byte-aligned slice of the transcript, so
32//! it is never re-scanned; the underlying `content.contains("```")` /
33//! value-density rules in [`super::classify`] still see it, unaffected):
34//! fenced code blocks ([`super::chunk::fence_segments`]) become atomic
35//! `code` segments; runs of at least 8 consecutive log-like lines (a
36//! volatile timestamp/pid prefix — [`super::log_normalize::mask_volatile_prefix`]
37//! — or a raw-text repeat) become `log` segments; everything else is
38//! `body`.
39//! 4. **Normalization**: an unsplittable fence over
40//! [`crate::limits::MAX_FRAGMENT_BYTES`] is a hard error (never silently
41//! truncated); an oversized `body` segment is re-split with
42//! [`super::chunk_text`]; segments under
43//! [`SegmentationPolicy::min_segment_bytes`] merge into an adjacent
44//! segment of the *same turn and kind*; more than
45//! [`crate::limits::MAX_FRAGMENTS`] segments after merging is a hard,
46//! actionable error ("raise `min_segment_bytes`") — never a silent drop.
47//!
48//! Every error surfaces as [`crate::error::MemoryError::ContextOverLimit`] or
49//! [`crate::error::MemoryError::IngestDisabled`]/[`crate::error::MemoryError::IngestOutsideRoots`]/
50//! [`crate::error::MemoryError::IngestPath`] (the last three only for a
51//! `path`-sourced transcript, via [`super::ingest::resolve_transcript_path`])
52//! — the same `INVALID_PARAMS`-category taxonomy `compile_context` already
53//! uses, deliberately not a new variant for this PR.
54
55use std::collections::BTreeMap;
56use std::ops::Range;
57
58use schemars::JsonSchema;
59use serde::{Deserialize, Serialize};
60use serde_json::{Map, Value};
61
62use super::chunk::{self, chunk_text, ChunkBoundary, ChunkPolicy};
63use super::log_normalize::mask_volatile_prefix;
64use super::model::ContextFragment;
65use crate::error::MemoryError;
66use crate::limits::{MAX_FRAGMENTS, MAX_FRAGMENT_BYTES, MAX_TRANSCRIPT_BYTES};
67
68/// A contiguous run of at least this many candidate log lines becomes a
69/// `log` segment (see the module docs' step 3). Chosen high enough that an
70/// ordinary short warning burst stays `body` (nothing to abstract), low
71/// enough that a real log dump — which `abstract.log_dedup` exists to
72/// collapse — is reliably recognized.
73const MIN_LOG_RUN_LINES: usize = 8;
74
75/// Which transcript format to assume, or detect automatically.
76#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
77#[serde(rename_all = "lowercase")]
78pub enum SegmentFormat {
79 /// Detect `jsonl` vs `plain` from the transcript itself (the default).
80 Auto,
81 /// Force plain-text, marker-based turn splitting — a transcript that
82 /// happens to also be valid JSONL is still segmented as plain text.
83 Plain,
84 /// Force one-line-one-turn JSONL parsing — a line that does not parse as
85 /// a `{role, content}` object is a hard error, never a silent fallback.
86 Jsonl,
87}
88
89/// What kind of content a sub-segment carries — decides whether it was cut
90/// out as an atomic fence, a detected log run, or ordinary prose/dialogue
91/// left for [`super::classify`]'s rule table to judge.
92#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
93#[serde(rename_all = "lowercase")]
94pub enum SegmentKind {
95 /// Ordinary text — [`ContextFragment::kind`] stays `None`, so the
96 /// existing classification rules (code fence, URL, negative constraint,
97 /// value density, …) decide its fate exactly as for `compile_context`.
98 Body,
99 /// A triple-backtick-fenced block, cut out atomically by
100 /// [`super::chunk::fence_segments`]. Tagged `kind = "code"` so
101 /// [`super::classify::classify`]'s `preserve.code_fence` rule matches
102 /// even for a fence whose content does not itself literally contain
103 /// `` ``` `` (defense in depth; it usually does).
104 Code,
105 /// A run of at least [`MIN_LOG_RUN_LINES`] log-like lines. Tagged
106 /// `kind = "log"` so `abstract.log_dedup` can consider it for
107 /// repeated-line collapsing exactly like a caller-declared `kind: "log"`
108 /// fragment in `compile_context`.
109 Log,
110}
111
112impl SegmentKind {
113 /// The [`ContextFragment::kind`] hint this segment kind maps to —
114 /// `None` for `body` (let the rule table decide unconstrained).
115 fn fragment_kind(self) -> Option<&'static str> {
116 match self {
117 Self::Body => None,
118 Self::Code => Some("code"),
119 Self::Log => Some("log"),
120 }
121 }
122}
123
124/// Tuning knobs for [`segment_transcript`]. `Default` is the recommended
125/// profile.
126#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
127#[serde(default)]
128#[schemars(transform = crate::schema::strip_int_formats)]
129pub struct SegmentationPolicy {
130 /// Which format to assume (see [`SegmentFormat`]). Default [`SegmentFormat::Auto`].
131 pub format: SegmentFormat,
132 /// Segments under this many bytes merge into an adjacent segment of the
133 /// same turn and kind (see the module docs' step 4). Default `256`.
134 pub min_segment_bytes: usize,
135 /// When `true` (the default) and [`SegmentationPolicy::format`]
136 /// determines the FIRST turn's role is `"system"` (case-insensitive),
137 /// every segment of that turn is marked `metadata.cache = true` — the
138 /// same signal `compile_context`'s `cache.stable_prefix` rule reads, so
139 /// a system prompt turn becomes the compiled output's stable,
140 /// cache-friendly prefix without the caller hand-annotating it.
141 pub cache_system_turn: bool,
142}
143
144impl Default for SegmentationPolicy {
145 fn default() -> Self {
146 Self {
147 format: SegmentFormat::Auto,
148 min_segment_bytes: 256,
149 cache_system_turn: true,
150 }
151 }
152}
153
154/// One segmented piece of the transcript: an ordinary [`ContextFragment`]
155/// (ready to feed [`super::ContextCompiler`]) plus the audit metadata the
156/// `compile_transcript` tool reports in its `segmentation.segments` list.
157#[derive(Debug, Clone)]
158pub struct TranscriptSegment {
159 /// The fragment this segment produces — feed it straight into a
160 /// [`super::CompileRequest::fragments`] list.
161 pub fragment: ContextFragment,
162 /// Which turn (0-based, transcript order) this segment belongs to.
163 pub turn: usize,
164 /// The turn's role, when one was determined (a marker match in `plain`
165 /// mode, or the parsed `role` field in `jsonl` mode). `None` for a
166 /// `plain` transcript with no matching marker at all.
167 pub role: Option<String>,
168 /// What kind of content this segment carries.
169 pub kind: SegmentKind,
170 /// Start byte offset (inclusive) of this segment in the ORIGINAL
171 /// transcript text.
172 pub byte_start: usize,
173 /// End byte offset (exclusive) of this segment in the ORIGINAL
174 /// transcript text.
175 pub byte_end: usize,
176}
177
178/// The full result of [`segment_transcript`]: the detected format, the
179/// segments, and how much normalization merging did.
180#[derive(Debug, Clone)]
181pub struct SegmentationOutcome {
182 /// `jsonl` or `plain` — never [`SegmentFormat::Auto`], which only ever
183 /// names a caller's REQUEST, not a detected outcome.
184 pub format_detected: SegmentFormat,
185 /// The final segments, in transcript order.
186 pub segments: Vec<TranscriptSegment>,
187 /// How many segments the [`SegmentationPolicy::min_segment_bytes`] merge
188 /// step eliminated (`pieces_before_merge - segments.len()`).
189 pub merged_segments: usize,
190}
191
192/// Segment `text` under `policy` — see the module docs for the full
193/// pipeline. Pure: no I/O, no clock, no randomness; the same `text` +
194/// `policy` always produce byte-identical output.
195///
196/// # Errors
197/// [`MemoryError::ContextOverLimit`] when `text` exceeds
198/// [`MAX_TRANSCRIPT_BYTES`], when [`SegmentFormat::Jsonl`] is forced but a
199/// line does not parse as a `{role, content}` object, when an unsplittable
200/// fence exceeds [`MAX_FRAGMENT_BYTES`], or when the segment count after
201/// merging still exceeds [`MAX_FRAGMENTS`].
202pub fn segment_transcript(
203 text: &str,
204 policy: &SegmentationPolicy,
205) -> Result<SegmentationOutcome, MemoryError> {
206 if text.len() > MAX_TRANSCRIPT_BYTES {
207 return Err(MemoryError::ContextOverLimit(format!(
208 "transcript of {} bytes exceeds the cap of {MAX_TRANSCRIPT_BYTES} bytes",
209 text.len()
210 )));
211 }
212
213 let (format_detected, pieces) = detect_and_segment(text, policy.format)?;
214 reject_oversized_fences(&pieces)?;
215 let pieces = resplit_oversized_bodies(text, pieces);
216 let pieces_before_merge = pieces.len();
217 let merged = merge_tiny(pieces, policy.min_segment_bytes);
218 if merged.len() > MAX_FRAGMENTS {
219 return Err(MemoryError::ContextOverLimit(format!(
220 "transcript segmented into {} fragments, exceeding the cap of {MAX_FRAGMENTS} — \
221 raise segmentation.min_segment_bytes to merge more small segments",
222 merged.len()
223 )));
224 }
225 let merged_segments = pieces_before_merge - merged.len();
226 let segments = merged
227 .into_iter()
228 .map(|piece| build_segment(text, piece, policy))
229 .collect();
230 Ok(SegmentationOutcome {
231 format_detected,
232 segments,
233 merged_segments,
234 })
235}
236
237// --- Raw (pre-normalization) pieces -----------------------------------------
238
239/// A sub-segment before normalization: still tied to the ORIGINAL text's byte
240/// range, except `content_override` — set only for a `jsonl` turn (and its
241/// re-split children), whose fragment content is a JSON-decoded string with
242/// no byte-aligned slice of the raw transcript (JSON escaping means the
243/// decoded text is not a substring of the source bytes). When set, `range`
244/// still names the raw JSON line's span (needed so the segmentation-wide
245/// byte ranges keep partitioning the transcript), but the fragment's
246/// `content` comes from `content_override`, never `text[range]`.
247struct RawPiece {
248 kind: SegmentKind,
249 range: Range<usize>,
250 turn: usize,
251 role: Option<String>,
252 content_override: Option<String>,
253}
254
255/// Detect the format and produce the initial (pre-normalization) pieces in
256/// one pass — for `jsonl` this avoids parsing every line twice (once to
257/// detect, once to build).
258fn detect_and_segment(
259 text: &str,
260 requested: SegmentFormat,
261) -> Result<(SegmentFormat, Vec<RawPiece>), MemoryError> {
262 match requested {
263 SegmentFormat::Plain => Ok((SegmentFormat::Plain, plain_pieces(text))),
264 SegmentFormat::Jsonl => {
265 let pieces = jsonl_pieces(text).map_err(MemoryError::ContextOverLimit)?;
266 Ok((SegmentFormat::Jsonl, pieces))
267 }
268 SegmentFormat::Auto => {
269 if !text.is_empty() {
270 if let Ok(pieces) = jsonl_pieces(text) {
271 return Ok((SegmentFormat::Jsonl, pieces));
272 }
273 }
274 Ok((SegmentFormat::Plain, plain_pieces(text)))
275 }
276 }
277}
278
279// --- JSONL -------------------------------------------------------------------
280
281/// One JSONL line's required shape. Both fields are mandatory: a line
282/// missing either — or not a JSON object at all — fails to parse, which
283/// [`detect_and_segment`] treats as "not jsonl" in [`SegmentFormat::Auto`]
284/// and as a hard error under a forced [`SegmentFormat::Jsonl`].
285#[derive(Deserialize)]
286struct JsonlLine {
287 role: String,
288 content: String,
289}
290
291/// Parse every non-blank line of `text` as one JSONL turn. A wholly empty
292/// line (`""` once the trailing `\r`/`\n` is stripped) never fails parsing
293/// and never opens a turn of its own — its bytes fold into the PRECEDING
294/// piece's range (or, for a leading blank run with no preceding piece yet,
295/// are deferred and prepended onto the first real turn once one arrives) so
296/// the byte ranges keep partitioning `text` exactly. Without this, a
297/// perfectly valid JSONL transcript that merely uses a blank line as a
298/// separator would fail to parse and (in [`SegmentFormat::Auto`]) silently
299/// fall back to a single roleless `plain` turn.
300///
301/// `Err` names the first (1-based) offending LINE — not turn — number: the
302/// first failure short-circuits, so a caller forcing `jsonl` on a bad
303/// transcript gets an actionable pointer instead of a generic "not jsonl".
304fn jsonl_pieces(text: &str) -> Result<Vec<RawPiece>, String> {
305 let mut pieces: Vec<RawPiece> = Vec::new();
306 let mut pending_prefix_start: Option<usize> = None;
307 let mut turn = 0_usize;
308 let mut cursor = 0_usize;
309 for (line_index, line) in text.split_inclusive('\n').enumerate() {
310 let start = cursor;
311 cursor += line.len();
312 let trimmed = line.trim_end_matches(['\r', '\n']);
313 if trimmed.is_empty() {
314 if let Some(last) = pieces.last_mut() {
315 last.range.end = cursor;
316 } else {
317 pending_prefix_start.get_or_insert(start);
318 }
319 continue;
320 }
321 let parsed: JsonlLine = serde_json::from_str(trimmed).map_err(|err| {
322 format!(
323 "jsonl line {}: not a valid {{role, content}} object: {err}",
324 line_index + 1
325 )
326 })?;
327 let piece_start = pending_prefix_start.take().unwrap_or(start);
328 pieces.push(RawPiece {
329 kind: SegmentKind::Body,
330 range: piece_start..cursor,
331 turn,
332 role: Some(parsed.role),
333 content_override: Some(parsed.content),
334 });
335 turn += 1;
336 }
337 if pieces.is_empty() {
338 // Every line (if any at all) was blank — nothing real to call
339 // jsonl; Auto mode falls back to plain, a forced jsonl request gets
340 // an honest error instead of a silently empty result.
341 return Err("no non-blank jsonl line found".to_owned());
342 }
343 Ok(pieces)
344}
345
346// --- Plain ---------------------------------------------------------------
347
348/// The CLOSED table of plain-text turn markers, checked in order — the first
349/// one a line starts with wins. Never a caller-supplied pattern, so turn
350/// detection stays deterministic and predictable (a "User:" cited in prose
351/// is a known, accepted false positive — see the crate README).
352const PLAIN_MARKERS: &[&str] = &[
353 "System:",
354 "User:",
355 "Human:",
356 "Assistant:",
357 "AI:",
358 "Tool:",
359 "### User",
360 "### Assistant",
361];
362
363/// The first [`PLAIN_MARKERS`] entry `line` starts with, if any.
364fn match_marker(line: &str) -> Option<&'static str> {
365 PLAIN_MARKERS
366 .iter()
367 .find(|marker| line.starts_with(*marker))
368 .copied()
369}
370
371/// A marker's role label: `"### User"` → `"User"`, `"System:"` → `"System"`.
372fn marker_role(marker: &str) -> String {
373 marker
374 .strip_prefix("### ")
375 .unwrap_or(marker)
376 .trim_end_matches(':')
377 .to_owned()
378}
379
380/// Split `text` into plain-format turns: `(byte_range, role)`, in order,
381/// partitioning `text` exactly. No marker anywhere in `text` yields exactly
382/// one turn covering the whole text with `role: None`.
383fn plain_turns(text: &str) -> Vec<(Range<usize>, Option<String>)> {
384 let mut turns = Vec::new();
385 let mut turn_start = 0_usize;
386 let mut pending_role: Option<String> = None;
387 let mut cursor = 0_usize;
388 for line in text.split_inclusive('\n') {
389 let line_start = cursor;
390 if let Some(marker) = match_marker(line) {
391 if line_start > turn_start {
392 turns.push((turn_start..line_start, pending_role.take()));
393 }
394 pending_role = Some(marker_role(marker));
395 turn_start = line_start;
396 }
397 cursor += line.len();
398 }
399 turns.push((turn_start..text.len(), pending_role));
400 turns
401}
402
403/// Build the initial pieces for a `plain` transcript: turns, then within
404/// each turn's slice, fences (atomic `code`) and log runs (`log`), the rest
405/// `body` — see the module docs' step 3.
406fn plain_pieces(text: &str) -> Vec<RawPiece> {
407 let mut pieces = Vec::new();
408 for (turn, (range, role)) in plain_turns(text).into_iter().enumerate() {
409 if range.is_empty() {
410 continue;
411 }
412 for segment in chunk::fence_segments(&text[range.clone()]) {
413 match segment {
414 chunk::Segment::Fence(relative) => pieces.push(RawPiece {
415 kind: SegmentKind::Code,
416 range: (range.start + relative.start)..(range.start + relative.end),
417 turn,
418 role: role.clone(),
419 content_override: None,
420 }),
421 chunk::Segment::Plain(relative) => {
422 let absolute = (range.start + relative.start)..(range.start + relative.end);
423 for (kind, sub_range) in log_split(text, absolute) {
424 pieces.push(RawPiece {
425 kind,
426 range: sub_range,
427 turn,
428 role: role.clone(),
429 content_override: None,
430 });
431 }
432 }
433 }
434 }
435 }
436 pieces
437}
438
439/// Split `range` of `text` into alternating `body`/`log` pieces: a maximal
440/// run of at least [`MIN_LOG_RUN_LINES`] consecutive "log-candidate" lines
441/// (a volatile timestamp/pid prefix, or a line that repeats elsewhere in
442/// `range`) becomes one `log` piece; every other line stays `body`,
443/// contiguous runs of it merged into one piece. Single linear scan.
444fn log_split(text: &str, range: Range<usize>) -> Vec<(SegmentKind, Range<usize>)> {
445 if range.is_empty() {
446 return Vec::new();
447 }
448 let slice = &text[range.clone()];
449 let mut lines: Vec<(Range<usize>, &str)> = Vec::new();
450 let mut cursor = range.start;
451 for line in slice.split_inclusive('\n') {
452 let end = cursor + line.len();
453 lines.push((cursor..end, line));
454 cursor = end;
455 }
456 if lines.is_empty() {
457 return Vec::new();
458 }
459
460 let trimmed: Vec<&str> = lines
461 .iter()
462 .map(|(_, line)| line.trim_end_matches(['\r', '\n']))
463 .collect();
464 let mut repeat_counts: BTreeMap<&str, usize> = BTreeMap::new();
465 for line in &trimmed {
466 *repeat_counts.entry(line).or_insert(0) += 1;
467 }
468 let candidate: Vec<bool> = trimmed
469 .iter()
470 .map(|line| {
471 !line.is_empty() && (mask_volatile_prefix(line).is_some() || repeat_counts[line] > 1)
472 })
473 .collect();
474
475 let mut pieces = Vec::new();
476 let mut body_start: Option<usize> = None;
477 let mut index = 0_usize;
478 while index < lines.len() {
479 if candidate[index] {
480 let run_start = index;
481 while index < lines.len() && candidate[index] {
482 index += 1;
483 }
484 if index - run_start >= MIN_LOG_RUN_LINES {
485 if let Some(start) = body_start.take() {
486 pieces.push((
487 SegmentKind::Body,
488 lines[start].0.start..lines[run_start - 1].0.end,
489 ));
490 }
491 pieces.push((
492 SegmentKind::Log,
493 lines[run_start].0.start..lines[index - 1].0.end,
494 ));
495 } else if body_start.is_none() {
496 body_start = Some(run_start);
497 }
498 } else {
499 if body_start.is_none() {
500 body_start = Some(index);
501 }
502 index += 1;
503 }
504 }
505 if let Some(start) = body_start {
506 pieces.push((
507 SegmentKind::Body,
508 lines[start].0.start..lines[lines.len() - 1].0.end,
509 ));
510 }
511 pieces
512}
513
514// --- Normalization -----------------------------------------------------------
515
516/// Reject an unsplittable fence over [`MAX_FRAGMENT_BYTES`] — a fence is
517/// always atomic (never cut, see [`super::chunk`]), so an oversized one
518/// cannot be brought under the cap the way a `body` piece can.
519///
520/// # Errors
521/// [`MemoryError::ContextOverLimit`] naming the first oversized fence found.
522fn reject_oversized_fences(pieces: &[RawPiece]) -> Result<(), MemoryError> {
523 if let Some(piece) = pieces
524 .iter()
525 .find(|piece| piece.kind == SegmentKind::Code && piece.range.len() > MAX_FRAGMENT_BYTES)
526 {
527 return Err(MemoryError::ContextOverLimit(format!(
528 "an unsplittable fenced code block of {} bytes exceeds the cap of {MAX_FRAGMENT_BYTES} bytes",
529 piece.range.len()
530 )));
531 }
532 Ok(())
533}
534
535/// Re-split every `body` or `log` piece over [`MAX_FRAGMENT_BYTES`] — see
536/// [`resplit_body`] and [`resplit_log`] for the two (deliberately different)
537/// strategies. A `code` piece is never touched here: it is atomic by
538/// construction (a fence is never cut, see [`super::chunk`]) and already
539/// rejected outright by [`reject_oversized_fences`] when oversized.
540fn resplit_oversized_bodies(text: &str, pieces: Vec<RawPiece>) -> Vec<RawPiece> {
541 let chunk_policy = ChunkPolicy {
542 max_chunk_bytes: MAX_FRAGMENT_BYTES,
543 overlap_bytes: 0,
544 boundary: ChunkBoundary::Paragraph,
545 };
546 pieces
547 .into_iter()
548 .flat_map(|piece| resplit_one(text, piece, &chunk_policy))
549 .collect()
550}
551
552fn resplit_one(text: &str, piece: RawPiece, chunk_policy: &ChunkPolicy) -> Vec<RawPiece> {
553 match piece.kind {
554 SegmentKind::Body => resplit_body(text, piece, chunk_policy),
555 SegmentKind::Log => resplit_log(text, piece),
556 SegmentKind::Code => vec![piece],
557 }
558}
559
560/// Re-split a `body` piece over [`MAX_FRAGMENT_BYTES`] with [`chunk_text`] —
561/// the same re-chunker `compile_context` itself uses for an oversized
562/// fragment. A `jsonl` piece's decoded `content_override` has no
563/// byte-aligned mapping back to the raw (JSON-escaped) source line, so its
564/// re-split children all keep the ORIGINAL line's full byte range — a
565/// documented, deliberately narrow trade-off: the byte-range-covers-the-
566/// transcript property holds at the turn level regardless, and a single
567/// JSONL line's `content` exceeding 1 MiB is an extreme edge case.
568fn resplit_body(text: &str, piece: RawPiece, chunk_policy: &ChunkPolicy) -> Vec<RawPiece> {
569 let effective_len = piece
570 .content_override
571 .as_ref()
572 .map_or(piece.range.len(), String::len);
573 if effective_len <= MAX_FRAGMENT_BYTES {
574 return vec![piece];
575 }
576 match &piece.content_override {
577 Some(content) => chunk_text(content, chunk_policy)
578 .into_iter()
579 .map(|chunk| RawPiece {
580 kind: SegmentKind::Body,
581 range: piece.range.clone(),
582 turn: piece.turn,
583 role: piece.role.clone(),
584 content_override: Some(chunk.text),
585 })
586 .collect(),
587 None => chunk_text(&text[piece.range.clone()], chunk_policy)
588 .into_iter()
589 .map(|chunk| RawPiece {
590 kind: SegmentKind::Body,
591 range: (piece.range.start + chunk.byte_range.start)
592 ..(piece.range.start + chunk.byte_range.end),
593 turn: piece.turn,
594 role: piece.role.clone(),
595 content_override: None,
596 })
597 .collect(),
598 }
599}
600
601/// Re-split a `log` piece over [`MAX_FRAGMENT_BYTES`] on LINE boundaries —
602/// never mid-line, so each resulting sub-run stays meaningful to
603/// `abstract.log_dedup` (which classifies and dedups per fragment, not
604/// across a cut line). Unlike [`resplit_body`], never [`chunk_text`]
605/// directly: paragraph-boundary chunking has no notion of "line", and would
606/// happily cut a log line in half. A `log` piece never carries a
607/// `content_override` (only `jsonl` pieces do, and `jsonl` never produces
608/// `log` — see the module docs), so this always reads straight from `text`.
609///
610/// Lines are packed greedily into chunks of at most [`MAX_FRAGMENT_BYTES`];
611/// a single line that alone exceeds the cap (extreme edge case — one log
612/// line over 1 MiB) is hard-split at char boundaries as a last resort, the
613/// same fallback [`super::chunk::chunk_text`] uses for an oversized atomic
614/// unit.
615fn resplit_log(text: &str, piece: RawPiece) -> Vec<RawPiece> {
616 if piece.range.len() <= MAX_FRAGMENT_BYTES {
617 return vec![piece];
618 }
619 let hard_split_policy = ChunkPolicy {
620 max_chunk_bytes: MAX_FRAGMENT_BYTES,
621 overlap_bytes: 0,
622 boundary: ChunkBoundary::Fixed,
623 };
624 let mut result = Vec::new();
625 let mut chunk_start = piece.range.start;
626 let mut cursor = piece.range.start;
627 for line in text[piece.range.clone()].split_inclusive('\n') {
628 let line_start = cursor;
629 let line_end = line_start + line.len();
630 cursor = line_end;
631
632 if line_end - line_start > MAX_FRAGMENT_BYTES {
633 // The line itself is oversized: seal whatever came before it,
634 // hard-split the line alone, then resume after it.
635 if chunk_start < line_start {
636 result.push(log_piece(&piece, chunk_start..line_start));
637 }
638 for hard in chunk_text(&text[line_start..line_end], &hard_split_policy) {
639 result.push(log_piece(
640 &piece,
641 (line_start + hard.byte_range.start)..(line_start + hard.byte_range.end),
642 ));
643 }
644 chunk_start = line_end;
645 continue;
646 }
647
648 if line_end - chunk_start > MAX_FRAGMENT_BYTES {
649 // Adding this line would overflow the open chunk: seal it
650 // first — `chunk_start..line_start` is guaranteed non-empty
651 // here (a lone line never exceeds the cap in this branch).
652 result.push(log_piece(&piece, chunk_start..line_start));
653 chunk_start = line_start;
654 }
655 }
656 if chunk_start < piece.range.end {
657 result.push(log_piece(&piece, chunk_start..piece.range.end));
658 }
659 result
660}
661
662/// A `log`-kind [`RawPiece`] over `range`, inheriting `source`'s turn/role —
663/// the shared constructor [`resplit_log`]'s two push sites use.
664fn log_piece(source: &RawPiece, range: Range<usize>) -> RawPiece {
665 RawPiece {
666 kind: SegmentKind::Log,
667 range,
668 turn: source.turn,
669 role: source.role.clone(),
670 content_override: None,
671 }
672}
673
674/// Merge adjacent pieces of the SAME turn and kind when either side is under
675/// `min_bytes` — see the module docs' step 4. A `jsonl` piece never merges
676/// with another (each holds its own unique `turn`, since `jsonl` is
677/// one-line-one-turn by construction), nor does any piece carrying a
678/// `content_override` (merging would require re-deriving a combined decoded
679/// string, which is not meaningful once JSON escaping is involved).
680///
681/// **Never merges past [`MAX_FRAGMENT_BYTES`]** — a piece that survived
682/// [`resplit_body`]/[`resplit_log`] is only guaranteed to be AT MOST the
683/// cap, so blindly recombining it with even a tiny neighbor can push the
684/// result back over (a ~1 MiB chunk plus a few trailing bytes, or two
685/// adjacent fences each individually under the cap). Merging is an
686/// optimization (fewer, more useful fragments), never allowed to violate the
687/// one invariant every other normalization step exists to uphold.
688fn merge_tiny(pieces: Vec<RawPiece>, min_bytes: usize) -> Vec<RawPiece> {
689 let mut merged: Vec<RawPiece> = Vec::new();
690 for piece in pieces {
691 let mergeable = merged.last().is_some_and(|last: &RawPiece| {
692 last.turn == piece.turn
693 && last.kind == piece.kind
694 && last.content_override.is_none()
695 && piece.content_override.is_none()
696 && last.range.end == piece.range.start
697 && last.range.len() + piece.range.len() <= MAX_FRAGMENT_BYTES
698 && (last.range.len() < min_bytes || piece.range.len() < min_bytes)
699 });
700 if mergeable {
701 // Safe: `mergeable` only true when `merged` is non-empty.
702 merged
703 .last_mut()
704 .expect("checked non-empty above")
705 .range
706 .end = piece.range.end;
707 } else {
708 merged.push(piece);
709 }
710 }
711 merged
712}
713
714// --- Assembly ------------------------------------------------------------
715
716/// Build the final [`TranscriptSegment`] for one normalized piece:
717/// `metadata = {role, turn}`, plus `cache: true` when
718/// [`SegmentationPolicy::cache_system_turn`] applies (turn 0, role
719/// case-insensitively `"system"`).
720fn build_segment(text: &str, piece: RawPiece, policy: &SegmentationPolicy) -> TranscriptSegment {
721 let content = piece
722 .content_override
723 .clone()
724 .unwrap_or_else(|| text[piece.range.clone()].to_owned());
725
726 let mut metadata = Map::new();
727 metadata.insert(
728 "role".to_owned(),
729 piece.role.clone().map_or(Value::Null, Value::String),
730 );
731 metadata.insert("turn".to_owned(), Value::Number(piece.turn.into()));
732 let is_first_turn_system = piece.turn == 0
733 && piece
734 .role
735 .as_deref()
736 .is_some_and(|role| role.eq_ignore_ascii_case("system"));
737 if policy.cache_system_turn && is_first_turn_system {
738 metadata.insert("cache".to_owned(), Value::Bool(true));
739 }
740
741 let fragment = ContextFragment {
742 id: None,
743 content,
744 path: None,
745 kind: piece.kind.fragment_kind().map(str::to_owned),
746 priority: None,
747 metadata: Some(metadata),
748 media: None,
749 };
750 TranscriptSegment {
751 fragment,
752 turn: piece.turn,
753 role: piece.role,
754 kind: piece.kind,
755 byte_start: piece.range.start,
756 byte_end: piece.range.end,
757 }
758}
759
760#[cfg(test)]
761#[path = "segment_tests.rs"]
762mod tests;