1use std::time::{Duration, SystemTime};
55
56use tokio::time::Instant;
57use tokio_util::sync::CancellationToken;
58
59use super::actor::SessionActor;
60use super::types::{AttachMode, ClientKind};
61use crate::core::config::{CacheKeepWarm, CacheTtl};
62use crate::runtime::keepwarm::{
63 CacheAnchor, KeepWarmProbe, KeepWarmUsage, PingBill, PingFailure, PingOutcome, PingReport,
64 PingTap,
65};
66
67pub const LATE_GUARD: Duration = Duration::from_secs(15);
70
71pub const BACKOFF: [Duration; 3] = [
73 Duration::from_secs(5),
74 Duration::from_secs(15),
75 Duration::from_secs(30),
76];
77
78pub const WALL_POLL: Duration = Duration::from_secs(15);
81
82pub const SHARE_LIMIT_PERCENT: u64 = 10;
88
89pub const PING_INPUT_ESTIMATE: u64 = 64;
92
93pub const PING_OUTPUT_ESTIMATE: u64 = 256;
101
102pub fn ping_output_estimate(max_tokens: u64) -> u64 {
106 max_tokens.min(PING_OUTPUT_ESTIMATE)
107}
108
109const BOUND_EPSILON: f64 = 1e-9;
111
112fn over_share(tokens: u64, prefix: u64) -> bool {
113 tokens.saturating_mul(100) > prefix.saturating_mul(SHARE_LIMIT_PERCENT)
114}
115
116pub fn governing_ttl(ttl: CacheTtl) -> Duration {
121 match ttl {
122 CacheTtl::OneHour => Duration::from_secs(3600),
123 CacheTtl::FiveMinutes | CacheTtl::Hybrid => Duration::from_secs(300),
124 }
125}
126
127pub fn safety_margin(ttl: CacheTtl) -> Duration {
129 match ttl {
130 CacheTtl::OneHour => Duration::from_secs(300),
131 CacheTtl::FiveMinutes | CacheTtl::Hybrid => Duration::from_secs(30),
132 }
133}
134
135pub fn ping_interval(ttl: CacheTtl) -> Duration {
137 governing_ttl(ttl) - safety_margin(ttl)
138}
139
140#[cfg(any(test, feature = "testing"))]
143static INTERVAL_OVERRIDE_MS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
144
145pub fn interval_override() -> Option<Duration> {
155 #[cfg(any(test, feature = "testing"))]
156 {
157 let ms = INTERVAL_OVERRIDE_MS.load(std::sync::atomic::Ordering::Relaxed);
158 if ms > 0 {
159 return Some(Duration::from_millis(ms));
160 }
161 }
162 None
163}
164
165#[cfg(any(test, feature = "testing"))]
167pub mod testing {
168 use std::sync::atomic::Ordering;
169 use std::time::Duration;
170
171 #[must_use = "the override is cleared when the guard drops"]
174 pub struct IntervalOverride(());
175
176 impl IntervalOverride {
177 pub fn set(interval: Duration) -> Self {
178 let ms = u64::try_from(interval.as_millis())
179 .unwrap_or(u64::MAX)
180 .max(1);
181 super::INTERVAL_OVERRIDE_MS.store(ms, Ordering::Relaxed);
182 Self(())
183 }
184 }
185
186 impl Drop for IntervalOverride {
187 fn drop(&mut self) {
188 super::INTERVAL_OVERRIDE_MS.store(0, Ordering::Relaxed);
189 }
190 }
191
192 static DUE_CHECKS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
193
194 pub fn due_checks() -> u64 {
198 DUE_CHECKS.load(Ordering::SeqCst)
199 }
200
201 pub(super) fn note_due_check() {
202 DUE_CHECKS.fetch_add(1, Ordering::SeqCst);
203 }
204}
205
206#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
208pub struct TtlEvidence {
209 pub downgraded: bool,
211 pub saw_1h: bool,
213 pub write_1h: u64,
215 pub write_5m: u64,
216}
217
218pub fn honoured_ttl(configured: CacheTtl, e: TtlEvidence) -> CacheTtl {
222 if configured != CacheTtl::OneHour {
223 return configured;
224 }
225 if e.downgraded {
226 return CacheTtl::FiveMinutes;
227 }
228 if e.write_1h > 0 {
229 return CacheTtl::OneHour;
230 }
231 if e.write_5m > 0 {
232 return CacheTtl::FiveMinutes;
233 }
234 if e.saw_1h {
235 CacheTtl::OneHour
236 } else {
237 CacheTtl::FiveMinutes
238 }
239}
240
241pub fn request_cost_usd(
244 model: &str,
245 ttl: CacheTtl,
246 read: u64,
247 write: u64,
248 input: u64,
249 output: u64,
250) -> f64 {
251 match ttl {
252 CacheTtl::OneHour => {
253 crate::pricing::calculate_cost_split(model, input, output, read, 0, write)
254 }
255 CacheTtl::FiveMinutes | CacheTtl::Hybrid => {
256 crate::pricing::calculate_cost_split(model, input, output, read, write, 0)
257 }
258 }
259}
260
261pub fn rewrite_cost_usd(model: &str, ttl: CacheTtl, prefix_tokens: u64) -> f64 {
264 request_cost_usd(model, ttl, 0, prefix_tokens, 0, 0)
265}
266
267#[derive(Debug, Clone, PartialEq)]
270pub enum PeriodEnd {
271 NothingCached,
274 UncachedInput {
276 input: u64,
277 prefix: u64,
278 },
279 BoundReached {
283 ledger_usd: f64,
284 bound_usd: f64,
285 reserve_usd: f64,
286 },
287 PingTooExpensive {
291 expected_usd: f64,
292 worst_usd: f64,
293 rewrite_usd: f64,
294 },
295 HoursReached(u32),
297 Expired {
300 age_secs: u64,
301 },
302 PrefixChanged,
304 ContextRollover(String),
307 CacheMiss {
309 written: u64,
310 },
311 PartialMiss {
313 written: u64,
314 prefix: u64,
315 },
316 Rejected(String),
318 TransientExhausted(String),
320 CostCap {
323 cap: f64,
324 },
325 TurnStarted,
326 Compaction,
327 Disabled,
329 ParkedByRequest,
331 SessionEnding,
332 Blocked(&'static str),
334}
335
336impl std::fmt::Display for PeriodEnd {
337 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
338 match self {
339 PeriodEnd::NothingCached => f.write_str("nothing cached"),
340 PeriodEnd::UncachedInput { input, prefix } => write!(
341 f,
342 "ping paid for {input} uncached input tokens (prefix {prefix})"
343 ),
344 PeriodEnd::BoundReached {
345 ledger_usd,
346 bound_usd,
347 reserve_usd,
348 } if *reserve_usd > 0.0 => write!(
349 f,
350 "ski-rental bound reached (${ledger_usd:.4} spent of ${bound_usd:.4}: \
351 ${:.4} rewrite + ${reserve_usd:.4} thinking reserve)",
352 bound_usd - reserve_usd
353 ),
354 PeriodEnd::BoundReached {
355 ledger_usd,
356 bound_usd,
357 ..
358 } => write!(
359 f,
360 "ski-rental bound reached (${ledger_usd:.4} spent of ${bound_usd:.4})"
361 ),
362 PeriodEnd::PingTooExpensive {
363 expected_usd,
364 worst_usd,
365 rewrite_usd,
366 } if worst_usd > expected_usd => write!(
367 f,
368 "cache-hit ping expected at ${expected_usd:.4} (${worst_usd:.4} worst case) \
369 exceeds the ${rewrite_usd:.4} rewrite it would save"
370 ),
371 PeriodEnd::PingTooExpensive {
372 worst_usd,
373 rewrite_usd,
374 ..
375 } => write!(
376 f,
377 "worst-case cache-hit ping ${worst_usd:.4} exceeds the ${rewrite_usd:.4} rewrite it would save"
378 ),
379 PeriodEnd::HoursReached(h) => write!(f, "{h}h window reached"),
380 PeriodEnd::Expired { age_secs } => write!(
381 f,
382 "cache expired or about to ({age_secs}s since last request; slept?)"
383 ),
384 PeriodEnd::PrefixChanged => f.write_str("prefix changed"),
385 PeriodEnd::ContextRollover(why) => write!(f, "{why}"),
386 PeriodEnd::CacheMiss { written } => {
387 write!(f, "ping missed the cache (wrote {written} tokens)")
388 }
389 PeriodEnd::PartialMiss { written, prefix } => write!(
390 f,
391 "ping rewrote {written} of {prefix} prefix tokens (history no longer cached)"
392 ),
393 PeriodEnd::Rejected(m) => write!(f, "ping rejected: {m}"),
394 PeriodEnd::TransientExhausted(m) => write!(f, "transient errors: {m}"),
395 PeriodEnd::CostCap { cap } => write!(f, "max_session_cost ${cap:.4} reached"),
396 PeriodEnd::TurnStarted => f.write_str("turn started"),
397 PeriodEnd::Compaction => f.write_str("compaction"),
398 PeriodEnd::Disabled => f.write_str("disabled"),
399 PeriodEnd::ParkedByRequest => f.write_str("parked by request"),
400 PeriodEnd::SessionEnding => f.write_str("session ending"),
401 PeriodEnd::Blocked(why) => write!(f, "{why}"),
402 }
403 }
404}
405
406#[derive(Debug, Clone)]
408pub struct PeriodConfig {
409 pub mode: CacheKeepWarm,
410 pub ttl: CacheTtl,
412 pub model: String,
413 pub max_tokens: u64,
415 pub output_estimate: u64,
420 pub interval_override: Option<Duration>,
422 pub auto_max_hours: u32,
425 pub reply_tokens: Option<u64>,
430}
431
432#[derive(Debug, Clone, Copy, PartialEq)]
439pub struct EpochCarry {
440 pub ledger_usd: f64,
441 pub pings: u32,
442 pub prefix_tokens: u64,
443 pub ping_input: u64,
444 pub idle_since: Instant,
445 pub idle_wall: SystemTime,
446}
447
448#[derive(Debug, Clone, PartialEq)]
450pub enum Fire {
451 NotYet,
452 Ping,
453 End(PeriodEnd),
454}
455
456#[derive(Debug, Clone)]
458pub struct Period {
459 pub id: u64,
460 cfg: PeriodConfig,
461 ttl: CacheTtl,
463 evidence: TtlEvidence,
464 idle_since: Instant,
465 idle_wall: SystemTime,
466 anchor: CacheAnchor,
467 fingerprint: u64,
468 prefix_tokens: u64,
469 first_write: u64,
471 rewrite_usd: f64,
473 reserve_usd: f64,
476 bound_usd: f64,
478 ledger_usd: f64,
479 expected_next_usd: f64,
482 worst_next_usd: f64,
485 exposure_next_usd: f64,
488 ping_input: u64,
490 pings: u32,
491 retries: usize,
492 next_at: Instant,
493 next_wall: SystemTime,
494 in_flight: bool,
495 fired_age: Duration,
497}
498
499fn age(anchor: &CacheAnchor, now: Instant, wall: SystemTime) -> Duration {
503 let mono = now.saturating_duration_since(anchor.mono);
504 let wall = wall.duration_since(anchor.wall).unwrap_or_default();
505 mono.max(wall)
506}
507
508impl Period {
509 #[allow(clippy::too_many_arguments)]
513 pub fn start(
514 id: u64,
515 cfg: PeriodConfig,
516 fingerprint: u64,
517 last: &KeepWarmUsage,
518 evidence: TtlEvidence,
519 anchor: CacheAnchor,
520 now: Instant,
521 wall: SystemTime,
522 ) -> Result<Period, PeriodEnd> {
523 Self::start_in_epoch(
524 id,
525 cfg,
526 fingerprint,
527 last,
528 evidence,
529 anchor,
530 now,
531 wall,
532 None,
533 )
534 }
535
536 #[allow(clippy::too_many_arguments)]
539 pub fn start_in_epoch(
540 id: u64,
541 cfg: PeriodConfig,
542 fingerprint: u64,
543 last: &KeepWarmUsage,
544 evidence: TtlEvidence,
545 anchor: CacheAnchor,
546 now: Instant,
547 wall: SystemTime,
548 carry: Option<&EpochCarry>,
549 ) -> Result<Period, PeriodEnd> {
550 let cached = last.cache_read_tokens + last.cache_write_tokens;
551 if cached == 0 {
552 return Err(PeriodEnd::NothingCached);
553 }
554 let ttl = honoured_ttl(cfg.ttl, evidence);
555 let age_now = age(&anchor, now, wall);
556 if age_now + LATE_GUARD >= governing_ttl(ttl) {
557 return Err(PeriodEnd::Expired {
558 age_secs: age_now.as_secs(),
559 });
560 }
561 let reply = cfg
566 .reply_tokens
567 .map_or(last.output_tokens, |r| r.min(last.output_tokens));
568 let first_write = last.input_tokens + reply;
569 let prefix_tokens = cached + first_write;
570 let mut p = Period {
571 id,
572 cfg,
573 ttl,
574 evidence,
575 idle_since: now,
576 idle_wall: wall,
577 anchor,
578 fingerprint,
579 prefix_tokens,
580 first_write,
581 rewrite_usd: 0.0,
582 reserve_usd: 0.0,
583 bound_usd: 0.0,
584 ledger_usd: 0.0,
585 expected_next_usd: 0.0,
586 worst_next_usd: 0.0,
587 exposure_next_usd: 0.0,
588 ping_input: PING_INPUT_ESTIMATE,
589 pings: 0,
590 retries: 0,
591 next_at: now,
592 next_wall: wall,
593 in_flight: false,
594 fired_age: Duration::ZERO,
595 };
596 if let Some(c) = carry {
597 p.ledger_usd = c.ledger_usd;
600 p.pings = c.pings;
601 p.idle_since = c.idle_since;
602 p.idle_wall = c.idle_wall;
603 if c.pings > 0 {
604 p.prefix_tokens = c.prefix_tokens;
605 p.ping_input = c.ping_input;
606 }
607 }
608 p.reprice();
609 let interval = p.interval();
611 p.next_at = (anchor.mono + interval).max(now);
612 p.next_wall = (anchor.wall + interval).max(wall);
613 if p.worst_next_usd > p.bound_usd * (1.0 + BOUND_EPSILON) {
616 return Err(PeriodEnd::PingTooExpensive {
617 expected_usd: p.expected_next_usd,
618 worst_usd: p.worst_next_usd,
619 rewrite_usd: p.rewrite_usd,
620 });
621 }
622 if let Some(r) = p
623 .past_window(p.next_at, p.next_wall)
624 .or_else(|| p.past_bound())
625 {
626 return Err(r);
627 }
628 Ok(p)
629 }
630
631 pub fn epoch_carry(&self) -> EpochCarry {
633 EpochCarry {
634 ledger_usd: self.ledger_usd,
635 pings: self.pings,
636 prefix_tokens: self.prefix_tokens,
637 ping_input: self.ping_input,
638 idle_since: self.idle_since,
639 idle_wall: self.idle_wall,
640 }
641 }
642
643 fn reprice(&mut self) {
655 let model = self.cfg.model.as_str();
656 let max_tokens = self.cfg.max_tokens;
657 let expected_out = self.cfg.output_estimate.min(max_tokens);
658 self.rewrite_usd = rewrite_cost_usd(model, self.ttl, self.prefix_tokens);
659 let write = if self.pings == 0 { self.first_write } else { 0 };
660 let hit = |out| {
661 request_cost_usd(
662 model,
663 self.ttl,
664 self.prefix_tokens.saturating_sub(write),
665 write,
666 self.ping_input,
667 out,
668 )
669 };
670 self.expected_next_usd = hit(expected_out);
671 self.worst_next_usd = hit(max_tokens);
672 self.reserve_usd = (self.worst_next_usd - self.expected_next_usd).max(0.0);
675 self.bound_usd = self.rewrite_usd + self.reserve_usd;
676 let miss = |ttl| {
680 request_cost_usd(
681 model,
682 ttl,
683 0,
684 self.prefix_tokens,
685 self.ping_input,
686 self.cfg.max_tokens,
687 )
688 };
689 let mut miss_usd = miss(self.ttl);
690 if matches!(self.cfg.ttl, CacheTtl::OneHour | CacheTtl::Hybrid) {
691 miss_usd = miss_usd.max(miss(CacheTtl::OneHour));
692 }
693 self.exposure_next_usd = self.worst_next_usd.max(miss_usd);
694 }
695
696 fn interval(&self) -> Duration {
697 self.cfg
698 .interval_override
699 .unwrap_or_else(|| ping_interval(self.ttl))
700 }
701
702 pub fn fingerprint(&self) -> u64 {
703 self.fingerprint
704 }
705 pub fn pings(&self) -> u32 {
706 self.pings
707 }
708 pub fn ledger_usd(&self) -> f64 {
709 self.ledger_usd
710 }
711 pub fn bound_usd(&self) -> f64 {
715 self.bound_usd
716 }
717 pub fn rewrite_usd(&self) -> f64 {
719 self.rewrite_usd
720 }
721 pub fn reserve_usd(&self) -> f64 {
724 self.reserve_usd
725 }
726 pub fn expected_next_usd(&self) -> f64 {
728 self.expected_next_usd
729 }
730 pub fn worst_next_usd(&self) -> f64 {
734 self.worst_next_usd
735 }
736 pub fn exposure_next_usd(&self) -> f64 {
740 self.exposure_next_usd
741 }
742 pub fn prefix_tokens(&self) -> u64 {
743 self.prefix_tokens
744 }
745 pub fn next_at(&self) -> Instant {
746 self.next_at
747 }
748 pub fn mode(&self) -> CacheKeepWarm {
749 self.cfg.mode
750 }
751 pub fn ttl(&self) -> CacheTtl {
753 self.ttl
754 }
755 pub fn fired_age(&self) -> Duration {
756 self.fired_age
757 }
758 pub fn in_flight(&self) -> bool {
759 self.in_flight
760 }
761
762 pub fn set_mode(&mut self, mode: CacheKeepWarm) {
764 self.cfg.mode = mode;
765 }
766
767 pub fn deadline(&self, now: Instant, wall: SystemTime) -> Option<Instant> {
771 if self.in_flight {
772 return None;
773 }
774 let wall_left = self.next_wall.duration_since(wall).unwrap_or_default();
775 Some(self.next_at.min(now + wall_left))
776 }
777
778 fn past_window(&self, at: Instant, wall: SystemTime) -> Option<PeriodEnd> {
780 let h = match self.cfg.mode {
781 CacheKeepWarm::Hours(h) => h,
782 CacheKeepWarm::Auto if self.cfg.auto_max_hours > 0 => self.cfg.auto_max_hours,
783 _ => return None,
784 };
785 let window = Duration::from_secs(u64::from(h) * 3600);
786 let walled = wall.duration_since(self.idle_wall).unwrap_or_default();
787 (at.saturating_duration_since(self.idle_since) >= window || walled >= window)
788 .then_some(PeriodEnd::HoursReached(h))
789 }
790
791 fn past_bound(&self) -> Option<PeriodEnd> {
792 (self.ledger_usd + self.worst_next_usd > self.bound_usd * (1.0 + BOUND_EPSILON)).then_some(
793 PeriodEnd::BoundReached {
794 ledger_usd: self.ledger_usd,
795 bound_usd: self.bound_usd,
796 reserve_usd: self.reserve_usd,
797 },
798 )
799 }
800
801 pub fn fire(
803 &mut self,
804 now: Instant,
805 wall: SystemTime,
806 session_cost: f64,
807 max_session_cost: Option<f64>,
808 ) -> Fire {
809 if self.in_flight || (now < self.next_at && wall < self.next_wall) {
810 return Fire::NotYet;
811 }
812 let age_now = age(&self.anchor, now, wall);
813 if age_now + LATE_GUARD >= governing_ttl(self.ttl) {
814 return Fire::End(PeriodEnd::Expired {
815 age_secs: age_now.as_secs(),
816 });
817 }
818 if let Some(r) = self.past_window(now, wall).or_else(|| self.past_bound()) {
819 return Fire::End(r);
820 }
821 if let Some(cap) = max_session_cost.filter(|c| c.is_finite()) {
822 if session_cost + self.exposure_next_usd >= cap {
824 return Fire::End(PeriodEnd::CostCap { cap });
825 }
826 }
827 self.in_flight = true;
828 self.fired_age = age_now;
829 Fire::Ping
830 }
831
832 pub fn charge(&mut self, cost_usd: f64) {
834 self.ledger_usd += cost_usd;
835 }
836
837 pub fn ping_sent(
842 &mut self,
843 usage: &KeepWarmUsage,
844 anchor: Option<CacheAnchor>,
845 downgraded: bool,
846 ) -> Result<(), PeriodEnd> {
847 self.in_flight = false;
848 self.retries = 0;
849 self.pings += 1;
850 let (read, write) = (usage.cache_read_tokens, usage.cache_write_tokens);
851 if read == 0 && write == 0 {
852 return Err(PeriodEnd::NothingCached);
853 }
854 if read == 0 {
855 return Err(PeriodEnd::CacheMiss { written: write });
856 }
857 if over_share(usage.input_tokens, self.prefix_tokens) {
858 return Err(PeriodEnd::UncachedInput {
859 input: usage.input_tokens,
860 prefix: self.prefix_tokens,
861 });
862 }
863 let allowed = if self.pings == 1 { self.first_write } else { 0 };
864 if over_share(write.saturating_sub(allowed), self.prefix_tokens) {
865 return Err(PeriodEnd::PartialMiss {
866 written: write,
867 prefix: self.prefix_tokens,
868 });
869 }
870 self.prefix_tokens = read + write;
872 self.ping_input = usage.input_tokens.max(1);
873 self.evidence.downgraded |= downgraded;
874 self.evidence.write_1h = usage.cache_write_1h.unwrap_or(0);
875 self.evidence.write_5m = usage.cache_write_5m.unwrap_or(0);
876 self.evidence.saw_1h |= self.evidence.write_1h > 0;
877 self.ttl = honoured_ttl(self.cfg.ttl, self.evidence);
878 self.reprice();
879 if let Some(a) = anchor.filter(|a| a.mono >= self.anchor.mono) {
880 self.anchor = a;
881 }
882 let interval = self.interval();
883 self.next_at = self.anchor.mono + interval;
884 self.next_wall = self.anchor.wall + interval;
885 if let Some(r) = self
886 .past_window(self.next_at, self.next_wall)
887 .or_else(|| self.past_bound())
888 {
889 return Err(r);
890 }
891 Ok(())
892 }
893
894 pub fn ping_failed(
898 &mut self,
899 failure: &PingFailure,
900 now: Instant,
901 wall: SystemTime,
902 ) -> Result<Duration, PeriodEnd> {
903 self.in_flight = false;
904 let msg = match failure {
905 PingFailure::Fatal(m) => return Err(PeriodEnd::Rejected(m.clone())),
906 PingFailure::Canceled => return Err(PeriodEnd::Rejected("canceled".into())),
907 PingFailure::Transient(m) => m.clone(),
908 };
909 let Some(delay) = BACKOFF.get(self.retries).copied() else {
910 return Err(PeriodEnd::TransientExhausted(msg));
911 };
912 let age_then = age(&self.anchor, now, wall) + delay;
913 if age_then + LATE_GUARD >= governing_ttl(self.ttl) {
914 return Err(PeriodEnd::TransientExhausted(format!(
915 "{msg}; no retry fits before the cache expires"
916 )));
917 }
918 if self.ledger_usd + self.worst_next_usd > self.bound_usd * (1.0 + BOUND_EPSILON) {
919 return Err(PeriodEnd::BoundReached {
920 ledger_usd: self.ledger_usd,
921 bound_usd: self.bound_usd,
922 reserve_usd: self.reserve_usd,
923 });
924 }
925 self.retries += 1;
926 self.next_at = now + delay;
927 self.next_wall = wall + delay;
928 Ok(delay)
929 }
930}
931
932pub(crate) struct PingTask {
934 pub(crate) handle: tokio::task::JoinHandle<PingReport>,
935 cancel: CancellationToken,
936 tap: PingTap,
937 model: String,
939 worst_usd: f64,
942}
943
944impl PingTask {
945 fn bill(&self) -> PingBill {
948 self.tap.bill(&self.model, self.worst_usd)
949 }
950}
951
952pub(crate) fn client_qualifies(kind: ClientKind, mode: AttachMode, headless: bool) -> bool {
969 if mode == AttachMode::Observe {
970 return false;
971 }
972 match kind {
973 ClientKind::Tui | ClientKind::Attach => true,
974 ClientKind::Chat | ClientKind::Rpc | ClientKind::Server => headless,
975 ClientKind::Test => false,
976 }
977}
978
979#[derive(Default)]
981pub(crate) struct KeepWarmState {
982 configured: CacheKeepWarm,
984 auto_max_hours: u32,
986 headless: bool,
989 epoch: Option<EpochCarry>,
993 pub(crate) override_mode: Option<CacheKeepWarm>,
995 pub(crate) override_slot: OverrideSlot,
998 pub(crate) period: Option<Period>,
999 pub(crate) task: Option<PingTask>,
1000 next_period_id: u64,
1001 turn_started_at: Option<Instant>,
1003 last_request: Option<KeepWarmUsage>,
1005 pub(crate) interactive_seen: bool,
1010 last_end: Option<String>,
1011 session_pings: u32,
1012 estimate_exceeded: bool,
1017}
1018
1019pub type OverrideSlot = std::sync::Arc<std::sync::Mutex<Option<CacheKeepWarm>>>;
1021
1022impl KeepWarmState {
1023 pub(crate) fn new(
1024 configured: CacheKeepWarm,
1025 auto_max_hours: u32,
1026 headless: bool,
1027 override_slot: OverrideSlot,
1028 ) -> Self {
1029 let override_mode = *override_slot
1030 .lock()
1031 .unwrap_or_else(std::sync::PoisonError::into_inner);
1032 Self {
1033 configured,
1034 auto_max_hours,
1035 headless,
1036 override_mode,
1037 override_slot,
1038 ..Self::default()
1039 }
1040 }
1041
1042 pub(crate) fn mode(&self) -> CacheKeepWarm {
1043 self.override_mode.unwrap_or(self.configured)
1044 }
1045
1046 pub(crate) fn set_override(&mut self, mode: CacheKeepWarm) {
1047 self.override_mode = Some(mode);
1048 *self
1049 .override_slot
1050 .lock()
1051 .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(mode);
1052 }
1053
1054 pub(crate) fn active(&self) -> bool {
1055 self.period.is_some()
1056 }
1057
1058 pub(crate) fn deadline(&self) -> Option<Instant> {
1062 self.deadline_at(Instant::now(), SystemTime::now())
1063 }
1064
1065 fn deadline_at(&self, now: Instant, wall: SystemTime) -> Option<Instant> {
1066 self.period
1067 .as_ref()
1068 .and_then(|p| p.deadline(now, wall))
1069 .map(|d| d.min(now + WALL_POLL))
1070 }
1071}
1072
1073pub(crate) async fn keepwarm_timer(deadline: Option<Instant>) {
1075 match deadline {
1076 Some(t) => tokio::time::sleep_until(t).await,
1077 None => std::future::pending().await,
1078 }
1079}
1080
1081pub(crate) async fn poll_keepwarm_ping(
1083 task: &mut Option<PingTask>,
1084) -> std::result::Result<PingReport, tokio::task::JoinError> {
1085 match task {
1086 Some(t) => (&mut t.handle).await,
1087 None => std::future::pending().await,
1088 }
1089}
1090
1091impl SessionActor {
1092 fn cache_keepwarm_blocker(&self) -> Option<&'static str> {
1094 if !self.runtime.is_live() || !self.conv.is_live() || self.is_parked() {
1095 return Some("session parked");
1096 }
1097 if self.streaming {
1098 return Some("turn running");
1099 }
1100 if self.compact.is_some() {
1101 return Some("compaction running");
1102 }
1103 if self.conv.queued_message.is_some() || !self.event_queue.is_empty() {
1104 return Some("work queued");
1105 }
1106 if !self.pending_prompts.is_empty() {
1107 return Some("prompt pending");
1108 }
1109 if self.driver.is_some() || self.driver_pending.is_some() {
1110 return Some("session driver armed");
1111 }
1112 None
1113 }
1114
1115 pub(crate) fn cache_keepwarm_note_attach(&mut self, kind: ClientKind, mode: AttachMode) {
1118 if client_qualifies(kind, mode, self.cache_keepwarm.headless) {
1119 self.cache_keepwarm.interactive_seen = true;
1120 }
1121 }
1122
1123 pub(crate) fn cache_keepwarm_note_usage(&mut self, usage: KeepWarmUsage) {
1126 self.cache_keepwarm.last_request = Some(usage);
1127 self.cache_keepwarm.epoch = None;
1129 }
1130
1131 pub(crate) fn cache_keepwarm_turn_started(&mut self) {
1134 self.cache_keepwarm_end(PeriodEnd::TurnStarted);
1135 self.cache_keepwarm.turn_started_at = Some(Instant::now());
1136 self.cache_keepwarm.last_request = None;
1137 }
1138
1139 pub(crate) fn cache_keepwarm_compaction(&mut self) {
1142 self.cache_keepwarm_end(PeriodEnd::Compaction);
1143 self.cache_keepwarm.turn_started_at = None;
1144 self.cache_keepwarm.last_request = None;
1145 self.cache_keepwarm.epoch = None;
1146 }
1147
1148 pub(crate) fn cache_keepwarm_conversation_replaced(&mut self) {
1155 self.cache_keepwarm_end(PeriodEnd::PrefixChanged);
1156 self.cache_keepwarm.turn_started_at = None;
1157 self.cache_keepwarm.last_request = None;
1158 self.cache_keepwarm.epoch = None;
1159 }
1160
1161 pub(crate) fn cache_keepwarm_park_requested(&mut self) {
1164 if self.cache_keepwarm.active() && self.can_park_ignoring_keepwarm() {
1165 self.cache_keepwarm_end(PeriodEnd::ParkedByRequest);
1166 }
1167 }
1168
1169 fn cache_keepwarm_book(&mut self, usage: &KeepWarmUsage, cost: f64) {
1174 if (cost <= 0.0 && *usage == KeepWarmUsage::default()) || !self.conv.is_live() {
1175 return;
1176 }
1177 self.conv.session_cost += cost;
1178 self.conv.keepwarm_cost += cost;
1179 self.conv.total_input_tokens += usage.input_tokens;
1180 self.conv.total_output_tokens += usage.output_tokens;
1181 self.conv.total_cache_read_tokens += usage.cache_read_tokens;
1182 self.conv.total_cache_creation_tokens += usage.cache_write_tokens;
1183 let t = &mut self.conv.keepwarm_tokens;
1184 t.input += usage.input_tokens;
1185 t.output += usage.output_tokens;
1186 t.cache_read += usage.cache_read_tokens;
1187 t.cache_creation += usage.cache_write_tokens;
1188 if let Some(p) = self.cache_keepwarm.period.as_mut() {
1189 p.charge(cost);
1190 }
1191 self.emit_conversation();
1192 self.request_save();
1198 }
1199
1200 fn cache_keepwarm_abort_ping(&mut self) {
1204 let Some(task) = self.cache_keepwarm.task.take() else {
1205 return;
1206 };
1207 task.cancel.cancel();
1208 task.handle.abort();
1209 let bill = task.bill();
1210 if bill.cost_usd > 0.0 {
1211 tracing::info!(
1212 session = %self.id,
1213 cost_usd = bill.cost_usd,
1214 booking = ?bill.booking,
1215 "cache keep-warm ping aborted in flight; booked"
1216 );
1217 }
1218 self.cache_keepwarm_book(&bill.usage, bill.cost_usd);
1219 }
1220
1221 pub(crate) fn cache_keepwarm_end(&mut self, reason: PeriodEnd) {
1223 self.cache_keepwarm_abort_ping();
1224 let Some(p) = self.cache_keepwarm.period.take() else {
1225 return;
1226 };
1227 tracing::info!(
1228 session = %self.id,
1229 period = p.id,
1230 pings = p.pings(),
1231 ledger_usd = p.ledger_usd(),
1232 bound_usd = p.bound_usd(),
1233 reason = %reason,
1234 "cache keep-warm period ended"
1235 );
1236 self.cache_keepwarm.last_end = Some(reason.to_string());
1237 self.cache_keepwarm.epoch = Some(p.epoch_carry());
1239 self.rearm_park();
1241 }
1242
1243 pub(crate) async fn cache_keepwarm_on_idle(&mut self) {
1246 let Some(turn_started_at) = self.cache_keepwarm.turn_started_at.take() else {
1247 return;
1248 };
1249 self.cache_keepwarm_try_start(Some(turn_started_at)).await;
1250 }
1251
1252 async fn cache_keepwarm_try_start(&mut self, turn_started_at: Option<Instant>) {
1257 if self.cache_keepwarm.active() {
1258 return;
1259 }
1260 let mode = self.cache_keepwarm.mode();
1261 if mode.is_off() || !(self.cache_keepwarm.interactive_seen || self.keep_warm) {
1264 return;
1265 }
1266 let Some(anchor) = self
1268 .runtime
1269 .cache_anchor()
1270 .filter(|a| turn_started_at.is_none_or(|t| a.mono >= t))
1271 else {
1272 return;
1273 };
1274 let Some(last) = self.cache_keepwarm.last_request.clone() else {
1275 return;
1276 };
1277 if let Some(why) = self.cache_keepwarm_blocker() {
1278 self.cache_keepwarm.last_end = Some(format!("not started: {why}"));
1279 return;
1280 }
1281 if let Some(why) = self.runtime.cache_keepwarm_ineligibility() {
1282 self.cache_keepwarm.last_end = Some(format!("not started: {why}"));
1283 return;
1284 }
1285 if !crate::runtime::keepwarm::history_ready(&self.conv.api_messages) {
1286 return;
1287 }
1288 let probe: KeepWarmProbe = match self
1289 .runtime
1290 .cache_keepwarm_probe(&self.conv.api_messages)
1291 .await
1292 {
1293 Ok(p) => p,
1294 Err(e) => {
1295 tracing::info!(session = %self.id, "cache keep-warm not started: {e}");
1296 self.cache_keepwarm.last_end = Some(format!("not started: {e}"));
1297 return;
1298 }
1299 };
1300 if let Some(why) = probe.context_blocks {
1301 let reason = PeriodEnd::ContextRollover(why);
1302 tracing::info!(session = %self.id, reason = %reason, "cache keep-warm not started");
1303 self.cache_keepwarm.last_end = Some(format!("not started: {reason}"));
1304 return;
1305 }
1306 let ttl = self.runtime.resolved_cache_ttl().await;
1311 let cfg = PeriodConfig {
1312 mode,
1313 ttl,
1314 model: self.runtime.model().to_string(),
1315 max_tokens: probe.max_tokens,
1316 output_estimate: if self.cache_keepwarm.estimate_exceeded {
1317 probe.max_tokens
1318 } else {
1319 ping_output_estimate(probe.max_tokens)
1320 },
1321 interval_override: interval_override(),
1322 auto_max_hours: self.cache_keepwarm.auto_max_hours,
1323 reply_tokens: Some(crate::runtime::keepwarm::reply_prefix_tokens(
1324 &self.conv.api_messages,
1325 )),
1326 };
1327 let evidence = TtlEvidence {
1328 downgraded: self.runtime.cache_ttl_downgraded(),
1329 saw_1h: self.runtime.cache_1h_honoured(),
1330 write_1h: last.cache_write_1h.unwrap_or(0),
1331 write_5m: last.cache_write_5m.unwrap_or(0),
1332 };
1333 self.cache_keepwarm.next_period_id += 1;
1334 let id = self.cache_keepwarm.next_period_id;
1335 let now = Instant::now();
1336 let carry = self.cache_keepwarm.epoch;
1337 match Period::start_in_epoch(
1338 id,
1339 cfg,
1340 probe.fingerprint,
1341 &last,
1342 evidence,
1343 anchor,
1344 now,
1345 SystemTime::now(),
1346 carry.as_ref(),
1347 ) {
1348 Ok(p) => {
1349 tracing::info!(
1350 session = %self.id,
1351 period = id,
1352 mode = %mode,
1353 ttl_secs = governing_ttl(p.ttl()).as_secs(),
1354 first_ping_in_secs = p.next_at().saturating_duration_since(now).as_secs(),
1355 prefix_tokens = p.prefix_tokens(),
1356 bound_usd = p.bound_usd(),
1357 rewrite_usd = p.rewrite_usd(),
1358 reserve_usd = p.reserve_usd(),
1359 expected_ping_usd = p.expected_next_usd(),
1360 worst_ping_usd = p.exposure_next_usd(),
1361 "cache keep-warm period started"
1362 );
1363 self.cache_keepwarm.period = Some(p);
1364 }
1365 Err(reason) => {
1366 tracing::info!(session = %self.id, reason = %reason, "cache keep-warm not started");
1367 self.cache_keepwarm.last_end = Some(format!("not started: {reason}"));
1368 }
1369 }
1370 }
1371
1372 pub(crate) async fn on_cache_keepwarm_due(&mut self) {
1374 #[cfg(any(test, feature = "testing"))]
1375 testing::note_due_check();
1376 if self.holds(super::actor::WorkSource::KeepWarmTimer)
1381 || self.cache_keepwarm.period.is_none()
1382 || self.cache_keepwarm.task.is_some()
1383 {
1384 return;
1385 }
1386 let (now, wall) = (Instant::now(), SystemTime::now());
1387 if self
1389 .cache_keepwarm
1390 .period
1391 .as_ref()
1392 .and_then(|p| p.deadline(now, wall))
1393 .is_some_and(|d| d > now)
1394 {
1395 return;
1396 }
1397 if let Some(why) = self.cache_keepwarm_blocker() {
1398 self.cache_keepwarm_end(PeriodEnd::Blocked(why));
1399 return;
1400 }
1401 if !crate::runtime::keepwarm::history_ready(&self.conv.api_messages) {
1402 self.cache_keepwarm_end(PeriodEnd::PrefixChanged);
1405 return;
1406 }
1407 match self
1410 .runtime
1411 .cache_keepwarm_probe(&self.conv.api_messages)
1412 .await
1413 {
1414 Ok(probe) => {
1415 if let Some(why) = probe.context_blocks {
1416 self.cache_keepwarm_end(PeriodEnd::ContextRollover(why));
1417 return;
1418 }
1419 }
1420 Err(e) => {
1421 self.cache_keepwarm_end(PeriodEnd::Rejected(e.to_string()));
1422 return;
1423 }
1424 }
1425 let session_cost = self.conv.session_cost;
1426 let cap = self.config.max_session_cost;
1427 let Some(p) = self.cache_keepwarm.period.as_mut() else {
1428 return;
1429 };
1430 match p.fire(Instant::now(), SystemTime::now(), session_cost, cap) {
1431 Fire::NotYet => {}
1432 Fire::End(reason) => self.cache_keepwarm_end(reason),
1433 Fire::Ping => {
1434 let expect = p.fingerprint();
1435 let worst_usd = p.worst_next_usd();
1436 let history = self.conv.api_messages.clone();
1440 let mut runtime = self.runtime.clone();
1441 runtime.share_ttl_latches(&self.runtime);
1442 let cancel = CancellationToken::new();
1443 let token = cancel.clone();
1444 let tap = PingTap::default();
1445 let task_tap = tap.clone();
1446 let handle = tokio::spawn(async move {
1447 runtime
1448 .cache_keepwarm_ping(&history, expect, &token, &task_tap)
1449 .await
1450 });
1451 self.cache_keepwarm.task = Some(PingTask {
1452 handle,
1453 cancel,
1454 tap,
1455 model: self.runtime.model().to_string(),
1456 worst_usd,
1457 });
1458 }
1459 }
1460 }
1461
1462 pub(crate) fn on_cache_keepwarm_done(
1464 &mut self,
1465 result: std::result::Result<PingReport, tokio::task::JoinError>,
1466 ) {
1467 let Some(task) = self.cache_keepwarm.task.take() else {
1468 return;
1469 };
1470 let report = match result {
1471 Ok(r) => r,
1472 Err(e) => PingReport {
1473 result: Err(if e.is_cancelled() {
1474 PingFailure::Canceled
1475 } else {
1476 PingFailure::Fatal(format!("ping task: {e}"))
1477 }),
1478 usage: task.tap.usage(),
1479 notices: Vec::new(),
1480 },
1481 };
1482 let bill = task.bill();
1484 let cost = bill.cost_usd;
1485 self.cache_keepwarm_book(&bill.usage, cost);
1486 if bill.usage.output_tokens > PING_OUTPUT_ESTIMATE && !self.cache_keepwarm.estimate_exceeded
1487 {
1488 self.cache_keepwarm.estimate_exceeded = true;
1489 tracing::warn!(
1490 session = %self.id,
1491 output = bill.usage.output_tokens,
1492 estimate = PING_OUTPUT_ESTIMATE,
1493 "cache keep-warm ping out-thought its output estimate; \
1494 later periods price pings at max_tokens"
1495 );
1496 }
1497 for notice in &report.notices {
1498 self.emit(crate::session::SessionEventWire::SystemNotice(
1499 notice.clone(),
1500 ));
1501 }
1502 if self.cache_keepwarm.period.is_none() {
1503 return;
1504 }
1505 match report.result {
1506 Ok(PingOutcome::Sent) => {
1507 let usage = report.usage;
1508 self.cache_keepwarm.session_pings += 1;
1509 let anchor = self.runtime.cache_anchor();
1510 let downgraded = self.runtime.cache_ttl_downgraded();
1511 let Some(p) = self.cache_keepwarm.period.as_mut() else {
1512 return;
1513 };
1514 let result = p.ping_sent(&usage, anchor, downgraded);
1515 tracing::info!(
1516 session = %self.id,
1517 period = p.id,
1518 ping = p.pings(),
1519 age_secs = p.fired_age().as_secs(),
1520 ttl_secs = governing_ttl(p.ttl()).as_secs(),
1521 cache_read = usage.cache_read_tokens,
1522 cache_write = usage.cache_write_tokens,
1523 input = usage.input_tokens,
1524 output = usage.output_tokens,
1525 cost_usd = cost,
1526 ledger_usd = p.ledger_usd(),
1527 bound_usd = p.bound_usd(),
1528 "cache keep-warm ping"
1529 );
1530 if let Err(reason) = result {
1531 if matches!(
1532 reason,
1533 PeriodEnd::PartialMiss { .. }
1534 | PeriodEnd::CacheMiss { .. }
1535 | PeriodEnd::UncachedInput { .. }
1536 | PeriodEnd::NothingCached
1537 ) {
1538 tracing::warn!(
1539 session = %self.id,
1540 reason = %reason,
1541 "cache keep-warm ping did not refresh the cache"
1542 );
1543 }
1544 self.cache_keepwarm_end(reason);
1545 return;
1546 }
1547 if let Some(cap) = self.config.max_session_cost.filter(|c| c.is_finite()) {
1548 if self.conv.session_cost >= cap {
1549 self.cache_keepwarm_end(PeriodEnd::CostCap { cap });
1550 }
1551 }
1552 }
1553 Ok(PingOutcome::PrefixChanged { .. }) => {
1554 self.cache_keepwarm_end(PeriodEnd::PrefixChanged);
1555 }
1556 Err(failure) => {
1557 let Some(p) = self.cache_keepwarm.period.as_mut() else {
1558 return;
1559 };
1560 match p.ping_failed(&failure, Instant::now(), SystemTime::now()) {
1561 Ok(delay) => tracing::info!(
1562 session = %self.id,
1563 period = p.id,
1564 retry_in_secs = delay.as_secs(),
1565 "cache keep-warm ping failed ({failure}); retrying"
1566 ),
1567 Err(reason) => self.cache_keepwarm_end(reason),
1568 }
1569 }
1570 }
1571 }
1572
1573 pub(crate) async fn cache_keepwarm_recheck(&mut self) -> Option<bool> {
1577 let expect = self.cache_keepwarm.period.as_ref()?.fingerprint();
1578 if self.cache_keepwarm.period.as_ref()?.in_flight() {
1579 return Some(true);
1580 }
1581 let matches = crate::runtime::keepwarm::history_ready(&self.conv.api_messages)
1582 && self
1583 .runtime
1584 .cache_keepwarm_probe(&self.conv.api_messages)
1585 .await
1586 .is_ok_and(|p| p.fingerprint == expect);
1587 if !matches {
1588 self.cache_keepwarm_end(PeriodEnd::PrefixChanged);
1589 }
1590 Some(matches)
1591 }
1592
1593 pub(crate) fn cache_keepwarm_status_json(
1595 &self,
1596 prefix_matches: Option<bool>,
1597 ) -> serde_json::Value {
1598 let kw = &self.cache_keepwarm;
1599 let (now, wall) = (Instant::now(), SystemTime::now());
1600 let period = kw.period.as_ref().map(|p| {
1601 let next = p
1603 .deadline(now, wall)
1604 .map(|d| d.saturating_duration_since(now).as_secs());
1605 serde_json::json!({
1606 "id": p.id,
1607 "ttl_secs": governing_ttl(p.ttl()).as_secs(),
1608 "pings": p.pings(),
1609 "ledger_usd": p.ledger_usd(),
1610 "bound_usd": p.bound_usd(),
1611 "rewrite_usd": p.rewrite_usd(),
1612 "reserve_usd": p.reserve_usd(),
1613 "expected_ping_usd": p.expected_next_usd(),
1614 "worst_ping_usd": p.exposure_next_usd(),
1615 "prefix_tokens": p.prefix_tokens(),
1616 "next_ping_in_secs": next.unwrap_or(0),
1617 "in_flight": p.in_flight(),
1618 })
1619 });
1620 serde_json::json!({
1621 "mode": kw.mode().to_string(),
1622 "active": kw.active(),
1623 "period": period,
1624 "prefix_matches": prefix_matches,
1625 "session_pings": kw.session_pings,
1626 "session_keepwarm_usd": self.conv.keepwarm_cost,
1627 "last_end": kw.last_end,
1628 })
1629 }
1630
1631 pub(crate) async fn cache_ttl_status_json(&mut self) -> serde_json::Value {
1638 let resolved = if self.runtime.cache_keepwarm_ineligibility().is_none() {
1640 Some(self.runtime.resolved_cache_ttl().await.as_str())
1641 } else {
1642 None
1643 };
1644 serde_json::json!({
1645 "setting": self.runtime.cache_ttl().as_str(),
1646 "resolved": resolved,
1647 })
1648 }
1649
1650 pub(crate) async fn cache_keepwarm_command(&mut self, arg: &str) -> serde_json::Value {
1652 let arg = arg.trim();
1653 if !(arg.is_empty() || arg.eq_ignore_ascii_case("status")) {
1654 let Some(mode) = CacheKeepWarm::parse(arg) else {
1655 return serde_json::json!({
1656 "kind": "error",
1657 "text": format!(
1658 "usage: /keepwarm [status|off|auto|<N>h] (N = 1–{}; unknown: {arg})",
1659 crate::core::config::MAX_CACHE_KEEPWARM_HOURS
1660 ),
1661 });
1662 };
1663 self.cache_keepwarm.set_override(mode);
1664 if self.keep_warm {
1665 self.note_pin(); }
1667 if mode.is_off() {
1668 self.cache_keepwarm_end(PeriodEnd::Disabled);
1669 } else if let Some(p) = self.cache_keepwarm.period.as_mut() {
1670 p.set_mode(mode);
1671 } else {
1672 self.cache_keepwarm_try_start(None).await;
1674 }
1675 }
1676 let prefix_matches = self.cache_keepwarm_recheck().await;
1677 let mut status = self.cache_keepwarm_status_json(prefix_matches);
1678 status["cache_ttl"] = self.cache_ttl_status_json().await;
1679 serde_json::json!({
1680 "kind": "output",
1681 "text": render_status(&status),
1682 "keepwarm": status,
1683 })
1684 }
1685}
1686
1687pub(crate) fn render_status(s: &serde_json::Value) -> String {
1689 let mode = s["mode"].as_str().unwrap_or("off");
1690 let spent = s["session_keepwarm_usd"].as_f64().unwrap_or(0.0);
1691 let pings = s["session_pings"].as_u64().unwrap_or(0);
1692 let mut out = format!("cache keep-warm: {mode} — this session: {pings} pings, ${spent:.4}");
1693 if let Some(resolved) = s["cache_ttl"]["resolved"].as_str() {
1695 out.push_str(&format!("\nprompt-cache TTL: {resolved}"));
1696 if s["cache_ttl"]["setting"].as_str() == Some("auto") {
1697 out.push_str(" (auto)");
1698 }
1699 }
1700 let p = &s["period"];
1701 if p.is_object() {
1702 let reserve = p["reserve_usd"].as_f64().unwrap_or(0.0);
1703 let bound = if reserve > 0.0 {
1704 format!(
1705 "${:.4} bound (${:.4} rewrite + ${reserve:.4} thinking reserve)",
1706 p["bound_usd"].as_f64().unwrap_or(0.0),
1707 p["rewrite_usd"].as_f64().unwrap_or(0.0),
1708 )
1709 } else {
1710 format!("${:.4} bound", p["bound_usd"].as_f64().unwrap_or(0.0))
1711 };
1712 out.push_str(&format!(
1713 "\nactive: ping {} due in {}s (TTL {}s); ${:.4} spent of {bound}; prefix {}",
1714 p["pings"].as_u64().unwrap_or(0) + 1,
1715 p["next_ping_in_secs"].as_u64().unwrap_or(0),
1716 p["ttl_secs"].as_u64().unwrap_or(0),
1717 p["ledger_usd"].as_f64().unwrap_or(0.0),
1718 match s["prefix_matches"].as_bool() {
1719 Some(true) => "unchanged",
1720 Some(false) => "changed",
1721 None => "unchecked",
1722 },
1723 ));
1724 } else {
1725 out.push_str("\nidle period: none");
1726 }
1727 if let Some(last) = s["last_end"].as_str() {
1728 out.push_str(&format!("\nlast: {last}"));
1729 }
1730 out
1731}
1732
1733#[cfg(test)]
1734mod tests {
1735 use super::*;
1736
1737 #[test]
1738 fn client_qualification_rule() {
1739 use AttachMode::{Mirror, Observe, Takeover};
1740 use ClientKind::{Attach, Chat, Rpc, Server, Test, Tui};
1741 for headless in [false, true] {
1742 for kind in [Tui, Chat, Attach, Rpc, Server, Test] {
1743 assert!(!client_qualifies(kind, Observe, headless), "{kind:?} observe");
1745 }
1746 for mode in [Mirror, Takeover] {
1747 assert!(client_qualifies(Tui, mode, headless));
1749 assert!(client_qualifies(Attach, mode, headless));
1750 assert!(!client_qualifies(Test, mode, headless));
1751 for kind in [Chat, Rpc, Server] {
1753 assert_eq!(client_qualifies(kind, mode, headless), headless, "{kind:?}");
1754 }
1755 }
1756 }
1757 }
1758
1759 const MIN: Duration = Duration::from_secs(60);
1760 const OPUS: &str = "claude-opus-4-7";
1761 const SONNET: &str = "claude-sonnet-4-6";
1762 const P: u64 = 300_000;
1763
1764 fn cfg(mode: CacheKeepWarm, ttl: CacheTtl, model: &str) -> PeriodConfig {
1766 cfg_max(mode, ttl, model, 1)
1767 }
1768
1769 fn cfg_max(mode: CacheKeepWarm, ttl: CacheTtl, model: &str, max_tokens: u64) -> PeriodConfig {
1770 PeriodConfig {
1771 mode,
1772 ttl,
1773 model: model.to_string(),
1774 max_tokens,
1775 output_estimate: ping_output_estimate(max_tokens),
1776 interval_override: None,
1777 auto_max_hours: 0,
1778 reply_tokens: None,
1779 }
1780 }
1781
1782 fn cfg_worst(mode: CacheKeepWarm, ttl: CacheTtl, model: &str, max_tokens: u64) -> PeriodConfig {
1785 PeriodConfig {
1786 output_estimate: max_tokens,
1787 ..cfg_max(mode, ttl, model, max_tokens)
1788 }
1789 }
1790
1791 fn ev_1h() -> TtlEvidence {
1793 TtlEvidence {
1794 saw_1h: true,
1795 ..TtlEvidence::default()
1796 }
1797 }
1798
1799 fn anchor_at(mono: Instant, wall: SystemTime) -> CacheAnchor {
1800 CacheAnchor { mono, wall }
1801 }
1802
1803 fn usage(model: &str, read: u64, write: u64, input: u64, out: u64) -> KeepWarmUsage {
1804 KeepWarmUsage {
1805 input_tokens: input,
1806 output_tokens: out,
1807 cache_read_tokens: read,
1808 cache_write_tokens: write,
1809 cache_write_5m: None,
1810 cache_write_1h: None,
1811 model: Some(model.to_string()),
1812 }
1813 }
1814
1815 fn last_read(model: &str) -> KeepWarmUsage {
1817 usage(model, P, 0, 0, 0)
1818 }
1819
1820 fn start(c: PeriodConfig, last: &KeepWarmUsage, t0: Instant, w0: SystemTime) -> Period {
1821 Period::start(1, c, 7, last, ev_1h(), anchor_at(t0, w0), t0, w0).expect("starts")
1822 }
1823
1824 fn run_from(
1828 cfg: PeriodConfig,
1829 last: &KeepWarmUsage,
1830 usages: &[KeepWarmUsage],
1831 ) -> (u32, f64, f64, PeriodEnd, Duration) {
1832 let t0 = Instant::now();
1833 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000);
1834 let model = cfg.model.clone();
1835 let mut p = start(cfg, last, t0, w0);
1836 let mut i = 0;
1837 loop {
1838 let at = p.next_at();
1839 let wall = w0 + at.saturating_duration_since(t0);
1840 match p.fire(at, wall, 0.0, None) {
1841 Fire::Ping => {}
1842 Fire::End(r) => {
1843 return (p.pings(), p.ledger_usd(), p.bound_usd(), r, at - t0);
1844 }
1845 Fire::NotYet => panic!("due but not fired"),
1846 }
1847 let u = &usages[i.min(usages.len() - 1)];
1848 i += 1;
1849 p.charge(u.cost_usd(&model));
1850 if let Err(r) = p.ping_sent(u, Some(anchor_at(at, wall)), false) {
1851 return (p.pings(), p.ledger_usd(), p.bound_usd(), r, at - t0);
1852 }
1853 assert!(i < 1000, "runaway");
1854 }
1855 }
1856
1857 fn run(cfg: PeriodConfig, usages: &[KeepWarmUsage]) -> (u32, f64, f64, PeriodEnd, Duration) {
1858 let last = last_read(&cfg.model);
1859 run_from(cfg, &last, usages)
1860 }
1861
1862 #[test]
1867 fn review_205_possible_cache_miss_must_fit_admission_cap() {
1868 let t0 = Instant::now();
1869 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000);
1870 let mut p = start(
1871 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
1872 &last_read(OPUS),
1873 t0,
1874 w0,
1875 );
1876 let at = p.next_at();
1877 let wall = w0 + at.saturating_duration_since(t0);
1878 let cap = 0.30;
1879 let decision = p.fire(at, wall, 0.0, Some(cap));
1880 let miss = KeepWarmUsage {
1881 cache_write_tokens: P,
1882 cache_write_1h: Some(P),
1883 ..KeepWarmUsage::default()
1884 };
1885 println!(
1886 "REVIEW estimated_worst={} possible_write={} cap={cap} decision={decision:?}",
1887 p.worst_next_usd,
1888 miss.cost_usd(OPUS)
1889 );
1890 if matches!(decision, Fire::Ping) {
1891 assert!(
1892 miss.cost_usd(OPUS) <= cap,
1893 "admitted request can exceed its session cost cap on a cache miss"
1894 );
1895 }
1896 }
1897
1898 #[test]
1902 fn a_cap_with_room_for_a_cache_miss_still_pings() {
1903 let t0 = Instant::now();
1904 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000);
1905 let mut p = start(
1906 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
1907 &last_read(OPUS),
1908 t0,
1909 w0,
1910 );
1911 let exposure = p.exposure_next_usd();
1913 let miss = usage(OPUS, 0, P, PING_INPUT_ESTIMATE, 1);
1914 let miss = KeepWarmUsage {
1915 cache_write_1h: Some(P),
1916 ..miss
1917 };
1918 assert!((exposure - miss.cost_usd(OPUS)).abs() < 1e-9, "{exposure}");
1919 assert!(p.worst_next_usd() < p.bound_usd(), "bound basis is the read");
1920 let at = p.next_at();
1921 let wall = w0 + at.saturating_duration_since(t0);
1922 assert_eq!(
1923 p.fire(at, wall, 0.0, Some(3.0)),
1924 Fire::End(PeriodEnd::CostCap { cap: 3.0 })
1925 );
1926 assert_eq!(p.fire(at, wall, 0.0, Some(3.01)), Fire::Ping);
1927 }
1928
1929 #[test]
1930 fn ski_rental_1h_is_nineteen_pure_reads_over_about_17h() {
1931 let (pings, ledger, bound, reason, elapsed) = run(
1935 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
1936 &[usage(OPUS, P, 0, 0, 0)],
1937 );
1938 assert!((bound - 3.0).abs() < 1e-9, "{bound}");
1939 assert_eq!(pings, 19);
1940 assert!(ledger <= bound * (1.0 + 1e-9));
1941 assert!(matches!(reason, PeriodEnd::BoundReached { .. }), "{reason}");
1942 assert_eq!(elapsed, 19 * 55 * MIN);
1943 }
1944
1945 #[test]
1946 fn ski_rental_5m_is_twelve_pure_reads_over_about_55_min() {
1947 let (pings, ledger, bound, reason, elapsed) = run(
1949 cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS),
1950 &[usage(OPUS, P, 0, 0, 0)],
1951 );
1952 assert!((bound - 1.875).abs() < 1e-9, "{bound}");
1953 assert_eq!(pings, 12);
1954 assert!((ledger - 1.8).abs() < 1e-9);
1955 assert!(matches!(reason, PeriodEnd::BoundReached { .. }), "{reason}");
1956 assert_eq!(elapsed, 12 * Duration::from_secs(270));
1957 }
1958
1959 #[test]
1960 fn measured_overhead_and_first_write_are_in_the_ledger() {
1961 let last = usage(OPUS, P - 2_000, 0, 0, 2_000);
1965 let first = usage(OPUS, P - 2_000, 2_000, 40, 1);
1966 let steady = usage(OPUS, P, 0, 40, 1);
1967 let (pings, ledger, bound, reason, _) = run_from(
1968 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
1969 &last,
1970 &[first.clone(), steady.clone()],
1971 );
1972 assert!(matches!(reason, PeriodEnd::BoundReached { .. }), "{reason}");
1973 let expected_ledger = first.cost_usd(OPUS) + f64::from(pings - 1) * steady.cost_usd(OPUS);
1974 assert!(
1975 (ledger - expected_ledger).abs() < 1e-9,
1976 "ledger is the measured sum"
1977 );
1978 assert!(ledger <= bound * (1.0 + 1e-9), "never past the bound");
1979 assert!(ledger + steady.cost_usd(OPUS) > bound);
1980 assert_eq!(pings, 19);
1981 }
1982
1983 #[test]
1984 fn an_expensive_ping_stops_the_period_early() {
1985 let heavy_ping = usage(SONNET, P, 0, 40, 5_000);
1988 let (heavy, ledger, bound, ..) = run(
1989 cfg_max(CacheKeepWarm::Auto, CacheTtl::OneHour, SONNET, 5_001),
1990 std::slice::from_ref(&heavy_ping),
1991 );
1992 let (worst, worst_ledger, worst_bound, ..) = run(
1993 cfg_worst(CacheKeepWarm::Auto, CacheTtl::OneHour, SONNET, 5_001),
1994 std::slice::from_ref(&heavy_ping),
1995 );
1996 let (light, ..) = run(
1997 cfg_max(CacheKeepWarm::Auto, CacheTtl::OneHour, SONNET, 31),
1998 &[usage(SONNET, P, 0, 40, 30)],
1999 );
2000 assert_eq!(worst, 10);
2003 assert!((worst_bound - 1.80).abs() < 1e-9 && worst_ledger <= 1.80);
2004 let reserve = (5_001 - PING_OUTPUT_ESTIMATE) as f64 * 15.0 / 1e6;
2007 assert_eq!(heavy, 11);
2008 assert!((bound - (1.80 + reserve)).abs() < 1e-9, "{bound}");
2009 assert!(ledger > 1.80 && ledger <= bound, "{ledger} of {bound}");
2010 assert_eq!(light, 19);
2011 }
2012
2013 #[test]
2014 fn budgeted_thinking_output_is_accounted() {
2015 let u = usage(SONNET, P, 0, 40, 300);
2016 let read_only = usage(SONNET, P, 0, 0, 0);
2017 let thinking_usd = u.cost_usd(SONNET) - read_only.cost_usd(SONNET);
2018 assert!((thinking_usd - (300.0 * 15.0 + 40.0 * 3.0) / 1e6).abs() < 1e-12);
2020 let (pings, ledger, ..) = run(
2021 cfg_max(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, SONNET, 301),
2022 std::slice::from_ref(&u),
2023 );
2024 assert!((ledger - f64::from(pings) * u.cost_usd(SONNET)).abs() < 1e-9);
2025 }
2026
2027 #[test]
2034 fn budgeted_thinking_worst_case_must_fit_the_bound() {
2035 let t0 = Instant::now();
2036 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000);
2037 let small = usage(SONNET, 20_000, 0, 0, 0);
2038 let start_5m = |c| {
2039 Period::start(1, c, 7, &small, TtlEvidence::default(), anchor_at(t0, w0), t0, w0)
2040 };
2041 let worst = cfg_worst(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, SONNET, 32_769);
2042 match start_5m(worst).unwrap_err() {
2043 PeriodEnd::PingTooExpensive {
2044 expected_usd,
2045 worst_usd,
2046 rewrite_usd,
2047 } => {
2048 assert!((rewrite_usd - 0.075).abs() < 1e-9, "{rewrite_usd}");
2049 assert!(worst_usd > 0.49, "{worst_usd}");
2050 assert_eq!(expected_usd, worst_usd, "no estimate: priced at worst");
2051 }
2052 other => panic!("{other}"),
2053 }
2054 let p = start_5m(cfg_max(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, SONNET, 32_769))
2055 .expect("expected ping fits the rewrite");
2056 assert!((p.rewrite_usd() - 0.075).abs() < 1e-9);
2057 let reserve = (32_769 - PING_OUTPUT_ESTIMATE) as f64 * 15.0 / 1e6;
2058 assert!((p.reserve_usd() - reserve).abs() < 1e-9, "{}", p.reserve_usd());
2059 assert!((p.bound_usd() - (0.075 + reserve)).abs() < 1e-9);
2060 assert!(p.expected_next_usd() <= p.rewrite_usd());
2061 assert!(p.worst_next_usd() <= p.bound_usd() * (1.0 + 1e-12));
2062 let p = start(
2064 cfg_worst(CacheKeepWarm::Auto, CacheTtl::OneHour, SONNET, 32_769),
2065 &usage(SONNET, P, 0, 0, 0),
2066 t0,
2067 w0,
2068 );
2069 assert!(p.worst_next_usd() > 0.58 && p.worst_next_usd() < p.bound_usd());
2070 assert_eq!(p.reserve_usd(), 0.0);
2071 }
2072
2073 fn break_even(c: &PeriodConfig) -> u64 {
2075 let t0 = Instant::now();
2076 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000);
2077 let starts = |prefix| {
2078 let last = usage(&c.model, prefix, 0, 0, 0);
2079 Period::start(1, c.clone(), 7, &last, ev_1h(), anchor_at(t0, w0), t0, w0).is_ok()
2080 };
2081 let (mut lo, mut hi) = (1, 1_000_000);
2082 assert!(starts(hi) && !starts(lo));
2083 while hi - lo > 1 {
2084 let mid = (lo + hi) / 2;
2085 if starts(mid) {
2086 hi = mid;
2087 } else {
2088 lo = mid;
2089 }
2090 }
2091 hi
2092 }
2093
2094 #[test]
2099 fn break_even_prefix_budgeted_vs_adaptive() {
2100 use CacheTtl::{FiveMinutes, OneHour};
2101 let auto = CacheKeepWarm::Auto;
2102 let worst_5m = break_even(&cfg_worst(auto, FiveMinutes, SONNET, 4_097));
2103 let worst_1h = break_even(&cfg_worst(auto, OneHour, SONNET, 4_097));
2104 let est_5m = break_even(&cfg_max(auto, FiveMinutes, SONNET, 4_097));
2105 let est_1h = break_even(&cfg_max(auto, OneHour, SONNET, 4_097));
2106 let adaptive_5m = break_even(&cfg(auto, FiveMinutes, OPUS));
2107 let adaptive_1h = break_even(&cfg(auto, OneHour, OPUS));
2108 println!(
2109 "BREAK-EVEN worst 5m={worst_5m} 1h={worst_1h}; estimate 5m={est_5m} 1h={est_1h}; \
2110 adaptive 5m={adaptive_5m} 1h={adaptive_1h}"
2111 );
2112 assert!((17_860..=17_880).contains(&worst_5m), "{worst_5m}");
2114 assert!((10_810..=10_820).contains(&worst_1h), "{worst_1h}");
2115 assert!((1_165..=1_175).contains(&est_5m), "{est_5m}");
2117 assert!((705..=712).contains(&est_1h), "{est_1h}");
2118 assert!(adaptive_5m < 1_024 && adaptive_1h < 1_024);
2120 let off = cfg_max(auto, FiveMinutes, SONNET, 1);
2122 let off_no_estimate = PeriodConfig {
2123 output_estimate: 1,
2124 ..off.clone()
2125 };
2126 assert_eq!(break_even(&off), break_even(&off_no_estimate));
2127 }
2128
2129 #[test]
2133 fn the_bound_holds_whatever_the_pings_think() {
2134 for ttl in [CacheTtl::FiveMinutes, CacheTtl::OneHour] {
2135 for prefix in [1_200, 2_000, 5_000, 20_000, 100_000] {
2136 for out in [1, 100, PING_OUTPUT_ESTIMATE, 1_000, 4_097] {
2137 let c = cfg_max(CacheKeepWarm::Auto, ttl, SONNET, 4_097);
2138 let last = usage(SONNET, prefix, 0, 0, 0);
2139 let ping = usage(SONNET, prefix, 0, 40, out);
2140 let (pings, ledger, bound, reason, _) = run_from(c, &last, &[ping]);
2141 let rewrite = rewrite_cost_usd(SONNET, ttl, prefix);
2142 let ctx = format!("{ttl:?} prefix {prefix} out {out}: {pings} pings, {reason}");
2143 assert!(pings >= 1, "{ctx}");
2144 assert!(ledger <= bound * (1.0 + 1e-9), "{ctx}: ${ledger} > ${bound}");
2145 if out <= PING_OUTPUT_ESTIMATE {
2146 assert!(ledger <= rewrite * (1.0 + 1e-9), "{ctx}: ${ledger} > ${rewrite}");
2147 }
2148 assert!(matches!(reason, PeriodEnd::BoundReached { .. }), "{ctx}");
2149 }
2150 }
2151 }
2152 }
2153
2154 #[test]
2155 fn max_session_cost_is_checked_against_the_worst_case() {
2156 let t0 = Instant::now();
2157 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(100_000);
2158 let mut p = start(
2161 cfg_max(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS, 10_001),
2162 &last_read(OPUS),
2163 t0,
2164 w0,
2165 );
2166 let at = p.next_at();
2167 let wat = w0 + (at - t0);
2168 assert_eq!(
2170 p.fire(at, wat, 4.70, Some(5.0)),
2171 Fire::End(PeriodEnd::CostCap { cap: 5.0 })
2172 );
2173 assert_eq!(
2176 p.fire(at, wat, 4.50, Some(5.0)),
2177 Fire::End(PeriodEnd::CostCap { cap: 5.0 })
2178 );
2179 assert_eq!(p.fire(at, wat, 1.70, Some(5.0)), Fire::Ping);
2181 }
2182
2183 #[test]
2188 fn small_prompt_below_the_cache_minimum_never_starts() {
2189 let t0 = Instant::now();
2190 let w0 = SystemTime::UNIX_EPOCH;
2191 let below_minimum = usage(OPUS, 0, 0, 800, 20);
2192 assert_eq!(
2193 Period::start(
2194 1,
2195 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
2196 7,
2197 &below_minimum,
2198 ev_1h(),
2199 anchor_at(t0, w0),
2200 t0,
2201 w0
2202 )
2203 .unwrap_err(),
2204 PeriodEnd::NothingCached
2205 );
2206 }
2207
2208 fn fired(c: PeriodConfig) -> (Period, Instant, SystemTime) {
2209 let t0 = Instant::now();
2210 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(90_000);
2211 let mut p = start(c, &last_read(OPUS), t0, w0);
2212 let at = p.next_at();
2213 let wat = w0 + (at - t0);
2214 assert_eq!(p.fire(at, wat, 0.0, None), Fire::Ping);
2215 (p, at, wat)
2216 }
2217
2218 #[test]
2219 fn a_ping_that_cached_nothing_ends_the_period() {
2220 let (mut p, ..) = fired(cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS));
2221 assert_eq!(
2222 p.ping_sent(&usage(OPUS, 0, 0, P, 1), None, false),
2223 Err(PeriodEnd::NothingCached)
2224 );
2225 }
2226
2227 #[test]
2228 fn a_ping_paying_for_a_large_uncached_share_ends_the_period() {
2229 let (mut p, ..) = fired(cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS));
2231 assert_eq!(
2232 p.ping_sent(&usage(OPUS, P - 60_000, 0, 60_000, 1), None, false),
2233 Err(PeriodEnd::UncachedInput {
2234 input: 60_000,
2235 prefix: P
2236 })
2237 );
2238 let (mut p, ..) = fired(cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS));
2240 assert_eq!(p.ping_sent(&usage(OPUS, P, 0, 40, 1), None, false), Ok(()));
2241 }
2242
2243 #[test]
2244 fn a_ping_that_writes_instead_of_reading_ends_the_period() {
2245 let (mut p, ..) = fired(cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS));
2246 assert_eq!(
2247 p.ping_sent(&usage(OPUS, 0, P, 40, 1), None, false),
2248 Err(PeriodEnd::CacheMiss { written: P })
2249 );
2250 }
2251
2252 #[test]
2255 fn a_partial_miss_after_ping_one_ends_the_period() {
2256 let (mut p, at, wat) = fired(cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS));
2257 p.ping_sent(&usage(OPUS, P, 0, 40, 1), Some(anchor_at(at, wat)), false)
2258 .unwrap();
2259 let at = p.next_at();
2260 let wat = wat + (at - p.anchor.mono);
2261 assert_eq!(p.fire(at, wat, 0.0, None), Fire::Ping);
2262 assert_eq!(
2263 p.ping_sent(&usage(OPUS, P - 50_000, 50_000, 40, 1), None, false),
2264 Err(PeriodEnd::PartialMiss {
2265 written: 50_000,
2266 prefix: P
2267 })
2268 );
2269 }
2270
2271 #[test]
2272 fn ping_one_may_write_the_last_reply() {
2273 let t0 = Instant::now();
2274 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(3_000);
2275 let last = usage(OPUS, 250_000, 0, 0, 50_000);
2277 let mut p = start(
2278 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
2279 &last,
2280 t0,
2281 w0,
2282 );
2283 assert_eq!(p.prefix_tokens(), P);
2284 let at = p.next_at();
2285 assert_eq!(p.fire(at, w0 + (at - t0), 0.0, None), Fire::Ping);
2286 assert_eq!(
2287 p.ping_sent(&usage(OPUS, 250_000, 50_000, 40, 1), None, false),
2288 Ok(()),
2289 "the reply's write is expected on ping 1"
2290 );
2291 }
2292
2293 #[test]
2296 fn honoured_ttl_needs_evidence_for_1h() {
2297 use CacheTtl::*;
2298 let none = TtlEvidence::default();
2299 assert_eq!(honoured_ttl(FiveMinutes, ev_1h()), FiveMinutes);
2300 assert_eq!(honoured_ttl(Hybrid, ev_1h()), Hybrid);
2301 assert_eq!(honoured_ttl(OneHour, none), FiveMinutes, "no evidence");
2302 assert_eq!(honoured_ttl(OneHour, ev_1h()), OneHour);
2303 let wrote_1h = TtlEvidence {
2304 write_1h: 10,
2305 ..none
2306 };
2307 assert_eq!(honoured_ttl(OneHour, wrote_1h), OneHour);
2308 let wrote_5m = TtlEvidence {
2309 saw_1h: true,
2310 write_5m: 10,
2311 ..none
2312 };
2313 assert_eq!(
2314 honoured_ttl(OneHour, wrote_5m),
2315 FiveMinutes,
2316 "5m-only write"
2317 );
2318 let downgraded = TtlEvidence {
2319 downgraded: true,
2320 write_1h: 10,
2321 saw_1h: true,
2322 ..none
2323 };
2324 assert_eq!(honoured_ttl(OneHour, downgraded), FiveMinutes, "latch wins");
2325 }
2326
2327 #[test]
2331 fn a_silent_downgrade_governs_by_5m() {
2332 let t0 = Instant::now();
2333 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(4_000);
2334 let downgraded = TtlEvidence {
2335 downgraded: true,
2336 ..TtlEvidence::default()
2337 };
2338 let p = Period::start(
2339 1,
2340 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
2341 7,
2342 &last_read(OPUS),
2343 downgraded,
2344 anchor_at(t0, w0),
2345 t0,
2346 w0,
2347 )
2348 .unwrap();
2349 assert_eq!(p.ttl(), CacheTtl::FiveMinutes);
2350 assert_eq!(p.next_at(), t0 + Duration::from_secs(270));
2351 assert!((p.bound_usd() - 1.875).abs() < 1e-9);
2352
2353 let mut p = Period::start(
2356 2,
2357 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
2358 7,
2359 &usage(OPUS, P - 1_000, 0, 0, 1_000),
2360 TtlEvidence::default(),
2361 anchor_at(t0, w0),
2362 t0,
2363 w0,
2364 )
2365 .unwrap();
2366 assert_eq!(p.ttl(), CacheTtl::FiveMinutes);
2367 let at = p.next_at();
2368 let wat = w0 + (at - t0);
2369 assert_eq!(p.fire(at, wat, 0.0, None), Fire::Ping);
2370 let mut first = usage(OPUS, P - 1_000, 1_000, 40, 1);
2371 first.cache_write_1h = Some(1_000);
2372 first.cache_write_5m = Some(0);
2373 p.ping_sent(&first, Some(anchor_at(at, wat)), false)
2374 .unwrap();
2375 assert_eq!(p.ttl(), CacheTtl::OneHour);
2376 assert_eq!(p.next_at(), at + 55 * MIN);
2377 let at = p.next_at();
2378 let wat = wat + 55 * MIN;
2379 assert_eq!(p.fire(at, wat, 0.0, None), Fire::Ping);
2380 p.ping_sent(&usage(OPUS, P, 0, 40, 1), Some(anchor_at(at, wat)), true)
2381 .unwrap();
2382 assert_eq!(p.ttl(), CacheTtl::FiveMinutes);
2383 assert_eq!(p.next_at(), at + Duration::from_secs(270));
2384 }
2385
2386 #[test]
2389 fn five_minute_pings_are_scheduled_from_the_anchor() {
2390 let t0 = Instant::now();
2391 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(5_000);
2392 let c = cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS);
2393 let idle = t0 + MIN;
2395 let mut p = Period::start(
2396 1,
2397 c,
2398 7,
2399 &last_read(OPUS),
2400 ev_1h(),
2401 anchor_at(t0, w0),
2402 idle,
2403 w0 + MIN,
2404 )
2405 .unwrap();
2406 assert_eq!(
2407 p.deadline(idle, w0 + MIN),
2408 Some(t0 + Duration::from_secs(270))
2409 );
2410 assert_eq!(
2411 p.fire(
2412 t0 + Duration::from_secs(200),
2413 w0 + Duration::from_secs(200),
2414 0.0,
2415 None
2416 ),
2417 Fire::NotYet
2418 );
2419 let at = t0 + Duration::from_secs(270);
2420 let wat = w0 + Duration::from_secs(270);
2421 assert_eq!(p.fire(at, wat, 0.0, None), Fire::Ping);
2422 assert_eq!(p.deadline(at, wat), None, "no second ping while in flight");
2423 assert_eq!(p.fire(at, wat, 0.0, None), Fire::NotYet);
2424 let sent = anchor_at(at + Duration::from_secs(2), wat + Duration::from_secs(2));
2426 p.ping_sent(&usage(OPUS, P, 0, 0, 1), Some(sent), false)
2427 .unwrap();
2428 assert_eq!(
2429 p.deadline(sent.mono, sent.wall),
2430 Some(sent.mono + Duration::from_secs(270))
2431 );
2432 }
2433
2434 #[test]
2435 fn hybrid_is_governed_by_the_5m_history_marker() {
2436 assert_eq!(governing_ttl(CacheTtl::Hybrid), Duration::from_secs(300));
2437 assert_eq!(governing_ttl(CacheTtl::OneHour), Duration::from_secs(3600));
2438 assert_eq!(ping_interval(CacheTtl::Hybrid), Duration::from_secs(270));
2439 assert!((rewrite_cost_usd(OPUS, CacheTtl::Hybrid, P) - 1.875).abs() < 1e-9);
2441 let (pings, ..) = run(
2442 cfg(CacheKeepWarm::Auto, CacheTtl::Hybrid, OPUS),
2443 &[usage(OPUS, P, 0, 0, 0)],
2444 );
2445 assert_eq!(pings, 12);
2446 }
2447
2448 #[test]
2449 fn a_turn_that_streamed_past_the_interval_pings_at_once_or_not_at_all() {
2450 let t0 = Instant::now();
2451 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(9_000);
2452 let c = || cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS);
2453 let at = |s: u64| (t0 + Duration::from_secs(s), w0 + Duration::from_secs(s));
2454 let (idle, widle) = at(280);
2455 let p = Period::start(
2456 1,
2457 c(),
2458 7,
2459 &last_read(OPUS),
2460 ev_1h(),
2461 anchor_at(t0, w0),
2462 idle,
2463 widle,
2464 )
2465 .unwrap();
2466 assert_eq!(p.deadline(idle, widle), Some(idle), "due immediately");
2467 let (late, wlate) = at(290);
2468 let err = Period::start(
2469 1,
2470 c(),
2471 7,
2472 &last_read(OPUS),
2473 ev_1h(),
2474 anchor_at(t0, w0),
2475 late,
2476 wlate,
2477 )
2478 .unwrap_err();
2479 assert!(matches!(err, PeriodEnd::Expired { age_secs: 290 }), "{err}");
2480 }
2481
2482 #[test]
2483 fn sleep_past_expiry_ends_the_period_without_pinging() {
2484 let t0 = Instant::now();
2485 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(50_000);
2486 for ttl in [CacheTtl::FiveMinutes, CacheTtl::OneHour] {
2487 let c = cfg(CacheKeepWarm::Auto, ttl, OPUS);
2488 let mut p = start(c, &last_read(OPUS), t0, w0);
2489 let due = p.next_at();
2490 let wall = w0 + Duration::from_secs(3 * 3600);
2493 match p.fire(due, wall, 0.0, None) {
2494 Fire::End(PeriodEnd::Expired { age_secs }) => assert_eq!(age_secs, 3 * 3600),
2495 other => panic!("{ttl:?}: {other:?}"),
2496 }
2497 }
2498 }
2499
2500 #[test]
2505 fn a_suspend_pulls_the_deadline_forward_by_the_wall_clock() {
2506 let t0 = Instant::now();
2507 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(60_000);
2508 let mut p = start(
2509 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
2510 &last_read(OPUS),
2511 t0,
2512 w0,
2513 );
2514 let (now, wall) = (t0 + 10 * MIN, w0 + 40 * MIN);
2516 assert_eq!(p.deadline(now, wall), Some(now + 15 * MIN));
2517 assert_eq!(p.fire(now, wall, 0.0, None), Fire::NotYet);
2518 let (now, wall) = (now + 15 * MIN, wall + 15 * MIN);
2521 assert_eq!(p.fire(now, wall, 0.0, None), Fire::Ping);
2522 assert_eq!(p.fired_age(), 55 * MIN);
2523 }
2524
2525 #[test]
2526 fn the_actor_polls_the_wall_clock_while_a_period_is_active() {
2527 assert!(WALL_POLL < safety_margin(CacheTtl::FiveMinutes));
2528 let t0 = Instant::now();
2529 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(61_000);
2530 let mut kw = KeepWarmState::default();
2531 assert_eq!(kw.deadline_at(t0, w0), None, "no period: no timer");
2532 kw.period = Some(start(
2534 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
2535 &last_read(OPUS),
2536 t0,
2537 w0,
2538 ));
2539 assert_eq!(kw.deadline_at(t0, w0), Some(t0 + WALL_POLL));
2540 let (near, wnear) = (
2542 t0 + 55 * MIN - Duration::from_secs(3),
2543 w0 + 55 * MIN - Duration::from_secs(3),
2544 );
2545 assert_eq!(kw.deadline_at(near, wnear), Some(t0 + 55 * MIN));
2546 let p = kw.period.as_mut().unwrap();
2548 assert_eq!(p.fire(t0 + 55 * MIN, w0 + 55 * MIN, 0.0, None), Fire::Ping);
2549 assert_eq!(kw.deadline_at(t0 + 55 * MIN, w0 + 55 * MIN), None);
2550 }
2551
2552 #[test]
2555 fn transient_errors_back_off_within_the_ttl_then_end() {
2556 let t0 = Instant::now();
2557 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(70_000);
2558 let mut p = start(
2559 cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS),
2560 &last_read(OPUS),
2561 t0,
2562 w0,
2563 );
2564 let mut now = p.next_at();
2565 let transient = PingFailure::Transient("HTTP 529".into());
2566 for want in BACKOFF {
2567 let wall = w0 + now.saturating_duration_since(t0);
2568 assert_eq!(p.fire(now, wall, 0.0, None), Fire::Ping);
2569 assert_eq!(p.ping_failed(&transient, now, wall), Ok(want));
2570 assert_eq!(p.deadline(now, wall), Some(now + want));
2571 now += want;
2572 }
2573 let wall = w0 + now.saturating_duration_since(t0);
2574 assert_eq!(p.fire(now, wall, 0.0, None), Fire::Ping);
2575 assert!(matches!(
2576 p.ping_failed(&transient, now, wall),
2577 Err(PeriodEnd::TransientExhausted(_))
2578 ));
2579
2580 let mut p = start(
2582 cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS),
2583 &last_read(OPUS),
2584 t0,
2585 w0,
2586 );
2587 let at = p.next_at();
2588 let wat = w0 + Duration::from_secs(270);
2589 assert_eq!(p.fire(at, wat, 0.0, None), Fire::Ping);
2590 assert_eq!(p.ping_failed(&transient, at, wat), Ok(BACKOFF[0]));
2591 let at = at + BACKOFF[0];
2592 let wat = wat + BACKOFF[0];
2593 assert_eq!(p.fire(at, wat, 0.0, None), Fire::Ping);
2594 assert!(matches!(
2595 p.ping_failed(&transient, at, wat),
2596 Err(PeriodEnd::TransientExhausted(_))
2597 ));
2598 }
2599
2600 #[test]
2602 fn failed_ping_spend_is_in_the_ledger() {
2603 let (mut p, at, wat) = fired(cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, OPUS));
2604 p.charge(1.80);
2605 assert!(matches!(
2606 p.ping_failed(&PingFailure::Transient("x".into()), at, wat),
2607 Err(PeriodEnd::BoundReached { .. })
2608 ));
2609 }
2610
2611 #[test]
2612 fn a_4xx_ends_the_period() {
2613 let (mut p, at, wat) = fired(cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS));
2614 let err = p
2615 .ping_failed(&PingFailure::Fatal("HTTP 400".into()), at, wat)
2616 .unwrap_err();
2617 assert_eq!(err, PeriodEnd::Rejected("HTTP 400".into()));
2618 }
2619
2620 #[test]
2621 fn hours_window_and_bound_whichever_first() {
2622 let (pings, _, _, reason, _) = run(
2624 cfg(CacheKeepWarm::Hours(1), CacheTtl::OneHour, OPUS),
2625 &[usage(OPUS, P, 0, 0, 0)],
2626 );
2627 assert_eq!((pings, reason), (1, PeriodEnd::HoursReached(1)));
2628 let (pings, _, _, reason, _) = run(
2630 cfg(CacheKeepWarm::Hours(2), CacheTtl::FiveMinutes, OPUS),
2631 &[usage(OPUS, P, 0, 0, 0)],
2632 );
2633 assert_eq!(pings, 12);
2634 assert!(matches!(reason, PeriodEnd::BoundReached { .. }));
2635 let t0 = Instant::now();
2637 let w0 = SystemTime::UNIX_EPOCH;
2638 let mut c = cfg(CacheKeepWarm::Hours(1), CacheTtl::OneHour, OPUS);
2639 c.interval_override = Some(Duration::from_secs(2 * 3600));
2640 assert_eq!(
2641 Period::start(
2642 1,
2643 c,
2644 7,
2645 &last_read(OPUS),
2646 ev_1h(),
2647 anchor_at(t0, w0),
2648 t0,
2649 w0
2650 )
2651 .unwrap_err(),
2652 PeriodEnd::HoursReached(1)
2653 );
2654 }
2655
2656 #[test]
2662 #[serial_test::serial(synaps_keepwarm_interval)]
2663 fn the_interval_override_is_set_only_by_code_never_by_the_environment() {
2664 let _env = crate::test_env::EnvVarGuard::set("SYNAPS_CACHE_KEEPWARM_INTERVAL_MS", "250");
2665 assert_eq!(interval_override(), None, "the old env var does nothing");
2666 {
2667 let _g = testing::IntervalOverride::set(Duration::from_millis(250));
2668 assert_eq!(interval_override(), Some(Duration::from_millis(250)));
2669 }
2670 assert_eq!(interval_override(), None, "cleared with the guard");
2671 }
2672
2673 #[test]
2678 fn a_re_enabled_period_continues_the_exhausted_ledger() {
2679 let t0 = Instant::now();
2680 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(80_000);
2681 let c = || cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS);
2682 let mut p = start(c(), &last_read(OPUS), t0, w0);
2683 let read = usage(OPUS, P, 0, 0, 0);
2684 let end = loop {
2685 let at = p.next_at();
2686 let wat = w0 + (at - t0);
2687 match p.fire(at, wat, 0.0, None) {
2688 Fire::Ping => {}
2689 Fire::End(r) => break r,
2690 Fire::NotYet => unreachable!(),
2691 }
2692 p.charge(read.cost_usd(OPUS));
2693 if let Err(r) = p.ping_sent(&read, Some(anchor_at(at, wat)), false) {
2694 break r;
2695 }
2696 };
2697 assert!(matches!(end, PeriodEnd::BoundReached { .. }), "{end}");
2698 let carry = p.epoch_carry();
2699 assert_eq!(carry.pings, 19);
2700 let anchor = p.anchor;
2702 let err = Period::start_in_epoch(
2703 2,
2704 c(),
2705 7,
2706 &last_read(OPUS),
2707 ev_1h(),
2708 anchor,
2709 anchor.mono,
2710 anchor.wall,
2711 Some(&carry),
2712 )
2713 .unwrap_err();
2714 assert!(matches!(err, PeriodEnd::BoundReached { .. }), "{err}");
2715 assert!(Period::start_in_epoch(
2717 3,
2718 c(),
2719 7,
2720 &last_read(OPUS),
2721 ev_1h(),
2722 anchor,
2723 anchor.mono,
2724 anchor.wall,
2725 None
2726 )
2727 .is_ok());
2728 }
2729
2730 #[test]
2733 fn ping_one_write_allowance_is_never_re_granted_in_an_epoch() {
2734 let t0 = Instant::now();
2735 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(81_000);
2736 let c = || cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS);
2737 let last = usage(OPUS, 250_000, 0, 0, 50_000);
2738 let mut p = start(c(), &last, t0, w0);
2739 let at = p.next_at();
2740 let wat = w0 + (at - t0);
2741 assert_eq!(p.fire(at, wat, 0.0, None), Fire::Ping);
2742 let first = usage(OPUS, 250_000, 50_000, 40, 1);
2743 p.charge(first.cost_usd(OPUS));
2744 p.ping_sent(&first, Some(anchor_at(at, wat)), false)
2745 .unwrap();
2746 let carry = p.epoch_carry();
2747 let mut q = Period::start_in_epoch(
2749 2,
2750 c(),
2751 7,
2752 &last,
2753 ev_1h(),
2754 anchor_at(at, wat),
2755 at,
2756 wat,
2757 Some(&carry),
2758 )
2759 .unwrap();
2760 assert_eq!(q.pings(), 1);
2761 assert!((q.ledger_usd() - first.cost_usd(OPUS)).abs() < 1e-12);
2762 let at2 = q.next_at();
2763 let wat2 = wat + (at2 - at);
2764 assert_eq!(q.fire(at2, wat2, 0.0, None), Fire::Ping);
2765 assert_eq!(
2767 q.ping_sent(&usage(OPUS, 250_000, 50_000, 40, 1), None, false),
2768 Err(PeriodEnd::PartialMiss {
2769 written: 50_000,
2770 prefix: P
2771 })
2772 );
2773 }
2774
2775 #[test]
2781 fn ping_counts_per_multiplier_and_the_auto_cap() {
2782 for (model, uncapped, capped) in [
2783 ("claude-opus-4-7", 19, 19),
2784 ("claude-opus-5-5", 39, 26),
2785 ("claude-fable-5-1", 79, 26),
2786 ] {
2787 let mut c = cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, model);
2788 let (pings, _, _, reason, elapsed) = run(c.clone(), &[usage(model, P, 0, 0, 0)]);
2789 assert_eq!(pings, uncapped, "{model}");
2790 assert!(
2791 matches!(reason, PeriodEnd::BoundReached { .. }),
2792 "{model}: {reason}"
2793 );
2794 assert_eq!(elapsed, uncapped * 55 * MIN, "{model}");
2795 c.auto_max_hours = 24;
2796 let (pings, _, _, reason, _) = run(c, &[usage(model, P, 0, 0, 0)]);
2797 assert_eq!(pings, capped, "{model}");
2798 if capped < uncapped {
2799 assert_eq!(reason, PeriodEnd::HoursReached(24), "{model}");
2800 }
2801 }
2802 for (model, want) in [
2804 ("claude-opus-4-7", 12),
2805 ("claude-opus-5-5", 24),
2806 ("claude-fable-5-1", 49),
2807 ] {
2808 let mut c = cfg(CacheKeepWarm::Auto, CacheTtl::FiveMinutes, model);
2809 c.auto_max_hours = 24;
2810 let (pings, ..) = run(c, &[usage(model, P, 0, 0, 0)]);
2811 assert_eq!(pings, want, "{model} 5m");
2812 }
2813 let mut c = cfg(
2815 CacheKeepWarm::Hours(48),
2816 CacheTtl::OneHour,
2817 "claude-fable-5-1",
2818 );
2819 c.auto_max_hours = 24;
2820 let (pings, ..) = run(c, &[usage("claude-fable-5-1", P, 0, 0, 0)]);
2821 assert_eq!(pings, 52, "48h window: 52 pings at 55 min");
2822 }
2823
2824 #[test]
2827 fn first_write_counts_only_what_the_prefix_will_hold() {
2828 let t0 = Instant::now();
2829 let w0 = SystemTime::UNIX_EPOCH + Duration::from_secs(82_000);
2830 let last = usage(OPUS, 250_000, 0, 1_000, 40_000);
2834 let mut c = cfg(CacheKeepWarm::Auto, CacheTtl::OneHour, OPUS);
2835 let billed = start(c.clone(), &last, t0, w0);
2836 assert_eq!(billed.prefix_tokens(), 291_000, "without the estimate");
2837 c.reply_tokens = Some(2_000);
2838 let p = start(c.clone(), &last, t0, w0);
2839 assert_eq!(p.prefix_tokens(), 253_000);
2840 c.reply_tokens = Some(90_000);
2842 assert_eq!(start(c, &last, t0, w0).prefix_tokens(), 291_000);
2843 }
2844
2845 #[test]
2846 fn status_renders_spend_and_period() {
2847 let s = serde_json::json!({
2848 "mode": "auto",
2849 "active": true,
2850 "period": {"id": 3, "ttl_secs": 300, "pings": 2, "ledger_usd": 0.3,
2851 "bound_usd": 1.875, "prefix_tokens": P, "next_ping_in_secs": 120,
2852 "in_flight": false},
2853 "prefix_matches": true,
2854 "session_pings": 5,
2855 "session_keepwarm_usd": 0.75,
2856 "last_end": "prefix changed",
2857 });
2858 let text = render_status(&s);
2859 assert!(text.contains("auto"), "{text}");
2860 assert!(text.contains("5 pings, $0.7500"), "{text}");
2861 assert!(text.contains("ping 3 due in 120s"), "{text}");
2862 assert!(text.contains("$0.3000 spent of $1.8750 bound"), "{text}");
2863 assert!(text.contains("prefix unchanged"), "{text}");
2864 assert!(text.contains("last: prefix changed"), "{text}");
2865 assert!(!text.contains("prompt-cache TTL"), "absent field, no line: {text}");
2866 }
2867
2868 #[test]
2869 fn status_renders_the_resolved_cache_ttl() {
2870 let mut s = serde_json::json!({
2871 "mode": "auto",
2872 "active": false,
2873 "period": null,
2874 "session_pings": 0,
2875 "session_keepwarm_usd": 0.0,
2876 "cache_ttl": {"setting": "auto", "resolved": "1h"},
2877 });
2878 let text = render_status(&s);
2879 assert!(text.contains("\nprompt-cache TTL: 1h (auto)\n"), "{text}");
2880 s["cache_ttl"] = serde_json::json!({"setting": "5m", "resolved": "5m"});
2881 let text = render_status(&s);
2882 assert!(text.contains("\nprompt-cache TTL: 5m\n"), "{text}");
2883 assert!(!text.contains("(auto)"), "{text}");
2884 }
2885}