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