返回 DeepSeek-TUI-2026
chunking.rs
根目录 / crates / tui / src / tui / streaming / chunking.rs
1 //! Adaptive stream chunking policy for two-gear streaming.
2 //!
3 //! Ported from `codex-rs/tui/src/streaming/chunking.rs`, adapted for deepseek-tui's
4 //! text-based streaming pipeline. The policy is queue-pressure driven and
5 //! source-agnostic.
6 //!
7 //! # Mental model
8 //!
9 //! Two gears:
10 //! - [`ChunkingMode::Smooth`]: drain one display chunk per commit tick (steady pacing).
11 //! - [`ChunkingMode::CatchUp`]: drain a bounded burst while pressure exists.
12 //!
13 //! # Hysteresis
14 //!
15 //! - Enter `CatchUp` when `queued_lines >= ENTER_QUEUE_DEPTH_LINES` OR
16 //! the oldest queued chunk is at least [`ENTER_OLDEST_AGE`].
17 //! - Exit `CatchUp` only after pressure stays below [`EXIT_QUEUE_DEPTH_LINES`]
18 //! AND [`EXIT_OLDEST_AGE`] for at least [`EXIT_HOLD`].
19 //! - After exit, suppress immediate re-entry for [`REENTER_CATCH_UP_HOLD`]
20 //! unless backlog is "severe" (queue >= [`SEVERE_QUEUE_DEPTH_LINES`] or
21 //! oldest >= [`SEVERE_OLDEST_AGE`]).
22
23 use std::time::Duration;
24 use std::time::Instant;
25
26 /// Queue-depth threshold that allows entering catch-up mode.
27 pub(crate) const ENTER_QUEUE_DEPTH_LINES: usize = 160;
28
29 /// Oldest-chunk age threshold that allows entering catch-up mode.
30 pub(crate) const ENTER_OLDEST_AGE: Duration = Duration::from_millis(1_200);
31
32 /// Queue-depth threshold used when evaluating catch-up exit hysteresis.
33 pub(crate) const EXIT_QUEUE_DEPTH_LINES: usize = 32;
34
35 /// Oldest-chunk age threshold used when evaluating catch-up exit hysteresis.
36 pub(crate) const EXIT_OLDEST_AGE: Duration = Duration::from_millis(300);
37
38 /// Minimum duration queue pressure must stay below exit thresholds to leave catch-up mode.
39 pub(crate) const EXIT_HOLD: Duration = Duration::from_millis(250);
40
41 /// Cooldown window after a catch-up exit that suppresses immediate re-entry.
42 pub(crate) const REENTER_CATCH_UP_HOLD: Duration = Duration::from_millis(250);
43
44 /// Queue-depth cutoff that marks backlog as severe (bypasses re-entry hold).
45 pub(crate) const SEVERE_QUEUE_DEPTH_LINES: usize = 640;
46
47 /// Oldest-line age cutoff that marks backlog as severe.
48 pub(crate) const SEVERE_OLDEST_AGE: Duration = Duration::from_millis(4_000);
49
50 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
51 pub enum ChunkingMode {
52 /// Drain one display chunk per baseline commit tick.
53 #[default]
54 Smooth,
55 /// Drain the queued backlog according to queue pressure.
56 CatchUp,
57 }
58
59 /// Captures queue pressure inputs used by adaptive chunking decisions.
60 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
61 pub struct QueueSnapshot {
62 /// Number of queued stream chunks waiting to be displayed.
63 pub queued_lines: usize,
64 /// Age of the oldest queued chunk at decision time.
65 pub oldest_age: Option<Duration>,
66 }
67
68 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
69 pub enum DrainPlan {
70 /// Emit exactly one queued line.
71 Single,
72 /// Emit up to `usize` queued lines.
73 Batch(usize),
74 }
75
76 /// Represents one policy decision for a specific queue snapshot.
77 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
78 pub struct ChunkingDecision {
79 /// Mode after applying hysteresis transitions for this decision.
80 pub mode: ChunkingMode,
81 /// Whether this decision transitioned from `Smooth` into `CatchUp`.
82 pub entered_catch_up: bool,
83 /// Drain plan to execute for the current commit tick.
84 pub drain_plan: DrainPlan,
85 }
86
87 /// Maintains adaptive chunking mode and hysteresis state across ticks.
88 #[derive(Debug, Default, Clone)]
89 pub struct AdaptiveChunkingPolicy {
90 mode: ChunkingMode,
91 below_exit_threshold_since: Option<Instant>,
92 last_catch_up_exit_at: Option<Instant>,
93 /// When true, the policy never enters `CatchUp` — it stays in `Smooth`
94 /// regardless of queue pressure, keeping the display calm for users who
95 /// prefer reduced visual churn.
96 low_motion: bool,
97 }
98
99 impl AdaptiveChunkingPolicy {
100 pub fn new() -> Self {
101 Self::default()
102 }
103
104 /// Returns the policy mode used by the most recent decision.
105 pub fn mode(&self) -> ChunkingMode {
106 self.mode
107 }
108
109 /// Resets state to baseline smooth mode.
110 pub fn reset(&mut self) {
111 self.mode = ChunkingMode::Smooth;
112 self.below_exit_threshold_since = None;
113 self.last_catch_up_exit_at = None;
114 }
115
116 /// When true, the policy never enters `CatchUp` — it stays in `Smooth`
117 /// regardless of queue pressure.
118 pub fn set_low_motion(&mut self, low_motion: bool) {
119 self.low_motion = low_motion;
120 if low_motion {
121 self.mode = ChunkingMode::Smooth;
122 self.below_exit_threshold_since = None;
123 self.last_catch_up_exit_at = None;
124 }
125 }
126
127 /// Computes a drain decision from the current queue snapshot.
128 pub fn decide(&mut self, snapshot: QueueSnapshot, now: Instant) -> ChunkingDecision {
129 // In low-motion mode, always use Smooth pacing regardless of queue
130 // pressure — the user asked for a calm, steady display.
131 if self.low_motion {
132 self.mode = ChunkingMode::Smooth;
133 self.below_exit_threshold_since = None;
134 return ChunkingDecision {
135 mode: self.mode,
136 entered_catch_up: false,
137 drain_plan: DrainPlan::Single,
138 };
139 }
140
141 if snapshot.queued_lines == 0 {
142 self.note_catch_up_exit(now);
143 self.mode = ChunkingMode::Smooth;
144 self.below_exit_threshold_since = None;
145 return ChunkingDecision {
146 mode: self.mode,
147 entered_catch_up: false,
148 drain_plan: DrainPlan::Single,
149 };
150 }
151
152 let entered_catch_up = match self.mode {
153 ChunkingMode::Smooth => self.maybe_enter_catch_up(snapshot, now),
154 ChunkingMode::CatchUp => {
155 self.maybe_exit_catch_up(snapshot, now);
156 false
157 }
158 };
159
160 let drain_plan = match self.mode {
161 ChunkingMode::Smooth => DrainPlan::Single,
162 ChunkingMode::CatchUp => DrainPlan::Batch(snapshot.queued_lines.max(1)),
163 };
164
165 ChunkingDecision {
166 mode: self.mode,
167 entered_catch_up,
168 drain_plan,
169 }
170 }
171
172 fn maybe_enter_catch_up(&mut self, snapshot: QueueSnapshot, now: Instant) -> bool {
173 if !should_enter_catch_up(snapshot) {
174 return false;
175 }
176 if self.reentry_hold_active(now) && !is_severe_backlog(snapshot) {
177 return false;
178 }
179 self.mode = ChunkingMode::CatchUp;
180 self.below_exit_threshold_since = None;
181 self.last_catch_up_exit_at = None;
182 true
183 }
184
185 fn maybe_exit_catch_up(&mut self, snapshot: QueueSnapshot, now: Instant) {
186 if !should_exit_catch_up(snapshot) {
187 self.below_exit_threshold_since = None;
188 return;
189 }
190
191 match self.below_exit_threshold_since {
192 Some(since) if now.saturating_duration_since(since) >= EXIT_HOLD => {
193 self.mode = ChunkingMode::Smooth;
194 self.below_exit_threshold_since = None;
195 self.last_catch_up_exit_at = Some(now);
196 }
197 Some(_) => {}
198 None => {
199 self.below_exit_threshold_since = Some(now);
200 }
201 }
202 }
203
204 fn note_catch_up_exit(&mut self, now: Instant) {
205 if self.mode == ChunkingMode::CatchUp {
206 self.last_catch_up_exit_at = Some(now);
207 }
208 }
209
210 fn reentry_hold_active(&self, now: Instant) -> bool {
211 self.last_catch_up_exit_at
212 .is_some_and(|exit| now.saturating_duration_since(exit) < REENTER_CATCH_UP_HOLD)
213 }
214 }
215
216 /// Returns whether current queue pressure warrants entering catch-up mode.
217 fn should_enter_catch_up(snapshot: QueueSnapshot) -> bool {
218 snapshot.queued_lines >= ENTER_QUEUE_DEPTH_LINES
219 || snapshot
220 .oldest_age
221 .is_some_and(|oldest| oldest >= ENTER_OLDEST_AGE)
222 }
223
224 /// Returns whether queue pressure is low enough to begin exit hysteresis.
225 fn should_exit_catch_up(snapshot: QueueSnapshot) -> bool {
226 snapshot.queued_lines <= EXIT_QUEUE_DEPTH_LINES
227 && snapshot
228 .oldest_age
229 .is_some_and(|oldest| oldest <= EXIT_OLDEST_AGE)
230 }
231
232 /// Returns whether backlog is severe enough to bypass the re-entry hold.
233 fn is_severe_backlog(snapshot: QueueSnapshot) -> bool {
234 snapshot.queued_lines >= SEVERE_QUEUE_DEPTH_LINES
235 || snapshot
236 .oldest_age
237 .is_some_and(|oldest| oldest >= SEVERE_OLDEST_AGE)
238 }
239
240 #[cfg(test)]
241 mod tests {
242 use super::*;
243
244 fn snap(queued_lines: usize, oldest_age_ms: u64) -> QueueSnapshot {
245 QueueSnapshot {
246 queued_lines,
247 oldest_age: Some(Duration::from_millis(oldest_age_ms)),
248 }
249 }
250
251 fn empty_snap() -> QueueSnapshot {
252 QueueSnapshot {
253 queued_lines: 0,
254 oldest_age: None,
255 }
256 }
257
258 #[test]
259 fn smooth_only_burst_emits_one_per_tick() {
260 // Five slowly-arriving lines, each well below enter thresholds, never
261 // flip the policy out of `Smooth`. Each decision should plan a single drain.
262 let mut policy = AdaptiveChunkingPolicy::new();
263 let t0 = Instant::now();
264
265 for i in 0..5 {
266 // 1 queued line, age 10 ms — far below ENTER thresholds.
267 let decision = policy.decide(snap(1, 10), t0 + Duration::from_millis(50 * i));
268 assert_eq!(decision.mode, ChunkingMode::Smooth);
269 assert!(!decision.entered_catch_up);
270 assert_eq!(decision.drain_plan, DrainPlan::Single);
271 }
272 }
273
274 #[test]
275 fn deep_burst_flips_to_catch_up_and_drains_backlog() {
276 // A burst crossing ENTER_QUEUE_DEPTH_LINES enters CatchUp. With
277 // single-grapheme chunks, the threshold stays high enough that
278 // ordinary prose still drips in visibly before catch-up engages.
279 // The policy should enter `CatchUp` and request a Batch drain matching
280 // the queue depth.
281 let mut policy = AdaptiveChunkingPolicy::new();
282 let now = Instant::now();
283
284 let decision = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 10), now);
285 assert_eq!(decision.mode, ChunkingMode::CatchUp);
286 assert!(decision.entered_catch_up);
287 assert_eq!(
288 decision.drain_plan,
289 DrainPlan::Batch(ENTER_QUEUE_DEPTH_LINES)
290 );
291
292 // Larger backlog requested next tick: still CatchUp, batch grows to match.
293 let larger_backlog = ENTER_QUEUE_DEPTH_LINES + 80;
294 let decision = policy.decide(snap(larger_backlog, 30), now + Duration::from_millis(10));
295 assert_eq!(decision.mode, ChunkingMode::CatchUp);
296 assert!(!decision.entered_catch_up, "no second transition signal");
297 assert_eq!(decision.drain_plan, DrainPlan::Batch(larger_backlog));
298 }
299
300 #[test]
301 fn age_threshold_alone_triggers_catch_up() {
302 // Queue depth is small, but the oldest chunk has crossed the age threshold.
303 // Either condition is sufficient to enter catch-up.
304 let mut policy = AdaptiveChunkingPolicy::new();
305 let now = Instant::now();
306
307 let decision = policy.decide(snap(2, ENTER_OLDEST_AGE.as_millis() as u64), now);
308 assert_eq!(decision.mode, ChunkingMode::CatchUp);
309 assert!(decision.entered_catch_up);
310 assert_eq!(decision.drain_plan, DrainPlan::Batch(2));
311 }
312
313 #[test]
314 fn catch_up_exits_after_low_activity_hold() {
315 // Enter CatchUp via depth burst, then drop pressure below exit
316 // thresholds. Policy must hold for >=EXIT_HOLD before returning to Smooth.
317 let mut policy = AdaptiveChunkingPolicy::new();
318 let t0 = Instant::now();
319
320 let _ = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 20), t0);
321 assert_eq!(policy.mode(), ChunkingMode::CatchUp);
322
323 // Pressure drops to the exit thresholds.
324 // Hold begins; not yet 250ms.
325 let pre_hold = policy.decide(
326 snap(EXIT_QUEUE_DEPTH_LINES, EXIT_OLDEST_AGE.as_millis() as u64),
327 t0 + Duration::from_millis(50),
328 );
329 assert_eq!(pre_hold.mode, ChunkingMode::CatchUp);
330
331 // Still under hold.
332 let mid_hold = policy.decide(
333 snap(EXIT_QUEUE_DEPTH_LINES, EXIT_OLDEST_AGE.as_millis() as u64),
334 t0 + Duration::from_millis(200),
335 );
336 assert_eq!(mid_hold.mode, ChunkingMode::CatchUp);
337
338 // Past EXIT_HOLD (250 ms) → return to Smooth.
339 let post_hold = policy.decide(
340 snap(EXIT_QUEUE_DEPTH_LINES, EXIT_OLDEST_AGE.as_millis() as u64),
341 t0 + Duration::from_millis(320),
342 );
343 assert_eq!(post_hold.mode, ChunkingMode::Smooth);
344 assert_eq!(post_hold.drain_plan, DrainPlan::Single);
345 }
346
347 #[test]
348 fn idle_resets_to_smooth_immediately() {
349 // An empty queue forces Smooth regardless of prior mode.
350 let mut policy = AdaptiveChunkingPolicy::new();
351 let now = Instant::now();
352
353 let _ = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 20), now);
354 assert_eq!(policy.mode(), ChunkingMode::CatchUp);
355
356 let decision = policy.decide(empty_snap(), now + Duration::from_millis(10));
357 assert_eq!(decision.mode, ChunkingMode::Smooth);
358 assert_eq!(decision.drain_plan, DrainPlan::Single);
359 }
360
361 #[test]
362 fn reentry_hold_blocks_immediate_flip_back() {
363 // After exiting CatchUp via idle, a threshold-sized burst that arrives within
364 // the re-entry hold window should not immediately re-enter CatchUp.
365 let mut policy = AdaptiveChunkingPolicy::new();
366 let t0 = Instant::now();
367
368 let _ = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 20), t0);
369 let _ = policy.decide(empty_snap(), t0 + Duration::from_millis(10));
370
371 // Within REENTER_CATCH_UP_HOLD (250 ms): hold blocks re-entry.
372 let held = policy.decide(
373 snap(ENTER_QUEUE_DEPTH_LINES, 20),
374 t0 + Duration::from_millis(100),
375 );
376 assert_eq!(held.mode, ChunkingMode::Smooth);
377 assert_eq!(held.drain_plan, DrainPlan::Single);
378
379 // Past the hold: re-entry permitted.
380 let reentered = policy.decide(
381 snap(ENTER_QUEUE_DEPTH_LINES, 20),
382 t0 + Duration::from_millis(400),
383 );
384 assert_eq!(reentered.mode, ChunkingMode::CatchUp);
385 assert_eq!(
386 reentered.drain_plan,
387 DrainPlan::Batch(ENTER_QUEUE_DEPTH_LINES)
388 );
389 }
390
391 #[test]
392 fn severe_backlog_bypasses_reentry_hold() {
393 // Even within the hold window, a "severe" backlog bypasses
394 // the gate so display lag doesn't unbounded-grow.
395 let mut policy = AdaptiveChunkingPolicy::new();
396 let t0 = Instant::now();
397
398 let _ = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES, 20), t0);
399 let _ = policy.decide(empty_snap(), t0 + Duration::from_millis(10));
400
401 let severe = policy.decide(
402 snap(SEVERE_QUEUE_DEPTH_LINES, 20),
403 t0 + Duration::from_millis(100),
404 );
405 assert_eq!(severe.mode, ChunkingMode::CatchUp);
406 assert_eq!(
407 severe.drain_plan,
408 DrainPlan::Batch(SEVERE_QUEUE_DEPTH_LINES)
409 );
410 }
411
412 #[test]
413 fn low_motion_always_smooth_regardless_of_pressure() {
414 let mut policy = AdaptiveChunkingPolicy::new();
415 policy.set_low_motion(true);
416 let t0 = Instant::now();
417
418 // Queue depth far above ENTER threshold.
419 let d1 = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES + 80, 10), t0);
420 assert_eq!(d1.mode, ChunkingMode::Smooth);
421 assert!(!d1.entered_catch_up);
422 assert_eq!(d1.drain_plan, DrainPlan::Single);
423
424 // Oldest age far above ENTER threshold.
425 let d2 = policy.decide(
426 snap(5, ENTER_OLDEST_AGE.as_millis() as u64),
427 t0 + Duration::from_millis(100),
428 );
429 assert_eq!(d2.mode, ChunkingMode::Smooth);
430 assert!(!d2.entered_catch_up);
431 assert_eq!(d2.drain_plan, DrainPlan::Single);
432
433 // Severe backlog — still Smooth.
434 let d3 = policy.decide(
435 snap(
436 SEVERE_QUEUE_DEPTH_LINES + 80,
437 SEVERE_OLDEST_AGE.as_millis() as u64,
438 ),
439 t0 + Duration::from_millis(200),
440 );
441 assert_eq!(d3.mode, ChunkingMode::Smooth);
442 assert_eq!(d3.drain_plan, DrainPlan::Single);
443 }
444
445 #[test]
446 fn low_motion_reset_resumes_normal_operation() {
447 let mut policy = AdaptiveChunkingPolicy::new();
448 policy.set_low_motion(true);
449 let t0 = Instant::now();
450
451 // Low motion blocks catch-up.
452 let d1 = policy.decide(snap(ENTER_QUEUE_DEPTH_LINES + 80, 10), t0);
453 assert_eq!(d1.mode, ChunkingMode::Smooth);
454
455 // Turn off low motion — next burst should enter CatchUp.
456 policy.set_low_motion(false);
457 let d2 = policy.decide(
458 snap(ENTER_QUEUE_DEPTH_LINES + 80, 10),
459 t0 + Duration::from_millis(10),
460 );
461 assert_eq!(d2.mode, ChunkingMode::CatchUp);
462 assert!(d2.entered_catch_up);
463 assert_eq!(
464 d2.drain_plan,
465 DrainPlan::Batch(ENTER_QUEUE_DEPTH_LINES + 80)
466 );
467 }
468 }
469
469 lines RUST