forge_workspace/conversation_window.rs
1//! The window a conversation is cut to when it is held or handed over.
2//!
3//! **One window, applied at every boundary.** The transport's copy of a
4//! seat's conversation lives in one, and a `Connected` cuts a resume's whole
5//! transcript to one before it rides an update - so a read that was 2.5 GiB on
6//! the seat #2150 was filed from is not copied per subscriber. The fan-out's
7//! ring is sized from the same caps, because a ring exists to fill a copy that
8//! is behind.
9//!
10//! **A window is bounded by frames and by bytes.** The count alone let a giant
11//! seat's window weigh 319 MiB (measured 2026-10-17 on the seat #1637 was
12//! filed from, in the frames a read serves - [`CONVERSATION_CAP_BYTES`] names
13//! the basis, and why the file's own lines read as about half of it) - and it
14//! is not the window's own weight that bites: a page is
15//! cut from the window, so a ceiling ABOVE the newest twenty turns changes
16//! nothing a reader feels. [`CONVERSATION_CAP_BYTES`] carries where the byte
17//! ceiling's number comes from and what it deliberately changes on the seats
18//! that reach it: a page there is the window whole rather than the newest
19//! twenty turns, and the history above it is reached the way any window's is,
20//! a `more` below the floor answered from the transcript.
21//!
22//! **And both cuts land on a turn's first frame.** A window starting part-way
23//! through a turn serves its first turn with the head missing, so what the cap
24//! sheds is counted to a frame a turn opens at - one cut, shared by both entry
25//! points, and [`crate::conversation_turns`] is where a turn opens. The one
26//! case with no whole turn to land on is a turn heavier than both ceilings
27//! and their slack: there the cut is made inside it, and the frames it keeps
28//! are the newest ones.
29
30use forge_primitives::Message;
31
32/// How many messages one copy of a conversation keeps.
33///
34/// **Sized from the read that consumes it.** Every page a client is handed is
35/// the newest twenty turns, and across the eight largest transcripts measured
36/// on a live machine that window spans 85 to 1,270 messages, with the heaviest
37/// twenty-turn run at 2,799 - so the cap holds every window any of them would
38/// serve. What it buys is a copy that stops tracking its transcript's length:
39/// the live heap grew a pair of ~90MB contiguous buffers an hour through the
40/// copies that used to exist, and a copy held in a cap-sized window cannot
41/// reach one.
42///
43/// **The count is one of two ceilings.** [`CONVERSATION_CAP_BYTES`] bounds the
44/// same window by weight, because on the seats that made #1637 four thousand
45/// frames is not a bound a reader can feel.
46pub const CONVERSATION_CAP: usize = 4_000;
47
48/// A cap under the window it serves cannot reach back over it: 2,799 messages
49/// is the heaviest twenty-turn run measured across the eight largest
50/// transcripts on the author's machine (2026-10-05), and every subscribe
51/// carries twenty turns. Checked at compile time, so a re-tune under it is a
52/// build that never runs rather than a page that quietly loses turns.
53const _: () = assert!(
54 CONVERSATION_CAP >= 2_799,
55 "the cap cannot reach back over the heaviest twenty-turn run measured",
56);
57
58/// How far past the cap a copy may grow before the drop runs.
59///
60/// The drop rebuilds what it keeps into an allocation of its own, so running
61/// it on every append past the cap would pay that rebuild per frame instead of
62/// once per slack - and the slack costs a proportion of the cap, not a second
63/// conversation.
64pub const CONVERSATION_SLACK: usize = 1_000;
65
66/// How many bytes one copy of a conversation keeps, as frames weigh on the
67/// wire.
68///
69/// **Sized from the seats that made #1637, and it binds only them.** Across
70/// the seven largest transcripts on the author's machine (2026-10-17) every
71/// window outside the two giant seats weighs 8.7-23.4 MiB and serves a
72/// twenty-turn page of 0.8-5.2 MiB, while the giants' windows weigh 319 and
73/// 467 MiB and their pages 95 and 105 MiB - about 2x under the client's
74/// 256 MiB frame cap rather than orders under it. 64 MiB leaves every ordinary
75/// seat untouched and comes down on exactly the giants, where the count alone
76/// could not: what those seats serve on open becomes the window whole rather
77/// than the newest twenty turns, and the rest of the history is reached the
78/// way any window's is - a `more` below the floor, answered from the
79/// transcript.
80///
81/// **Every figure above weighs the frames a read serves, the shape a page
82/// carries.** A served frame runs within a percent of the transcript row it
83/// came from (measured over the newest four thousand), while roughly half of a
84/// giant transcript's LINES are rows no reader is served: control bookkeeping
85/// (`worktree-state`, `tag`, `last-prompt`, `mode`, `queue-operation` and the
86/// like - about 32,000 of 114,000 on the seat measured) and attachment rows
87/// that are not queued commands (`hook_success`, `total_tokens_reminder` and
88/// the like). A measurement taken from the file's newest lines therefore finds
89/// about half the frames and about half the bytes, which is the disagreement a
90/// re-measurement from disk runs into.
91pub const CONVERSATION_CAP_BYTES: usize = 64 << 20;
92
93/// How far past the byte cap a copy may grow before the drop runs. A quarter
94/// of the cap, as the count's slack is, and it is also the room a turn
95/// crossing the byte line is kept whole in.
96pub const CONVERSATION_SLACK_BYTES: usize = 16 << 20;
97
98/// The two ceilings one cut answers to, as a value.
99///
100/// The shipped numbers are [`WINDOW`]; the cut is a parameter over them so a
101/// test can drive it at a ceiling a fixture can reach.
102#[derive(Clone, Copy)]
103struct Budget {
104 /// Frames the window keeps, and how far past that a drop may run at.
105 count: usize,
106 count_slack: usize,
107 /// Wire bytes the window keeps, and how far past that a drop may run at.
108 bytes: usize,
109 byte_slack: usize,
110}
111
112/// The ceilings the shipped window answers to.
113const WINDOW: Budget = Budget {
114 count: CONVERSATION_CAP,
115 count_slack: CONVERSATION_SLACK,
116 bytes: CONVERSATION_CAP_BYTES,
117 byte_slack: CONVERSATION_SLACK_BYTES,
118};
119
120/// What a window's cut shed: the frames it dropped, and the thinking growth
121/// the drop took out of the newest turn.
122///
123/// **The second number is a reader's debt.** A cut inside a running turn takes
124/// its counter frames, and a page of what survives sums to less than the turn
125/// spent - so the share the cut took is carried to the page that serves the
126/// turn, which folds it back into the first counter it carries.
127#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
128pub struct Shed {
129 /// The frames the cut dropped.
130 pub frames: usize,
131 /// The `estimated_tokens_delta` of the dropped counters belonging to the
132 /// turn open at the window's end.
133 pub thinking: i64,
134}
135
136/// One conversation's newest window, and what each frame in it weighs.
137///
138/// **The weights are taken once, where a frame enters.** A copy's byte ceiling
139/// is only a ceiling if it is known per frame: measuring the window at each
140/// drop would serialize what the drop is about to throw away, and measuring
141/// the history at each read is the transcript's own cost. So a frame is
142/// measured as it is pushed, [`CONVERSATION_CAP_BYTES`] past which the drop
143/// runs before the push, and what a seed keeps is what its cut measured on the
144/// way in.
145#[derive(Default)]
146pub struct Window {
147 messages: Vec<Message>,
148 /// How each message weighs on the wire, one entry per message.
149 sizes: Vec<usize>,
150 /// The running total of `sizes`, so a push never walks the window.
151 bytes: usize,
152 /// Where the newest turn's opening frame sits in `messages`, `None` when
153 /// the window holds no opening frame for it - either a cut dropped it or
154 /// the scan named none.
155 ///
156 /// **It is how a drop tells the newest turn's counters from older ones**:
157 /// a cut at a turn's first frame takes settled turns, whose counters no
158 /// page owes, while a cut inside the newest turn owes every counter after
159 /// its opening. **No opening frame means the whole window is one turn**,
160 /// so every dropped counter from its front belongs to that turn.
161 newest_open: Option<usize>,
162}
163
164impl Window {
165 /// A window holding nothing.
166 pub fn empty() -> Self {
167 Self::default()
168 }
169
170 pub fn messages(&self) -> &[Message] {
171 &self.messages
172 }
173
174 pub fn len(&self) -> usize {
175 self.messages.len()
176 }
177
178 pub fn is_empty(&self) -> bool {
179 self.messages.is_empty()
180 }
181
182 /// One frame the session emitted, and what the ceilings shed for it.
183 ///
184 /// **Before the push, so the drop pays for the room it makes.** After it,
185 /// this one frame is what tips the store into a doubling the drop is about
186 /// to throw away. `opened` says the frame starts a turn, which is what
187 /// tells the cut where the newest turn begins.
188 pub fn push(&mut self, message: Message, opened: bool) -> Shed {
189 self.push_with(message, opened, WINDOW)
190 }
191
192 /// Take everything `other` holds onto this window's end.
193 ///
194 /// **Not re-cut.** What a fold took out plus what arrived while it ran, so
195 /// until the next push sheds them a merged window may sit past a ceiling's
196 /// slack by whatever the fold's run carried - the split the fold has always
197 /// had, and the push that follows it is what brings the ceilings back.
198 pub fn append(&mut self, other: &mut Window) {
199 let base = self.messages.len();
200 self.bytes = self.bytes.saturating_add(other.bytes);
201 self.messages.append(&mut other.messages);
202 self.sizes.append(&mut other.sizes);
203 other.bytes = 0;
204 // The tail's openings were read off the stream, so its own answer is
205 // exact; it is only its position that moves.
206 if let Some(at) = other.newest_open {
207 self.newest_open = Some(base + at);
208 }
209 // The donor is left empty, so its own answer must be too.
210 other.newest_open = None;
211 }
212
213 /// Fold a cut's thinking debt into the first counter the newest turn still
214 /// carries.
215 ///
216 /// **A handover has no page to fold into.** A resume hands the window to
217 /// the update stream, so the debt rides the frames rather than a serve:
218 /// the first counter the newest turn carries absorbs it, and every reader
219 /// of the handover sums the whole turn.
220 pub fn carry_thinking(&mut self, shed: Shed) {
221 if shed.thinking == 0 {
222 return;
223 }
224 let from = self.newest_open.unwrap_or(0);
225 let Some(at) =
226 self.messages[from..].iter().position(|message| turn_counter(message).is_some())
227 else {
228 return;
229 };
230 if let Message::ThinkingTokens { estimated_tokens_delta, .. } =
231 &mut self.messages[from + at]
232 {
233 *estimated_tokens_delta = estimated_tokens_delta.saturating_add(shed.thinking);
234 }
235 }
236
237 /// The newest window of `history`, as a copy to keep or hand over, and what
238 /// the cut shed.
239 ///
240 /// **Taken from the slice rather than cloned whole and cut**, which is the
241 /// difference between a window-sized allocation and a transcript-sized
242 /// one: what a connect hands over is the whole conversation, and the copy
243 /// that keeps it wants the tail. **And nothing older than the ceilings is
244 /// ever weighed**: the walk starts at the newest frame and stops once
245 /// either ceiling's slack is spent, so a 2.8 GB read pays a window's worth
246 /// of measuring rather than a transcript's.
247 ///
248 /// The openings are the exception to the measure's stop: a scan from the
249 /// history's own start is the only exact answer, and it weighs nothing.
250 pub fn tail_of(history: &[Message]) -> (Window, Shed) {
251 Self::tail_of_with(history, WINDOW)
252 }
253
254 /// A frame on the end, weighed once, dropping first when a ceiling is
255 /// already passed - the whole cut, drivable at a smaller ceiling.
256 fn push_with(&mut self, message: Message, opened: bool, budget: Budget) -> Shed {
257 let shed = if self.messages.len() >= budget.count + budget.count_slack
258 || self.bytes >= budget.bytes + budget.byte_slack
259 {
260 self.drop_with(budget)
261 } else {
262 Shed::default()
263 };
264 let size = wire_bytes(&message);
265 self.bytes = self.bytes.saturating_add(size);
266 self.sizes.push(size);
267 self.messages.push(message);
268 if opened {
269 self.newest_open = Some(self.messages.len() - 1);
270 }
271 shed
272 }
273
274 /// Drop the oldest messages until the window is back inside the ceilings,
275 /// and report what went.
276 ///
277 /// **A fresh allocation rather than a `drain`**, because a `Vec` keeps the
278 /// store it grew to: the point of the cap is that a copy stops holding the
279 /// backing store of a conversation it no longer has, and a drain would
280 /// leave it holding all of it with a window's worth of messages in front.
281 fn drop_with(&mut self, budget: Budget) -> Shed {
282 let front = front_of(&self.messages, &self.sizes, 0, budget);
283 if front == 0 {
284 return Shed::default();
285 }
286 // The debt is only the newest turn's: a cut at or before its opening
287 // takes settled turns, whose counters no page owes.
288 let thinking = match self.newest_open {
289 Some(at) if front <= at => 0,
290 Some(at) => thinking_in(&self.messages[at..front]),
291 None => thinking_in(&self.messages[..front]),
292 };
293 let taken = std::mem::take(&mut self.messages);
294 let taken_sizes = std::mem::take(&mut self.sizes);
295 // Sized for the slack as well, so the frames that refill it do not
296 // double the store on the way back up.
297 let mut kept: Vec<Message> = Vec::with_capacity(CONVERSATION_CAP + CONVERSATION_SLACK);
298 kept.extend(taken.into_iter().skip(front));
299 let mut kept_sizes: Vec<usize> = Vec::with_capacity(CONVERSATION_CAP + CONVERSATION_SLACK);
300 kept_sizes.extend(taken_sizes.into_iter().skip(front));
301 self.bytes = kept_sizes.iter().sum();
302 self.messages = kept;
303 self.sizes = kept_sizes;
304 self.newest_open = self.newest_open.and_then(|at| at.checked_sub(front));
305 Shed { frames: front, thinking }
306 }
307
308 /// [`Self::tail_of`]'s body, at a ceiling a caller names.
309 ///
310 /// `sizes` covers `history[measured_from..]`: what the walk weighed, which
311 /// is the newest frames up to whichever ceiling stopped it.
312 fn tail_of_with(history: &[Message], budget: Budget) -> (Window, Shed) {
313 let mut sizes_newest_first: Vec<usize> = Vec::new();
314 let mut bytes = 0usize;
315 for message in history.iter().rev() {
316 if sizes_newest_first.len() >= budget.count + budget.count_slack {
317 break;
318 }
319 let size = wire_bytes(message);
320 bytes = bytes.saturating_add(size);
321 sizes_newest_first.push(size);
322 if bytes > budget.bytes + budget.byte_slack {
323 break;
324 }
325 }
326 let measured_from = history.len() - sizes_newest_first.len();
327 let mut sizes = sizes_newest_first;
328 sizes.reverse();
329 let front = front_of(history, &sizes, measured_from, budget);
330 let kept_sizes = sizes[(front - measured_from)..].to_vec();
331 let mut scan = crate::conversation_turns::TurnScan::default();
332 let mut newest_open = None;
333 for (at, message) in history.iter().enumerate() {
334 if crate::conversation_turns::opens_for_debt(&mut scan, message) {
335 newest_open = Some(at);
336 }
337 }
338 let thinking = match newest_open {
339 Some(at) if front <= at => 0,
340 Some(at) => thinking_in(&history[at..front]),
341 None => thinking_in(&history[..front]),
342 };
343 let kept_open = newest_open.and_then(|at| at.checked_sub(front));
344 let window = Window {
345 bytes: kept_sizes.iter().sum(),
346 sizes: kept_sizes,
347 messages: history[front..].to_vec(),
348 newest_open: kept_open,
349 };
350 (window, Shed { frames: front, thinking })
351 }
352}
353
354/// The thinking growth a run of frames stands for: the sum of the deltas on
355/// the turn's own counters among them.
356fn thinking_in(messages: &[Message]) -> i64 {
357 messages.iter().filter_map(turn_counter).sum()
358}
359
360/// The growth a frame stands for when it is the turn's own thinking counter,
361/// not a dispatched sub-agent's.
362///
363/// **The same guard `compact_counters` keeps**: a sub-agent's counters belong
364/// to the subagents surface, and adding them to the turn's debt would put
365/// them on the parent's figure.
366fn turn_counter(message: &Message) -> Option<i64> {
367 let Message::ThinkingTokens { estimated_tokens_delta, extras, .. } = message else {
368 return None;
369 };
370 let dispatched = extras
371 .get("parent_tool_use_id")
372 .and_then(serde_json::Value::as_str)
373 .is_some_and(|id| !id.trim().is_empty());
374 (!dispatched).then_some(*estimated_tokens_delta)
375}
376
377/// What a frame weighs on the wire: its JSON, which is what a page carries.
378///
379/// A `Message` serializes by construction - every shape crossed to get here -
380/// so a short count is a frame that cannot cross either way, and it stays as
381/// the count it reached rather than failing the frame around it.
382pub(crate) fn wire_bytes(message: &Message) -> usize {
383 let mut sink = Counted(0);
384 serde_json::to_writer(&mut sink, message).map_or(sink.0, |()| sink.0)
385}
386
387/// A writer that only counts, so weighing a frame does not allocate a copy of
388/// it.
389struct Counted(usize);
390
391impl std::io::Write for Counted {
392 fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
393 self.0 = self.0.saturating_add(buf.len());
394 Ok(buf.len())
395 }
396
397 fn flush(&mut self) -> std::io::Result<()> {
398 Ok(())
399 }
400}
401
402/// Where the window's front moves to in `messages`: the first frame of the
403/// oldest turn it keeps.
404///
405/// **The window ends at the end of a turn, so its front is a turn's first
406/// frame.** `line` is where a count-based cut would land - `len` less the cap -
407/// and the front is the first turn open at or above it, which keeps the window
408/// whole turns and can come in slightly under the cap. The byte ceiling lands
409/// the same way: its own line is the first frame whose tail weighs no more
410/// than the ceiling, and the cut is made at whichever line reaches later, so
411/// what a drop leaves is inside both ceilings or a turn short of the later one.
412/// (A copy grows toward a ceiling's slack again between drops, and the append
413/// at its edge rides one frame past before the next push checks - the shape the
414/// count alone has always had.)
415///
416/// **A turn crossing the line rides into the slack.** The newest turn holds
417/// the newest frames, so a window cannot drop it: when it alone is longer than
418/// the cap, the window keeps it whole up to the cap's slack rather than cutting
419/// inside it - and only when that single turn alone would push the window past
420/// the slack is the cut made inside it, the one case with no whole turn to land
421/// on. [`crate::conversation_turns`] answers where the turns open, and the walk
422/// stops at the first one at or above the line.
423///
424/// **The window is never left empty.** A single frame heavier than the whole
425/// byte budget keeps the window at that one frame rather than at none: the
426/// newest frame cannot be dropped - a page would open on nothing - and the
427/// appends that follow shed it as soon as the tail fits again.
428///
429/// `sizes` is what each frame of `history[measured_from..]` weighs; nothing
430/// older than that was measured, and nothing older than it can be kept (the
431/// ceilings that would reach further back are inside the measure's own stop).
432fn front_of(messages: &[Message], sizes: &[usize], measured_from: usize, budget: Budget) -> usize {
433 let count_line = messages.len().saturating_sub(budget.count);
434 let byte_line = measured_from + line_by_bytes(sizes, budget.bytes);
435 let line = count_line.max(byte_line).min(messages.len().saturating_sub(1));
436 if line == 0 {
437 return 0;
438 }
439 let mut scan = crate::conversation_turns::TurnScan::default();
440 let mut newest = None;
441 for (at, message) in messages.iter().enumerate() {
442 if scan.opens(message) {
443 if at >= line {
444 return at;
445 }
446 newest = Some(at);
447 }
448 }
449 match newest {
450 Some(at)
451 if messages.len() - at <= budget.count + budget.count_slack
452 && at >= measured_from
453 && weighted_from(sizes, at - measured_from) <= budget.bytes + budget.byte_slack =>
454 {
455 at
456 }
457 // No turn THE SCAN CAN NAME leaves the window inside the slack: either
458 // none opens at or above the line, or the newest one it names is
459 // further back than the slack. The fold can still hold a head there -
460 // a queued prompt opens a turn at it and this cannot read one - and
461 // the count's own cut is what is left, inside whichever turn holds it.
462 //
463 // `at < measured_from` is the same answer: the newest turn opened
464 // below everything the walk weighed, so it alone is heavier than the
465 // ceiling's slack and cannot ride.
466 _ => line,
467 }
468}
469
470/// The first frame of `sizes` whose tail weighs no more than `cap`.
471///
472/// The walk runs from the end and stops at the frame that tips the cap, so a
473/// frame's own size is never summed past what the cut could keep.
474fn line_by_bytes(sizes: &[usize], cap: usize) -> usize {
475 let mut acc = 0usize;
476 for (at, size) in sizes.iter().enumerate().rev() {
477 acc = acc.saturating_add(*size);
478 if acc > cap {
479 return at + 1;
480 }
481 }
482 0
483}
484
485/// What `sizes[from..]` weighs together.
486fn weighted_from(sizes: &[usize], from: usize) -> usize {
487 sizes[from..].iter().sum()
488}
489
490#[cfg(test)]
491mod tests {
492 use super::*;
493
494 fn a_frame(at: usize) -> Message {
495 a_said(&format!("frame {at}"))
496 }
497
498 /// A user row carrying `text`.
499 fn a_said(text: &str) -> Message {
500 serde_json::from_value(serde_json::json!({
501 "type": "user",
502 "message": {"role": "user", "content": text},
503 "session_id": "5b1c2d3e-4f50-4a61-b728-9c0d1e2f3a45",
504 }))
505 .expect("a user frame")
506 }
507
508 fn said(message: &Message) -> String {
509 let Message::User { message, .. } = message else {
510 return String::new();
511 };
512 message
513 .content
514 .iter()
515 .find_map(|block| match block {
516 forge_primitives::ContentBlock::Text { text, .. } => Some(text.clone()),
517 _ => None,
518 })
519 .unwrap_or_default()
520 }
521
522 /// A user row carrying a queued prompt, which the fold opens a turn at.
523 fn a_queued(text: &str) -> Message {
524 serde_json::from_value(serde_json::json!({
525 "type": "user",
526 "message": {
527 "role": "user",
528 "content": [{"type": "queued_command", "prompt": text}],
529 },
530 "session_id": "5b1c2d3e-4f50-4a61-b728-9c0d1e2f3a45",
531 }))
532 .expect("a user frame")
533 }
534
535 /// A frame inside a turn: a result answering the turn's own call, which
536 /// the fold draws no turn of its own from.
537 fn a_work_frame(at: usize, within: usize) -> Message {
538 serde_json::from_value(serde_json::json!({
539 "type": "user",
540 "message": {
541 "role": "user",
542 "content": [{
543 "type": "tool_result",
544 "tool_use_id": format!("tu{at}-{within}"),
545 "content": "ok",
546 }],
547 },
548 "session_id": "5b1c2d3e-4f50-4a61-b728-9c0d1e2f3a45",
549 }))
550 .expect("a user frame")
551 }
552
553 /// One thinking-token counter, with a cumulative absolute that is not its
554 /// delta - so summing the wrong field fails.
555 fn a_counter(at: usize, delta: i64) -> Message {
556 let absolute = at.saturating_add(1) * usize::try_from(delta).expect("a positive delta");
557 serde_json::from_value(serde_json::json!({
558 "type": "system",
559 "subtype": "thinking_tokens",
560 "estimated_tokens": absolute,
561 "estimated_tokens_delta": delta,
562 "uuid": format!("t-{at}"),
563 "session_id": "5b1c2d3e-4f50-4a61-b728-9c0d1e2f3a45",
564 }))
565 .expect("a counter frame")
566 }
567
568 /// The frame that ends a turn.
569 fn a_result() -> Message {
570 serde_json::from_value(serde_json::json!({
571 "type": "result",
572 "subtype": "success",
573 "uuid": "r-1",
574 "session_id": "5b1c2d3e-4f50-4a61-b728-9c0d1e2f3a45",
575 "is_error": false,
576 "num_turns": 1,
577 "duration_ms": 12,
578 "duration_api_ms": 9,
579 }))
580 .expect("a result frame")
581 }
582
583 /// `turns` turns of `frames_each` frames: the user row that opens one,
584 /// then frames that belong to it and open none of their own.
585 fn a_history_of(turns: usize, frames_each: usize) -> Vec<Message> {
586 let mut messages = Vec::with_capacity(turns * frames_each);
587 for at in 0..turns {
588 messages.push(a_said(&format!("turn {at}")));
589 for within in 1..frames_each {
590 messages.push(a_work_frame(at, within));
591 }
592 }
593 messages
594 }
595
596 fn opens_a_turn(message: &Message) -> bool {
597 crate::conversation_turns::TurnScan::default().opens(message)
598 }
599
600 /// A ceiling a fixture can reach: the shape of the shipped pair, at
601 /// numbers a test's history can cross.
602 fn a_small_ceiling() -> Budget {
603 Budget { count: 8, count_slack: 2, bytes: 300, byte_slack: 60 }
604 }
605
606 /// **The window's front is a turn's first frame.** The frame the cap's own
607 /// count would land on is inside a turn, so the drop walks up to the next
608 /// turn and the window comes in under the cap.
609 #[test]
610 fn the_drop_lands_on_the_first_turn_at_or_above_the_line() {
611 let mut messages = a_history_of(1_666, 3);
612 messages.extend((0..2).map(|within| a_work_frame(9_999, within)));
613 assert_eq!(
614 messages.len(),
615 CONVERSATION_CAP + CONVERSATION_SLACK,
616 "precondition: the list is at the slack's edge, where an append drops",
617 );
618
619 let (window, dropped) = Window::tail_of(&messages);
620
621 assert_eq!(
622 dropped.frames, 1_002,
623 "the drop passes the count it had to shed to land on a turn's first frame",
624 );
625 assert_eq!(
626 window.len(),
627 3_998,
628 "and the window it leaves is whole turns at or under the cap",
629 );
630 assert!(window.len() <= CONVERSATION_CAP, "never over the cap");
631 assert_eq!(
632 said(&window.messages()[0]),
633 "turn 334",
634 "the oldest kept opens the window's oldest turn"
635 );
636 assert!(
637 opens_a_turn(&window.messages()[0]),
638 "the frame the window starts on opens a turn, not the one the count landed in",
639 );
640 }
641
642 /// A single turn crossing the line rides into the slack: the newest turn
643 /// holds the newest frames and cannot be dropped, so a window that would
644 /// cut inside it keeps it whole up toward the cap's slack instead.
645 #[test]
646 fn a_turn_crossing_the_line_rides_into_the_slack() {
647 let mut messages = a_history_of(125, 4);
648 messages.extend(a_history_of(1, 4_501));
649 assert_eq!(messages.len(), 5_001, "precondition: one turn alone reaches over the cap");
650
651 let (window, dropped) = Window::tail_of(&messages);
652
653 assert_eq!(dropped.frames, 500, "the crossing turn is kept whole");
654 assert_eq!(
655 window.len(),
656 4_501,
657 "so the window rides over the cap rather than being cut inside the turn",
658 );
659 assert!(window.len() <= CONVERSATION_CAP + CONVERSATION_SLACK, "and stays in the slack");
660 assert_eq!(said(&window.messages()[0]), "turn 0", "the window is that one turn");
661 }
662
663 /// **And a turn larger than the slack is cut inside, which is the one case
664 /// with no whole turn to land on.** Keeping it would push the window past
665 /// the slack the cap bounds it by; the newest turn alone is longer than
666 /// that, so the cut is made inside it.
667 #[test]
668 fn a_turn_larger_than_the_slack_is_cut_inside() {
669 let mut messages = a_history_of(125, 4);
670 messages.extend(a_history_of(1, 5_500));
671 assert_eq!(messages.len(), 6_000, "precondition: one turn is longer than the cap's slack");
672
673 let (window, dropped) = Window::tail_of(&messages);
674
675 assert_eq!(dropped.frames, 2_000, "the cut is the count the cap holds");
676 assert_eq!(window.len(), CONVERSATION_CAP, "and the window comes out at the cap");
677 assert!(
678 !opens_a_turn(&window.messages()[0]),
679 "inside the turn, which is the only option when a single turn alone exceeds the slack",
680 );
681 }
682
683 /// **A cut inside a turn reports the counters it took.** The debt is the
684 /// newest turn's own counters, not the settled turn's before it, and a
685 /// cut at a turn's first frame owes nothing.
686 #[test]
687 fn a_cut_inside_a_turn_reports_the_thinking_it_took() {
688 let mut history = vec![a_said("settled"), a_counter(1, 5), a_result(), a_said("running")];
689 for at in 0..10 {
690 history.push(a_counter(at + 10, 3));
691 }
692
693 let inside = Budget { count: 5, count_slack: 0, bytes: usize::MAX, byte_slack: 0 };
694 let (_, shed) = Window::tail_of_with(&history, inside);
695 assert_eq!(shed.frames, 9, "the cut passes the settled turn and the running head");
696 assert_eq!(shed.thinking, 15, "only the running turn's dropped counters count");
697
698 let boundary = Budget { count: 11, count_slack: 0, bytes: usize::MAX, byte_slack: 0 };
699 let (_, shed) = Window::tail_of_with(&history, boundary);
700 assert_eq!(shed.frames, 3, "the cut lands on the running turn's first frame");
701 assert_eq!(shed.thinking, 0, "a cut at a turn's first frame owes nothing");
702 }
703
704 /// **An append carries the tail's newest opening onto the head.** The fold
705 /// hands over the frames that arrived while it ran, and a turn opened in
706 /// them is the newest turn the drop must attribute to.
707 #[test]
708 fn an_append_carries_the_tails_newest_opening() {
709 let ceiling = Budget { count: 8, count_slack: 2, bytes: usize::MAX, byte_slack: 0 };
710 let mut head = Window::empty();
711 let opening = a_said("head");
712 let opened = opens_a_turn(&opening);
713 head.push_with(opening, opened, ceiling);
714 assert_eq!(head.newest_open, Some(0), "precondition: the head's own opening");
715
716 let mut tail = Window::empty();
717 let frame = a_work_frame(1, 1);
718 let opened = opens_a_turn(&frame);
719 tail.push_with(frame, opened, ceiling);
720 head.append(&mut tail);
721 assert_eq!(head.newest_open, Some(0), "a tail with no opening leaves the head's answer");
722 assert_eq!(tail.newest_open, None, "and the donor is left empty");
723
724 let mut tail = Window::empty();
725 let opening = a_said("tail");
726 let opened = opens_a_turn(&opening);
727 tail.push_with(opening, opened, ceiling);
728 let at = head.len();
729 head.append(&mut tail);
730 assert_eq!(head.newest_open, Some(at), "a tail's opening becomes the newest turn's");
731 }
732
733 /// **A head the scan cannot read leaves the count's own cut, deliberately.**
734 /// A queued prompt opens a turn in the fold, and the window's scan cannot
735 /// see the card that decides whether it does - so a queued prompt at or
736 /// above the line is a head the fold has and the scan does not. When the
737 /// newest frame the scan CAN name is further back than the slack, the cut
738 /// falls at the count and starts inside whatever turn holds it, which is
739 /// the one thing the window's front is otherwise never allowed to do. That
740 /// is what the pre-window cut did everywhere, and no real conversation
741 /// measured has reached it.
742 #[test]
743 fn a_head_the_scan_cannot_name_leaves_the_count_cut() {
744 let mut messages = vec![a_said("the only head the scan can name")];
745 messages.extend((0..1_999).map(|within| a_work_frame(9_999, within)));
746 messages.push(a_queued("a head the scan cannot read"));
747 messages.extend((2_001..5_001).map(|within| a_work_frame(9_998, within)));
748 assert_eq!(messages.len(), 5_001, "precondition: one queued head sits above the line");
749
750 let (window, dropped) = Window::tail_of(&messages);
751
752 assert_eq!(
753 dropped.frames, 1_001,
754 "the cut is the count the cap holds, not a turn's first frame"
755 );
756 assert_eq!(window.len(), CONVERSATION_CAP, "and the window comes out at the cap");
757 assert!(
758 !opens_a_turn(&window.messages()[0]),
759 "inside the turn the fold opened at the head the scan could name, because the head \
760 above the line is a queued prompt it cannot read",
761 );
762 }
763
764 /// The cut never starts on a frame the window's scan leaves alone: it
765 /// steps over every one of them, so the window's first frame is a frame
766 /// the fold opens a turn at.
767 ///
768 /// The run below is what the scan cannot name - a delivery's envelope row,
769 /// which the fold opens a turn at, and a queued prompt - so the cut has to
770 /// step past all of them.
771 #[test]
772 fn a_frame_the_scan_leaves_alone_is_never_the_cut() {
773 let mut messages = a_history_of(333, 3);
774 // The run of frames the scan leaves alone: a delivery's envelope row,
775 // which the fold opens a turn at, and the queued prompt after them.
776 messages
777 .extend((0..4).map(|at| a_said(&format!("[Gotify - app 'x', priority 1]\n\n{at}"))));
778 messages.push(
779 serde_json::from_value(serde_json::json!({
780 "type": "user",
781 "message": {
782 "role": "user",
783 "content": [{"type": "queued_command", "prompt": "and this one"}],
784 },
785 "session_id": "5b1c2d3e-4f50-4a61-b728-9c0d1e2f3a45",
786 }))
787 .expect("a user frame"),
788 );
789 messages.extend(a_history_of(1_332, 3));
790 assert_eq!(
791 messages.len(),
792 CONVERSATION_CAP + CONVERSATION_SLACK,
793 "precondition: the line lands inside the run of frames that open no turn",
794 );
795
796 let (window, dropped) = Window::tail_of(&messages);
797
798 assert_eq!(dropped.frames, 1_004, "the cut passes every frame the scan leaves alone");
799 assert!(
800 opens_a_turn(&window.messages()[0]),
801 "and the window starts on a turn's own first frame"
802 );
803 }
804
805 /// The cap keeps the newest and reports what it dropped.
806 #[test]
807 fn a_copy_past_the_cap_drops_its_oldest_messages() {
808 let at_cap: Vec<Message> = (0..CONVERSATION_CAP).map(a_frame).collect();
809 assert_eq!(Window::tail_of(&at_cap).1.frames, 0, "a copy at the cap is left alone");
810
811 let messages: Vec<Message> =
812 (0..CONVERSATION_CAP + CONVERSATION_SLACK).map(a_frame).collect();
813 let (window, dropped) = Window::tail_of(&messages);
814
815 assert_eq!(
816 dropped.frames, CONVERSATION_SLACK,
817 "the drop keeps the cap and reports the rest"
818 );
819 assert_eq!(
820 window.len(),
821 CONVERSATION_CAP,
822 "and what it left is the cap's worth of messages",
823 );
824 assert_eq!(
825 said(&window.messages()[0]),
826 format!("frame {CONVERSATION_SLACK}"),
827 "the oldest kept is the one the cap reaches back to",
828 );
829 assert_eq!(
830 said(window.messages().last().expect("a newest")),
831 format!("frame {}", CONVERSATION_CAP + CONVERSATION_SLACK - 1),
832 "and the newest is the frame the copy ended on",
833 );
834 }
835
836 /// **The store the drop frees is the point of it**, so this pins the shape
837 /// rather than the length: a `drain` trims to the cap and keeps the
838 /// transcript's buffer under it.
839 #[test]
840 fn a_copy_over_the_cap_is_held_in_a_window_sized_store() {
841 let messages: Vec<Message> = (0..CONVERSATION_CAP * 4).map(a_frame).collect();
842
843 let (window, _) = Window::tail_of(&messages);
844
845 assert_eq!(window.len(), CONVERSATION_CAP, "the drop still keeps the cap");
846 assert!(
847 window.messages.capacity() <= CONVERSATION_CAP + CONVERSATION_SLACK,
848 "the store it left is the window's rather than the history's: capacity for {} \
849 messages over a history of {}",
850 window.messages.capacity(),
851 CONVERSATION_CAP * 4,
852 );
853 }
854
855 /// A history anywhere near the cap is taken whole, and one over it loses
856 /// only its front.
857 #[test]
858 fn the_tail_is_the_newest_window_and_nothing_else() {
859 let short: Vec<Message> = (0..10).map(a_frame).collect();
860 assert_eq!(Window::tail_of(&short).0.len(), 10, "a history under the cap is kept whole");
861
862 let long: Vec<Message> = (0..CONVERSATION_CAP * 2).map(a_frame).collect();
863 let (window, dropped) = Window::tail_of(&long);
864 assert_eq!(window.len(), CONVERSATION_CAP, "an over-long history is taken as a window");
865 assert_eq!(dropped.frames, CONVERSATION_CAP, "and the front it left is reported");
866 assert_eq!(
867 said(&window.messages()[0]),
868 format!("frame {CONVERSATION_CAP}"),
869 "and the window is the newest one",
870 );
871 assert_eq!(
872 said(window.messages().last().expect("a newest")),
873 format!("frame {}", CONVERSATION_CAP * 2 - 1),
874 "ending on the history's own last frame",
875 );
876 }
877
878 /// **A frame is weighed as the JSON a page carries it as**, which is what
879 /// the byte ceiling counts: the measured weight of anything else (a
880 /// `Value`, an estimate) would be a ceiling over a number nothing pays.
881 #[test]
882 fn a_frame_is_weighed_as_the_json_it_crosses_as() {
883 for message in [a_said("hello"), a_work_frame(1, 1), a_queued("queued")] {
884 assert_eq!(
885 wire_bytes(&message),
886 serde_json::to_vec(&message).expect("a frame encodes").len(),
887 "the weight is the frame's own JSON: {message:?}",
888 );
889 }
890 }
891
892 /// The shipped ceilings are the budget the window actually consults - a
893 /// wiring that named the count cap where the byte cap belongs would type
894 /// check and silently uncap the bytes.
895 #[test]
896 fn the_shipped_budget_is_the_shipped_ceilings() {
897 assert_eq!(WINDOW.count, CONVERSATION_CAP);
898 assert_eq!(WINDOW.count_slack, CONVERSATION_SLACK);
899 assert_eq!(WINDOW.bytes, CONVERSATION_CAP_BYTES);
900 assert_eq!(WINDOW.byte_slack, CONVERSATION_SLACK_BYTES);
901 }
902
903 /// **The byte ceiling binds where the count never could.** A history well
904 /// under the count cap but over the byte ceiling keeps only the newest
905 /// frames that fit, landing on a turn's first frame like the count's own
906 /// cut does.
907 #[test]
908 fn the_byte_ceiling_binds_under_the_count_cap() {
909 let messages = a_history_of(10, 2);
910 let sizes = vec![100usize; messages.len()];
911 let ceiling = a_small_ceiling();
912
913 let front = front_of(&messages, &sizes, 0, ceiling);
914
915 assert_eq!(front, 18, "the newest two frames are what the byte ceiling fits");
916 assert_eq!(
917 weighted_from(&sizes, front),
918 200,
919 "and the count cap had nothing to say here: it would have kept eight",
920 );
921 assert!(opens_a_turn(&messages[front]), "landed on a turn's own first frame");
922 }
923
924 /// The count still takes the cut when it is the lighter ceiling, and the
925 /// byte ceiling whatever a fixture makes it: the two lines compose, and
926 /// the later one wins.
927 #[test]
928 fn the_later_ceiling_takes_the_cut() {
929 let messages = a_history_of(50, 2);
930 let ceiling = a_small_ceiling();
931
932 let by_count = front_of(&messages, &vec![1usize; messages.len()], 0, ceiling);
933 assert_eq!(by_count, messages.len() - 8, "small frames leave the count cap to bind");
934
935 let by_bytes = front_of(&messages, &vec![100usize; messages.len()], 0, ceiling);
936 assert_eq!(by_bytes, messages.len() - 2, "heavy ones leave the byte ceiling to bind");
937 assert!(
938 opens_a_turn(&messages[by_bytes]),
939 "and the byte's cut lands on a turn's own first frame too",
940 );
941 }
942
943 /// **A turn crossing the byte line rides into the byte slack**, the way a
944 /// crossing turn rides the count's: the newest turn holds the newest
945 /// frames, so a window that would cut inside it keeps it whole up to the
946 /// ceiling's slack.
947 #[test]
948 fn a_turn_crossing_the_byte_line_rides_into_the_byte_slack() {
949 let messages = a_history_of(3, 5);
950 // The newest turn is frames 10..15; 350 bytes is over the 300 ceiling
951 // but inside its slack.
952 let mut sizes = vec![1usize; 10];
953 sizes.extend([70usize; 5]);
954 let ceiling = a_small_ceiling();
955
956 let front = front_of(&messages, &sizes, 0, ceiling);
957
958 assert_eq!(front, 10, "the crossing turn is kept whole, over the ceiling rather than cut");
959 assert_eq!(weighted_from(&sizes, front), 350, "and the window is that one turn");
960 }
961
962 /// **A turn heavier than the byte ceiling's slack is cut inside**, and the
963 /// count's own slack is not what decides it: a turn inside the count's
964 /// slack but past the byte's cannot ride, or the byte ceiling would be
965 /// defeated by exactly the turns the seats it exists for run.
966 #[test]
967 fn a_turn_past_the_byte_slack_is_cut_inside() {
968 let messages = a_history_of(3, 5);
969 // The newest turn is frames 10..15; 600 bytes is past the 300 ceiling
970 // and past its slack.
971 let mut sizes = vec![1usize; 10];
972 sizes.extend([120usize; 5]);
973 let ceiling = a_small_ceiling();
974
975 let front = front_of(&messages, &sizes, 0, ceiling);
976
977 assert_eq!(front, 13, "the cut is made inside the turn, at the byte ceiling's own line");
978 assert!(
979 !opens_a_turn(&messages[front]),
980 "which the count's slack had nothing to say about",
981 );
982 }
983
984 /// **A single frame heavier than the whole budget keeps the window at that
985 /// one frame rather than at none.** A page opening on nothing is worse
986 /// than a window one frame over: the appends that follow shed it as soon
987 /// as the tail fits again.
988 #[test]
989 fn a_frame_heavier_than_the_whole_budget_stays_alone() {
990 let messages = a_history_of(2, 2);
991 let mut sizes = vec![10usize; messages.len()];
992 *sizes.last_mut().expect("a newest") = 1_000;
993 let ceiling = a_small_ceiling();
994
995 let front = front_of(&messages, &sizes, 0, ceiling);
996
997 assert_eq!(front, messages.len() - 1, "the newest frame is never dropped");
998 }
999
1000 /// **A push past the byte ceiling drops before it, keeping the newest
1001 /// frames that fit.** The ceiling is consulted on the running total, so it
1002 /// binds between drops rather than only when the count's own slack runs
1003 /// out - which on a giant seat is where the window's weight lives.
1004 #[test]
1005 fn a_push_past_the_byte_ceiling_drops_the_oldest() {
1006 let ceiling = Budget { count: 1_000, count_slack: 100, bytes: 2_000, byte_slack: 500 };
1007 let mut window = Window::empty();
1008 for _ in 0..40 {
1009 let frame = a_said(&"x".repeat(200));
1010 let opened = opens_a_turn(&frame);
1011 window.push_with(frame, opened, ceiling);
1012 }
1013
1014 let held: usize = window.messages().iter().map(wire_bytes).sum();
1015 assert_eq!(window.bytes, held, "the running total is the window's own weight");
1016 assert!(
1017 window.bytes <= ceiling.bytes + ceiling.byte_slack,
1018 "a push never leaves the window past the ceiling's slack: {} bytes",
1019 window.bytes,
1020 );
1021 assert!(window.len() < 40, "so frames have been shed");
1022 assert_eq!(
1023 said(window.messages().last().expect("a newest")),
1024 "x".repeat(200),
1025 "and the newest frame is the one just pushed",
1026 );
1027 }
1028
1029 /// **A seed weighs only what a cut could keep.** The walk starts at the
1030 /// newest frame and stops once a ceiling's slack is spent, so the frames
1031 /// above the stop are copied and sized while the ones below are neither -
1032 /// which is what makes a 2.8 GB read cost a window rather than a
1033 /// transcript.
1034 #[test]
1035 fn a_seed_keeps_what_the_ceilings_reach_and_no_more() {
1036 let messages = a_history_of(100, 1);
1037 let ceiling = Budget { count: 8, count_slack: 2, bytes: usize::MAX, byte_slack: 0 };
1038
1039 let (window, dropped) = Window::tail_of_with(&messages, ceiling);
1040
1041 assert_eq!(window.len(), 8, "the count cap is what the window keeps");
1042 assert_eq!(dropped.frames, 92, "and everything below the measure's stop is reported gone");
1043 assert_eq!(
1044 said(&window.messages()[0]),
1045 "turn 92",
1046 "the window is the newest turns, sized by the ceiling rather than the history",
1047 );
1048 assert_eq!(window.sizes.len(), window.len(), "every kept frame carries its weight");
1049 assert_eq!(window.bytes, window.sizes.iter().sum::<usize>(), "and the total is theirs");
1050 }
1051
1052 /// **The push path weighs a frame once and keeps the weights aligned.**
1053 /// The drop rebuilds the window from what was measured, so a sizes vec out
1054 /// of step with the messages would make every later cut land on the wrong
1055 /// frame - after an append, after a drop, and after a merge.
1056 #[test]
1057 fn a_window_keeps_the_weights_aligned_with_the_messages() {
1058 let ceiling = Budget { count: 6, count_slack: 2, bytes: usize::MAX, byte_slack: 0 };
1059 let mut window = Window::empty();
1060 for at in 0..20 {
1061 let frame = a_frame(at);
1062 let opened = opens_a_turn(&frame);
1063 window.push_with(frame, opened, ceiling);
1064 }
1065 assert!(window.len() <= 8, "precondition: the count ceiling bound the window");
1066
1067 let mut arrivals = Window::empty();
1068 for at in [100, 101] {
1069 let frame = a_frame(at);
1070 let opened = opens_a_turn(&frame);
1071 arrivals.push_with(frame, opened, ceiling);
1072 }
1073 window.append(&mut arrivals);
1074
1075 assert_eq!(window.sizes.len(), window.messages().len(), "one weight per message");
1076 assert_eq!(
1077 window.sizes.clone(),
1078 window.messages().iter().map(wire_bytes).collect::<Vec<_>>(),
1079 "each weight is that message's own",
1080 );
1081 assert_eq!(window.bytes, window.sizes.iter().sum::<usize>(), "and the total is theirs");
1082 assert!(arrivals.is_empty() && arrivals.bytes == 0, "the donor is left empty");
1083 }
1084}