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