1use std::{
8 collections::{BTreeMap, HashMap, HashSet},
9 path::Path,
10};
11
12use anyhow::{Context as _, ensure};
13pub use kcode_context_cache_policy::CacheExpectation;
14use kcode_context_cache_policy::{
15 MarkerObservation, MarkerState, PreviousProjection, StaleMarker, classify, decide_markers,
16};
17use kcode_session_log::{
18 EventPosition, Role, Session as DurableSession, SessionStore as DurableSessionStore,
19};
20use serde::{Deserialize, Serialize};
21use serde_json::{Value, json};
22
23pub const FORMAT_VERSION: u32 = 1;
24pub const MAX_OBJECT_BYTES: u64 = 32 * 1024 * 1024 * 1024;
25pub const ESTIMATED_BYTES_PER_TOKEN: u64 = 4;
26
27const HISTORY_INGRESS_ATTEMPT_RESET_NOTE: &str = "history_ingress_attempt_reset";
28const INGRESS_TIME_MARKER_NOTE: &str = "ingress_time_marker";
29
30#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
32pub struct ProviderTokenUsage {
33 pub input_tokens: u64,
34 pub cached_input_tokens: u64,
35 pub thinking_tokens: u64,
36 pub output_tokens: u64,
37}
38
39#[derive(Clone, Copy, Debug, PartialEq)]
41pub enum ProviderMetering {
42 Tokens(ProviderTokenUsage),
43 DurationSeconds { seconds: f64 },
44 Unavailable,
45}
46
47#[derive(Clone, Debug, Eq, PartialEq)]
49pub struct ProviderCostEstimate {
50 pub usd_nanos: u64,
51 pub accuracy: Value,
52 pub pricing_version: String,
53}
54
55pub type ProviderCostEstimator = fn(&str, &ProviderMetering) -> Option<ProviderCostEstimate>;
57
58#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
60pub struct ProviderCostSummary {
61 pub estimated_cost_usd_nanos: u64,
62 pub unpriced_provider_calls: u64,
63}
64
65pub(crate) fn legacy_provider_cost_summary_for_archive(
70 archive: &Value,
71 default_provider_model: Option<&str>,
72 estimator: ProviderCostEstimator,
73) -> anyhow::Result<ProviderCostSummary> {
74 let metadata = archive
75 .get("metadata")
76 .cloned()
77 .context("session archive is missing Chatend metadata")?;
78 let metadata = serde_json::from_value(metadata).context("decoding session archive metadata")?;
79 let log =
80 serde_json::from_value(archive.clone()).context("decoding immutable session archive")?;
81 let chatend = Chatend::replay_with_provider_costs(
82 metadata,
83 &log,
84 default_provider_model,
85 Some(estimator),
86 )?;
87 let status = chatend.projection().status;
88 Ok(ProviderCostSummary {
89 estimated_cost_usd_nanos: status.estimated_cost_usd_nanos,
90 unpriced_provider_calls: status.unpriced_provider_calls,
91 })
92}
93
94#[derive(
95 Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize,
96)]
97#[serde(transparent)]
98pub struct EventId(pub u64);
99
100impl std::fmt::Display for EventId {
101 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102 self.0.fmt(formatter)
103 }
104}
105
106#[derive(
107 Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize,
108)]
109#[serde(transparent)]
110pub struct BoxId(pub u64);
111
112impl std::fmt::Display for BoxId {
113 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
114 self.0.fmt(formatter)
115 }
116}
117
118#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
119#[serde(transparent)]
120pub struct PendingId(String);
121
122impl PendingId {
123 pub fn from_event(id: EventId) -> Self {
124 Self(format!("pending:{}", id.0))
125 }
126
127 pub fn parse(value: impl Into<String>) -> anyhow::Result<Self> {
128 let value = value.into();
129 let number = value
130 .strip_prefix("pending:")
131 .context("pending identity must begin with `pending:`")?
132 .parse::<u64>()
133 .context("pending identity must end in an unsigned integer")?;
134 ensure!(number > 0, "pending identity zero is reserved");
135 Ok(Self(value))
136 }
137
138 pub fn number(&self) -> u64 {
139 self.0["pending:".len()..]
140 .parse()
141 .expect("validated PendingId")
142 }
143}
144
145impl std::fmt::Display for PendingId {
146 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
147 formatter.write_str(&self.0)
148 }
149}
150
151#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
152#[serde(rename_all = "snake_case")]
153pub enum SessionKind {
154 Conversation,
155 Telegram,
156 TelegramGroup,
157 SelfTime,
158 AudioIngress,
159 HistoryIngress,
160 Other(String),
161}
162
163#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
164#[serde(rename_all = "camelCase")]
165pub struct SessionMetadata {
166 pub session_id: String,
167 pub kind: SessionKind,
168 pub created_at: String,
169 pub effective_context_tokens: u64,
170 #[serde(default)]
171 pub channel: Value,
172}
173
174#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
175#[serde(tag = "kind", rename_all = "snake_case")]
176pub enum BoxOwner {
177 User,
178 Kennedy,
179 Controller,
180 System,
181 Tool { tool_instance: String, slot: String },
182}
183
184#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
185#[serde(rename_all = "camelCase")]
186pub struct BoxContent {
187 #[serde(default)]
188 pub text: String,
189 #[serde(default)]
190 pub objects: Vec<String>,
191 #[serde(default)]
192 pub metadata: Value,
193}
194
195impl BoxContent {
196 pub fn text(value: impl Into<String>) -> Self {
197 Self {
198 text: value.into(),
199 ..Self::default()
200 }
201 }
202
203 pub fn use_concise_header(&mut self) {
206 if !self.metadata.is_object() {
207 self.metadata = json!({});
208 }
209 self.metadata["chatendConciseHeader"] = json!(true);
210 }
211
212 fn render(&self) -> String {
213 let mut rendered = self.text.clone();
214 for object in &self.objects {
215 if !rendered.is_empty() && !rendered.ends_with('\n') {
216 rendered.push('\n');
217 }
218 rendered.push_str("Object provided: ");
219 rendered.push_str(object);
220 }
221 rendered
222 }
223}
224
225#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
226#[serde(tag = "kind", rename_all = "snake_case")]
227pub enum Representation {
228 Hydrated { canonical_event: EventId },
229 Dehydrated { based_on: EventId },
230 Summarized { based_on: EventId, text: String },
231}
232
233#[derive(Clone, Debug, Eq, PartialEq)]
234pub enum BoxRepresentation {
235 Hydrated,
236 Dehydrated,
237 Summarized(String),
238}
239
240impl Representation {
241 fn based_on(&self) -> EventId {
242 match self {
243 Self::Hydrated { canonical_event } => *canonical_event,
244 Self::Dehydrated { based_on } | Self::Summarized { based_on, .. } => *based_on,
245 }
246 }
247}
248
249#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
250#[serde(rename_all = "camelCase")]
251pub struct CanonicalRevision {
252 pub event_id: EventId,
253 pub content: BoxContent,
254}
255
256#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
257#[serde(rename_all = "camelCase")]
258pub struct BoxState {
259 pub id: BoxId,
260 pub name: String,
261 pub owner: BoxOwner,
262 pub created_at: EventId,
263 pub canonical: CanonicalRevision,
264 pub representation: Representation,
265 pub occurrence_events: Vec<EventId>,
266 pub active: bool,
267}
268
269impl BoxState {
270 pub fn stale(&self) -> bool {
271 self.representation.based_on() != self.canonical.event_id
272 }
273}
274
275#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
276#[serde(rename_all = "snake_case")]
277pub enum PendingKind {
278 Node,
279 Object,
280}
281
282#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
283#[serde(tag = "type", rename_all = "snake_case")]
284pub enum EventKind {
285 SessionConfigured {
286 effective_context_tokens: u64,
287 kind: SessionKind,
288 },
289 BoxCreated {
290 box_id: BoxId,
291 name: String,
292 owner: BoxOwner,
293 content: BoxContent,
294 },
295 CanonicalUpdated {
296 box_id: BoxId,
297 content: BoxContent,
298 },
299 CanonicalAdvanced {
300 box_id: BoxId,
301 content: BoxContent,
302 },
303 BoxRenamed {
304 box_id: BoxId,
305 name: String,
306 },
307 BoxDehydrated {
308 box_id: BoxId,
309 },
310 BoxSummarized {
311 box_id: BoxId,
312 text: String,
313 },
314 BoxRehydrated {
315 box_id: BoxId,
316 },
317 BoxRetired {
318 box_id: BoxId,
319 },
320 PendingAllocated {
321 pending_id: PendingId,
322 resource: PendingKind,
323 },
324 ToolInvoked {
325 tool_instance: String,
326 tool_name: String,
327 arguments: Value,
328 #[serde(default, skip_serializing_if = "Option::is_none")]
329 invocation_id: Option<String>,
330 },
331 ToolCompleted {
332 tool_instance: String,
333 tool_name: String,
334 outcome: Value,
335 #[serde(default, skip_serializing_if = "Option::is_none")]
336 invocation_id: Option<String>,
337 },
338 ToolLayoutChanged {
339 tool_instance: String,
340 box_ids: Vec<BoxId>,
341 },
342 InferenceSubmitted {
343 manifest_hash: String,
344 estimated_input_tokens: u64,
345 #[serde(default)]
346 raw_estimated_input_tokens: Option<u64>,
347 },
348 ProviderInputSubmitted {
349 round: u64,
350 context: ProviderContext,
351 #[serde(default, skip_serializing_if = "Option::is_none")]
352 transport_input_hash: Option<String>,
353 #[serde(default, skip_serializing_if = "Option::is_none")]
354 transport_input_bytes: Option<u64>,
355 #[serde(default, skip_serializing_if = "Option::is_none")]
356 thread_action: Option<String>,
357 #[serde(default, skip_serializing_if = "Option::is_none")]
358 thread_reset_reason: Option<String>,
359 cacheable_prefix_bytes: u64,
360 material_fingerprint: String,
361 cache_expectation: String,
362 #[serde(default, skip_serializing_if = "Option::is_none")]
363 planned_invalidation_reason: Option<String>,
364 },
365 ProjectionMarkersReset,
366 StaleBoxesMarked {
367 box_ids: Vec<BoxId>,
368 consolidated: bool,
369 },
370 ContextSizeMarked {
371 estimated_tokens: u64,
372 },
373 ProviderReceipt {
374 manifest_hash: String,
375 input_tokens: Option<u64>,
376 output_tokens: Option<u64>,
377 #[serde(default)]
378 context_bytes: Option<u64>,
379 #[serde(default)]
380 raw_context_tokens: Option<u64>,
381 provider_data: Value,
382 },
383 CapacityError {
384 attempted_operation: String,
385 projected_tokens: u64,
386 limit_tokens: u64,
387 },
388 SourceTerminated {
389 reason: String,
390 },
391 HistoryIngressStarted,
392 HistoryEventInspected {
393 source_event: EventId,
394 },
395 HistoryEventReleased {
396 source_event: EventId,
397 },
398 KwebPlanChanged {
399 operation: Value,
400 },
401 KwebCommitted {
402 transaction_id: String,
403 session_object_id: String,
404 mappings: Value,
405 },
406 SessionCompleted {
407 session_object_id: String,
408 },
409 Note {
410 label: String,
411 value: Value,
412 },
413}
414
415#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
416#[serde(rename_all = "camelCase")]
417pub struct Event {
418 pub id: EventId,
419 pub recorded_at: String,
420 pub kind: EventKind,
421}
422
423#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
424#[serde(rename_all = "camelCase")]
425pub struct Transition {
426 pub recorded_at: String,
427 pub events: Vec<Event>,
428}
429
430#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
431#[serde(rename_all = "camelCase")]
432struct PersistedContextEvent {
433 context_event_version: u32,
434 recorded_at: String,
435 kind: EventKind,
436}
437
438#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
439#[serde(rename_all = "camelCase")]
440struct PersistedContextEventWire {
441 context_event_version: u32,
442 recorded_at: String,
443 kind: Value,
444}
445
446#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
447#[serde(rename_all = "camelCase")]
448pub struct ObjectMetadata {
449 pub pending_id: PendingId,
450 pub event_id: EventId,
451 pub recorded_at: String,
452 pub media_type: String,
453 pub file_name: Option<String>,
454 #[serde(default)]
455 pub transport: Value,
456}
457
458#[derive(Clone, Debug, Eq, PartialEq)]
459pub struct ObjectLocation {
460 pub metadata: ObjectMetadata,
461 pub payload_offset: u64,
462 pub payload_len: u64,
463}
464
465#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
466#[serde(rename_all = "camelCase")]
467pub struct ToolSlot {
468 pub slot: String,
469 pub box_id: BoxId,
470 pub retired: bool,
471}
472
473#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
474#[serde(rename_all = "camelCase")]
475pub struct ToolState {
476 pub slots: Vec<ToolSlot>,
477}
478
479#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
480#[serde(rename_all = "camelCase")]
481pub struct ProviderToolDefinition {
482 pub name: String,
483 pub description: String,
484 pub input_schema: Value,
485}
486
487#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
488#[serde(rename_all = "camelCase")]
489pub struct ProviderContext {
490 pub input: String,
491 pub provider: String,
492 pub model: String,
493 pub reasoning_effort: String,
494 pub base_instructions: Option<String>,
495 pub developer_instructions: Option<String>,
496 pub tools: Vec<ProviderToolDefinition>,
497}
498
499#[derive(Clone, Debug, Eq, PartialEq)]
500pub struct PreparedProviderProjection {
501 pub projection: ContextProjection,
502 pub provider_input: String,
503 pub thread_reset_reason: Option<String>,
504 pub cacheable_prefix_bytes: u64,
505 pub expectation: CacheExpectation,
506}
507
508#[derive(Clone, Debug, Eq, PartialEq)]
509pub struct PreparedProviderResume {
510 pub marker_lines: Vec<String>,
511 pub thread_reset_reason: Option<String>,
512}
513
514#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
515struct IngressTimeMarkerState {
516 any: bool,
517 at_or_below_thirty_minutes: bool,
518 at_or_below_ten_minutes: bool,
519}
520
521fn should_append_ingress_time_marker(
522 state: IngressTimeMarkerState,
523 remaining_seconds: u64,
524 context_size_marked: bool,
525) -> bool {
526 !state.any
527 || context_size_marked
528 || (remaining_seconds <= 10 * 60 && !state.at_or_below_ten_minutes)
529 || (remaining_seconds <= 20 * 60 && !state.at_or_below_thirty_minutes)
530}
531
532#[derive(Clone, Debug, Eq, PartialEq)]
533pub struct ToolSlotInput {
534 pub slot: String,
535 pub name: String,
536 pub content: BoxContent,
537 pub retired: bool,
538}
539
540#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
541#[serde(rename_all = "camelCase")]
542pub struct Chatend {
543 pub metadata: SessionMetadata,
544 pub next_id: u64,
545 pub events: Vec<Event>,
546 pub boxes: BTreeMap<BoxId, BoxState>,
547 pub pending: BTreeMap<PendingId, PendingKind>,
548 pub tools: BTreeMap<String, ToolState>,
549 #[serde(default)]
550 pub tool_layouts: BTreeMap<String, Vec<BoxId>>,
551 pub source_terminated: bool,
552 pub history_ingress_started: bool,
553 pub completed_session_object: Option<String>,
554}
555
556impl Chatend {
557 fn opened(metadata: SessionMetadata) -> Self {
558 Self {
559 metadata,
560 next_id: 1,
561 events: Vec::new(),
562 boxes: BTreeMap::new(),
563 pending: BTreeMap::new(),
564 tools: BTreeMap::new(),
565 tool_layouts: BTreeMap::new(),
566 source_terminated: false,
567 history_ingress_started: false,
568 completed_session_object: None,
569 }
570 }
571
572 fn restore_history_ingress_baseline(&mut self) -> anyhow::Result<()> {
573 let baseline_end = self
574 .events
575 .iter()
576 .position(|event| matches!(event.kind, EventKind::HistoryIngressStarted))
577 .context("history ingress attempt reset has no initial ingress boundary")?;
578 let retained_events = std::mem::take(&mut self.events);
579 let retained_next_id = self.next_id;
580 *self = Self::opened(self.metadata.clone());
581 for event in &retained_events[..=baseline_end] {
582 self.apply_transition(&Transition {
583 recorded_at: event.recorded_at.clone(),
584 events: vec![event.clone()],
585 })?;
586 }
587 self.events = retained_events;
588 self.next_id = retained_next_id;
589 Ok(())
590 }
591
592 fn current_ingress_attempt_start(&self) -> usize {
593 self.events
594 .iter()
595 .rposition(|event| {
596 matches!(
597 &event.kind,
598 EventKind::Note { label, .. }
599 if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE
600 )
601 })
602 .map_or(0, |index| index + 1)
603 }
604
605 pub fn current_ingress_attempt_events(&self) -> &[Event] {
606 &self.events[self.current_ingress_attempt_start()..]
607 }
608
609 pub(crate) fn replay(
610 metadata: SessionMetadata,
611 log: &kcode_session_log::SessionLog,
612 ) -> anyhow::Result<Self> {
613 Self::replay_with_provider_costs(metadata, log, None, None)
614 }
615
616 pub(crate) fn replay_with_provider_costs(
617 metadata: SessionMetadata,
618 log: &kcode_session_log::SessionLog,
619 default_provider_model: Option<&str>,
620 estimator: Option<ProviderCostEstimator>,
621 ) -> anyhow::Result<Self> {
622 ensure!(
623 metadata.session_id == log.header.session_id,
624 "session metadata and session-log identities differ"
625 );
626 ensure!(
627 metadata.created_at == log.header.created_at,
628 "session metadata and session-log creation times differ"
629 );
630 let mut chatend = Self::opened(metadata);
631 for (position, stored) in log.events.iter().enumerate() {
632 let persisted = decode_context_event(stored)?;
633 let id = EventId(position as u64 + 1);
634 let mut kind = persisted.kind;
635 normalize_derived_identity(&mut kind, id)?;
636 chatend.apply_transition(&Transition {
637 recorded_at: persisted.recorded_at.clone(),
638 events: vec![Event {
639 id,
640 recorded_at: persisted.recorded_at,
641 kind,
642 }],
643 })?;
644 }
645 if let Some(estimator) = estimator {
646 chatend.apply_legacy_provider_costs(default_provider_model, estimator);
647 }
648 Ok(chatend)
649 }
650
651 fn apply_legacy_provider_costs(
652 &mut self,
653 default_provider_model: Option<&str>,
654 estimator: ProviderCostEstimator,
655 ) {
656 let mut active_tool_models = Vec::<ActiveToolModel>::new();
657 let mut subagent_model = None::<String>;
658 for event in &mut self.events {
659 match &mut event.kind {
660 EventKind::ToolInvoked {
661 tool_instance,
662 tool_name,
663 arguments,
664 invocation_id,
665 } => {
666 let model = arguments
667 .get("model")
668 .and_then(Value::as_str)
669 .filter(|model| !model.trim().is_empty())
670 .map(normalize_requested_model);
671 active_tool_models.push(ActiveToolModel {
672 invocation_id: invocation_id.clone(),
673 tool_instance: tool_instance.clone(),
674 tool_name: tool_name.clone(),
675 model,
676 });
677 }
678 EventKind::ToolCompleted {
679 tool_instance,
680 tool_name,
681 invocation_id,
682 ..
683 } => {
684 if let Some(index) = active_tool_models.iter().rposition(|active| {
685 if let Some(invocation_id) = invocation_id {
686 active.invocation_id.as_ref() == Some(invocation_id)
687 } else {
688 active.tool_instance == *tool_instance && active.tool_name == *tool_name
689 }
690 }) {
691 active_tool_models.remove(index);
692 }
693 if tool_name == "RunSubagent" {
694 subagent_model = None;
695 }
696 }
697 EventKind::Note { label, value } if label == "subagent_started" => {
698 subagent_model = value
699 .get("providerModel")
700 .or_else(|| value.get("provider_model"))
701 .and_then(Value::as_str)
702 .filter(|model| !model.trim().is_empty())
703 .map(str::to_owned);
704 }
705 EventKind::ProviderReceipt { provider_data, .. }
706 if provider_data
707 .get("estimatedCostUsdNanos")
708 .and_then(Value::as_u64)
709 .is_none() =>
710 {
711 let source = provider_data.get("source").and_then(Value::as_str);
712 let model = provider_data
713 .get("providerModel")
714 .or_else(|| provider_data.get("provider_model"))
715 .and_then(Value::as_str)
716 .filter(|model| !model.trim().is_empty())
717 .map(str::to_owned)
718 .or_else(|| {
719 (source == Some("subagent"))
720 .then(|| subagent_model.clone())
721 .flatten()
722 })
723 .or_else(|| {
724 source
725 .is_some()
726 .then(|| {
727 active_tool_models
728 .iter()
729 .rev()
730 .find_map(|active| active.model.clone())
731 })
732 .flatten()
733 })
734 .or_else(|| {
735 source
736 .is_none()
737 .then(|| default_provider_model.map(str::to_owned))
738 .flatten()
739 });
740 let Some(model) = model else {
741 continue;
742 };
743 let metering = provider_metering(provider_data);
744 let Some(cost) = estimator(&model, &metering) else {
745 continue;
746 };
747 let Some(data) = provider_data.as_object_mut() else {
748 continue;
749 };
750 data.entry("providerModel")
751 .or_insert_with(|| Value::String(model));
752 data.insert("estimatedCostUsdNanos".into(), Value::from(cost.usd_nanos));
753 data.insert("costAccuracy".into(), cost.accuracy);
754 data.insert("pricingVersion".into(), Value::String(cost.pricing_version));
755 }
756 _ => {}
757 }
758 }
759 }
760
761 pub fn event(&self, id: EventId) -> Option<&Event> {
762 self.events.iter().find(|event| event.id == id)
763 }
764
765 pub fn box_state(&self, id: BoxId) -> Option<&BoxState> {
766 self.boxes.get(&id)
767 }
768
769 pub fn active_boxes(&self) -> impl Iterator<Item = &BoxState> {
770 self.boxes.values().filter(|state| state.active)
771 }
772
773 fn canonical_content_at(&self, box_id: BoxId, event_id: EventId) -> Option<&BoxContent> {
774 let event = self.event(event_id)?;
775 match &event.kind {
776 EventKind::BoxCreated {
777 box_id: event_box,
778 content,
779 ..
780 }
781 | EventKind::CanonicalUpdated {
782 box_id: event_box,
783 content,
784 }
785 | EventKind::CanonicalAdvanced {
786 box_id: event_box,
787 content,
788 } if *event_box == box_id => Some(content),
789 _ => None,
790 }
791 }
792
793 pub fn live_context_limit(&self) -> u64 {
794 self.metadata.effective_context_tokens.saturating_mul(70) / 100
795 }
796
797 pub fn forced_ingress_context_limit(&self) -> u64 {
798 self.metadata.effective_context_tokens.saturating_mul(75) / 100
799 }
800
801 pub fn ingress_initial_context_limit(&self) -> u64 {
802 self.metadata.effective_context_tokens.saturating_mul(75) / 100
803 }
804
805 pub fn ingress_context_limit(&self) -> u64 {
806 self.metadata.effective_context_tokens.saturating_mul(90) / 100
807 }
808
809 pub fn active_context_limit(&self) -> u64 {
810 if matches!(
811 self.metadata.kind,
812 SessionKind::SelfTime | SessionKind::HistoryIngress | SessionKind::AudioIngress
813 ) {
814 self.ingress_context_limit()
815 } else {
816 self.live_context_limit()
817 }
818 }
819
820 pub fn projection_with_new_boxes(
821 &self,
822 boxes: &[(String, BoxOwner, BoxContent)],
823 ) -> anyhow::Result<ContextProjection> {
824 self.projection_with_new_boxes_at("preview", boxes)
825 }
826
827 pub fn projection_with_new_boxes_at(
828 &self,
829 recorded_at: &str,
830 boxes: &[(String, BoxOwner, BoxContent)],
831 ) -> anyhow::Result<ContextProjection> {
832 self.projection_with_new_boxes_and_updates_at(recorded_at, boxes, &BTreeMap::new())
833 }
834
835 pub fn projection_with_new_boxes_and_updates(
836 &self,
837 boxes: &[(String, BoxOwner, BoxContent)],
838 updates: &BTreeMap<BoxId, BoxContent>,
839 ) -> anyhow::Result<ContextProjection> {
840 self.projection_with_new_boxes_and_updates_at("preview", boxes, updates)
841 }
842
843 pub fn projection_with_new_boxes_and_updates_at(
844 &self,
845 recorded_at: &str,
846 boxes: &[(String, BoxOwner, BoxContent)],
847 updates: &BTreeMap<BoxId, BoxContent>,
848 ) -> anyhow::Result<ContextProjection> {
849 let mut preview = self.clone();
850 let mut next = preview.next_id;
851 let mut events = Vec::with_capacity(boxes.len() + updates.len());
852 for (name, owner, content) in boxes {
853 let id = EventId(next);
854 let box_id = BoxId(next);
855 next = next.checked_add(1).context("event identity overflow")?;
856 events.push(Event {
857 id,
858 recorded_at: recorded_at.into(),
859 kind: EventKind::BoxCreated {
860 box_id,
861 name: name.clone(),
862 owner: owner.clone(),
863 content: content.clone(),
864 },
865 });
866 }
867 for (box_id, content) in updates {
868 let state = preview
869 .box_state(*box_id)
870 .with_context(|| format!("box {box_id} does not exist"))?;
871 ensure!(state.active, "box {box_id} is retired");
872 if state.canonical.content == *content {
873 continue;
874 }
875 let id = EventId(next);
876 next = next.checked_add(1).context("event identity overflow")?;
877 events.push(Event {
878 id,
879 recorded_at: recorded_at.into(),
880 kind: EventKind::CanonicalAdvanced {
881 box_id: *box_id,
882 content: content.clone(),
883 },
884 });
885 }
886 if !events.is_empty() {
887 preview.apply_transition(&Transition {
888 recorded_at: recorded_at.into(),
889 events,
890 })?;
891 }
892 Ok(preview.projection())
893 }
894
895 pub fn projection_with_box_representations(
896 &self,
897 desired: &BTreeMap<BoxId, BoxRepresentation>,
898 ) -> anyhow::Result<ContextProjection> {
899 let mut preview = self.clone();
900 let events = preview.box_representation_events("preview", desired)?;
901 if !events.is_empty() {
902 preview.apply_transition(&Transition {
903 recorded_at: "preview".into(),
904 events,
905 })?;
906 }
907 Ok(preview.projection())
908 }
909
910 fn box_representation_events(
911 &self,
912 recorded_at: &str,
913 desired: &BTreeMap<BoxId, BoxRepresentation>,
914 ) -> anyhow::Result<Vec<Event>> {
915 let mut next = self.next_id;
916 let mut events = Vec::new();
917 for (box_id, desired) in desired {
918 let state = self
919 .box_state(*box_id)
920 .with_context(|| format!("box {box_id} does not exist"))?;
921 ensure!(state.active, "box {box_id} is retired");
922 let kind = match (desired, &state.representation) {
923 (BoxRepresentation::Hydrated, Representation::Hydrated { canonical_event })
924 if *canonical_event == state.canonical.event_id =>
925 {
926 None
927 }
928 (BoxRepresentation::Dehydrated, Representation::Dehydrated { .. }) => None,
929 (
930 BoxRepresentation::Summarized(desired),
931 Representation::Summarized { text, .. },
932 ) if desired == text => None,
933 (BoxRepresentation::Hydrated, _) => {
934 Some(EventKind::BoxRehydrated { box_id: *box_id })
935 }
936 (BoxRepresentation::Dehydrated, _) => {
937 Some(EventKind::BoxDehydrated { box_id: *box_id })
938 }
939 (BoxRepresentation::Summarized(text), _) => Some(EventKind::BoxSummarized {
940 box_id: *box_id,
941 text: text.clone(),
942 }),
943 };
944 let Some(kind) = kind else {
945 continue;
946 };
947 events.push(Event {
948 id: EventId(next),
949 recorded_at: recorded_at.into(),
950 kind,
951 });
952 next = next.checked_add(1).context("event ID overflow")?;
953 }
954 Ok(events)
955 }
956
957 fn latest_provider_input(&self) -> Option<(EventId, &ProviderContext, u64, &str)> {
958 self.current_ingress_attempt_events()
959 .iter()
960 .rev()
961 .find_map(|event| {
962 let EventKind::ProviderInputSubmitted {
963 context,
964 cacheable_prefix_bytes,
965 material_fingerprint,
966 ..
967 } = &event.kind
968 else {
969 return None;
970 };
971 Some((
972 event.id,
973 context,
974 *cacheable_prefix_bytes,
975 material_fingerprint.as_str(),
976 ))
977 })
978 }
979
980 fn ingress_time_marker_state(&self) -> IngressTimeMarkerState {
981 let mut state = IngressTimeMarkerState::default();
982 for event in self.current_ingress_attempt_events() {
983 let EventKind::Note { label, value } = &event.kind else {
984 continue;
985 };
986 if label != INGRESS_TIME_MARKER_NOTE {
987 continue;
988 }
989 let Some(remaining_seconds) = value.get("remainingSeconds").and_then(Value::as_u64)
990 else {
991 continue;
992 };
993 state.any = true;
994 state.at_or_below_thirty_minutes |= remaining_seconds <= 30 * 60;
995 state.at_or_below_ten_minutes |= remaining_seconds <= 10 * 60;
996 }
997 state
998 }
999
1000 fn marker_state(&self) -> MarkerState {
1001 let reset = self.events.iter().rposition(|event| {
1002 matches!(event.kind, EventKind::ProjectionMarkersReset)
1003 || matches!(
1004 &event.kind,
1005 EventKind::Note { label, .. }
1006 if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE
1007 )
1008 });
1009 let mut state = MarkerState::default();
1010 for event in &self.events[reset.unwrap_or(0)..] {
1011 match &event.kind {
1012 EventKind::StaleBoxesMarked {
1013 box_ids,
1014 consolidated,
1015 } => {
1016 if *consolidated {
1017 state.reported_stale_boxes.clear();
1018 }
1019 state
1020 .reported_stale_boxes
1021 .extend(box_ids.iter().map(|id| id.0));
1022 }
1023 EventKind::ContextSizeMarked { estimated_tokens } => {
1024 state.last_context_size_tokens = Some(*estimated_tokens);
1025 }
1026 _ => {}
1027 }
1028 }
1029 state
1030 }
1031
1032 fn projection_rewrite_reason_after(&self, event_id: Option<EventId>) -> String {
1033 let mut reasons = std::collections::BTreeSet::new();
1034 for event in self
1035 .events
1036 .iter()
1037 .filter(|event| event_id.is_none_or(|id| event.id > id))
1038 {
1039 let reason = match event.kind {
1040 EventKind::BoxDehydrated { .. } => Some("dehydration"),
1041 EventKind::BoxRehydrated { .. } => Some("rehydration"),
1042 EventKind::BoxSummarized { .. } => Some("summarization"),
1043 EventKind::CanonicalUpdated { .. }
1044 | EventKind::BoxRenamed { .. }
1045 | EventKind::BoxRetired { .. } => Some("canonical_state_replacement"),
1046 EventKind::ToolLayoutChanged { .. } => Some("projection_reordered"),
1047 EventKind::SessionConfigured { .. } => Some("context_limit_changed"),
1048 EventKind::Note { ref label, .. }
1049 if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE =>
1050 {
1051 Some("ingress_attempt_reset")
1052 }
1053 _ => None,
1054 };
1055 if let Some(reason) = reason {
1056 reasons.insert(reason);
1057 }
1058 }
1059 reasons.into_iter().collect::<Vec<_>>().join("+")
1060 }
1061
1062 pub fn projection(&self) -> ContextProjection {
1063 self.projection_with_footer_lines(&[])
1064 }
1065
1066 pub fn projection_with_footer_lines(&self, footer_lines: &[String]) -> ContextProjection {
1067 self.projection_at(footer_lines)
1068 }
1069
1070 fn projection_at(&self, footer_lines: &[String]) -> ContextProjection {
1071 let items = self.projection_items(false);
1072 let stale_boxes = self
1073 .active_boxes()
1074 .filter(|state| state.stale())
1075 .map(|state| state.id)
1076 .collect::<Vec<_>>();
1077 let footer = footer_lines.join("\n");
1078 let context_bytes = projected_context_bytes(&items, &footer);
1079 let raw_estimated_tokens = estimate_bytes(context_bytes);
1080 let estimated_tokens = self.calibrated_estimate(context_bytes, raw_estimated_tokens);
1081 let fully_hydrated_context_tokens = self.fully_hydrated_context_tokens_at(footer_lines);
1082 let usage = self.cumulative_token_usage();
1083 let cost = self.cumulative_cost();
1084 let status = SessionStatus {
1085 current_context_tokens: estimated_tokens,
1086 fully_hydrated_context_tokens,
1087 context_limit_tokens: self.active_context_limit(),
1088 current_context_bytes: context_bytes,
1089 cached_input_tokens: usage.cached_input_tokens,
1090 non_cached_input_tokens: usage.non_cached_input_tokens,
1091 thinking_tokens: usage.thinking_tokens,
1092 output_tokens: usage.output_tokens,
1093 estimated_cost_usd_nanos: cost.estimated_cost_usd_nanos,
1094 unpriced_provider_calls: cost.unpriced_provider_calls,
1095 };
1096 ContextProjection {
1097 items,
1098 stale_boxes,
1099 footer,
1100 estimated_tokens,
1101 raw_estimated_tokens,
1102 context_bytes,
1103 status,
1104 }
1105 }
1106
1107 fn projection_items(&self, fully_hydrated: bool) -> Vec<ProjectionItem> {
1108 let mut next_occurrence = HashMap::new();
1109 for state in self.boxes.values() {
1110 for pair in state.occurrence_events.windows(2) {
1111 next_occurrence.insert(pair[0], pair[1]);
1112 }
1113 }
1114 let latest_marker_reset = self.events.iter().rev().find_map(|event| {
1115 (matches!(event.kind, EventKind::ProjectionMarkersReset)
1116 || matches!(
1117 &event.kind,
1118 EventKind::Note { label, .. }
1119 if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE
1120 ))
1121 .then_some(event.id)
1122 });
1123 let current_ingress_attempt_start = self.current_ingress_attempt_start();
1124 let mut items = Vec::new();
1125 for (index, event) in self.events.iter().enumerate() {
1126 if index >= current_ingress_attempt_start
1127 && let EventKind::Note { label, value } = &event.kind
1128 && label == INGRESS_TIME_MARKER_NOTE
1129 && let Some(text) = ingress_time_marker_text(value)
1130 {
1131 items.push(ProjectionItem::projection_marker(event.id, text));
1132 continue;
1133 }
1134 if latest_marker_reset.is_none_or(|reset| event.id >= reset) {
1135 match &event.kind {
1136 EventKind::StaleBoxesMarked {
1137 box_ids,
1138 consolidated,
1139 } => {
1140 let marker = if *consolidated {
1141 StaleMarker::Consolidated(box_ids.iter().map(|id| id.0).collect())
1142 } else {
1143 StaleMarker::New(box_ids.iter().map(|id| id.0).collect())
1144 };
1145 items.push(ProjectionItem::projection_marker(event.id, marker.render()));
1146 continue;
1147 }
1148 EventKind::ContextSizeMarked { estimated_tokens } => {
1149 items.push(ProjectionItem::projection_marker(
1150 event.id,
1151 context_size_marker_text(
1152 *estimated_tokens,
1153 self.active_context_limit(),
1154 ),
1155 ));
1156 continue;
1157 }
1158 _ => {}
1159 }
1160 }
1161 let Some(box_id) = event_box_id(&event.kind) else {
1162 continue;
1163 };
1164 let Some(state) = self.boxes.get(&box_id) else {
1165 continue;
1166 };
1167 if next_occurrence.contains_key(&event.id) {
1168 items.push(ProjectionItem::marker(
1169 event.id,
1170 box_id,
1171 "[box updated]".into(),
1172 ));
1173 continue;
1174 }
1175 if !state.active || state.occurrence_events.last() != Some(&event.id) {
1176 continue;
1177 }
1178 let (representation, body) = if fully_hydrated {
1179 ("hydrated", state.canonical.content.render())
1180 } else {
1181 match &state.representation {
1182 Representation::Hydrated { canonical_event } => {
1183 let content = self
1184 .canonical_content_at(box_id, *canonical_event)
1185 .expect("hydrated representation references a canonical event");
1186 ("hydrated", content.render())
1187 }
1188 Representation::Dehydrated { .. } => (
1189 "dehydrated",
1190 format!(
1191 "[contents dehydrated; hydrate box {} to inspect the latest canonical revision]",
1192 box_id
1193 ),
1194 ),
1195 Representation::Summarized { text, .. } => ("summarized", text.clone()),
1196 }
1197 };
1198 let stale = !fully_hydrated && state.stale();
1199 let mut header = vec![format!("box {box_id}"), state.name.clone()];
1200 if matches!(
1201 (&state.owner, state.name.as_str()),
1202 (BoxOwner::User, "User message") | (BoxOwner::Kennedy, "Kennedy message")
1203 ) {
1204 header.push(format!("timestamp={}", event.recorded_at));
1205 }
1206 header.push(representation.into());
1207 let mut text = format!("[{}]\n{}", header.join(" | "), body);
1208 if text.ends_with('\n') {
1209 text.pop();
1210 }
1211 items.push(ProjectionItem {
1212 event_id: event.id,
1213 box_id,
1214 marker: false,
1215 stale,
1216 approximate_tokens: estimate_tokens(&text),
1217 text,
1218 });
1219 }
1220 for box_ids in self.tool_layouts.values() {
1221 arrange_tool_projection(&mut items, box_ids);
1222 }
1223 items
1224 }
1225
1226 fn fully_hydrated_context_tokens_at(&self, footer_lines: &[String]) -> u64 {
1227 let items = self.projection_items(true);
1228 let footer = footer_lines.join("\n");
1229 let context_bytes = projected_context_bytes(&items, &footer);
1230 self.calibrated_estimate(context_bytes, estimate_bytes(context_bytes))
1231 }
1232
1233 fn calibrated_estimate(&self, current_bytes: u64, raw_current: u64) -> u64 {
1234 let Some((manifest_hash, measured, bytes_at_receipt, raw_at_receipt)) =
1235 self.events.iter().rev().find_map(|event| {
1236 let EventKind::ProviderReceipt {
1237 manifest_hash,
1238 input_tokens: Some(input_tokens),
1239 context_bytes,
1240 raw_context_tokens,
1241 ..
1242 } = &event.kind
1243 else {
1244 return None;
1245 };
1246 Some((
1247 manifest_hash,
1248 *input_tokens,
1249 *context_bytes,
1250 *raw_context_tokens,
1251 ))
1252 })
1253 else {
1254 return raw_current;
1255 };
1256 if let Some(bytes_at_receipt) = bytes_at_receipt {
1257 let token_delta = current_bytes.abs_diff(bytes_at_receipt) / ESTIMATED_BYTES_PER_TOKEN;
1258 return if current_bytes >= bytes_at_receipt {
1259 measured.saturating_add(token_delta)
1260 } else {
1261 measured.saturating_sub(token_delta)
1262 };
1263 }
1264 let raw_at_measurement = match raw_at_receipt {
1265 Some(raw) => raw,
1266 None => {
1267 let Some(raw) = self.events.iter().rev().find_map(|event| {
1268 let EventKind::InferenceSubmitted {
1269 manifest_hash: submitted,
1270 estimated_input_tokens,
1271 raw_estimated_input_tokens,
1272 } = &event.kind
1273 else {
1274 return None;
1275 };
1276 (submitted == manifest_hash)
1277 .then_some(raw_estimated_input_tokens.unwrap_or(*estimated_input_tokens))
1278 }) else {
1279 return raw_current;
1280 };
1281 raw
1282 }
1283 };
1284 if raw_current >= raw_at_measurement {
1285 measured.saturating_add(raw_current - raw_at_measurement)
1286 } else {
1287 measured.saturating_sub(raw_at_measurement - raw_current)
1288 }
1289 }
1290
1291 fn cumulative_token_usage(&self) -> CumulativeTokenUsage {
1292 let mut total = CumulativeTokenUsage::default();
1293 for event in &self.events {
1294 let EventKind::ProviderReceipt { provider_data, .. } = &event.kind else {
1295 continue;
1296 };
1297 let cached = provider_u64(provider_data, &["cachedInputTokens", "cached_input_tokens"]);
1298 let thinking = provider_u64(
1299 provider_data,
1300 &[
1301 "thinkingTokens",
1302 "thinking_tokens",
1303 "reasoningOutputTokens",
1304 "reasoning_output_tokens",
1305 ],
1306 );
1307 let normalized_delta = provider_data
1308 .get("usageIsDelta")
1309 .and_then(Value::as_bool)
1310 .unwrap_or(false);
1311 let non_cached = if normalized_delta {
1312 provider_u64(
1313 provider_data,
1314 &["nonCachedInputTokens", "non_cached_input_tokens"],
1315 )
1316 } else {
1317 provider_u64(provider_data, &["inputTokens", "input_tokens"]).saturating_sub(cached)
1318 };
1319 let output = if normalized_delta {
1320 provider_u64(provider_data, &["outputTokens", "output_tokens"])
1321 } else {
1322 provider_u64(provider_data, &["outputTokens", "output_tokens"])
1323 .saturating_sub(thinking)
1324 };
1325 total.cached_input_tokens = total.cached_input_tokens.saturating_add(cached);
1326 total.non_cached_input_tokens =
1327 total.non_cached_input_tokens.saturating_add(non_cached);
1328 total.thinking_tokens = total.thinking_tokens.saturating_add(thinking);
1329 total.output_tokens = total.output_tokens.saturating_add(output);
1330 }
1331 total
1332 }
1333
1334 fn cumulative_cost(&self) -> CumulativeCost {
1335 let mut total = CumulativeCost::default();
1336 for event in &self.events {
1337 let EventKind::ProviderReceipt { provider_data, .. } = &event.kind else {
1338 continue;
1339 };
1340 if let Some(cost) = provider_data
1341 .get("estimatedCostUsdNanos")
1342 .and_then(Value::as_u64)
1343 {
1344 total.estimated_cost_usd_nanos =
1345 total.estimated_cost_usd_nanos.saturating_add(cost);
1346 } else {
1347 total.unpriced_provider_calls = total.unpriced_provider_calls.saturating_add(1);
1348 }
1349 }
1350 total
1351 }
1352
1353 pub fn render(&self) -> String {
1354 self.projection().render()
1355 }
1356
1357 fn apply_transition(&mut self, transition: &Transition) -> anyhow::Result<()> {
1358 ensure!(
1359 !transition.events.is_empty(),
1360 "a transition cannot be empty"
1361 );
1362 for event in &transition.events {
1363 self.apply_event(event)?;
1364 }
1365 Ok(())
1366 }
1367
1368 fn apply_event(&mut self, event: &Event) -> anyhow::Result<()> {
1369 ensure!(
1370 event.id.0 >= self.next_id,
1371 "event {} reuses an allocated identity (next is {})",
1372 event.id,
1373 self.next_id
1374 );
1375 self.next_id = event.id.0.checked_add(1).context("event ID overflow")?;
1376 match &event.kind {
1377 EventKind::SessionConfigured {
1378 effective_context_tokens,
1379 kind,
1380 } => {
1381 ensure!(
1382 *effective_context_tokens > 0,
1383 "effective context window must be positive"
1384 );
1385 self.metadata.effective_context_tokens = *effective_context_tokens;
1386 self.metadata.kind = kind.clone();
1387 }
1388 EventKind::BoxCreated {
1389 box_id,
1390 name,
1391 owner,
1392 content,
1393 } => {
1394 ensure!(
1395 box_id.0 == event.id.0,
1396 "BoxId must equal its creation EventId"
1397 );
1398 ensure!(
1399 !self.boxes.contains_key(box_id),
1400 "box {} already exists",
1401 box_id
1402 );
1403 self.boxes.insert(
1404 *box_id,
1405 BoxState {
1406 id: *box_id,
1407 name: name.clone(),
1408 owner: owner.clone(),
1409 created_at: event.id,
1410 canonical: CanonicalRevision {
1411 event_id: event.id,
1412 content: content.clone(),
1413 },
1414 representation: Representation::Hydrated {
1415 canonical_event: event.id,
1416 },
1417 occurrence_events: vec![event.id],
1418 active: true,
1419 },
1420 );
1421 if let BoxOwner::Tool {
1422 tool_instance,
1423 slot,
1424 } = owner
1425 {
1426 self.tools
1427 .entry(tool_instance.clone())
1428 .or_default()
1429 .slots
1430 .push(ToolSlot {
1431 slot: slot.clone(),
1432 box_id: *box_id,
1433 retired: false,
1434 });
1435 }
1436 }
1437 EventKind::CanonicalUpdated { box_id, content } => {
1438 let state = active_box_mut(&mut self.boxes, *box_id)?;
1439 state.canonical = CanonicalRevision {
1440 event_id: event.id,
1441 content: content.clone(),
1442 };
1443 if matches!(state.representation, Representation::Hydrated { .. }) {
1444 state.representation = Representation::Hydrated {
1445 canonical_event: event.id,
1446 };
1447 }
1448 state.occurrence_events.push(event.id);
1449 }
1450 EventKind::CanonicalAdvanced { box_id, content } => {
1451 let state = active_box_mut(&mut self.boxes, *box_id)?;
1452 state.canonical = CanonicalRevision {
1453 event_id: event.id,
1454 content: content.clone(),
1455 };
1456 }
1457 EventKind::BoxRenamed { box_id, name } => {
1458 ensure!(!name.trim().is_empty(), "a box name cannot be empty");
1459 let state = active_box_mut(&mut self.boxes, *box_id)?;
1460 state.name = name.clone();
1461 state.occurrence_events.push(event.id);
1462 }
1463 EventKind::BoxDehydrated { box_id } => {
1464 let state = active_box_mut(&mut self.boxes, *box_id)?;
1465 state.representation = Representation::Dehydrated {
1466 based_on: state.canonical.event_id,
1467 };
1468 state.occurrence_events.push(event.id);
1469 }
1470 EventKind::BoxSummarized { box_id, text } => {
1471 ensure!(!text.trim().is_empty(), "a box summary cannot be empty");
1472 let state = active_box_mut(&mut self.boxes, *box_id)?;
1473 state.representation = Representation::Summarized {
1474 based_on: state.canonical.event_id,
1475 text: text.clone(),
1476 };
1477 state.occurrence_events.push(event.id);
1478 }
1479 EventKind::BoxRehydrated { box_id } => {
1480 let state = active_box_mut(&mut self.boxes, *box_id)?;
1481 state.representation = Representation::Hydrated {
1482 canonical_event: state.canonical.event_id,
1483 };
1484 state.occurrence_events.push(event.id);
1485 }
1486 EventKind::BoxRetired { box_id } => {
1487 let tool_instance = self.boxes.get(box_id).and_then(|state| {
1488 let BoxOwner::Tool { tool_instance, .. } = &state.owner else {
1489 return None;
1490 };
1491 Some(tool_instance.clone())
1492 });
1493 let state = active_box_mut(&mut self.boxes, *box_id)?;
1494 state.active = false;
1495 state.occurrence_events.push(event.id);
1496 if let Some(tool_instance) = tool_instance {
1497 let slot = self
1498 .tools
1499 .get_mut(&tool_instance)
1500 .and_then(|tool| tool.slots.iter_mut().find(|slot| slot.box_id == *box_id))
1501 .with_context(|| {
1502 format!("tool box {box_id} is missing from {tool_instance}")
1503 })?;
1504 slot.retired = true;
1505 }
1506 }
1507 EventKind::ToolLayoutChanged {
1508 tool_instance,
1509 box_ids,
1510 } => {
1511 let mut unique = std::collections::HashSet::new();
1512 for box_id in box_ids {
1513 ensure!(
1514 unique.insert(*box_id),
1515 "tool layout contains duplicate box {box_id}"
1516 );
1517 let state = self
1518 .boxes
1519 .get(box_id)
1520 .with_context(|| format!("tool layout references missing box {box_id}"))?;
1521 ensure!(state.active, "tool layout references retired box {box_id}");
1522 ensure!(
1523 matches!(
1524 &state.owner,
1525 BoxOwner::Tool {
1526 tool_instance: owner,
1527 ..
1528 } if owner == tool_instance
1529 ),
1530 "tool layout box {box_id} belongs to another tool"
1531 );
1532 }
1533 self.tool_layouts
1534 .insert(tool_instance.clone(), box_ids.clone());
1535 }
1536 EventKind::PendingAllocated {
1537 pending_id,
1538 resource,
1539 } => {
1540 ensure!(
1541 pending_id.number() == event.id.0,
1542 "pending identity must equal its allocation EventId"
1543 );
1544 ensure!(
1545 self.pending
1546 .insert(pending_id.clone(), resource.clone())
1547 .is_none(),
1548 "pending identity {} already exists",
1549 pending_id
1550 );
1551 }
1552 EventKind::SourceTerminated { .. } => self.source_terminated = true,
1553 EventKind::HistoryIngressStarted => {
1554 ensure!(
1555 self.source_terminated,
1556 "history ingress requires source termination"
1557 );
1558 self.history_ingress_started = true;
1559 }
1560 EventKind::Note { label, .. } if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE => {
1561 self.restore_history_ingress_baseline()?;
1562 }
1563 EventKind::SessionCompleted { session_object_id } => {
1564 self.completed_session_object = Some(session_object_id.clone());
1565 }
1566 EventKind::ToolInvoked { .. }
1567 | EventKind::ToolCompleted { .. }
1568 | EventKind::InferenceSubmitted { .. }
1569 | EventKind::ProviderInputSubmitted { .. }
1570 | EventKind::ProjectionMarkersReset
1571 | EventKind::StaleBoxesMarked { .. }
1572 | EventKind::ContextSizeMarked { .. }
1573 | EventKind::ProviderReceipt { .. }
1574 | EventKind::CapacityError { .. }
1575 | EventKind::HistoryEventInspected { .. }
1576 | EventKind::HistoryEventReleased { .. }
1577 | EventKind::KwebPlanChanged { .. }
1578 | EventKind::KwebCommitted { .. }
1579 | EventKind::Note { .. } => {}
1580 }
1581 self.events.push(event.clone());
1582 Ok(())
1583 }
1584}
1585
1586fn active_box_mut(
1587 boxes: &mut BTreeMap<BoxId, BoxState>,
1588 box_id: BoxId,
1589) -> anyhow::Result<&mut BoxState> {
1590 let state = boxes
1591 .get_mut(&box_id)
1592 .with_context(|| format!("box {box_id} does not exist"))?;
1593 ensure!(state.active, "box {box_id} is retired");
1594 Ok(state)
1595}
1596
1597fn arrange_tool_projection(items: &mut Vec<ProjectionItem>, box_ids: &[BoxId]) {
1598 if box_ids.is_empty() {
1599 return;
1600 }
1601 let ranks = box_ids
1602 .iter()
1603 .enumerate()
1604 .map(|(rank, box_id)| (*box_id, rank))
1605 .collect::<HashMap<_, _>>();
1606 let insertion = items
1607 .iter()
1608 .position(|item| !item.marker && ranks.contains_key(&item.box_id));
1609 let Some(insertion) = insertion else {
1610 return;
1611 };
1612 let mut arranged = Vec::with_capacity(box_ids.len());
1613 let mut retained = Vec::with_capacity(items.len());
1614 for item in std::mem::take(items) {
1615 if !item.marker && ranks.contains_key(&item.box_id) {
1616 arranged.push(item);
1617 } else {
1618 retained.push(item);
1619 }
1620 }
1621 arranged.sort_by_key(|item| ranks[&item.box_id]);
1622 let insertion = insertion.min(retained.len());
1623 retained.splice(insertion..insertion, arranged);
1624 *items = retained;
1625}
1626
1627fn event_box_id(kind: &EventKind) -> Option<BoxId> {
1628 match kind {
1629 EventKind::BoxCreated { box_id, .. }
1630 | EventKind::CanonicalUpdated { box_id, .. }
1631 | EventKind::CanonicalAdvanced { box_id, .. }
1632 | EventKind::BoxRenamed { box_id, .. }
1633 | EventKind::BoxDehydrated { box_id }
1634 | EventKind::BoxSummarized { box_id, .. }
1635 | EventKind::BoxRehydrated { box_id }
1636 | EventKind::BoxRetired { box_id } => Some(*box_id),
1637 _ => None,
1638 }
1639}
1640
1641fn ingress_time_marker_text(value: &Value) -> Option<String> {
1642 let remaining_seconds = value.get("remainingSeconds")?.as_u64()?;
1643 let previous_attempt_timed_out = value
1644 .get("previousAttemptTimedOut")
1645 .and_then(Value::as_bool)
1646 .unwrap_or(false);
1647 let previous = if previous_attempt_timed_out {
1648 "Previous attempt ran out of time; please budget your ingress time carefully. "
1649 } else {
1650 ""
1651 };
1652 let label = if previous_attempt_timed_out {
1653 "Ingress"
1654 } else {
1655 "ingress"
1656 };
1657 let warning = if remaining_seconds <= 10 * 60 {
1658 ". If you do not call EndSession before time expires, the ingress will fail and all Kmap updates will be lost."
1659 } else {
1660 ""
1661 };
1662 Some(format!(
1663 "[{previous}{label} time remaining: {remaining_seconds} seconds{warning}]"
1664 ))
1665}
1666
1667fn context_size_marker_text(estimated_tokens: u64, context_limit: u64) -> String {
1668 format!(
1669 "[current context size: approximately {estimated_tokens} tokens | session limit: {context_limit} tokens]"
1670 )
1671}
1672
1673#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1674#[serde(rename_all = "camelCase")]
1675pub struct ProjectionItem {
1676 pub event_id: EventId,
1677 pub box_id: BoxId,
1678 pub marker: bool,
1679 pub stale: bool,
1680 pub approximate_tokens: u64,
1681 pub text: String,
1682}
1683
1684impl ProjectionItem {
1685 fn marker(event_id: EventId, box_id: BoxId, text: String) -> Self {
1686 Self {
1687 event_id,
1688 box_id,
1689 marker: true,
1690 stale: false,
1691 approximate_tokens: estimate_tokens(&text),
1692 text,
1693 }
1694 }
1695
1696 fn projection_marker(event_id: EventId, text: String) -> Self {
1697 Self::marker(event_id, BoxId(0), text)
1698 }
1699}
1700
1701#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1702#[serde(rename_all = "camelCase")]
1703pub struct ContextProjection {
1704 pub items: Vec<ProjectionItem>,
1705 pub stale_boxes: Vec<BoxId>,
1706 pub footer: String,
1707 pub estimated_tokens: u64,
1708 pub raw_estimated_tokens: u64,
1709 pub context_bytes: u64,
1710 pub status: SessionStatus,
1711}
1712
1713impl ContextProjection {
1714 pub fn render(&self) -> String {
1715 let mut blocks = self
1716 .items
1717 .iter()
1718 .map(|item| item.text.as_str())
1719 .collect::<Vec<_>>();
1720 if !self.footer.is_empty() {
1721 blocks.push(&self.footer);
1722 }
1723 blocks.join("\n\n")
1724 }
1725}
1726
1727#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
1728#[serde(rename_all = "camelCase")]
1729pub struct SessionStatus {
1730 pub current_context_tokens: u64,
1731 #[serde(default)]
1732 pub fully_hydrated_context_tokens: u64,
1733 pub context_limit_tokens: u64,
1734 pub current_context_bytes: u64,
1735 pub cached_input_tokens: u64,
1736 pub non_cached_input_tokens: u64,
1737 pub thinking_tokens: u64,
1738 pub output_tokens: u64,
1739 #[serde(default)]
1740 pub estimated_cost_usd_nanos: u64,
1741 #[serde(default)]
1742 pub unpriced_provider_calls: u64,
1743}
1744
1745#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
1746struct CumulativeTokenUsage {
1747 cached_input_tokens: u64,
1748 non_cached_input_tokens: u64,
1749 thinking_tokens: u64,
1750 output_tokens: u64,
1751}
1752
1753#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
1754struct CumulativeCost {
1755 estimated_cost_usd_nanos: u64,
1756 unpriced_provider_calls: u64,
1757}
1758
1759pub fn estimate_tokens(text: &str) -> u64 {
1760 estimate_bytes(text.len() as u64)
1761}
1762
1763fn estimate_bytes(bytes: u64) -> u64 {
1764 bytes.div_ceil(ESTIMATED_BYTES_PER_TOKEN)
1765}
1766
1767fn projected_context_bytes(items: &[ProjectionItem], footer: &str) -> u64 {
1768 let blocks = items.len() as u64 + u64::from(!footer.is_empty());
1769 items
1770 .iter()
1771 .fold(footer.len() as u64, |total, item| {
1772 total.saturating_add(item.text.len() as u64)
1773 })
1774 .saturating_add(blocks.saturating_sub(1).saturating_mul(2))
1775}
1776
1777fn provider_u64(value: &Value, keys: &[&str]) -> u64 {
1778 keys.iter()
1779 .find_map(|key| value.get(*key).and_then(Value::as_u64))
1780 .unwrap_or_default()
1781}
1782
1783struct ActiveToolModel {
1784 invocation_id: Option<String>,
1785 tool_instance: String,
1786 tool_name: String,
1787 model: Option<String>,
1788}
1789
1790fn normalize_requested_model(model: &str) -> String {
1791 model.strip_prefix("codex/").unwrap_or(model).to_owned()
1792}
1793
1794fn provider_metering(provider_data: &Value) -> ProviderMetering {
1795 let metering = provider_data.get("metering");
1796 if metering
1797 .and_then(Value::as_str)
1798 .is_some_and(|kind| kind == "unavailable")
1799 {
1800 return ProviderMetering::Unavailable;
1801 }
1802 if let Some(metering_value) = metering
1803 && let Some(metering) = metering_value.as_object()
1804 {
1805 match metering.get("kind").and_then(Value::as_str) {
1806 Some("duration_seconds") => {
1807 return metering
1808 .get("seconds")
1809 .and_then(Value::as_f64)
1810 .map(|seconds| ProviderMetering::DurationSeconds { seconds })
1811 .unwrap_or(ProviderMetering::Unavailable);
1812 }
1813 Some("unavailable") => return ProviderMetering::Unavailable,
1814 Some("tokens") => {
1815 return ProviderMetering::Tokens(token_metering(metering_value));
1816 }
1817 _ => {}
1818 }
1819 }
1820 let has_tokens = metering.and_then(Value::as_str) == Some("tokens")
1821 || [
1822 "inputTokens",
1823 "input_tokens",
1824 "nonCachedInputTokens",
1825 "non_cached_input_tokens",
1826 "cachedInputTokens",
1827 "cached_input_tokens",
1828 "thinkingTokens",
1829 "thinking_tokens",
1830 "outputTokens",
1831 "output_tokens",
1832 ]
1833 .iter()
1834 .any(|key| provider_data.get(*key).and_then(Value::as_u64).is_some());
1835 if has_tokens {
1836 ProviderMetering::Tokens(token_metering(provider_data))
1837 } else {
1838 ProviderMetering::Unavailable
1839 }
1840}
1841
1842fn token_metering(value: &Value) -> ProviderTokenUsage {
1843 let cached_input_tokens = provider_u64(value, &["cachedInputTokens", "cached_input_tokens"]);
1844 let thinking_tokens = provider_u64(
1845 value,
1846 &[
1847 "thinkingTokens",
1848 "thinking_tokens",
1849 "reasoningOutputTokens",
1850 "reasoning_output_tokens",
1851 ],
1852 );
1853 let normalized_delta = value
1854 .get("usageIsDelta")
1855 .or_else(|| value.get("usage_is_delta"))
1856 .and_then(Value::as_bool)
1857 .unwrap_or(false);
1858 let input_tokens = if normalized_delta {
1859 provider_u64(value, &["nonCachedInputTokens", "non_cached_input_tokens"])
1860 } else {
1861 provider_u64(value, &["inputTokens", "input_tokens"]).saturating_sub(cached_input_tokens)
1862 };
1863 let output_tokens = if normalized_delta {
1864 provider_u64(value, &["outputTokens", "output_tokens"])
1865 } else {
1866 provider_u64(value, &["outputTokens", "output_tokens"]).saturating_sub(thinking_tokens)
1867 };
1868 ProviderTokenUsage {
1869 input_tokens,
1870 cached_input_tokens,
1871 thinking_tokens,
1872 output_tokens,
1873 }
1874}
1875
1876fn context_event_role(kind: &EventKind) -> Role {
1877 match kind {
1878 EventKind::BoxCreated { owner, content, .. } => match owner {
1879 BoxOwner::System | BoxOwner::Controller => {
1880 if content
1881 .metadata
1882 .get("capacityError")
1883 .and_then(Value::as_bool)
1884 .unwrap_or(false)
1885 {
1886 Role::SystemError
1887 } else {
1888 Role::SystemMessage
1889 }
1890 }
1891 BoxOwner::User => Role::UserMessage,
1892 BoxOwner::Kennedy => Role::KennedyMessage,
1893 BoxOwner::Tool { .. } => Role::ToolResult,
1894 },
1895 EventKind::ToolInvoked { .. } => Role::KennedyToolCall,
1896 EventKind::ToolCompleted { outcome, .. } => {
1897 if outcome.get("ok").and_then(Value::as_bool).unwrap_or(true) {
1898 Role::ToolResult
1899 } else {
1900 Role::ToolError
1901 }
1902 }
1903 EventKind::CapacityError { .. } => Role::SystemError,
1904 EventKind::PendingAllocated {
1905 resource: PendingKind::Object,
1906 ..
1907 } => Role::PendingObject,
1908 EventKind::BoxDehydrated { .. }
1909 | EventKind::BoxSummarized { .. }
1910 | EventKind::BoxRehydrated { .. }
1911 | EventKind::BoxRetired { .. } => Role::KennedyToolCall,
1912 _ => Role::SystemMessage,
1913 }
1914}
1915
1916fn encode_context_event(recorded_at: &str, kind: &EventKind) -> anyhow::Result<String> {
1917 let mut kind = serde_json::to_value(kind)?;
1918 let kind_object = kind
1919 .as_object_mut()
1920 .context("Kennedy context event kind must encode as an object")?;
1921 match kind_object.get("type").and_then(Value::as_str) {
1922 Some("box_created") => {
1923 kind_object.remove("box_id");
1924 }
1925 Some("pending_allocated") => {
1926 kind_object.remove("pending_id");
1927 }
1928 _ => {}
1929 }
1930 Ok(serde_json::to_string(&PersistedContextEventWire {
1931 context_event_version: FORMAT_VERSION,
1932 recorded_at: recorded_at.into(),
1933 kind,
1934 })?)
1935}
1936
1937fn decode_context_event(
1938 event: &kcode_session_log::SessionEvent,
1939) -> anyhow::Result<PersistedContextEvent> {
1940 let wire: PersistedContextEventWire = match serde_json::from_str(&event.text) {
1941 Ok(persisted) => persisted,
1942 Err(_) if event.role == Role::PendingObject => {
1943 return Ok(PersistedContextEvent {
1944 context_event_version: FORMAT_VERSION,
1945 recorded_at: String::new(),
1946 kind: EventKind::PendingAllocated {
1947 pending_id: PendingId::from_event(EventId(1)),
1948 resource: PendingKind::Object,
1949 },
1950 });
1951 }
1952 Err(error) => {
1953 return Err(error).context("decoding Kennedy context event from session log");
1954 }
1955 };
1956 ensure!(
1957 wire.context_event_version == FORMAT_VERSION,
1958 "unsupported Kennedy context event version {}",
1959 wire.context_event_version
1960 );
1961 let mut kind = wire.kind;
1962 let kind_object = kind
1963 .as_object_mut()
1964 .context("Kennedy context event kind must be an object")?;
1965 match kind_object.get("type").and_then(Value::as_str) {
1966 Some("box_created") if !kind_object.contains_key("box_id") => {
1967 kind_object.insert("box_id".into(), Value::from(0));
1968 }
1969 Some("pending_allocated") if !kind_object.contains_key("pending_id") => {
1970 kind_object.insert("pending_id".into(), Value::String("pending:1".into()));
1971 }
1972 _ => {}
1973 }
1974 Ok(PersistedContextEvent {
1975 context_event_version: wire.context_event_version,
1976 recorded_at: wire.recorded_at,
1977 kind: serde_json::from_value(kind).context("decoding Kennedy context event kind")?,
1978 })
1979}
1980
1981fn normalize_derived_identity(kind: &mut EventKind, id: EventId) -> anyhow::Result<()> {
1982 match kind {
1983 EventKind::BoxCreated { box_id, .. } => *box_id = BoxId(id.0),
1984 EventKind::PendingAllocated { pending_id, .. } => {
1985 *pending_id = PendingId::from_event(id);
1986 }
1987 _ => {}
1988 }
1989 Ok(())
1990}
1991
1992pub struct Session {
1993 durable: DurableSession,
1994 chatend: Chatend,
1995 objects: BTreeMap<PendingId, ObjectLocation>,
1996}
1997
1998struct UnfinishedToolInvocation {
1999 invocation_id: Option<String>,
2000 tool_instance: String,
2001 tool_name: String,
2002}
2003
2004impl Session {
2005 pub(crate) fn create(
2006 path: impl AsRef<Path>,
2007 metadata: SessionMetadata,
2008 ) -> anyhow::Result<Self> {
2009 ensure!(
2010 metadata.effective_context_tokens > 0,
2011 "effective context window must be positive"
2012 );
2013 let requested = path.as_ref();
2014 let directory = requested
2015 .parent()
2016 .filter(|path| !path.as_os_str().is_empty())
2017 .unwrap_or_else(|| Path::new("."));
2018 let durable = DurableSessionStore::new(directory)
2019 .create_session(&metadata.session_id, &metadata.created_at)?;
2020 Ok(Self {
2021 durable,
2022 chatend: Chatend::opened(metadata),
2023 objects: BTreeMap::new(),
2024 })
2025 }
2026
2027 pub(crate) fn open_with_metadata(
2028 path: impl AsRef<Path>,
2029 metadata: SessionMetadata,
2030 ) -> anyhow::Result<Self> {
2031 Self::open_with_metadata_and_provider_costs(path, metadata, None, None)
2032 }
2033
2034 pub(crate) fn open_with_metadata_and_provider_costs(
2035 path: impl AsRef<Path>,
2036 metadata: SessionMetadata,
2037 default_provider_model: Option<&str>,
2038 estimator: Option<ProviderCostEstimator>,
2039 ) -> anyhow::Result<Self> {
2040 let requested = path.as_ref();
2041 ensure!(
2042 requested.extension().and_then(|value| value.to_str()) == Some("session-log"),
2043 "{} is not a session-log path",
2044 requested.display()
2045 );
2046 let directory = requested
2047 .parent()
2048 .filter(|path| !path.as_os_str().is_empty())
2049 .unwrap_or_else(|| Path::new("."));
2050 let session_id = requested
2051 .file_stem()
2052 .and_then(|value| value.to_str())
2053 .context("session-log filename is not valid UTF-8")?;
2054 let durable = DurableSessionStore::new(directory).open_session(session_id)?;
2055 let log = durable.list();
2056 let chatend =
2057 Chatend::replay_with_provider_costs(metadata, &log, default_provider_model, estimator)?;
2058 let mut objects = BTreeMap::new();
2059 for (position, stored) in log.events.iter().enumerate() {
2060 let persisted = decode_context_event(stored)?;
2061 let id = EventId(position as u64 + 1);
2062 let mut kind = persisted.kind;
2063 normalize_derived_identity(&mut kind, id)?;
2064 if let EventKind::PendingAllocated {
2065 pending_id,
2066 resource: PendingKind::Object,
2067 } = kind
2068 {
2069 let object = durable.read_pending_object(EventPosition(position as u64))?;
2070 let metadata = ObjectMetadata {
2071 pending_id: pending_id.clone(),
2072 event_id: id,
2073 recorded_at: chatend
2074 .event(id)
2075 .map(|event| event.recorded_at.clone())
2076 .unwrap_or_default(),
2077 media_type: object.media_type,
2078 file_name: Some(object.file_name),
2079 transport: Value::Null,
2080 };
2081 objects.insert(
2082 pending_id,
2083 ObjectLocation {
2084 metadata,
2085 payload_offset: position as u64,
2086 payload_len: object.bytes.len() as u64,
2087 },
2088 );
2089 }
2090 }
2091 Ok(Self {
2092 durable,
2093 chatend,
2094 objects,
2095 })
2096 }
2097
2098 pub fn id(&self) -> &str {
2099 &self.chatend.metadata.session_id
2100 }
2101
2102 pub fn state(&self) -> &Chatend {
2103 &self.chatend
2104 }
2105
2106 pub fn objects(&self) -> &BTreeMap<PendingId, ObjectLocation> {
2107 &self.objects
2108 }
2109
2110 #[cfg(test)]
2111 fn session_log(&self) -> kcode_session_log::SessionLog {
2112 self.durable.list()
2113 }
2114
2115 pub fn archive_bytes(&self) -> anyhow::Result<Vec<u8>> {
2116 let mut archive =
2117 serde_json::to_value(self.durable.list()).context("serializing the session log")?;
2118 let object = archive
2119 .as_object_mut()
2120 .context("serialized session log is not an object")?;
2121 object.insert(
2122 "metadata".into(),
2123 serde_json::to_value(&self.chatend.metadata)?,
2124 );
2125 object.insert("boxes".into(), serde_json::to_value(&self.chatend.boxes)?);
2126 let projection = self.chatend.projection();
2127 let submitted = self
2128 .chatend
2129 .current_ingress_attempt_events()
2130 .iter()
2131 .rev()
2132 .find_map(|event| {
2133 let EventKind::ProviderInputSubmitted { round, context, .. } = &event.kind else {
2134 return None;
2135 };
2136 Some((event.recorded_at.as_str(), *round, context))
2137 });
2138 if let Some((submitted_at, round, submitted)) = submitted {
2139 object.insert("chatendText".into(), Value::String(submitted.input.clone()));
2140 object.insert(
2141 "chatendTextSource".into(),
2142 Value::String("submitted".into()),
2143 );
2144 object.insert(
2145 "structuredMaterial".into(),
2146 json!({
2147 "provider":submitted.provider,
2148 "model":submitted.model,
2149 "reasoningEffort":submitted.reasoning_effort,
2150 "baseInstructions":submitted.base_instructions,
2151 "developerInstructions":submitted.developer_instructions,
2152 "tools":submitted.tools,
2153 "round":round,
2154 "submittedAt":submitted_at,
2155 }),
2156 );
2157 } else {
2158 object.insert("chatendText".into(), Value::String(projection.render()));
2159 object.insert(
2160 "chatendTextSource".into(),
2161 Value::String("reconstructed".into()),
2162 );
2163 object.insert("structuredMaterial".into(), Value::Null);
2164 }
2165 object.insert("context".into(), serde_json::to_value(projection)?);
2166 serde_json::to_vec(&archive).context("serializing the session archive")
2167 }
2168
2169 pub fn is_sealed(&self) -> bool {
2170 self.durable.is_sealed()
2171 }
2172
2173 pub fn seal(&mut self) -> anyhow::Result<()> {
2174 let unfinished_tools = self.unfinished_tool_invocations()?;
2175 ensure!(
2176 unfinished_tools.is_empty(),
2177 "session ends with unfinished tools {}",
2178 unfinished_tools
2179 .iter()
2180 .map(|tool| tool.tool_name.as_str())
2181 .collect::<Vec<_>>()
2182 .join(", ")
2183 );
2184 self.durable.seal()?;
2185 Ok(())
2186 }
2187
2188 pub fn repair_unfinished_tools(
2189 &mut self,
2190 recorded_at: impl Into<String>,
2191 ) -> anyhow::Result<Vec<EventId>> {
2192 let unfinished = self.unfinished_tool_invocations()?;
2193 if unfinished.is_empty() {
2194 return Ok(Vec::new());
2195 }
2196 let recorded_at = recorded_at.into();
2197 let mut repaired = Vec::with_capacity(unfinished.len());
2198 for tool in unfinished.iter().rev() {
2199 let message = format!(
2200 "{} was interrupted before a durable completion was recorded; the abandoned invocation was closed during session recovery.",
2201 tool.tool_name
2202 );
2203 let kind = if let Some(invocation_id) = &tool.invocation_id {
2204 EventKind::ToolCompleted {
2205 tool_instance: tool.tool_instance.clone(),
2206 tool_name: tool.tool_name.clone(),
2207 outcome: serde_json::json!({"ok":false,"recovered":true,"result":message}),
2208 invocation_id: Some(invocation_id.clone()),
2209 }
2210 } else {
2211 EventKind::ToolCompleted {
2212 tool_instance: "call_ktool".into(),
2213 tool_name: "call_ktool".into(),
2214 outcome: serde_json::json!({"ok":false,"recovered":true,"result":message}),
2215 invocation_id: None,
2216 }
2217 };
2218 repaired.push(self.record(recorded_at.clone(), kind)?);
2219 }
2220 ensure!(
2221 self.unfinished_tool_invocations()?.is_empty(),
2222 "session tool recovery left unfinished invocations"
2223 );
2224 Ok(repaired)
2225 }
2226
2227 fn unfinished_tool_invocations(&self) -> anyhow::Result<Vec<UnfinishedToolInvocation>> {
2228 let mut identified = BTreeMap::<String, UnfinishedToolInvocation>::new();
2229 let mut legacy = Vec::<UnfinishedToolInvocation>::new();
2230 for event in self.chatend.current_ingress_attempt_events() {
2231 match &event.kind {
2232 EventKind::ToolInvoked {
2233 tool_instance,
2234 tool_name,
2235 invocation_id,
2236 ..
2237 } => {
2238 let pending = UnfinishedToolInvocation {
2239 invocation_id: invocation_id.clone(),
2240 tool_instance: tool_instance.clone(),
2241 tool_name: tool_name.clone(),
2242 };
2243 if let Some(invocation_id) = invocation_id {
2244 ensure!(
2245 identified.insert(invocation_id.clone(), pending).is_none(),
2246 "duplicate tool invocation identity {invocation_id}"
2247 );
2248 } else {
2249 legacy.push(pending);
2250 }
2251 }
2252 EventKind::ToolCompleted {
2253 tool_instance,
2254 tool_name,
2255 invocation_id,
2256 ..
2257 } => {
2258 if let Some(invocation_id) = invocation_id {
2259 let invoked = identified.remove(invocation_id).with_context(|| {
2260 format!(
2261 "tool {tool_name} completed without matching invocation {invocation_id}"
2262 )
2263 })?;
2264 ensure!(
2265 invoked.tool_instance.as_str() == tool_instance
2266 && invoked.tool_name.as_str() == tool_name,
2267 "tool completion {invocation_id} does not match its invocation"
2268 );
2269 } else if tool_name == "call_ktool" {
2270 legacy.pop();
2271 } else {
2272 let before = legacy.len();
2273 legacy.retain(|unfinished| unfinished.tool_name.as_str() != tool_name);
2274 ensure!(
2275 legacy.len() != before,
2276 "tool {tool_name} completed without a matching invocation"
2277 );
2278 }
2279 }
2280 _ => {}
2281 }
2282 }
2283 legacy.extend(identified.into_values());
2284 Ok(legacy)
2285 }
2286
2287 pub fn mark_completed(&mut self, session_object_id: String) {
2288 self.chatend.completed_session_object = Some(session_object_id);
2289 }
2290
2291 pub fn configure_context(&mut self, kind: SessionKind, effective_context_tokens: u64) {
2292 self.chatend.metadata.kind = kind;
2293 self.chatend.metadata.effective_context_tokens = effective_context_tokens;
2294 }
2295
2296 pub fn create_box(
2297 &mut self,
2298 recorded_at: impl Into<String>,
2299 name: impl Into<String>,
2300 owner: BoxOwner,
2301 content: BoxContent,
2302 ) -> anyhow::Result<BoxId> {
2303 let recorded_at = recorded_at.into();
2304 let id = EventId(self.chatend.next_id);
2305 let box_id = BoxId(id.0);
2306 self.commit_events(
2307 recorded_at.clone(),
2308 vec![Event {
2309 id,
2310 recorded_at,
2311 kind: EventKind::BoxCreated {
2312 box_id,
2313 name: name.into(),
2314 owner,
2315 content,
2316 },
2317 }],
2318 )?;
2319 Ok(box_id)
2320 }
2321
2322 pub fn update_box(
2323 &mut self,
2324 recorded_at: impl Into<String>,
2325 box_id: BoxId,
2326 content: BoxContent,
2327 ) -> anyhow::Result<Option<EventId>> {
2328 let state = self
2329 .chatend
2330 .boxes
2331 .get(&box_id)
2332 .with_context(|| format!("box {box_id} does not exist"))?;
2333 ensure!(state.active, "box {box_id} is retired");
2334 if state.canonical.content == content {
2335 return Ok(None);
2336 }
2337 let id = EventId(self.chatend.next_id);
2338 let recorded_at = recorded_at.into();
2339 self.commit_events(
2340 recorded_at.clone(),
2341 vec![Event {
2342 id,
2343 recorded_at,
2344 kind: EventKind::CanonicalAdvanced { box_id, content },
2345 }],
2346 )?;
2347 Ok(Some(id))
2348 }
2349
2350 pub fn dehydrate_boxes(
2351 &mut self,
2352 recorded_at: impl Into<String>,
2353 box_ids: &[BoxId],
2354 ) -> anyhow::Result<Vec<EventId>> {
2355 ensure!(
2356 !box_ids.is_empty(),
2357 "at least one box must be selected for dehydration"
2358 );
2359 ensure!(
2360 box_ids.iter().copied().collect::<HashSet<_>>().len() == box_ids.len(),
2361 "box dehydration cannot contain duplicate box IDs"
2362 );
2363 let recorded_at = recorded_at.into();
2364 let events = box_ids
2365 .iter()
2366 .enumerate()
2367 .map(|(offset, box_id)| {
2368 let offset = u64::try_from(offset).context("box dehydration batch is too large")?;
2369 let id = self
2370 .chatend
2371 .next_id
2372 .checked_add(offset)
2373 .context("event ID overflow")?;
2374 Ok(Event {
2375 id: EventId(id),
2376 recorded_at: recorded_at.clone(),
2377 kind: EventKind::BoxDehydrated { box_id: *box_id },
2378 })
2379 })
2380 .collect::<anyhow::Result<Vec<_>>>()?;
2381 let ids = events.iter().map(|event| event.id).collect();
2382 self.commit_events(recorded_at, events)?;
2383 Ok(ids)
2384 }
2385
2386 pub fn summarize_box(
2387 &mut self,
2388 recorded_at: impl Into<String>,
2389 box_id: BoxId,
2390 text: impl Into<String>,
2391 ) -> anyhow::Result<EventId> {
2392 self.box_operation(
2393 recorded_at,
2394 EventKind::BoxSummarized {
2395 box_id,
2396 text: text.into(),
2397 },
2398 )
2399 }
2400
2401 pub fn rehydrate_box(
2402 &mut self,
2403 recorded_at: impl Into<String>,
2404 box_id: BoxId,
2405 ) -> anyhow::Result<EventId> {
2406 let state = self
2407 .chatend
2408 .box_state(box_id)
2409 .with_context(|| format!("box {box_id} does not exist"))?;
2410 ensure!(state.active, "box {box_id} is retired");
2411 if matches!(
2412 state.representation,
2413 Representation::Hydrated { canonical_event }
2414 if canonical_event == state.canonical.event_id
2415 ) {
2416 return Ok(state.canonical.event_id);
2417 }
2418 self.box_operation(recorded_at, EventKind::BoxRehydrated { box_id })
2419 }
2420
2421 pub fn retire_box(
2422 &mut self,
2423 recorded_at: impl Into<String>,
2424 box_id: BoxId,
2425 ) -> anyhow::Result<EventId> {
2426 self.box_operation(recorded_at, EventKind::BoxRetired { box_id })
2427 }
2428
2429 fn box_operation(
2430 &mut self,
2431 recorded_at: impl Into<String>,
2432 kind: EventKind,
2433 ) -> anyhow::Result<EventId> {
2434 let box_id = event_box_id(&kind).context("box operation has no box identity")?;
2435 let state = self
2436 .chatend
2437 .box_state(box_id)
2438 .with_context(|| format!("box {box_id} does not exist"))?;
2439 ensure!(state.active, "box {box_id} is retired");
2440 if let EventKind::BoxSummarized { text, .. } = &kind {
2441 ensure!(!text.trim().is_empty(), "a box summary cannot be empty");
2442 }
2443 if matches!(kind, EventKind::BoxRetired { .. })
2444 && let BoxOwner::Tool { tool_instance, .. } = &state.owner
2445 {
2446 ensure!(
2447 self.chatend
2448 .tools
2449 .get(tool_instance)
2450 .is_some_and(|tool| tool.slots.iter().any(|slot| slot.box_id == box_id)),
2451 "tool box {box_id} is missing from {tool_instance}"
2452 );
2453 }
2454 let id = EventId(self.chatend.next_id);
2455 let recorded_at = recorded_at.into();
2456 self.commit_events(
2457 recorded_at.clone(),
2458 vec![Event {
2459 id,
2460 recorded_at,
2461 kind,
2462 }],
2463 )?;
2464 Ok(id)
2465 }
2466
2467 pub fn allocate_pending_node(
2468 &mut self,
2469 recorded_at: impl Into<String>,
2470 ) -> anyhow::Result<PendingId> {
2471 let id = EventId(self.chatend.next_id);
2472 let pending_id = PendingId::from_event(id);
2473 let recorded_at = recorded_at.into();
2474 self.commit_events(
2475 recorded_at.clone(),
2476 vec![Event {
2477 id,
2478 recorded_at,
2479 kind: EventKind::PendingAllocated {
2480 pending_id: pending_id.clone(),
2481 resource: PendingKind::Node,
2482 },
2483 }],
2484 )?;
2485 Ok(pending_id)
2486 }
2487
2488 pub fn stage_object(
2489 &mut self,
2490 recorded_at: impl Into<String>,
2491 media_type: impl Into<String>,
2492 file_name: Option<String>,
2493 transport: Value,
2494 bytes: &[u8],
2495 ) -> anyhow::Result<PendingId> {
2496 ensure!(
2497 bytes.len() as u64 <= MAX_OBJECT_BYTES,
2498 "object exceeds the 32 GiB V1 limit"
2499 );
2500 let aggregate = self
2501 .objects
2502 .values()
2503 .try_fold(bytes.len() as u64, |total, object| {
2504 total.checked_add(object.payload_len)
2505 })
2506 .context("staged object aggregate length overflow")?;
2507 ensure!(
2508 aggregate <= MAX_OBJECT_BYTES,
2509 "session object payload total exceeds the 32 GiB V1 limit"
2510 );
2511 let event_id = EventId(self.chatend.next_id);
2512 let pending_id = PendingId::from_event(event_id);
2513 let metadata = ObjectMetadata {
2514 pending_id: pending_id.clone(),
2515 event_id,
2516 recorded_at: recorded_at.into(),
2517 media_type: media_type.into(),
2518 file_name,
2519 transport,
2520 };
2521 let kind = EventKind::PendingAllocated {
2522 pending_id: pending_id.clone(),
2523 resource: PendingKind::Object,
2524 };
2525 let text = encode_context_event(&metadata.recorded_at, &kind)?;
2526 let object_file_name = metadata
2527 .file_name
2528 .clone()
2529 .unwrap_or_else(|| format!("object-{}", event_id.0));
2530 let position = self.durable.add_pending_object(
2531 text,
2532 object_file_name,
2533 metadata.media_type.clone(),
2534 bytes,
2535 )?;
2536 ensure!(
2537 position.0 + 1 == event_id.0,
2538 "session-log event position diverged from Kennedy context identity"
2539 );
2540 let allocation = Event {
2541 id: event_id,
2542 recorded_at: metadata.recorded_at.clone(),
2543 kind,
2544 };
2545 self.chatend.apply_transition(&Transition {
2546 recorded_at: metadata.recorded_at.clone(),
2547 events: vec![allocation],
2548 })?;
2549 self.objects.insert(
2550 pending_id.clone(),
2551 ObjectLocation {
2552 metadata,
2553 payload_offset: position.0,
2554 payload_len: bytes.len() as u64,
2555 },
2556 );
2557 Ok(pending_id)
2558 }
2559
2560 pub fn read_object(&mut self, id: &PendingId) -> anyhow::Result<Vec<u8>> {
2561 let location = self
2562 .objects
2563 .get(id)
2564 .with_context(|| format!("staged object {id} does not exist"))?
2565 .clone();
2566 Ok(self
2567 .durable
2568 .read_pending_object(EventPosition(location.payload_offset))?
2569 .bytes)
2570 }
2571
2572 pub fn reset_history_ingress_attempt(
2573 &mut self,
2574 recorded_at: impl Into<String>,
2575 ) -> anyhow::Result<EventId> {
2576 ensure!(
2577 !self.is_sealed(),
2578 "a sealed session cannot start another ingress attempt"
2579 );
2580 ensure!(
2581 self.chatend.history_ingress_started,
2582 "history ingress must have started before an attempt can be reset"
2583 );
2584 let recorded_at = recorded_at.into();
2585 self.record(
2586 recorded_at,
2587 EventKind::Note {
2588 label: HISTORY_INGRESS_ATTEMPT_RESET_NOTE.into(),
2589 value: json!({}),
2590 },
2591 )
2592 }
2593
2594 pub fn prepare_provider_projection(
2595 &mut self,
2596 recorded_at: impl Into<String>,
2597 footer_lines: &[String],
2598 material_fingerprint: &str,
2599 resume_after: Option<EventId>,
2600 ) -> anyhow::Result<PreparedProviderProjection> {
2601 self.prepare_provider_projection_at(
2602 recorded_at,
2603 footer_lines,
2604 material_fingerprint,
2605 resume_after,
2606 None,
2607 )
2608 }
2609
2610 pub fn prepare_provider_projection_with_ingress_time(
2611 &mut self,
2612 recorded_at: impl Into<String>,
2613 footer_lines: &[String],
2614 material_fingerprint: &str,
2615 resume_after: Option<EventId>,
2616 remaining_seconds: u64,
2617 previous_attempt_timed_out: bool,
2618 ) -> anyhow::Result<PreparedProviderProjection> {
2619 self.prepare_provider_projection_at(
2620 recorded_at,
2621 footer_lines,
2622 material_fingerprint,
2623 resume_after,
2624 Some((remaining_seconds, previous_attempt_timed_out)),
2625 )
2626 }
2627
2628 pub fn prepare_provider_resume(
2629 &mut self,
2630 recorded_at: impl Into<String>,
2631 synchronized_after: EventId,
2632 ingress_time: Option<(u64, bool)>,
2633 ) -> anyhow::Result<PreparedProviderResume> {
2634 let rewrite_reason = self
2635 .chatend
2636 .projection_rewrite_reason_after(Some(synchronized_after));
2637 let thread_reset_reason = (!rewrite_reason.is_empty()).then_some(rewrite_reason);
2638 let expectation = match &thread_reset_reason {
2639 Some(reason) => CacheExpectation::PlannedInvalidation {
2640 reason: reason.clone(),
2641 },
2642 None => CacheExpectation::ExpectedWarm,
2643 };
2644 let marker_lines =
2645 self.append_provider_markers(recorded_at.into(), expectation, ingress_time)?;
2646 Ok(PreparedProviderResume {
2647 marker_lines,
2648 thread_reset_reason,
2649 })
2650 }
2651
2652 pub fn prepare_provider_resume_markers(
2653 &mut self,
2654 recorded_at: impl Into<String>,
2655 synchronized_after: EventId,
2656 ingress_time: Option<(u64, bool)>,
2657 ) -> anyhow::Result<Vec<String>> {
2658 self.prepare_provider_resume(recorded_at, synchronized_after, ingress_time)
2659 .map(|prepared| prepared.marker_lines)
2660 }
2661
2662 fn prepare_provider_projection_at(
2663 &mut self,
2664 recorded_at: impl Into<String>,
2665 footer_lines: &[String],
2666 material_fingerprint: &str,
2667 resume_after: Option<EventId>,
2668 ingress_time: Option<(u64, bool)>,
2669 ) -> anyhow::Result<PreparedProviderProjection> {
2670 let cacheable_projection = self.chatend.projection().render();
2671 let previous = self.chatend.latest_provider_input();
2672 let previous_projection = previous
2673 .map(|(_, context, prefix_bytes, fingerprint)| {
2674 let prefix_bytes = usize::try_from(prefix_bytes)
2675 .context("cacheable provider-prefix length does not fit usize")?;
2676 ensure!(
2677 prefix_bytes <= context.input.len()
2678 && context.input.is_char_boundary(prefix_bytes),
2679 "cacheable provider-prefix length is outside the exact input"
2680 );
2681 Ok(PreviousProjection {
2682 text: &context.input[..prefix_bytes],
2683 material_fingerprint: fingerprint,
2684 })
2685 })
2686 .transpose()?;
2687 let expectation = classify(
2688 previous_projection,
2689 &cacheable_projection,
2690 material_fingerprint,
2691 &self
2692 .chatend
2693 .projection_rewrite_reason_after(previous.map(|(id, ..)| id)),
2694 );
2695 self.append_provider_markers(recorded_at.into(), expectation.clone(), ingress_time)?;
2696 let cacheable_projection = self.chatend.projection().render();
2697 let projection = self.chatend.projection_with_footer_lines(footer_lines);
2698 let rewrite_reason = self.chatend.projection_rewrite_reason_after(resume_after);
2699 let thread_reset_reason = resume_after
2700 .is_some()
2701 .then_some(rewrite_reason)
2702 .filter(|reason| !reason.is_empty());
2703 let provider_input = if let Some(resume_after) = resume_after
2704 && thread_reset_reason.is_none()
2705 {
2706 let mut blocks = projection
2707 .items
2708 .iter()
2709 .filter(|item| item.event_id > resume_after)
2710 .map(|item| item.text.as_str())
2711 .collect::<Vec<_>>();
2712 if !projection.footer.is_empty() {
2713 blocks.push(&projection.footer);
2714 }
2715 blocks.join("\n\n")
2716 } else {
2717 projection.render()
2718 };
2719 Ok(PreparedProviderProjection {
2720 projection,
2721 provider_input,
2722 thread_reset_reason,
2723 cacheable_prefix_bytes: cacheable_projection.len() as u64,
2724 expectation,
2725 })
2726 }
2727
2728 fn append_provider_markers(
2729 &mut self,
2730 recorded_at: String,
2731 expectation: CacheExpectation,
2732 ingress_time: Option<(u64, bool)>,
2733 ) -> anyhow::Result<Vec<String>> {
2734 let projection = if expectation.expects_cache_hit() {
2735 self.chatend.projection()
2736 } else {
2737 let mut preview = self.chatend.clone();
2738 preview.apply_transition(&Transition {
2739 recorded_at: "marker-preview".into(),
2740 events: vec![Event {
2741 id: EventId(preview.next_id),
2742 recorded_at: "marker-preview".into(),
2743 kind: EventKind::ProjectionMarkersReset,
2744 }],
2745 })?;
2746 preview.projection()
2747 };
2748 let decision = decide_markers(
2749 &self.chatend.marker_state(),
2750 MarkerObservation {
2751 stale_boxes: projection.stale_boxes.iter().map(|id| id.0).collect(),
2752 current_context_tokens: projection.estimated_tokens,
2753 context_limit_tokens: projection.status.context_limit_tokens,
2754 expectation: expectation.clone(),
2755 },
2756 );
2757 let mut next = self.chatend.next_id;
2758 let mut events = Vec::new();
2759 let mut marker_lines = Vec::new();
2760 let ingress_time_marker_state = self.chatend.ingress_time_marker_state();
2761 let append_ingress_time = ingress_time.is_some_and(|(remaining_seconds, _)| {
2762 should_append_ingress_time_marker(
2763 ingress_time_marker_state,
2764 remaining_seconds,
2765 decision.context_size_tokens.is_some(),
2766 )
2767 });
2768 {
2769 let mut push = |kind| -> anyhow::Result<()> {
2770 events.push(Event {
2771 id: EventId(next),
2772 recorded_at: recorded_at.clone(),
2773 kind,
2774 });
2775 next = next.checked_add(1).context("event ID overflow")?;
2776 Ok(())
2777 };
2778 if decision.reset_epoch {
2779 push(EventKind::ProjectionMarkersReset)?;
2780 }
2781 if let Some(stale) = decision.stale {
2782 marker_lines.push(stale.render());
2783 let (box_ids, consolidated) = match stale {
2784 StaleMarker::New(ids) => (ids, false),
2785 StaleMarker::Consolidated(ids) => (ids, true),
2786 };
2787 push(EventKind::StaleBoxesMarked {
2788 box_ids: box_ids.into_iter().map(BoxId).collect(),
2789 consolidated,
2790 })?;
2791 }
2792 if let Some(estimated_tokens) = decision.context_size_tokens {
2793 marker_lines.push(context_size_marker_text(
2794 estimated_tokens,
2795 projection.status.context_limit_tokens,
2796 ));
2797 push(EventKind::ContextSizeMarked { estimated_tokens })?;
2798 }
2799 if append_ingress_time
2800 && let Some((remaining_seconds, previous_attempt_timed_out)) = ingress_time
2801 {
2802 let value = json!({
2803 "remainingSeconds":remaining_seconds,
2804 "previousAttemptTimedOut":previous_attempt_timed_out
2805 && !ingress_time_marker_state.any,
2806 });
2807 let marker_line = ingress_time_marker_text(&value)
2808 .expect("constructed ingress marker value is valid");
2809 push(EventKind::Note {
2810 label: INGRESS_TIME_MARKER_NOTE.into(),
2811 value,
2812 })?;
2813 marker_lines.push(marker_line);
2814 }
2815 }
2816 if !events.is_empty() {
2817 self.commit_events(recorded_at, events)?;
2818 }
2819 Ok(marker_lines)
2820 }
2821
2822 pub fn record(
2823 &mut self,
2824 recorded_at: impl Into<String>,
2825 kind: EventKind,
2826 ) -> anyhow::Result<EventId> {
2827 let id = EventId(self.chatend.next_id);
2828 let recorded_at = recorded_at.into();
2829 self.commit_events(
2830 recorded_at.clone(),
2831 vec![Event {
2832 id,
2833 recorded_at,
2834 kind,
2835 }],
2836 )?;
2837 Ok(id)
2838 }
2839
2840 pub fn commit_events(
2841 &mut self,
2842 recorded_at: impl Into<String>,
2843 events: Vec<Event>,
2844 ) -> anyhow::Result<()> {
2845 let transition = Transition {
2846 recorded_at: recorded_at.into(),
2847 events,
2848 };
2849 ensure!(
2850 !transition.events.is_empty(),
2851 "a transition cannot be empty"
2852 );
2853 ensure!(
2854 !transition.events.iter().any(|event| {
2855 matches!(
2856 event.kind,
2857 EventKind::PendingAllocated {
2858 resource: PendingKind::Object,
2859 ..
2860 }
2861 )
2862 }),
2863 "pending objects must be added through stage_object"
2864 );
2865 let mut preview = self.chatend.clone();
2866 preview.apply_transition(&transition)?;
2867 for event in &transition.events {
2868 let expected = self.durable.list().events.len() as u64 + 1;
2869 ensure!(
2870 event.id.0 == expected,
2871 "Kennedy context event {} does not match session-log position {}",
2872 event.id,
2873 expected - 1
2874 );
2875 self.durable.add_event(
2876 context_event_role(&event.kind),
2877 encode_context_event(&event.recorded_at, &event.kind)?,
2878 )?;
2879 }
2880 self.chatend = preview;
2881 Ok(())
2882 }
2883
2884 pub fn apply_tool_slots(
2885 &mut self,
2886 recorded_at: impl Into<String>,
2887 tool_instance: impl Into<String>,
2888 slots: Vec<ToolSlotInput>,
2889 ) -> anyhow::Result<Vec<EventId>> {
2890 self.apply_tool_slots_inner(recorded_at, tool_instance, slots, None)
2891 }
2892
2893 pub fn apply_tool_slots_with_layout(
2894 &mut self,
2895 recorded_at: impl Into<String>,
2896 tool_instance: impl Into<String>,
2897 slots: Vec<ToolSlotInput>,
2898 layout_slots: &[String],
2899 ) -> anyhow::Result<Vec<EventId>> {
2900 self.apply_tool_slots_inner(recorded_at, tool_instance, slots, Some(layout_slots))
2901 }
2902
2903 fn apply_tool_slots_inner(
2904 &mut self,
2905 recorded_at: impl Into<String>,
2906 tool_instance: impl Into<String>,
2907 slots: Vec<ToolSlotInput>,
2908 layout_slots: Option<&[String]>,
2909 ) -> anyhow::Result<Vec<EventId>> {
2910 let recorded_at = recorded_at.into();
2911 let tool_instance = tool_instance.into();
2912 let current = self
2913 .chatend
2914 .tools
2915 .get(&tool_instance)
2916 .cloned()
2917 .unwrap_or_default();
2918 ensure!(
2919 slots.len() >= current.slots.len(),
2920 "stateful tool slot sequence was truncated"
2921 );
2922 for (index, existing) in current.slots.iter().enumerate() {
2923 ensure!(
2924 slots[index].slot == existing.slot,
2925 "stateful tool slot sequence was reordered at index {index}"
2926 );
2927 ensure!(
2928 !existing.retired || slots[index].retired,
2929 "retired tool slot {} cannot be reactivated",
2930 existing.slot
2931 );
2932 }
2933 let mut events = Vec::new();
2934 let mut next = self.chatend.next_id;
2935 let mut next_state = current.clone();
2936 for (index, input) in slots.iter().enumerate() {
2937 if let Some(existing) = current.slots.get(index) {
2938 let state = self
2939 .chatend
2940 .boxes
2941 .get(&existing.box_id)
2942 .context("tool slot references a missing box")?;
2943 if input.retired && !existing.retired {
2944 let id = EventId(next);
2945 next += 1;
2946 events.push(Event {
2947 id,
2948 recorded_at: recorded_at.clone(),
2949 kind: EventKind::BoxRetired {
2950 box_id: existing.box_id,
2951 },
2952 });
2953 next_state.slots[index].retired = true;
2954 } else if !input.retired {
2955 if state.name != input.name {
2956 let id = EventId(next);
2957 next += 1;
2958 events.push(Event {
2959 id,
2960 recorded_at: recorded_at.clone(),
2961 kind: EventKind::BoxRenamed {
2962 box_id: existing.box_id,
2963 name: input.name.clone(),
2964 },
2965 });
2966 }
2967 if state.canonical.content != input.content {
2968 let id = EventId(next);
2969 next += 1;
2970 events.push(Event {
2971 id,
2972 recorded_at: recorded_at.clone(),
2973 kind: EventKind::CanonicalAdvanced {
2974 box_id: existing.box_id,
2975 content: input.content.clone(),
2976 },
2977 });
2978 }
2979 }
2980 } else {
2981 ensure!(
2982 !input.retired,
2983 "a newly appended tool slot cannot start retired"
2984 );
2985 let id = EventId(next);
2986 next += 1;
2987 let box_id = BoxId(id.0);
2988 events.push(Event {
2989 id,
2990 recorded_at: recorded_at.clone(),
2991 kind: EventKind::BoxCreated {
2992 box_id,
2993 name: input.name.clone(),
2994 owner: BoxOwner::Tool {
2995 tool_instance: tool_instance.clone(),
2996 slot: input.slot.clone(),
2997 },
2998 content: input.content.clone(),
2999 },
3000 });
3001 next_state.slots.push(ToolSlot {
3002 slot: input.slot.clone(),
3003 box_id,
3004 retired: false,
3005 });
3006 }
3007 }
3008 if let Some(layout_slots) = layout_slots {
3009 let mut unique = std::collections::HashSet::new();
3010 let box_ids = layout_slots
3011 .iter()
3012 .map(|slot_name| {
3013 ensure!(
3014 unique.insert(slot_name),
3015 "tool layout contains duplicate slot {slot_name}"
3016 );
3017 let slot = next_state
3018 .slots
3019 .iter()
3020 .find(|slot| &slot.slot == slot_name)
3021 .with_context(|| {
3022 format!("tool layout references missing slot {slot_name}")
3023 })?;
3024 ensure!(
3025 !slot.retired,
3026 "tool layout references retired slot {slot_name}"
3027 );
3028 Ok(slot.box_id)
3029 })
3030 .collect::<anyhow::Result<Vec<_>>>()?;
3031 if self.chatend.tool_layouts.get(&tool_instance) != Some(&box_ids) {
3032 let id = EventId(next);
3033 events.push(Event {
3034 id,
3035 recorded_at: recorded_at.clone(),
3036 kind: EventKind::ToolLayoutChanged {
3037 tool_instance: tool_instance.clone(),
3038 box_ids,
3039 },
3040 });
3041 }
3042 }
3043 if events.is_empty() {
3044 return Ok(Vec::new());
3045 }
3046 let ids = events.iter().map(|event| event.id).collect::<Vec<_>>();
3047 self.commit_events(recorded_at, events)?;
3048 Ok(ids)
3049 }
3050
3051 pub fn apply_box_representations(
3052 &mut self,
3053 recorded_at: impl Into<String>,
3054 desired: &BTreeMap<BoxId, BoxRepresentation>,
3055 ) -> anyhow::Result<Vec<EventId>> {
3056 let recorded_at = recorded_at.into();
3057 let events = self
3058 .chatend
3059 .box_representation_events(&recorded_at, desired)?;
3060 if events.is_empty() {
3061 return Ok(Vec::new());
3062 }
3063 let ids = events.iter().map(|event| event.id).collect::<Vec<_>>();
3064 self.commit_events(recorded_at, events)?;
3065 Ok(ids)
3066 }
3067}
3068
3069pub struct SessionHistoryIntegration;
3070
3071impl SessionHistoryIntegration {
3072 pub fn create_session(
3073 path: impl AsRef<Path>,
3074 metadata: SessionMetadata,
3075 ) -> anyhow::Result<Session> {
3076 Session::create(path, metadata)
3077 }
3078
3079 pub fn open_session(
3080 path: impl AsRef<Path>,
3081 metadata: SessionMetadata,
3082 default_provider_model: Option<&str>,
3083 estimator: Option<ProviderCostEstimator>,
3084 ) -> anyhow::Result<Session> {
3085 match (default_provider_model, estimator) {
3086 (None, None) => Session::open_with_metadata(path, metadata),
3087 (default_provider_model, estimator) => Session::open_with_metadata_and_provider_costs(
3088 path,
3089 metadata,
3090 default_provider_model,
3091 estimator,
3092 ),
3093 }
3094 }
3095
3096 pub fn replay(
3097 metadata: SessionMetadata,
3098 log: &kcode_session_log::SessionLog,
3099 default_provider_model: Option<&str>,
3100 estimator: Option<ProviderCostEstimator>,
3101 ) -> anyhow::Result<Chatend> {
3102 match (default_provider_model, estimator) {
3103 (None, None) => Chatend::replay(metadata, log),
3104 (default_provider_model, estimator) => Chatend::replay_with_provider_costs(
3105 metadata,
3106 log,
3107 default_provider_model,
3108 estimator,
3109 ),
3110 }
3111 }
3112
3113 pub fn legacy_provider_cost_summary_for_archive(
3114 archive: &Value,
3115 default_provider_model: Option<&str>,
3116 estimator: ProviderCostEstimator,
3117 ) -> anyhow::Result<ProviderCostSummary> {
3118 legacy_provider_cost_summary_for_archive(archive, default_provider_model, estimator)
3119 }
3120}
3121
3122#[cfg(test)]
3123type SessionJournal = Session;
3124
3125#[cfg(test)]
3126mod tests {
3127 use std::path::PathBuf;
3128 use std::time::{SystemTime, UNIX_EPOCH};
3129
3130 use serde_json::json;
3131
3132 use super::*;
3133
3134 fn path(label: &str) -> PathBuf {
3135 std::env::temp_dir()
3136 .join(format!(
3137 "kennedy-chatend-{label}-{}-{}",
3138 std::process::id(),
3139 SystemTime::now()
3140 .duration_since(UNIX_EPOCH)
3141 .unwrap()
3142 .as_nanos()
3143 ))
3144 .join("session-1.session-log")
3145 }
3146
3147 fn metadata() -> SessionMetadata {
3148 SessionMetadata {
3149 session_id: "session-1".into(),
3150 kind: SessionKind::Conversation,
3151 created_at: "2026-07-23T00:00:00Z".into(),
3152 effective_context_tokens: 1_000,
3153 channel: json!({"kind":"test"}),
3154 }
3155 }
3156
3157 #[test]
3158 fn box_identity_continuations_staleness_and_replay_are_exact() {
3159 let path = path("boxes");
3160 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
3161 let id = journal
3162 .create_box(
3163 "t1",
3164 "message",
3165 BoxOwner::User,
3166 BoxContent::text("original"),
3167 )
3168 .unwrap();
3169 assert_eq!(id, BoxId(1));
3170 journal.summarize_box("t2", id, "summary").unwrap();
3171 journal
3172 .update_box("t3", id, BoxContent::text("changed"))
3173 .unwrap();
3174 let state = journal.state().box_state(id).unwrap().clone();
3175 assert!(state.stale());
3176 assert_eq!(
3177 state.representation,
3178 Representation::Summarized {
3179 based_on: EventId(1),
3180 text: "summary".into()
3181 }
3182 );
3183 let projection = journal.state().projection();
3184 assert_eq!(projection.items[0].text, "[box updated]");
3185 assert!(projection.items[1].text.contains("summary"));
3186 assert!(projection.items[1].stale);
3187 assert!(projection.footer.is_empty());
3188 assert!(projection.render().ends_with("summary"));
3189
3190 let transient = vec![
3191 "[provider calls remaining after this call: 249 | max provider calls: 250]".into(),
3192 "[provider call time remaining: 3h 45m]".into(),
3193 ];
3194 let with_limits = journal.state().projection_with_footer_lines(&transient);
3195 assert_eq!(with_limits.footer, transient.join("\n"));
3196 assert!(with_limits.render().ends_with(&transient.join("\n")));
3197 assert!(with_limits.context_bytes > projection.context_bytes);
3198 drop(journal);
3199 let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
3200 assert_eq!(reopened.state().box_state(id), Some(&state));
3201 std::fs::remove_file(path).unwrap();
3202 }
3203
3204 #[test]
3205 fn cache_safe_advance_preserves_hydrated_summary_and_dehydrated_bytes() {
3206 for representation in ["hydrated", "summarized", "dehydrated"] {
3207 let path = path(&format!("cache-safe-{representation}"));
3208 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
3209 let box_id = journal
3210 .create_box(
3211 "t1",
3212 "cache-sensitive",
3213 BoxOwner::Controller,
3214 BoxContent::text("provider-visible original"),
3215 )
3216 .unwrap();
3217 match representation {
3218 "summarized" => {
3219 journal
3220 .summarize_box("t2", box_id, "provider-visible summary")
3221 .unwrap();
3222 }
3223 "dehydrated" => {
3224 journal.dehydrate_boxes("t2", &[box_id]).unwrap();
3225 }
3226 _ => {}
3227 }
3228 let before = journal.state().projection().render();
3229 let occurrence_events = journal
3230 .state()
3231 .box_state(box_id)
3232 .unwrap()
3233 .occurrence_events
3234 .clone();
3235
3236 let advanced = journal
3237 .update_box("t3", box_id, BoxContent::text("latest canonical"))
3238 .unwrap()
3239 .unwrap();
3240
3241 let state = journal.state().box_state(box_id).unwrap();
3242 assert_eq!(state.canonical.event_id, advanced);
3243 assert_eq!(
3244 state.canonical.content,
3245 BoxContent::text("latest canonical")
3246 );
3247 assert_eq!(state.occurrence_events, occurrence_events);
3248 assert!(state.stale());
3249 let projection = journal.state().projection();
3250 assert_eq!(projection.render().as_bytes(), before.as_bytes());
3251 assert_eq!(projection.stale_boxes, [box_id]);
3252 assert!(
3253 projection
3254 .items
3255 .iter()
3256 .find(|item| item.box_id == box_id && !item.marker)
3257 .unwrap()
3258 .stale
3259 );
3260 assert!(!projection.render().contains("| stale"));
3261 std::fs::remove_file(path).unwrap();
3262 }
3263 }
3264
3265 #[test]
3266 fn multiple_advances_keep_original_bytes_and_rehydrate_to_latest() {
3267 let path = path("multiple-cache-safe-advances");
3268 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
3269 let box_id = journal
3270 .create_box(
3271 "t1",
3272 "cache-sensitive",
3273 BoxOwner::Controller,
3274 BoxContent::text("original"),
3275 )
3276 .unwrap();
3277 let original = journal.state().projection().render();
3278 let first = journal
3279 .update_box("t2", box_id, BoxContent::text("middle"))
3280 .unwrap()
3281 .unwrap();
3282 let latest = journal
3283 .update_box("t3", box_id, BoxContent::text("latest"))
3284 .unwrap()
3285 .unwrap();
3286
3287 let state = journal.state().box_state(box_id).unwrap();
3288 assert_eq!(state.canonical.event_id, latest);
3289 assert_eq!(state.canonical.content, BoxContent::text("latest"));
3290 assert!(state.stale());
3291 assert_eq!(journal.state().projection().render(), original);
3292 assert!(matches!(
3293 journal.state().event(first).unwrap().kind,
3294 EventKind::CanonicalAdvanced { .. }
3295 ));
3296
3297 let rehydrated = journal.rehydrate_box("t4", box_id).unwrap();
3298 assert_ne!(rehydrated, latest);
3299 let state = journal.state().box_state(box_id).unwrap();
3300 assert_eq!(
3301 state.representation,
3302 Representation::Hydrated {
3303 canonical_event: latest
3304 }
3305 );
3306 assert!(!state.stale());
3307 assert!(journal.state().projection().render().contains("latest"));
3308 assert!(!journal.state().projection().render().contains("original"));
3309 std::fs::remove_file(path).unwrap();
3310 }
3311
3312 #[test]
3313 fn current_hydration_is_noop_but_stale_hydrated_plans_rehydration() {
3314 let path = path("hydration-noop");
3315 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
3316 let box_id = journal
3317 .create_box(
3318 "t1",
3319 "box",
3320 BoxOwner::Controller,
3321 BoxContent::text("original"),
3322 )
3323 .unwrap();
3324 let next = journal.state().next_id;
3325 assert_eq!(journal.rehydrate_box("t2", box_id).unwrap(), EventId(1));
3326 assert_eq!(journal.state().next_id, next);
3327 assert!(
3328 journal
3329 .apply_box_representations(
3330 "t3",
3331 &BTreeMap::from([(box_id, BoxRepresentation::Hydrated)])
3332 )
3333 .unwrap()
3334 .is_empty()
3335 );
3336
3337 journal
3338 .update_box("t4", box_id, BoxContent::text("latest"))
3339 .unwrap();
3340 let preview = journal
3341 .state()
3342 .projection_with_box_representations(&BTreeMap::from([(
3343 box_id,
3344 BoxRepresentation::Hydrated,
3345 )]))
3346 .unwrap();
3347 let ids = journal
3348 .apply_box_representations(
3349 "t5",
3350 &BTreeMap::from([(box_id, BoxRepresentation::Hydrated)]),
3351 )
3352 .unwrap();
3353 assert_eq!(ids.len(), 1);
3354 assert_eq!(journal.state().projection(), preview);
3355 assert!(!journal.state().box_state(box_id).unwrap().stale());
3356 std::fs::remove_file(path).unwrap();
3357 }
3358
3359 #[test]
3360 fn historical_update_replays_visible_replacement_while_advance_reopens_stale() {
3361 let legacy_path = path("historical-canonical-update");
3362 let mut legacy = SessionJournal::create(&legacy_path, metadata()).unwrap();
3363 let legacy_box = legacy
3364 .create_box(
3365 "t1",
3366 "legacy",
3367 BoxOwner::Controller,
3368 BoxContent::text("before"),
3369 )
3370 .unwrap();
3371 legacy
3372 .record(
3373 "t2",
3374 EventKind::CanonicalUpdated {
3375 box_id: legacy_box,
3376 content: BoxContent::text("after"),
3377 },
3378 )
3379 .unwrap();
3380 let legacy_projection = legacy.state().projection().clone();
3381 assert_eq!(legacy_projection.items[0].text, "[box updated]");
3382 assert!(legacy_projection.render().contains("after"));
3383 assert!(!legacy.state().box_state(legacy_box).unwrap().stale());
3384 drop(legacy);
3385 let legacy_reopened = SessionJournal::open_with_metadata(&legacy_path, metadata()).unwrap();
3386 assert_eq!(legacy_reopened.state().projection(), legacy_projection);
3387 std::fs::remove_file(legacy_path).unwrap();
3388
3389 let advanced_path = path("canonical-advance-reopen");
3390 let mut advanced = SessionJournal::create(&advanced_path, metadata()).unwrap();
3391 let advanced_box = advanced
3392 .create_box(
3393 "t1",
3394 "advanced",
3395 BoxOwner::Controller,
3396 BoxContent::text("before"),
3397 )
3398 .unwrap();
3399 let visible = advanced.state().projection().render();
3400 advanced
3401 .update_box("t2", advanced_box, BoxContent::text("after"))
3402 .unwrap();
3403 let advanced_state = advanced.state().box_state(advanced_box).unwrap().clone();
3404 assert!(advanced_state.stale());
3405 assert_eq!(advanced.state().projection().render(), visible);
3406 drop(advanced);
3407 let advanced_reopened =
3408 SessionJournal::open_with_metadata(&advanced_path, metadata()).unwrap();
3409 assert_eq!(
3410 advanced_reopened.state().box_state(advanced_box),
3411 Some(&advanced_state)
3412 );
3413 assert_eq!(advanced_reopened.state().projection().render(), visible);
3414 std::fs::remove_file(advanced_path).unwrap();
3415 }
3416
3417 #[test]
3418 fn update_preview_and_tool_slot_updates_are_cache_safe_advances() {
3419 let path = path("cache-safe-call-sites");
3420 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
3421 let box_id = journal
3422 .create_box(
3423 "t1",
3424 "ordinary",
3425 BoxOwner::Controller,
3426 BoxContent::text("before"),
3427 )
3428 .unwrap();
3429 let before = journal.state().projection().render();
3430 let preview = journal
3431 .state()
3432 .projection_with_new_boxes_and_updates(
3433 &[],
3434 &BTreeMap::from([(box_id, BoxContent::text("after"))]),
3435 )
3436 .unwrap();
3437 assert_eq!(preview.render(), before);
3438 assert_eq!(preview.stale_boxes, [box_id]);
3439
3440 journal
3441 .apply_tool_slots(
3442 "t2",
3443 "tool",
3444 vec![ToolSlotInput {
3445 slot: "slot".into(),
3446 name: "slot".into(),
3447 content: BoxContent::text("tool before"),
3448 retired: false,
3449 }],
3450 )
3451 .unwrap();
3452 let tool_box = journal.state().tools["tool"].slots[0].box_id;
3453 let tool_before = journal.state().projection().render();
3454 let events = journal
3455 .apply_tool_slots(
3456 "t3",
3457 "tool",
3458 vec![ToolSlotInput {
3459 slot: "slot".into(),
3460 name: "slot".into(),
3461 content: BoxContent::text("tool after"),
3462 retired: false,
3463 }],
3464 )
3465 .unwrap();
3466 assert!(matches!(
3467 journal.state().event(events[0]).unwrap().kind,
3468 EventKind::CanonicalAdvanced { .. }
3469 ));
3470 assert_eq!(journal.state().projection().render(), tool_before);
3471 assert!(journal.state().box_state(tool_box).unwrap().stale());
3472 std::fs::remove_file(path).unwrap();
3473 }
3474
3475 #[test]
3476 fn box_headers_hide_internal_ownership_and_timestamp_messages() {
3477 let path = path("box-headers");
3478 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
3479 journal
3480 .create_box(
3481 "2026-07-23T00:00:01Z",
3482 "User message",
3483 BoxOwner::User,
3484 BoxContent::text("Hello"),
3485 )
3486 .unwrap();
3487 journal
3488 .create_box(
3489 "2026-07-23T00:00:02Z",
3490 "Kennedy message",
3491 BoxOwner::Kennedy,
3492 BoxContent::text("Hi"),
3493 )
3494 .unwrap();
3495 let mut content = BoxContent::text("Node ID: AAAAAAAB");
3496 content.use_concise_header();
3497 journal
3498 .create_box(
3499 "2026-07-23T00:00:03Z",
3500 "Kweb loaded node",
3501 BoxOwner::Tool {
3502 tool_instance: "kweb".into(),
3503 slot: "loaded".into(),
3504 },
3505 content,
3506 )
3507 .unwrap();
3508
3509 let projection = journal.state().projection();
3510 assert_eq!(
3511 projection.items[0].text,
3512 "[box 1 | User message | timestamp=2026-07-23T00:00:01Z | hydrated]\nHello"
3513 );
3514 assert_eq!(
3515 projection.items[1].text,
3516 "[box 2 | Kennedy message | timestamp=2026-07-23T00:00:02Z | hydrated]\nHi"
3517 );
3518 assert_eq!(
3519 projection.items[2].text,
3520 "[box 3 | Kweb loaded node | hydrated]\nNode ID: AAAAAAAB"
3521 );
3522 assert!(
3523 projection
3524 .items
3525 .iter()
3526 .all(|item| !item.text.contains("owner="))
3527 );
3528 assert!(projection.footer.is_empty());
3529 std::fs::remove_file(path).unwrap();
3530 }
3531
3532 #[test]
3533 fn provider_preparation_anchors_sparse_markers_and_resets_rewritten_epochs() {
3534 let path = path("provider-markers");
3535 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
3536 let first_box = journal
3537 .create_box(
3538 "t1",
3539 "System prompt",
3540 BoxOwner::System,
3541 BoxContent::text("x".repeat(1_000)),
3542 )
3543 .unwrap();
3544 let first = journal
3545 .prepare_provider_projection(
3546 "t2",
3547 &["[provider calls remaining: 9]".into()],
3548 "m1",
3549 None,
3550 )
3551 .unwrap();
3552 assert_eq!(first.expectation, CacheExpectation::ColdStart);
3553 assert!(first.projection.render().contains("[current context size:"));
3554 assert!(
3555 first
3556 .projection
3557 .render()
3558 .ends_with("[provider calls remaining: 9]")
3559 );
3560 let first_input = first.projection.render();
3561 let synchronized_event_id = journal
3562 .record(
3563 "t3",
3564 EventKind::ProviderInputSubmitted {
3565 round: 1,
3566 context: provider_context(first_input.clone()),
3567 transport_input_hash: None,
3568 transport_input_bytes: None,
3569 thread_action: None,
3570 thread_reset_reason: None,
3571 cacheable_prefix_bytes: first.cacheable_prefix_bytes,
3572 material_fingerprint: "m1".into(),
3573 cache_expectation: first.expectation.label().into(),
3574 planned_invalidation_reason: None,
3575 },
3576 )
3577 .unwrap();
3578 let archive: Value = serde_json::from_slice(&journal.archive_bytes().unwrap()).unwrap();
3579 assert_eq!(archive["chatendText"], first_input);
3580 assert_eq!(archive["chatendTextSource"], "submitted");
3581 assert_eq!(archive["structuredMaterial"]["provider"], "codex");
3582 assert_eq!(archive["structuredMaterial"]["model"], "test");
3583 journal
3584 .create_box(
3585 "t4",
3586 "User message",
3587 BoxOwner::User,
3588 BoxContent::text("next"),
3589 )
3590 .unwrap();
3591 let warm = journal
3592 .prepare_provider_projection("t5", &[], "m1", Some(synchronized_event_id))
3593 .unwrap();
3594 assert_eq!(warm.expectation, CacheExpectation::ExpectedWarm);
3595 assert!(warm.provider_input.contains("next"));
3596 assert!(!warm.provider_input.contains(&"x".repeat(1_000)));
3597 assert!(warm.thread_reset_reason.is_none());
3598
3599 journal.summarize_box("t6", first_box, "short").unwrap();
3600 let reset = journal
3601 .prepare_provider_projection("t7", &[], "m1", Some(synchronized_event_id))
3602 .unwrap();
3603 assert_eq!(
3604 reset.expectation,
3605 CacheExpectation::PlannedInvalidation {
3606 reason: "summarization".into()
3607 }
3608 );
3609 assert_eq!(reset.thread_reset_reason.as_deref(), Some("summarization"));
3610 assert_eq!(reset.provider_input, reset.projection.render());
3611 assert!(!reset.projection.render().contains("[current context size:"));
3612 assert_eq!(
3613 journal
3614 .state()
3615 .events
3616 .iter()
3617 .filter(|event| matches!(event.kind, EventKind::ProjectionMarkersReset))
3618 .count(),
3619 2
3620 );
3621 std::fs::remove_file(path).unwrap();
3622 }
3623
3624 fn prepared_resume_after<T>(
3625 label: &str,
3626 setup: impl FnOnce(&mut SessionJournal) -> T,
3627 mutate: impl FnOnce(&mut SessionJournal, T),
3628 ) -> PreparedProviderResume {
3629 let path = path(label);
3630 let mut journal = SessionJournal::create(
3631 &path,
3632 SessionMetadata {
3633 effective_context_tokens: 100_000,
3634 ..metadata()
3635 },
3636 )
3637 .unwrap();
3638 let value = setup(&mut journal);
3639 let synchronized_after = journal.state().events.last().unwrap().id;
3640 mutate(&mut journal, value);
3641 let prepared = journal
3642 .prepare_provider_resume("resume", synchronized_after, None)
3643 .unwrap();
3644 std::fs::remove_file(path).unwrap();
3645 prepared
3646 }
3647
3648 #[test]
3649 fn structured_provider_resume_reports_append_only_and_exact_rewrite_reasons() {
3650 let append_only = prepared_resume_after(
3651 "resume-append-only",
3652 |journal| {
3653 journal
3654 .create_box("t1", "first", BoxOwner::User, BoxContent::text("first"))
3655 .unwrap()
3656 },
3657 |journal, _| {
3658 journal
3659 .create_box("t2", "second", BoxOwner::User, BoxContent::text("second"))
3660 .unwrap();
3661 },
3662 );
3663 assert!(append_only.marker_lines.is_empty());
3664 assert_eq!(append_only.thread_reset_reason, None);
3665
3666 let dehydration = prepared_resume_after(
3667 "resume-dehydration",
3668 |journal| {
3669 journal
3670 .create_box("t1", "box", BoxOwner::Controller, BoxContent::text("body"))
3671 .unwrap()
3672 },
3673 |journal, box_id| {
3674 journal.dehydrate_boxes("t2", &[box_id]).unwrap();
3675 },
3676 );
3677 assert_eq!(
3678 dehydration.thread_reset_reason.as_deref(),
3679 Some("dehydration")
3680 );
3681
3682 let summarization = prepared_resume_after(
3683 "resume-summarization",
3684 |journal| {
3685 journal
3686 .create_box("t1", "box", BoxOwner::Controller, BoxContent::text("body"))
3687 .unwrap()
3688 },
3689 |journal, box_id| {
3690 journal.summarize_box("t2", box_id, "summary").unwrap();
3691 },
3692 );
3693 assert_eq!(
3694 summarization.thread_reset_reason.as_deref(),
3695 Some("summarization")
3696 );
3697
3698 let rehydration = prepared_resume_after(
3699 "resume-rehydration",
3700 |journal| {
3701 let box_id = journal
3702 .create_box("t1", "box", BoxOwner::Controller, BoxContent::text("body"))
3703 .unwrap();
3704 journal.dehydrate_boxes("t2", &[box_id]).unwrap();
3705 box_id
3706 },
3707 |journal, box_id| {
3708 journal.rehydrate_box("t3", box_id).unwrap();
3709 },
3710 );
3711 assert_eq!(
3712 rehydration.thread_reset_reason.as_deref(),
3713 Some("rehydration")
3714 );
3715
3716 let canonical_advance = prepared_resume_after(
3717 "resume-canonical-advance",
3718 |journal| {
3719 journal
3720 .create_box(
3721 "t1",
3722 "box",
3723 BoxOwner::Controller,
3724 BoxContent::text("before"),
3725 )
3726 .unwrap()
3727 },
3728 |journal, box_id| {
3729 journal
3730 .update_box("t2", box_id, BoxContent::text("after"))
3731 .unwrap();
3732 },
3733 );
3734 assert_eq!(canonical_advance.thread_reset_reason, None);
3735 assert_eq!(canonical_advance.marker_lines, ["[new stale boxes: 1]"]);
3736
3737 let historical_replacement = prepared_resume_after(
3738 "resume-historical-replacement",
3739 |journal| {
3740 journal
3741 .create_box(
3742 "t1",
3743 "box",
3744 BoxOwner::Controller,
3745 BoxContent::text("before"),
3746 )
3747 .unwrap()
3748 },
3749 |journal, box_id| {
3750 journal
3751 .record(
3752 "t2",
3753 EventKind::CanonicalUpdated {
3754 box_id,
3755 content: BoxContent::text("after"),
3756 },
3757 )
3758 .unwrap();
3759 },
3760 );
3761 assert_eq!(
3762 historical_replacement.thread_reset_reason.as_deref(),
3763 Some("canonical_state_replacement")
3764 );
3765
3766 let projection_reordered = prepared_resume_after(
3767 "resume-projection-reordered",
3768 |journal| {
3769 journal
3770 .apply_tool_slots_with_layout(
3771 "t1",
3772 "tool",
3773 vec![
3774 ToolSlotInput {
3775 slot: "a".into(),
3776 name: "a".into(),
3777 content: BoxContent::text("a"),
3778 retired: false,
3779 },
3780 ToolSlotInput {
3781 slot: "b".into(),
3782 name: "b".into(),
3783 content: BoxContent::text("b"),
3784 retired: false,
3785 },
3786 ],
3787 &["a".into(), "b".into()],
3788 )
3789 .unwrap();
3790 },
3791 |journal, ()| {
3792 journal
3793 .apply_tool_slots_with_layout(
3794 "t2",
3795 "tool",
3796 vec![
3797 ToolSlotInput {
3798 slot: "a".into(),
3799 name: "a".into(),
3800 content: BoxContent::text("a"),
3801 retired: false,
3802 },
3803 ToolSlotInput {
3804 slot: "b".into(),
3805 name: "b".into(),
3806 content: BoxContent::text("b"),
3807 retired: false,
3808 },
3809 ],
3810 &["b".into(), "a".into()],
3811 )
3812 .unwrap();
3813 },
3814 );
3815 assert_eq!(
3816 projection_reordered.thread_reset_reason.as_deref(),
3817 Some("projection_reordered")
3818 );
3819 }
3820
3821 #[test]
3822 fn provider_resume_marker_wrapper_preserves_growth_and_rewritten_epochs() {
3823 let path = path("provider-resume-markers");
3824 let mut journal = SessionJournal::create(
3825 &path,
3826 SessionMetadata {
3827 effective_context_tokens: 100_000,
3828 ..metadata()
3829 },
3830 )
3831 .unwrap();
3832 let first_box = journal
3833 .create_box(
3834 "t1",
3835 "System prompt",
3836 BoxOwner::System,
3837 BoxContent::text("small"),
3838 )
3839 .unwrap();
3840 let initial = journal
3841 .prepare_provider_projection("t2", &[], "m1", None)
3842 .unwrap();
3843 assert!(
3844 !initial
3845 .projection
3846 .render()
3847 .contains("[current context size:")
3848 );
3849 let mut synchronized_after = journal
3850 .record(
3851 "t3",
3852 EventKind::ProviderInputSubmitted {
3853 round: 1,
3854 context: provider_context(initial.projection.render()),
3855 transport_input_hash: None,
3856 transport_input_bytes: None,
3857 thread_action: None,
3858 thread_reset_reason: None,
3859 cacheable_prefix_bytes: initial.cacheable_prefix_bytes,
3860 material_fingerprint: "m1".into(),
3861 cache_expectation: initial.expectation.label().into(),
3862 planned_invalidation_reason: None,
3863 },
3864 )
3865 .unwrap();
3866
3867 journal
3868 .create_box(
3869 "t4",
3870 "large result",
3871 BoxOwner::Controller,
3872 BoxContent::text("x".repeat(130_000)),
3873 )
3874 .unwrap();
3875 let first = journal
3876 .prepare_provider_resume_markers("t5", synchronized_after, None)
3877 .unwrap();
3878 let (marker_event, marked_tokens) = journal
3879 .state()
3880 .events
3881 .iter()
3882 .rev()
3883 .find_map(|event| match &event.kind {
3884 EventKind::ContextSizeMarked { estimated_tokens } => {
3885 Some((event.id, *estimated_tokens))
3886 }
3887 _ => None,
3888 })
3889 .unwrap();
3890 let expected = format!(
3891 "[current context size: approximately {marked_tokens} tokens | session limit: 70000 tokens]"
3892 );
3893 assert_eq!(first.as_slice(), std::slice::from_ref(&expected));
3894 assert_eq!(
3895 journal
3896 .state()
3897 .projection()
3898 .items
3899 .into_iter()
3900 .find(|item| item.event_id == marker_event)
3901 .unwrap()
3902 .text,
3903 expected
3904 );
3905
3906 synchronized_after = journal.state().events.last().unwrap().id;
3907 journal
3908 .create_box(
3909 "t6",
3910 "small growth",
3911 BoxOwner::Controller,
3912 BoxContent::text("y".repeat(20_000)),
3913 )
3914 .unwrap();
3915 assert!(
3916 journal
3917 .prepare_provider_resume_markers("t7", synchronized_after, None)
3918 .unwrap()
3919 .is_empty()
3920 );
3921
3922 synchronized_after = journal.state().events.last().unwrap().id;
3923 journal
3924 .create_box(
3925 "t8",
3926 "large growth",
3927 BoxOwner::Controller,
3928 BoxContent::text("z".repeat(40_000)),
3929 )
3930 .unwrap();
3931 let second = journal
3932 .prepare_provider_resume_markers("t9", synchronized_after, None)
3933 .unwrap();
3934 assert_eq!(second.len(), 1);
3935 assert!(second[0].starts_with("[current context size: approximately "));
3936
3937 synchronized_after = journal.state().events.last().unwrap().id;
3938 journal
3939 .update_box(
3940 "t10",
3941 first_box,
3942 BoxContent::text("rewritten system prompt"),
3943 )
3944 .unwrap();
3945 let stale = journal
3946 .prepare_provider_resume_markers("t11", synchronized_after, None)
3947 .unwrap();
3948 assert_eq!(stale, ["[new stale boxes: 1]"]);
3949 assert_eq!(
3950 journal
3951 .state()
3952 .events
3953 .iter()
3954 .filter(|event| matches!(event.kind, EventKind::ProjectionMarkersReset))
3955 .count(),
3956 1,
3957 );
3958 std::fs::remove_file(path).unwrap();
3959 }
3960
3961 fn provider_context(input: String) -> ProviderContext {
3962 ProviderContext {
3963 input,
3964 provider: "codex".into(),
3965 model: "test".into(),
3966 reasoning_effort: "xhigh".into(),
3967 base_instructions: Some("base".into()),
3968 developer_instructions: Some(String::new()),
3969 tools: Vec::new(),
3970 }
3971 }
3972
3973 #[test]
3974 fn shared_pending_and_box_identity_space_never_overlaps() {
3975 let path = path("pending");
3976 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
3977 let first = journal.allocate_pending_node("t1").unwrap();
3978 let box_id = journal
3979 .create_box("t2", "box", BoxOwner::Kennedy, BoxContent::text("hello"))
3980 .unwrap();
3981 let object = journal
3982 .stage_object(
3983 "t3",
3984 "application/octet-stream",
3985 None,
3986 Value::Null,
3987 b"\0binary\xff",
3988 )
3989 .unwrap();
3990 assert_eq!(first.to_string(), "pending:1");
3991 assert_eq!(box_id, BoxId(2));
3992 assert_eq!(object.to_string(), "pending:3");
3993 assert_eq!(journal.read_object(&object).unwrap(), b"\0binary\xff");
3994 let stored = journal.session_log();
3995 assert!(!stored.events[0].text.contains("pending_id"));
3996 assert!(!stored.events[1].text.contains("box_id"));
3997 assert!(!stored.events[2].text.contains("pending_id"));
3998 drop(journal);
3999 let mut reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
4000 assert_eq!(reopened.read_object(&object).unwrap(), b"\0binary\xff");
4001 assert_eq!(reopened.state().next_id, 4);
4002 std::fs::remove_file(path).unwrap();
4003 }
4004
4005 #[test]
4006 fn kennedy_tool_completion_is_validated_before_storage_seals() {
4007 let path = path("seal-tool");
4008 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4009 journal
4010 .record(
4011 "t1",
4012 EventKind::ToolInvoked {
4013 tool_instance: "CreateNode:1".into(),
4014 tool_name: "CreateNode".into(),
4015 arguments: json!({}),
4016 invocation_id: None,
4017 },
4018 )
4019 .unwrap();
4020 assert!(journal.seal().is_err());
4021 journal
4022 .record(
4023 "t2",
4024 EventKind::ToolCompleted {
4025 tool_instance: "call_ktool".into(),
4026 tool_name: "call_ktool".into(),
4027 outcome: json!({"ok":true}),
4028 invocation_id: None,
4029 },
4030 )
4031 .unwrap();
4032 journal.seal().unwrap();
4033 assert!(journal.is_sealed());
4034 std::fs::remove_file(path).unwrap();
4035 }
4036
4037 #[test]
4038 fn interrupted_tools_are_recovered_by_identity_without_losing_legacy_journals() {
4039 let path = path("repair-tools");
4040 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4041 journal
4042 .record(
4043 "t1",
4044 EventKind::ToolInvoked {
4045 tool_instance: "WebSearch:legacy".into(),
4046 tool_name: "WebSearch".into(),
4047 arguments: json!({"question":"legacy"}),
4048 invocation_id: None,
4049 },
4050 )
4051 .unwrap();
4052 journal
4053 .record(
4054 "t2",
4055 EventKind::ToolInvoked {
4056 tool_instance: "WebSearch:new".into(),
4057 tool_name: "WebSearch".into(),
4058 arguments: json!({"question":"identified"}),
4059 invocation_id: Some("call-1".into()),
4060 },
4061 )
4062 .unwrap();
4063 assert!(journal.seal().is_err());
4064
4065 let repaired = journal.repair_unfinished_tools("recovery").unwrap();
4066 assert_eq!(repaired.len(), 2);
4067 assert!(
4068 journal
4069 .state()
4070 .events
4071 .iter()
4072 .rev()
4073 .take(2)
4074 .all(|event| matches!(
4075 &event.kind,
4076 EventKind::ToolCompleted { outcome, .. }
4077 if outcome.get("recovered").and_then(Value::as_bool) == Some(true)
4078 ))
4079 );
4080 journal.seal().unwrap();
4081 assert!(journal.is_sealed());
4082 std::fs::remove_file(path).unwrap();
4083 }
4084
4085 #[test]
4086 fn identified_tool_completions_can_arrive_out_of_order() {
4087 let path = path("tool-identity");
4088 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4089 for id in ["call-1", "call-2"] {
4090 journal
4091 .record(
4092 id,
4093 EventKind::ToolInvoked {
4094 tool_instance: format!("WebSearch:{id}"),
4095 tool_name: "WebSearch".into(),
4096 arguments: json!({"question":id}),
4097 invocation_id: Some(id.into()),
4098 },
4099 )
4100 .unwrap();
4101 }
4102 for id in ["call-1", "call-2"] {
4103 journal
4104 .record(
4105 format!("{id}-complete"),
4106 EventKind::ToolCompleted {
4107 tool_instance: format!("WebSearch:{id}"),
4108 tool_name: "WebSearch".into(),
4109 outcome: json!({"ok":true}),
4110 invocation_id: Some(id.into()),
4111 },
4112 )
4113 .unwrap();
4114 }
4115 journal.seal().unwrap();
4116 std::fs::remove_file(path).unwrap();
4117 }
4118
4119 #[test]
4120 fn invalid_box_operations_do_not_poison_the_append_only_journal() {
4121 let path = path("invalid-box-operation");
4122 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4123 let box_id = journal
4124 .create_box(
4125 "t1",
4126 "valid",
4127 BoxOwner::Controller,
4128 BoxContent::text("canonical"),
4129 )
4130 .unwrap();
4131 let valid_length = std::fs::metadata(&path).unwrap().len();
4132
4133 let result = journal.dehydrate_boxes("t2", &[box_id, BoxId(97)]);
4134 assert_eq!(result.unwrap_err().to_string(), "box 97 does not exist");
4135 assert_eq!(std::fs::metadata(&path).unwrap().len(), valid_length);
4136 assert!(matches!(
4137 journal.state().box_state(box_id).unwrap().representation,
4138 Representation::Hydrated { .. }
4139 ));
4140
4141 for result in [
4142 journal.summarize_box("t3", BoxId(97), "summary"),
4143 journal.rehydrate_box("t4", BoxId(97)),
4144 journal.retire_box("t5", BoxId(97)),
4145 ] {
4146 assert_eq!(result.unwrap_err().to_string(), "box 97 does not exist");
4147 assert_eq!(std::fs::metadata(&path).unwrap().len(), valid_length);
4148 }
4149
4150 drop(journal);
4151 let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
4152 assert_eq!(
4153 reopened
4154 .state()
4155 .box_state(box_id)
4156 .unwrap()
4157 .canonical
4158 .content,
4159 BoxContent::text("canonical")
4160 );
4161 std::fs::remove_file(path).unwrap();
4162 }
4163
4164 #[test]
4165 fn multiple_boxes_dehydrate_in_one_replayable_batch() {
4166 let path = path("dehydrate-boxes");
4167 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4168 let boxes = ["first", "second", "third"]
4169 .into_iter()
4170 .map(|text| {
4171 journal
4172 .create_box("t1", text, BoxOwner::Controller, BoxContent::text(text))
4173 .unwrap()
4174 })
4175 .collect::<Vec<_>>();
4176
4177 let event_ids = journal
4178 .dehydrate_boxes("t2", &[boxes[0], boxes[2]])
4179 .unwrap();
4180 assert_eq!(event_ids, [EventId(4), EventId(5)]);
4181 assert!(matches!(
4182 journal.state().box_state(boxes[0]).unwrap().representation,
4183 Representation::Dehydrated { .. }
4184 ));
4185 assert!(matches!(
4186 journal.state().box_state(boxes[1]).unwrap().representation,
4187 Representation::Hydrated { .. }
4188 ));
4189 assert!(matches!(
4190 journal.state().box_state(boxes[2]).unwrap().representation,
4191 Representation::Dehydrated { .. }
4192 ));
4193
4194 let valid_length = std::fs::metadata(&path).unwrap().len();
4195 assert_eq!(
4196 journal.dehydrate_boxes("t3", &[]).unwrap_err().to_string(),
4197 "at least one box must be selected for dehydration"
4198 );
4199 assert_eq!(
4200 journal
4201 .dehydrate_boxes("t3", &[boxes[1], boxes[1]])
4202 .unwrap_err()
4203 .to_string(),
4204 "box dehydration cannot contain duplicate box IDs"
4205 );
4206 assert_eq!(std::fs::metadata(&path).unwrap().len(), valid_length);
4207
4208 drop(journal);
4209 let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
4210 assert!(matches!(
4211 reopened.state().box_state(boxes[0]).unwrap().representation,
4212 Representation::Dehydrated { .. }
4213 ));
4214 assert!(matches!(
4215 reopened.state().box_state(boxes[2]).unwrap().representation,
4216 Representation::Dehydrated { .. }
4217 ));
4218 std::fs::remove_file(path).unwrap();
4219 }
4220
4221 #[test]
4222 fn status_estimates_the_context_with_every_active_box_hydrated() {
4223 let path = path("fully-hydrated-estimate");
4224 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4225 let box_id = journal
4226 .create_box(
4227 "t1",
4228 "large",
4229 BoxOwner::Controller,
4230 BoxContent::text("canonical material ".repeat(1_000)),
4231 )
4232 .unwrap();
4233 journal.dehydrate_boxes("t2", &[box_id]).unwrap();
4234
4235 let status = journal.state().projection().status;
4236 assert!(
4237 status.fully_hydrated_context_tokens > status.current_context_tokens,
4238 "the hydrated estimate should include the canonical body"
4239 );
4240 std::fs::remove_file(path).unwrap();
4241 }
4242
4243 #[test]
4244 fn provider_measurements_recalibrate_the_matching_manifest_then_track_deltas() {
4245 let path = path("calibration");
4246 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4247 journal
4248 .create_box(
4249 "t1",
4250 "system",
4251 BoxOwner::System,
4252 BoxContent::text("baseline provider content"),
4253 )
4254 .unwrap();
4255 let submitted = journal.state().projection();
4256 journal
4257 .record(
4258 "t2",
4259 EventKind::InferenceSubmitted {
4260 manifest_hash: "manifest-1".into(),
4261 estimated_input_tokens: submitted.estimated_tokens,
4262 raw_estimated_input_tokens: Some(submitted.raw_estimated_tokens),
4263 },
4264 )
4265 .unwrap();
4266 let tool_box = journal
4267 .create_box(
4268 "t3",
4269 "tool result",
4270 BoxOwner::Controller,
4271 BoxContent::text("x".repeat(3_000)),
4272 )
4273 .unwrap();
4274 let measured_context = journal.state().projection();
4275 journal
4276 .record(
4277 "t4",
4278 EventKind::ProviderReceipt {
4279 manifest_hash: "manifest-1".into(),
4280 input_tokens: Some(777),
4281 output_tokens: Some(3),
4282 context_bytes: Some(measured_context.context_bytes),
4283 raw_context_tokens: Some(measured_context.raw_estimated_tokens),
4284 provider_data: Value::Null,
4285 },
4286 )
4287 .unwrap();
4288 assert_eq!(journal.state().projection().estimated_tokens, 777);
4289 journal
4290 .create_box(
4291 "t5",
4292 "new material",
4293 BoxOwner::User,
4294 BoxContent::text("x".repeat(300)),
4295 )
4296 .unwrap();
4297 let expanded = journal.state().projection();
4298 assert_eq!(
4299 expanded.estimated_tokens,
4300 777 + expanded
4301 .context_bytes
4302 .abs_diff(measured_context.context_bytes)
4303 / ESTIMATED_BYTES_PER_TOKEN
4304 );
4305 assert!(expanded.raw_estimated_tokens > submitted.raw_estimated_tokens);
4306 journal.dehydrate_boxes("t6", &[tool_box]).unwrap();
4307 let shrunken = journal.state().projection();
4308 assert_eq!(
4309 shrunken.estimated_tokens,
4310 777_u64.saturating_sub(
4311 measured_context
4312 .context_bytes
4313 .abs_diff(shrunken.context_bytes)
4314 / ESTIMATED_BYTES_PER_TOKEN
4315 )
4316 );
4317 std::fs::remove_file(path).unwrap();
4318 }
4319
4320 #[test]
4321 fn session_status_keeps_provider_usage_categories_exact_and_exclusive() {
4322 let path = path("session-status");
4323 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4324 journal
4325 .create_box("t1", "system", BoxOwner::System, BoxContent::text("é"))
4326 .unwrap();
4327 let before_usage = journal.state().projection();
4328 assert_eq!(
4329 before_usage.raw_estimated_tokens,
4330 before_usage
4331 .context_bytes
4332 .div_ceil(ESTIMATED_BYTES_PER_TOKEN)
4333 );
4334 journal
4335 .record(
4336 "t2",
4337 EventKind::ProviderReceipt {
4338 manifest_hash: "manifest-1".into(),
4339 input_tokens: Some(100),
4340 output_tokens: Some(20),
4341 context_bytes: Some(before_usage.context_bytes),
4342 raw_context_tokens: Some(before_usage.raw_estimated_tokens),
4343 provider_data: json!({
4344 "usageIsDelta":true,
4345 "cachedInputTokens":40,
4346 "nonCachedInputTokens":60,
4347 "thinkingTokens":8,
4348 "outputTokens":12,
4349 "estimatedCostUsdNanos":12345
4350 }),
4351 },
4352 )
4353 .unwrap();
4354 let second_anchor = journal.state().projection();
4355 journal
4356 .record(
4357 "t3",
4358 EventKind::ProviderReceipt {
4359 manifest_hash: "manifest-1".into(),
4360 input_tokens: Some(120),
4361 output_tokens: Some(5),
4362 context_bytes: Some(second_anchor.context_bytes),
4363 raw_context_tokens: Some(second_anchor.raw_estimated_tokens),
4364 provider_data: json!({
4365 "usageIsDelta":true,
4366 "cachedInputTokens":10,
4367 "nonCachedInputTokens":10,
4368 "thinkingTokens":2,
4369 "outputTokens":3,
4370 "estimatedCostUsdNanos":6789
4371 }),
4372 },
4373 )
4374 .unwrap();
4375 let projection = journal.state().projection();
4376 assert_eq!(
4377 projection.status,
4378 SessionStatus {
4379 current_context_tokens: 120,
4380 fully_hydrated_context_tokens: 120,
4381 context_limit_tokens: 700,
4382 current_context_bytes: projection.context_bytes,
4383 cached_input_tokens: 50,
4384 non_cached_input_tokens: 70,
4385 thinking_tokens: 10,
4386 output_tokens: 15,
4387 estimated_cost_usd_nanos: 19_134,
4388 unpriced_provider_calls: 0,
4389 }
4390 );
4391 let archive: Value = serde_json::from_slice(&journal.archive_bytes().unwrap()).unwrap();
4392 assert_eq!(archive["context"]["status"]["cachedInputTokens"], 50);
4393 assert!(
4394 archive["chatendText"]
4395 .as_str()
4396 .unwrap()
4397 .ends_with(archive["context"]["footer"].as_str().unwrap())
4398 );
4399 std::fs::remove_file(path).unwrap();
4400 }
4401
4402 fn compatibility_cost(
4403 model: &str,
4404 metering: &ProviderMetering,
4405 ) -> Option<ProviderCostEstimate> {
4406 let ProviderMetering::Tokens(usage) = metering else {
4407 return None;
4408 };
4409 let base = match model {
4410 "gpt-5.6-sol" => 1_000,
4411 "gemini-3.1-pro-preview" => 2_000,
4412 _ => return None,
4413 };
4414 Some(ProviderCostEstimate {
4415 usd_nanos: base + usage.input_tokens,
4416 accuracy: json!("exact"),
4417 pricing_version: "test-prices".into(),
4418 })
4419 }
4420
4421 #[test]
4422 fn legacy_provider_costs_are_reconstructed_without_rewriting_history() {
4423 let path = path("legacy-provider-costs");
4424 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4425 journal
4426 .record(
4427 "t1",
4428 EventKind::ProviderReceipt {
4429 manifest_hash: "top".into(),
4430 input_tokens: Some(10),
4431 output_tokens: Some(1),
4432 context_bytes: None,
4433 raw_context_tokens: None,
4434 provider_data: json!({
4435 "usageIsDelta":true,
4436 "nonCachedInputTokens":10,
4437 "cachedInputTokens":0,
4438 "thinkingTokens":0,
4439 "outputTokens":1
4440 }),
4441 },
4442 )
4443 .unwrap();
4444 journal
4445 .record(
4446 "t2",
4447 EventKind::ToolInvoked {
4448 tool_instance: "RunSubagent:1".into(),
4449 tool_name: "RunSubagent".into(),
4450 arguments: json!({"model":"codex/gpt-5.6-sol"}),
4451 invocation_id: Some("subagent-1".into()),
4452 },
4453 )
4454 .unwrap();
4455 journal
4456 .record(
4457 "t3",
4458 EventKind::Note {
4459 label: "subagent_started".into(),
4460 value: json!({"providerModel":"gpt-5.6-sol"}),
4461 },
4462 )
4463 .unwrap();
4464 journal
4465 .record(
4466 "t4",
4467 EventKind::ProviderReceipt {
4468 manifest_hash: "subagent".into(),
4469 input_tokens: None,
4470 output_tokens: None,
4471 context_bytes: None,
4472 raw_context_tokens: None,
4473 provider_data: json!({
4474 "source":"subagent",
4475 "usageIsDelta":true,
4476 "nonCachedInputTokens":20,
4477 "cachedInputTokens":0,
4478 "thinkingTokens":0,
4479 "outputTokens":1
4480 }),
4481 },
4482 )
4483 .unwrap();
4484 journal
4485 .record(
4486 "t5",
4487 EventKind::ToolInvoked {
4488 tool_instance: "AnnotateMedia:1".into(),
4489 tool_name: "AnnotateMedia".into(),
4490 arguments: json!({"model":"gemini-3.1-pro-preview"}),
4491 invocation_id: Some("media-1".into()),
4492 },
4493 )
4494 .unwrap();
4495 journal
4496 .record(
4497 "t6",
4498 EventKind::ProviderReceipt {
4499 manifest_hash: "media".into(),
4500 input_tokens: None,
4501 output_tokens: None,
4502 context_bytes: None,
4503 raw_context_tokens: None,
4504 provider_data: json!({
4505 "source":"media_annotation",
4506 "usageIsDelta":true,
4507 "nonCachedInputTokens":30,
4508 "cachedInputTokens":0,
4509 "thinkingTokens":0,
4510 "outputTokens":1
4511 }),
4512 },
4513 )
4514 .unwrap();
4515 journal
4516 .record(
4517 "t7",
4518 EventKind::ToolCompleted {
4519 tool_instance: "AnnotateMedia:1".into(),
4520 tool_name: "AnnotateMedia".into(),
4521 outcome: json!({"ok":true}),
4522 invocation_id: Some("media-1".into()),
4523 },
4524 )
4525 .unwrap();
4526 journal
4527 .record(
4528 "t8",
4529 EventKind::ToolCompleted {
4530 tool_instance: "RunSubagent:1".into(),
4531 tool_name: "RunSubagent".into(),
4532 outcome: json!({"ok":true}),
4533 invocation_id: Some("subagent-1".into()),
4534 },
4535 )
4536 .unwrap();
4537 journal
4538 .record(
4539 "t9",
4540 EventKind::ProviderReceipt {
4541 manifest_hash: "unknown".into(),
4542 input_tokens: None,
4543 output_tokens: None,
4544 context_bytes: None,
4545 raw_context_tokens: None,
4546 provider_data: json!({
4547 "source":"web_search",
4548 "usageIsDelta":true,
4549 "nonCachedInputTokens":40,
4550 "cachedInputTokens":0,
4551 "thinkingTokens":0,
4552 "outputTokens":1
4553 }),
4554 },
4555 )
4556 .unwrap();
4557 drop(journal);
4558
4559 let compatible = SessionJournal::open_with_metadata_and_provider_costs(
4560 &path,
4561 metadata(),
4562 Some("gpt-5.6-sol"),
4563 Some(compatibility_cost),
4564 )
4565 .unwrap();
4566 let status = &compatible.state().projection().status;
4567 assert_eq!(status.estimated_cost_usd_nanos, 4_060);
4568 assert_eq!(status.unpriced_provider_calls, 1);
4569 let inferred_models = compatible
4570 .state()
4571 .events
4572 .iter()
4573 .filter_map(|event| match &event.kind {
4574 EventKind::ProviderReceipt { provider_data, .. } => provider_data
4575 .get("providerModel")
4576 .and_then(Value::as_str)
4577 .map(str::to_owned),
4578 _ => None,
4579 })
4580 .collect::<Vec<_>>();
4581 assert_eq!(
4582 inferred_models,
4583 vec!["gpt-5.6-sol", "gpt-5.6-sol", "gemini-3.1-pro-preview"]
4584 );
4585 drop(compatible);
4586
4587 let unchanged = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
4588 assert_eq!(
4589 unchanged
4590 .state()
4591 .projection()
4592 .status
4593 .estimated_cost_usd_nanos,
4594 0
4595 );
4596 assert_eq!(
4597 unchanged
4598 .state()
4599 .projection()
4600 .status
4601 .unpriced_provider_calls,
4602 4
4603 );
4604 let archive: Value = serde_json::from_slice(&unchanged.archive_bytes().unwrap()).unwrap();
4605 let archive_summary = legacy_provider_cost_summary_for_archive(
4606 &archive,
4607 Some("gpt-5.6-sol"),
4608 compatibility_cost,
4609 )
4610 .unwrap();
4611 assert_eq!(archive_summary.estimated_cost_usd_nanos, 4_060);
4612 assert_eq!(archive_summary.unpriced_provider_calls, 1);
4613 assert_eq!(archive["context"]["status"]["estimatedCostUsdNanos"], 0);
4614 assert_eq!(archive["context"]["status"]["unpricedProviderCalls"], 4);
4615 drop(unchanged);
4616 std::fs::remove_file(path).unwrap();
4617 }
4618
4619 #[test]
4620 fn legacy_provider_receipts_without_a_raw_context_anchor_remain_readable() {
4621 let receipt: EventKind = serde_json::from_value(json!({
4622 "type":"provider_receipt",
4623 "manifest_hash":"legacy-manifest",
4624 "input_tokens":123,
4625 "output_tokens":4,
4626 "provider_data":null
4627 }))
4628 .unwrap();
4629 assert!(matches!(
4630 receipt,
4631 EventKind::ProviderReceipt {
4632 context_bytes: None,
4633 raw_context_tokens: None,
4634 ..
4635 }
4636 ));
4637 }
4638
4639 #[test]
4640 fn stateful_tool_slots_are_batched_append_only_and_do_not_see_summaries() {
4641 let path = path("slots");
4642 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4643 journal
4644 .apply_tool_slots(
4645 "t1",
4646 "rust-1",
4647 vec![
4648 ToolSlotInput {
4649 slot: "a.rs".into(),
4650 name: "a.rs".into(),
4651 content: BoxContent::text("a"),
4652 retired: false,
4653 },
4654 ToolSlotInput {
4655 slot: "b.rs".into(),
4656 name: "b.rs".into(),
4657 content: BoxContent::text("b"),
4658 retired: false,
4659 },
4660 ],
4661 )
4662 .unwrap();
4663 let a = journal.state().tools["rust-1"].slots[0].box_id;
4664 journal.summarize_box("t2", a, "Kennedy summary").unwrap();
4665 let before = journal.state().clone();
4666 assert!(
4667 journal
4668 .apply_tool_slots(
4669 "t3",
4670 "rust-1",
4671 vec![ToolSlotInput {
4672 slot: "b.rs".into(),
4673 name: "b.rs".into(),
4674 content: BoxContent::text("b"),
4675 retired: false,
4676 }]
4677 )
4678 .is_err()
4679 );
4680 assert_eq!(journal.state(), &before);
4681 journal
4682 .apply_tool_slots(
4683 "t4",
4684 "rust-1",
4685 vec![
4686 ToolSlotInput {
4687 slot: "a.rs".into(),
4688 name: "a.rs".into(),
4689 content: BoxContent::text("a2"),
4690 retired: false,
4691 },
4692 ToolSlotInput {
4693 slot: "b.rs".into(),
4694 name: "b.rs".into(),
4695 content: BoxContent::text("b"),
4696 retired: true,
4697 },
4698 ToolSlotInput {
4699 slot: "c.rs".into(),
4700 name: "c.rs".into(),
4701 content: BoxContent::text("c"),
4702 retired: false,
4703 },
4704 ],
4705 )
4706 .unwrap();
4707 let a_state = journal.state().box_state(a).unwrap();
4708 assert_eq!(a_state.canonical.content.text, "a2");
4709 assert!(matches!(
4710 a_state.representation,
4711 Representation::Summarized { .. }
4712 ));
4713 drop(journal);
4714 let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
4715 assert_eq!(reopened.state().tools["rust-1"].slots.len(), 3);
4716 assert!(reopened.state().tools["rust-1"].slots[1].retired);
4717 std::fs::remove_file(path).unwrap();
4718 }
4719
4720 #[test]
4721 fn tool_layout_orders_current_boxes_without_changing_their_identities() {
4722 let path = path("tool-layout");
4723 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4724 journal
4725 .apply_tool_slots_with_layout(
4726 "t1",
4727 "kweb",
4728 vec![
4729 ToolSlotInput {
4730 slot: "active".into(),
4731 name: "Active node".into(),
4732 content: BoxContent::text("ACTIVE NODE"),
4733 retired: false,
4734 },
4735 ToolSlotInput {
4736 slot: "direct".into(),
4737 name: "Direct node".into(),
4738 content: BoxContent::text("DIRECT NODE"),
4739 retired: false,
4740 },
4741 ],
4742 &["direct".into(), "active".into()],
4743 )
4744 .unwrap();
4745 let tool = &journal.state().tools["kweb"];
4746 let active_id = tool.slots[0].box_id;
4747 let direct_id = tool.slots[1].box_id;
4748 let projected_items = journal.state().projection().items;
4749 let rendered = projected_items
4750 .iter()
4751 .map(|item| item.text.as_str())
4752 .collect::<Vec<_>>()
4753 .join("\n\n");
4754 assert!(rendered.find("DIRECT NODE") < rendered.find("ACTIVE NODE"));
4755 assert_eq!(
4756 journal.state().tool_layouts["kweb"],
4757 vec![direct_id, active_id]
4758 );
4759 drop(journal);
4760 let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
4761 assert_eq!(reopened.state().projection().items, projected_items);
4762 std::fs::remove_file(path).unwrap();
4763 }
4764
4765 #[test]
4766 fn ingress_attempt_reset_restores_the_initial_provider_visible_state() {
4767 let path = path("ingress-attempt-reset");
4768 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4769 let baseline_box = journal
4770 .create_box(
4771 "t1",
4772 "source",
4773 BoxOwner::User,
4774 BoxContent::text("original source"),
4775 )
4776 .unwrap();
4777 journal
4778 .record(
4779 "t2",
4780 EventKind::SourceTerminated {
4781 reason: "history_ingress".into(),
4782 },
4783 )
4784 .unwrap();
4785 journal
4786 .record("t3", EventKind::HistoryIngressStarted)
4787 .unwrap();
4788 let baseline = journal.state().projection().render();
4789
4790 journal
4791 .update_box(
4792 "t4",
4793 baseline_box,
4794 BoxContent::text("failed attempt update"),
4795 )
4796 .unwrap();
4797 journal
4798 .create_box(
4799 "t5",
4800 "failed attempt",
4801 BoxOwner::Kennedy,
4802 BoxContent::text("discard me"),
4803 )
4804 .unwrap();
4805 journal.reset_history_ingress_attempt("t6").unwrap();
4806
4807 assert_eq!(journal.state().projection().render(), baseline);
4808 assert!(journal.state().current_ingress_attempt_events().is_empty());
4809 drop(journal);
4810 let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
4811 assert_eq!(reopened.state().projection().render(), baseline);
4812 std::fs::remove_file(path).unwrap();
4813 }
4814
4815 #[test]
4816 fn ingress_time_markers_are_sparse_resettable_and_system_bracketed() {
4817 let path = path("ingress-time-markers");
4818 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4819 journal
4820 .create_box(
4821 "t1",
4822 "source",
4823 BoxOwner::User,
4824 BoxContent::text("short source"),
4825 )
4826 .unwrap();
4827 journal
4828 .record(
4829 "t2",
4830 EventKind::SourceTerminated {
4831 reason: "history_ingress".into(),
4832 },
4833 )
4834 .unwrap();
4835 journal
4836 .record("t3", EventKind::HistoryIngressStarted)
4837 .unwrap();
4838
4839 journal
4840 .prepare_provider_projection_with_ingress_time("t4", &[], "m1", None, 2_700, false)
4841 .unwrap();
4842 let mut synchronized_after = journal.state().events.last().unwrap().id;
4843 let twenty = journal
4844 .prepare_provider_resume_markers("t5", synchronized_after, Some((1_200, false)))
4845 .unwrap();
4846 assert_eq!(twenty, ["[ingress time remaining: 1200 seconds]"]);
4847 synchronized_after = journal.state().events.last().unwrap().id;
4848 assert!(
4849 journal
4850 .prepare_provider_resume_markers("t6", synchronized_after, Some((1_199, false)),)
4851 .unwrap()
4852 .is_empty()
4853 );
4854 let ten = journal
4855 .prepare_provider_resume_markers("t7", synchronized_after, Some((600, false)))
4856 .unwrap();
4857 assert_eq!(
4858 ten,
4859 [
4860 "[ingress time remaining: 600 seconds. If you do not call EndSession before time expires, the ingress will fail and all Kmap updates will be lost.]"
4861 ],
4862 );
4863 let markers = journal
4864 .state()
4865 .projection()
4866 .items
4867 .into_iter()
4868 .filter(|item| item.text.contains("ingress time remaining"))
4869 .map(|item| item.text)
4870 .collect::<Vec<_>>();
4871 assert_eq!(markers.len(), 3);
4872 assert_eq!(markers[0], "[ingress time remaining: 2700 seconds]");
4873 assert_eq!(markers[1], "[ingress time remaining: 1200 seconds]");
4874 assert_eq!(
4875 markers[2],
4876 "[ingress time remaining: 600 seconds. If you do not call EndSession before time expires, the ingress will fail and all Kmap updates will be lost.]"
4877 );
4878 assert!(should_append_ingress_time_marker(
4879 IngressTimeMarkerState {
4880 any: true,
4881 at_or_below_thirty_minutes: true,
4882 at_or_below_ten_minutes: true,
4883 },
4884 500,
4885 true,
4886 ));
4887
4888 journal.reset_history_ingress_attempt("t8").unwrap();
4889 let retry = journal
4890 .prepare_provider_projection_with_ingress_time("t9", &[], "m1", None, 2_700, true)
4891 .unwrap();
4892 assert!(retry.projection.render().contains(
4893 "[Previous attempt ran out of time; please budget your ingress time carefully. Ingress time remaining: 2700 seconds]"
4894 ));
4895 assert!(
4896 !retry
4897 .projection
4898 .render()
4899 .contains("[ingress time remaining: 600 seconds. If you do not call EndSession")
4900 );
4901 std::fs::remove_file(path).unwrap();
4902 }
4903
4904 #[test]
4905 fn limits_use_exact_floor_percentages() {
4906 let state = Chatend::opened(SessionMetadata {
4907 effective_context_tokens: 101,
4908 ..metadata()
4909 });
4910 assert_eq!(state.live_context_limit(), 70);
4911 assert_eq!(state.forced_ingress_context_limit(), 75);
4912 assert_eq!(state.ingress_initial_context_limit(), 75);
4913 assert_eq!(state.ingress_context_limit(), 90);
4914 assert_eq!(state.active_context_limit(), 70);
4915 for kind in [
4916 SessionKind::SelfTime,
4917 SessionKind::HistoryIngress,
4918 SessionKind::AudioIngress,
4919 ] {
4920 let writer = Chatend::opened(SessionMetadata {
4921 kind,
4922 effective_context_tokens: 101,
4923 ..metadata()
4924 });
4925 assert_eq!(writer.active_context_limit(), 90);
4926 }
4927 for kind in [SessionKind::Telegram, SessionKind::Other("test".into())] {
4928 let ordinary = Chatend::opened(SessionMetadata {
4929 kind,
4930 effective_context_tokens: 101,
4931 ..metadata()
4932 });
4933 assert_eq!(ordinary.active_context_limit(), 70);
4934 }
4935 assert_eq!(estimate_tokens("1234"), 1);
4936 }
4937
4938 #[test]
4939 fn new_box_projection_preview_matches_the_committed_projection() {
4940 let path = path("new-box-preview");
4941 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4942 let boxes = vec![
4943 (
4944 "User message".into(),
4945 BoxOwner::User,
4946 BoxContent::text("prospective user text"),
4947 ),
4948 (
4949 "attachment".into(),
4950 BoxOwner::User,
4951 BoxContent::text("prospective attachment text"),
4952 ),
4953 ];
4954 let preview = journal
4955 .state()
4956 .projection_with_new_boxes_at("t1", &boxes)
4957 .unwrap();
4958 for (name, owner, content) in boxes {
4959 journal.create_box("t1", name, owner, content).unwrap();
4960 }
4961 assert_eq!(journal.state().projection(), preview);
4962 std::fs::remove_file(path).unwrap();
4963 }
4964
4965 #[test]
4966 fn box_representation_preview_matches_the_batched_append() {
4967 let path = path("representation-plan");
4968 let mut journal = SessionJournal::create(&path, metadata()).unwrap();
4969 let hydrated = journal
4970 .create_box(
4971 "t1",
4972 "large result",
4973 BoxOwner::Controller,
4974 BoxContent::text("x".repeat(3_000)),
4975 )
4976 .unwrap();
4977 let summarized = journal
4978 .create_box(
4979 "t2",
4980 "Kennedy summary",
4981 BoxOwner::Kennedy,
4982 BoxContent::text("y".repeat(3_000)),
4983 )
4984 .unwrap();
4985 journal
4986 .summarize_box("t3", summarized, "important points")
4987 .unwrap();
4988 let desired = BTreeMap::from([
4989 (hydrated, BoxRepresentation::Dehydrated),
4990 (
4991 summarized,
4992 BoxRepresentation::Summarized("important points".into()),
4993 ),
4994 ]);
4995 let preview = journal
4996 .state()
4997 .projection_with_box_representations(&desired)
4998 .unwrap();
4999 let next_id = journal.state().next_id;
5000 let ids = journal.apply_box_representations("t4", &desired).unwrap();
5001 assert_eq!(ids, vec![EventId(next_id)]);
5002 assert_eq!(journal.state().projection(), preview);
5003 assert!(matches!(
5004 journal.state().box_state(hydrated).unwrap().representation,
5005 Representation::Dehydrated { .. }
5006 ));
5007 assert_eq!(
5008 journal
5009 .state()
5010 .box_state(summarized)
5011 .unwrap()
5012 .representation,
5013 Representation::Summarized {
5014 based_on: EventId(2),
5015 text: "important points".into(),
5016 }
5017 );
5018 std::fs::remove_file(path).unwrap();
5019 }
5020}