kcode_chatend/

lib.rs

//! Kennedy's reconstructed, provider-independent durable session context.
//!
//! This module owns Kennedy's box model, context projection, context
//! representations, token policy, and replay rules. Durable session history
//! is owned by the separate `kcode-session-log` package.

use std::{
    collections::{BTreeMap, HashMap, HashSet},
    path::Path,
};

use anyhow::{Context as _, ensure};
pub use kcode_context_cache_policy::CacheExpectation;
use kcode_context_cache_policy::{
    MarkerObservation, MarkerState, PreviousProjection, StaleMarker, classify, decide_markers,
};
use kcode_session_log::{
    EventPosition, Role, Session as DurableSession, SessionStore as DurableSessionStore,
};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

pub const FORMAT_VERSION: u32 = 1;
pub const MAX_OBJECT_BYTES: u64 = 32 * 1024 * 1024 * 1024;
pub const ESTIMATED_BYTES_PER_TOKEN: u64 = 4;

const HISTORY_INGRESS_ATTEMPT_RESET_NOTE: &str = "history_ingress_attempt_reset";
const INGRESS_TIME_MARKER_NOTE: &str = "ingress_time_marker";

/// Provider token dimensions retained by historical Chatend receipts.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ProviderTokenUsage {
    pub input_tokens: u64,
    pub cached_input_tokens: u64,
    pub thinking_tokens: u64,
    pub output_tokens: u64,
}

/// Provider metering that can be reconstructed without changing durable history.
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum ProviderMetering {
    Tokens(ProviderTokenUsage),
    DurationSeconds { seconds: f64 },
    Unavailable,
}

/// One compatibility price returned by the application-owned provider catalog.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ProviderCostEstimate {
    pub usd_nanos: u64,
    pub accuracy: Value,
    pub pricing_version: String,
}

/// Application callback used only for legacy receipts that did not store a cost.
pub type ProviderCostEstimator = fn(&str, &ProviderMetering) -> Option<ProviderCostEstimate>;

/// Read-time cost fields reconstructed from one immutable session archive.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ProviderCostSummary {
    pub estimated_cost_usd_nanos: u64,
    pub unpriced_provider_calls: u64,
}

/// Replays an immutable session archive and returns only its compatible cost fields.
///
/// The input value is never modified. Callers may overlay the returned fields on a
/// response projection without changing the archive's checksummed event stream.
pub(crate) fn legacy_provider_cost_summary_for_archive(
    archive: &Value,
    default_provider_model: Option<&str>,
    estimator: ProviderCostEstimator,
) -> anyhow::Result<ProviderCostSummary> {
    let metadata = archive
        .get("metadata")
        .cloned()
        .context("session archive is missing Chatend metadata")?;
    let metadata = serde_json::from_value(metadata).context("decoding session archive metadata")?;
    let log =
        serde_json::from_value(archive.clone()).context("decoding immutable session archive")?;
    let chatend = Chatend::replay_with_provider_costs(
        metadata,
        &log,
        default_provider_model,
        Some(estimator),
    )?;
    let status = chatend.projection().status;
    Ok(ProviderCostSummary {
        estimated_cost_usd_nanos: status.estimated_cost_usd_nanos,
        unpriced_provider_calls: status.unpriced_provider_calls,
    })
}

#[derive(
    Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize,
)]
#[serde(transparent)]
pub struct EventId(pub u64);

impl std::fmt::Display for EventId {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.fmt(formatter)
    }
}

#[derive(
    Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize,
)]
#[serde(transparent)]
pub struct BoxId(pub u64);

impl std::fmt::Display for BoxId {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.fmt(formatter)
    }
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)]
#[serde(transparent)]
pub struct PendingId(String);

impl PendingId {
    pub fn from_event(id: EventId) -> Self {
        Self(format!("pending:{}", id.0))
    }

    pub fn parse(value: impl Into<String>) -> anyhow::Result<Self> {
        let value = value.into();
        let number = value
            .strip_prefix("pending:")
            .context("pending identity must begin with `pending:`")?
            .parse::<u64>()
            .context("pending identity must end in an unsigned integer")?;
        ensure!(number > 0, "pending identity zero is reserved");
        Ok(Self(value))
    }

    pub fn number(&self) -> u64 {
        self.0["pending:".len()..]
            .parse()
            .expect("validated PendingId")
    }
}

impl std::fmt::Display for PendingId {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.0)
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionKind {
    Conversation,
    Telegram,
    TelegramGroup,
    SelfTime,
    AudioIngress,
    HistoryIngress,
    Other(String),
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionMetadata {
    pub session_id: String,
    pub kind: SessionKind,
    pub created_at: String,
    pub effective_context_tokens: u64,
    #[serde(default)]
    pub channel: Value,
}

#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum BoxOwner {
    User,
    Kennedy,
    Controller,
    System,
    Tool { tool_instance: String, slot: String },
}

#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BoxContent {
    #[serde(default)]
    pub text: String,
    #[serde(default)]
    pub objects: Vec<String>,
    #[serde(default)]
    pub metadata: Value,
}

impl BoxContent {
    pub fn text(value: impl Into<String>) -> Self {
        Self {
            text: value.into(),
            ..Self::default()
        }
    }

    /// Retained for source compatibility. All Chatend box headers now omit
    /// the internal owner label.
    pub fn use_concise_header(&mut self) {
        if !self.metadata.is_object() {
            self.metadata = json!({});
        }
        self.metadata["chatendConciseHeader"] = json!(true);
    }

    fn render(&self) -> String {
        let mut rendered = self.text.clone();
        for object in &self.objects {
            if !rendered.is_empty() && !rendered.ends_with('\n') {
                rendered.push('\n');
            }
            rendered.push_str("Object provided: ");
            rendered.push_str(object);
        }
        rendered
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Representation {
    Hydrated { canonical_event: EventId },
    Dehydrated { based_on: EventId },
    Summarized { based_on: EventId, text: String },
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum BoxRepresentation {
    Hydrated,
    Dehydrated,
    Summarized(String),
}

impl Representation {
    fn based_on(&self) -> EventId {
        match self {
            Self::Hydrated { canonical_event } => *canonical_event,
            Self::Dehydrated { based_on } | Self::Summarized { based_on, .. } => *based_on,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CanonicalRevision {
    pub event_id: EventId,
    pub content: BoxContent,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BoxState {
    pub id: BoxId,
    pub name: String,
    pub owner: BoxOwner,
    pub created_at: EventId,
    pub canonical: CanonicalRevision,
    pub representation: Representation,
    pub occurrence_events: Vec<EventId>,
    pub active: bool,
}

impl BoxState {
    pub fn stale(&self) -> bool {
        !matches!(self.representation, Representation::Hydrated { .. })
            && self.representation.based_on() != self.canonical.event_id
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PendingKind {
    Node,
    Object,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum EventKind {
    SessionConfigured {
        effective_context_tokens: u64,
        kind: SessionKind,
    },
    BoxCreated {
        box_id: BoxId,
        name: String,
        owner: BoxOwner,
        content: BoxContent,
    },
    CanonicalUpdated {
        box_id: BoxId,
        content: BoxContent,
    },
    BoxRenamed {
        box_id: BoxId,
        name: String,
    },
    BoxDehydrated {
        box_id: BoxId,
    },
    BoxSummarized {
        box_id: BoxId,
        text: String,
    },
    BoxRehydrated {
        box_id: BoxId,
    },
    BoxRetired {
        box_id: BoxId,
    },
    PendingAllocated {
        pending_id: PendingId,
        resource: PendingKind,
    },
    ToolInvoked {
        tool_instance: String,
        tool_name: String,
        arguments: Value,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        invocation_id: Option<String>,
    },
    ToolCompleted {
        tool_instance: String,
        tool_name: String,
        outcome: Value,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        invocation_id: Option<String>,
    },
    ToolLayoutChanged {
        tool_instance: String,
        box_ids: Vec<BoxId>,
    },
    InferenceSubmitted {
        manifest_hash: String,
        estimated_input_tokens: u64,
        #[serde(default)]
        raw_estimated_input_tokens: Option<u64>,
    },
    ProviderInputSubmitted {
        round: u64,
        context: ProviderContext,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        transport_input_hash: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        transport_input_bytes: Option<u64>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        thread_action: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        thread_reset_reason: Option<String>,
        cacheable_prefix_bytes: u64,
        material_fingerprint: String,
        cache_expectation: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        planned_invalidation_reason: Option<String>,
    },
    ProjectionMarkersReset,
    StaleBoxesMarked {
        box_ids: Vec<BoxId>,
        consolidated: bool,
    },
    ContextSizeMarked {
        estimated_tokens: u64,
    },
    ProviderReceipt {
        manifest_hash: String,
        input_tokens: Option<u64>,
        output_tokens: Option<u64>,
        /// Exact UTF-8 byte length of Chatend's rendered context when this
        /// provider input-token measurement arrived.
        #[serde(default)]
        context_bytes: Option<u64>,
        /// Legacy uncalibrated token anchor retained for archive replay.
        #[serde(default)]
        raw_context_tokens: Option<u64>,
        provider_data: Value,
    },
    CapacityError {
        attempted_operation: String,
        projected_tokens: u64,
        limit_tokens: u64,
    },
    SourceTerminated {
        reason: String,
    },
    HistoryIngressStarted,
    HistoryEventInspected {
        source_event: EventId,
    },
    HistoryEventReleased {
        source_event: EventId,
    },
    KwebPlanChanged {
        operation: Value,
    },
    KwebCommitted {
        transaction_id: String,
        session_object_id: String,
        mappings: Value,
    },
    SessionCompleted {
        session_object_id: String,
    },
    Note {
        label: String,
        value: Value,
    },
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Event {
    pub id: EventId,
    pub recorded_at: String,
    pub kind: EventKind,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Transition {
    pub recorded_at: String,
    pub events: Vec<Event>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PersistedContextEvent {
    context_event_version: u32,
    recorded_at: String,
    kind: EventKind,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PersistedContextEventWire {
    context_event_version: u32,
    recorded_at: String,
    kind: Value,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ObjectMetadata {
    pub pending_id: PendingId,
    pub event_id: EventId,
    pub recorded_at: String,
    pub media_type: String,
    pub file_name: Option<String>,
    #[serde(default)]
    pub transport: Value,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ObjectLocation {
    pub metadata: ObjectMetadata,
    pub payload_offset: u64,
    pub payload_len: u64,
}

#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ToolSlot {
    pub slot: String,
    pub box_id: BoxId,
    pub retired: bool,
}

#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ToolState {
    pub slots: Vec<ToolSlot>,
}

/// One provider-visible dynamic tool contract retained with submitted context.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProviderToolDefinition {
    pub name: String,
    pub description: String,
    pub input_schema: Value,
}

/// Exact caller-controlled material submitted to one model inference.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProviderContext {
    pub input: String,
    pub provider: String,
    pub model: String,
    pub reasoning_effort: String,
    pub base_instructions: Option<String>,
    pub developer_instructions: Option<String>,
    pub tools: Vec<ProviderToolDefinition>,
}

/// Provider projection prepared after sparse marker reconciliation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PreparedProviderProjection {
    pub projection: ContextProjection,
    pub provider_input: String,
    pub thread_reset_reason: Option<String>,
    pub cacheable_prefix_bytes: u64,
    pub expectation: CacheExpectation,
}

/// Persistent markers and provider-thread reset decision prepared for a native resume.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PreparedProviderResume {
    /// Newly committed model-visible marker lines to append to the provider response.
    pub marker_lines: Vec<String>,
    /// Exact projection rewrite reason requiring a provider-thread reset, if any.
    pub thread_reset_reason: Option<String>,
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
struct IngressTimeMarkerState {
    any: bool,
    at_or_below_thirty_minutes: bool,
    at_or_below_ten_minutes: bool,
}

fn should_append_ingress_time_marker(
    state: IngressTimeMarkerState,
    remaining_seconds: u64,
    context_size_marked: bool,
) -> bool {
    !state.any
        || context_size_marked
        || (remaining_seconds <= 10 * 60 && !state.at_or_below_ten_minutes)
        || (remaining_seconds <= 20 * 60 && !state.at_or_below_thirty_minutes)
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ToolSlotInput {
    pub slot: String,
    pub name: String,
    pub content: BoxContent,
    pub retired: bool,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Chatend {
    pub metadata: SessionMetadata,
    pub next_id: u64,
    pub events: Vec<Event>,
    pub boxes: BTreeMap<BoxId, BoxState>,
    pub pending: BTreeMap<PendingId, PendingKind>,
    pub tools: BTreeMap<String, ToolState>,
    #[serde(default)]
    pub tool_layouts: BTreeMap<String, Vec<BoxId>>,
    pub source_terminated: bool,
    pub history_ingress_started: bool,
    pub completed_session_object: Option<String>,
}

impl Chatend {
    fn opened(metadata: SessionMetadata) -> Self {
        Self {
            metadata,
            next_id: 1,
            events: Vec::new(),
            boxes: BTreeMap::new(),
            pending: BTreeMap::new(),
            tools: BTreeMap::new(),
            tool_layouts: BTreeMap::new(),
            source_terminated: false,
            history_ingress_started: false,
            completed_session_object: None,
        }
    }

    fn restore_history_ingress_baseline(&mut self) -> anyhow::Result<()> {
        let baseline_end = self
            .events
            .iter()
            .position(|event| matches!(event.kind, EventKind::HistoryIngressStarted))
            .context("history ingress attempt reset has no initial ingress boundary")?;
        let retained_events = std::mem::take(&mut self.events);
        let retained_next_id = self.next_id;
        *self = Self::opened(self.metadata.clone());
        for event in &retained_events[..=baseline_end] {
            self.apply_transition(&Transition {
                recorded_at: event.recorded_at.clone(),
                events: vec![event.clone()],
            })?;
        }
        self.events = retained_events;
        self.next_id = retained_next_id;
        Ok(())
    }

    fn current_ingress_attempt_start(&self) -> usize {
        self.events
            .iter()
            .rposition(|event| {
                matches!(
                    &event.kind,
                    EventKind::Note { label, .. }
                        if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE
                )
            })
            .map_or(0, |index| index + 1)
    }

    /// Returns events belonging to the current provider-visible ingress attempt.
    pub fn current_ingress_attempt_events(&self) -> &[Event] {
        &self.events[self.current_ingress_attempt_start()..]
    }

    pub(crate) fn replay(
        metadata: SessionMetadata,
        log: &kcode_session_log::SessionLog,
    ) -> anyhow::Result<Self> {
        Self::replay_with_provider_costs(metadata, log, None, None)
    }

    pub(crate) fn replay_with_provider_costs(
        metadata: SessionMetadata,
        log: &kcode_session_log::SessionLog,
        default_provider_model: Option<&str>,
        estimator: Option<ProviderCostEstimator>,
    ) -> anyhow::Result<Self> {
        ensure!(
            metadata.session_id == log.header.session_id,
            "session metadata and session-log identities differ"
        );
        ensure!(
            metadata.created_at == log.header.created_at,
            "session metadata and session-log creation times differ"
        );
        let mut chatend = Self::opened(metadata);
        for (position, stored) in log.events.iter().enumerate() {
            let persisted = decode_context_event(stored)?;
            let id = EventId(position as u64 + 1);
            let mut kind = persisted.kind;
            normalize_derived_identity(&mut kind, id)?;
            chatend.apply_transition(&Transition {
                recorded_at: persisted.recorded_at.clone(),
                events: vec![Event {
                    id,
                    recorded_at: persisted.recorded_at,
                    kind,
                }],
            })?;
        }
        if let Some(estimator) = estimator {
            chatend.apply_legacy_provider_costs(default_provider_model, estimator);
        }
        Ok(chatend)
    }

    fn apply_legacy_provider_costs(
        &mut self,
        default_provider_model: Option<&str>,
        estimator: ProviderCostEstimator,
    ) {
        let mut active_tool_models = Vec::<ActiveToolModel>::new();
        let mut subagent_model = None::<String>;
        for event in &mut self.events {
            match &mut event.kind {
                EventKind::ToolInvoked {
                    tool_instance,
                    tool_name,
                    arguments,
                    invocation_id,
                } => {
                    let model = arguments
                        .get("model")
                        .and_then(Value::as_str)
                        .filter(|model| !model.trim().is_empty())
                        .map(normalize_requested_model);
                    active_tool_models.push(ActiveToolModel {
                        invocation_id: invocation_id.clone(),
                        tool_instance: tool_instance.clone(),
                        tool_name: tool_name.clone(),
                        model,
                    });
                }
                EventKind::ToolCompleted {
                    tool_instance,
                    tool_name,
                    invocation_id,
                    ..
                } => {
                    if let Some(index) = active_tool_models.iter().rposition(|active| {
                        if let Some(invocation_id) = invocation_id {
                            active.invocation_id.as_ref() == Some(invocation_id)
                        } else {
                            active.tool_instance == *tool_instance && active.tool_name == *tool_name
                        }
                    }) {
                        active_tool_models.remove(index);
                    }
                    if tool_name == "RunSubagent" {
                        subagent_model = None;
                    }
                }
                EventKind::Note { label, value } if label == "subagent_started" => {
                    subagent_model = value
                        .get("providerModel")
                        .or_else(|| value.get("provider_model"))
                        .and_then(Value::as_str)
                        .filter(|model| !model.trim().is_empty())
                        .map(str::to_owned);
                }
                EventKind::ProviderReceipt { provider_data, .. }
                    if provider_data
                        .get("estimatedCostUsdNanos")
                        .and_then(Value::as_u64)
                        .is_none() =>
                {
                    let source = provider_data.get("source").and_then(Value::as_str);
                    let model = provider_data
                        .get("providerModel")
                        .or_else(|| provider_data.get("provider_model"))
                        .and_then(Value::as_str)
                        .filter(|model| !model.trim().is_empty())
                        .map(str::to_owned)
                        .or_else(|| {
                            (source == Some("subagent"))
                                .then(|| subagent_model.clone())
                                .flatten()
                        })
                        .or_else(|| {
                            source
                                .is_some()
                                .then(|| {
                                    active_tool_models
                                        .iter()
                                        .rev()
                                        .find_map(|active| active.model.clone())
                                })
                                .flatten()
                        })
                        .or_else(|| {
                            source
                                .is_none()
                                .then(|| default_provider_model.map(str::to_owned))
                                .flatten()
                        });
                    let Some(model) = model else {
                        continue;
                    };
                    let metering = provider_metering(provider_data);
                    let Some(cost) = estimator(&model, &metering) else {
                        continue;
                    };
                    let Some(data) = provider_data.as_object_mut() else {
                        continue;
                    };
                    data.entry("providerModel")
                        .or_insert_with(|| Value::String(model));
                    data.insert("estimatedCostUsdNanos".into(), Value::from(cost.usd_nanos));
                    data.insert("costAccuracy".into(), cost.accuracy);
                    data.insert("pricingVersion".into(), Value::String(cost.pricing_version));
                }
                _ => {}
            }
        }
    }

    pub fn event(&self, id: EventId) -> Option<&Event> {
        self.events.iter().find(|event| event.id == id)
    }

    pub fn box_state(&self, id: BoxId) -> Option<&BoxState> {
        self.boxes.get(&id)
    }

    pub fn active_boxes(&self) -> impl Iterator<Item = &BoxState> {
        self.boxes.values().filter(|state| state.active)
    }

    pub fn live_context_limit(&self) -> u64 {
        self.metadata.effective_context_tokens.saturating_mul(70) / 100
    }

    pub fn forced_ingress_context_limit(&self) -> u64 {
        self.metadata.effective_context_tokens.saturating_mul(75) / 100
    }

    pub fn ingress_initial_context_limit(&self) -> u64 {
        self.metadata.effective_context_tokens.saturating_mul(75) / 100
    }

    pub fn ingress_context_limit(&self) -> u64 {
        self.metadata.effective_context_tokens.saturating_mul(90) / 100
    }

    pub fn active_context_limit(&self) -> u64 {
        if matches!(
            self.metadata.kind,
            SessionKind::SelfTime | SessionKind::HistoryIngress | SessionKind::AudioIngress
        ) {
            self.ingress_context_limit()
        } else {
            self.live_context_limit()
        }
    }

    pub fn projection_with_new_boxes(
        &self,
        boxes: &[(String, BoxOwner, BoxContent)],
    ) -> anyhow::Result<ContextProjection> {
        self.projection_with_new_boxes_at("preview", boxes)
    }

    pub fn projection_with_new_boxes_at(
        &self,
        recorded_at: &str,
        boxes: &[(String, BoxOwner, BoxContent)],
    ) -> anyhow::Result<ContextProjection> {
        self.projection_with_new_boxes_and_updates_at(recorded_at, boxes, &BTreeMap::new())
    }

    pub fn projection_with_new_boxes_and_updates(
        &self,
        boxes: &[(String, BoxOwner, BoxContent)],
        updates: &BTreeMap<BoxId, BoxContent>,
    ) -> anyhow::Result<ContextProjection> {
        self.projection_with_new_boxes_and_updates_at("preview", boxes, updates)
    }

    pub fn projection_with_new_boxes_and_updates_at(
        &self,
        recorded_at: &str,
        boxes: &[(String, BoxOwner, BoxContent)],
        updates: &BTreeMap<BoxId, BoxContent>,
    ) -> anyhow::Result<ContextProjection> {
        let mut preview = self.clone();
        let mut next = preview.next_id;
        let mut events = Vec::with_capacity(boxes.len() + updates.len());
        for (name, owner, content) in boxes {
            let id = EventId(next);
            let box_id = BoxId(next);
            next = next.checked_add(1).context("event identity overflow")?;
            events.push(Event {
                id,
                recorded_at: recorded_at.into(),
                kind: EventKind::BoxCreated {
                    box_id,
                    name: name.clone(),
                    owner: owner.clone(),
                    content: content.clone(),
                },
            });
        }
        for (box_id, content) in updates {
            let state = preview
                .box_state(*box_id)
                .with_context(|| format!("box {box_id} does not exist"))?;
            ensure!(state.active, "box {box_id} is retired");
            if state.canonical.content == *content {
                continue;
            }
            let id = EventId(next);
            next = next.checked_add(1).context("event identity overflow")?;
            events.push(Event {
                id,
                recorded_at: recorded_at.into(),
                kind: EventKind::CanonicalUpdated {
                    box_id: *box_id,
                    content: content.clone(),
                },
            });
        }
        if !events.is_empty() {
            preview.apply_transition(&Transition {
                recorded_at: recorded_at.into(),
                events,
            })?;
        }
        Ok(preview.projection())
    }

    pub fn projection_with_box_representations(
        &self,
        desired: &BTreeMap<BoxId, BoxRepresentation>,
    ) -> anyhow::Result<ContextProjection> {
        let mut preview = self.clone();
        let events = preview.box_representation_events("preview", desired)?;
        if !events.is_empty() {
            preview.apply_transition(&Transition {
                recorded_at: "preview".into(),
                events,
            })?;
        }
        Ok(preview.projection())
    }

    fn box_representation_events(
        &self,
        recorded_at: &str,
        desired: &BTreeMap<BoxId, BoxRepresentation>,
    ) -> anyhow::Result<Vec<Event>> {
        let mut next = self.next_id;
        let mut events = Vec::new();
        for (box_id, desired) in desired {
            let state = self
                .box_state(*box_id)
                .with_context(|| format!("box {box_id} does not exist"))?;
            ensure!(state.active, "box {box_id} is retired");
            let kind = match (desired, &state.representation) {
                (BoxRepresentation::Hydrated, Representation::Hydrated { .. })
                | (BoxRepresentation::Dehydrated, Representation::Dehydrated { .. }) => None,
                (
                    BoxRepresentation::Summarized(desired),
                    Representation::Summarized { text, .. },
                ) if desired == text => None,
                (BoxRepresentation::Hydrated, _) => {
                    Some(EventKind::BoxRehydrated { box_id: *box_id })
                }
                (BoxRepresentation::Dehydrated, _) => {
                    Some(EventKind::BoxDehydrated { box_id: *box_id })
                }
                (BoxRepresentation::Summarized(text), _) => Some(EventKind::BoxSummarized {
                    box_id: *box_id,
                    text: text.clone(),
                }),
            };
            let Some(kind) = kind else {
                continue;
            };
            events.push(Event {
                id: EventId(next),
                recorded_at: recorded_at.into(),
                kind,
            });
            next = next.checked_add(1).context("event ID overflow")?;
        }
        Ok(events)
    }

    fn latest_provider_input(&self) -> Option<(EventId, &ProviderContext, u64, &str)> {
        self.current_ingress_attempt_events()
            .iter()
            .rev()
            .find_map(|event| {
                let EventKind::ProviderInputSubmitted {
                    context,
                    cacheable_prefix_bytes,
                    material_fingerprint,
                    ..
                } = &event.kind
                else {
                    return None;
                };
                Some((
                    event.id,
                    context,
                    *cacheable_prefix_bytes,
                    material_fingerprint.as_str(),
                ))
            })
    }

    fn ingress_time_marker_state(&self) -> IngressTimeMarkerState {
        let mut state = IngressTimeMarkerState::default();
        for event in self.current_ingress_attempt_events() {
            let EventKind::Note { label, value } = &event.kind else {
                continue;
            };
            if label != INGRESS_TIME_MARKER_NOTE {
                continue;
            }
            let Some(remaining_seconds) = value.get("remainingSeconds").and_then(Value::as_u64)
            else {
                continue;
            };
            state.any = true;
            state.at_or_below_thirty_minutes |= remaining_seconds <= 30 * 60;
            state.at_or_below_ten_minutes |= remaining_seconds <= 10 * 60;
        }
        state
    }

    fn marker_state(&self) -> MarkerState {
        let reset = self.events.iter().rposition(|event| {
            matches!(event.kind, EventKind::ProjectionMarkersReset)
                || matches!(
                    &event.kind,
                    EventKind::Note { label, .. }
                        if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE
                )
        });
        let mut state = MarkerState::default();
        for event in &self.events[reset.unwrap_or(0)..] {
            match &event.kind {
                EventKind::StaleBoxesMarked {
                    box_ids,
                    consolidated,
                } => {
                    if *consolidated {
                        state.reported_stale_boxes.clear();
                    }
                    state
                        .reported_stale_boxes
                        .extend(box_ids.iter().map(|id| id.0));
                }
                EventKind::ContextSizeMarked { estimated_tokens } => {
                    state.last_context_size_tokens = Some(*estimated_tokens);
                }
                _ => {}
            }
        }
        state
    }

    fn projection_rewrite_reason_after(&self, event_id: Option<EventId>) -> String {
        let mut reasons = std::collections::BTreeSet::new();
        for event in self
            .events
            .iter()
            .filter(|event| event_id.is_none_or(|id| event.id > id))
        {
            let reason = match event.kind {
                EventKind::BoxDehydrated { .. } => Some("dehydration"),
                EventKind::BoxRehydrated { .. } => Some("rehydration"),
                EventKind::BoxSummarized { .. } => Some("summarization"),
                EventKind::CanonicalUpdated { .. }
                | EventKind::BoxRenamed { .. }
                | EventKind::BoxRetired { .. } => Some("canonical_state_replacement"),
                EventKind::ToolLayoutChanged { .. } => Some("projection_reordered"),
                EventKind::SessionConfigured { .. } => Some("context_limit_changed"),
                EventKind::Note { ref label, .. }
                    if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE =>
                {
                    Some("ingress_attempt_reset")
                }
                _ => None,
            };
            if let Some(reason) = reason {
                reasons.insert(reason);
            }
        }
        reasons.into_iter().collect::<Vec<_>>().join("+")
    }

    pub fn projection(&self) -> ContextProjection {
        self.projection_with_footer_lines(&[])
    }

    /// Projects provider-facing context with caller-owned transient footer lines.
    pub fn projection_with_footer_lines(&self, footer_lines: &[String]) -> ContextProjection {
        self.projection_at(footer_lines)
    }

    fn projection_at(&self, footer_lines: &[String]) -> ContextProjection {
        let items = self.projection_items(false);
        let stale_boxes = self
            .active_boxes()
            .filter(|state| state.stale())
            .map(|state| state.id)
            .collect::<Vec<_>>();
        let footer = footer_lines.join("\n");
        let context_bytes = projected_context_bytes(&items, &footer);
        let raw_estimated_tokens = estimate_bytes(context_bytes);
        let estimated_tokens = self.calibrated_estimate(context_bytes, raw_estimated_tokens);
        let fully_hydrated_context_tokens = self.fully_hydrated_context_tokens_at(footer_lines);
        let usage = self.cumulative_token_usage();
        let cost = self.cumulative_cost();
        let status = SessionStatus {
            current_context_tokens: estimated_tokens,
            fully_hydrated_context_tokens,
            context_limit_tokens: self.active_context_limit(),
            current_context_bytes: context_bytes,
            cached_input_tokens: usage.cached_input_tokens,
            non_cached_input_tokens: usage.non_cached_input_tokens,
            thinking_tokens: usage.thinking_tokens,
            output_tokens: usage.output_tokens,
            estimated_cost_usd_nanos: cost.estimated_cost_usd_nanos,
            unpriced_provider_calls: cost.unpriced_provider_calls,
        };
        ContextProjection {
            items,
            stale_boxes,
            footer,
            estimated_tokens,
            raw_estimated_tokens,
            context_bytes,
            status,
        }
    }

    fn projection_items(&self, fully_hydrated: bool) -> Vec<ProjectionItem> {
        let mut next_occurrence = HashMap::new();
        for state in self.boxes.values() {
            for pair in state.occurrence_events.windows(2) {
                next_occurrence.insert(pair[0], pair[1]);
            }
        }
        let latest_marker_reset = self.events.iter().rev().find_map(|event| {
            (matches!(event.kind, EventKind::ProjectionMarkersReset)
                || matches!(
                    &event.kind,
                    EventKind::Note { label, .. }
                        if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE
                ))
            .then_some(event.id)
        });
        let current_ingress_attempt_start = self.current_ingress_attempt_start();
        let mut items = Vec::new();
        for (index, event) in self.events.iter().enumerate() {
            if index >= current_ingress_attempt_start
                && let EventKind::Note { label, value } = &event.kind
                && label == INGRESS_TIME_MARKER_NOTE
                && let Some(text) = ingress_time_marker_text(value)
            {
                items.push(ProjectionItem::projection_marker(event.id, text));
                continue;
            }
            if latest_marker_reset.is_none_or(|reset| event.id >= reset) {
                match &event.kind {
                    EventKind::StaleBoxesMarked {
                        box_ids,
                        consolidated,
                    } => {
                        let marker = if *consolidated {
                            StaleMarker::Consolidated(box_ids.iter().map(|id| id.0).collect())
                        } else {
                            StaleMarker::New(box_ids.iter().map(|id| id.0).collect())
                        };
                        items.push(ProjectionItem::projection_marker(event.id, marker.render()));
                        continue;
                    }
                    EventKind::ContextSizeMarked { estimated_tokens } => {
                        items.push(ProjectionItem::projection_marker(
                            event.id,
                            context_size_marker_text(
                                *estimated_tokens,
                                self.active_context_limit(),
                            ),
                        ));
                        continue;
                    }
                    _ => {}
                }
            }
            let Some(box_id) = event_box_id(&event.kind) else {
                continue;
            };
            let Some(state) = self.boxes.get(&box_id) else {
                continue;
            };
            if next_occurrence.contains_key(&event.id) {
                items.push(ProjectionItem::marker(
                    event.id,
                    box_id,
                    "[box updated]".into(),
                ));
                continue;
            }
            if !state.active || state.occurrence_events.last() != Some(&event.id) {
                continue;
            }
            let (representation, body) = if fully_hydrated {
                ("hydrated", state.canonical.content.render())
            } else {
                match &state.representation {
                    Representation::Hydrated { .. } => {
                        ("hydrated", state.canonical.content.render())
                    }
                    Representation::Dehydrated { .. } => (
                        "dehydrated",
                        format!(
                            "[contents dehydrated; hydrate box {} to inspect the latest canonical revision]",
                            box_id
                        ),
                    ),
                    Representation::Summarized { text, .. } => ("summarized", text.clone()),
                }
            };
            let stale = !fully_hydrated && state.stale();
            let mut header = vec![format!("box {box_id}"), state.name.clone()];
            if matches!(
                (&state.owner, state.name.as_str()),
                (BoxOwner::User, "User message") | (BoxOwner::Kennedy, "Kennedy message")
            ) {
                header.push(format!("timestamp={}", event.recorded_at));
            }
            header.push(representation.into());
            if stale {
                header.push("stale".into());
            }
            let mut text = format!("[{}]\n{}", header.join(" | "), body);
            if text.ends_with('\n') {
                text.pop();
            }
            items.push(ProjectionItem {
                event_id: event.id,
                box_id,
                marker: false,
                stale,
                approximate_tokens: estimate_tokens(&text),
                text,
            });
        }
        for box_ids in self.tool_layouts.values() {
            arrange_tool_projection(&mut items, box_ids);
        }
        items
    }

    fn fully_hydrated_context_tokens_at(&self, footer_lines: &[String]) -> u64 {
        let items = self.projection_items(true);
        let footer = footer_lines.join("\n");
        let context_bytes = projected_context_bytes(&items, &footer);
        self.calibrated_estimate(context_bytes, estimate_bytes(context_bytes))
    }

    fn calibrated_estimate(&self, current_bytes: u64, raw_current: u64) -> u64 {
        let Some((manifest_hash, measured, bytes_at_receipt, raw_at_receipt)) =
            self.events.iter().rev().find_map(|event| {
                let EventKind::ProviderReceipt {
                    manifest_hash,
                    input_tokens: Some(input_tokens),
                    context_bytes,
                    raw_context_tokens,
                    ..
                } = &event.kind
                else {
                    return None;
                };
                Some((
                    manifest_hash,
                    *input_tokens,
                    *context_bytes,
                    *raw_context_tokens,
                ))
            })
        else {
            return raw_current;
        };
        if let Some(bytes_at_receipt) = bytes_at_receipt {
            let token_delta = current_bytes.abs_diff(bytes_at_receipt) / ESTIMATED_BYTES_PER_TOKEN;
            return if current_bytes >= bytes_at_receipt {
                measured.saturating_add(token_delta)
            } else {
                measured.saturating_sub(token_delta)
            };
        }
        let raw_at_measurement = match raw_at_receipt {
            Some(raw) => raw,
            None => {
                let Some(raw) = self.events.iter().rev().find_map(|event| {
                    let EventKind::InferenceSubmitted {
                        manifest_hash: submitted,
                        estimated_input_tokens,
                        raw_estimated_input_tokens,
                    } = &event.kind
                    else {
                        return None;
                    };
                    (submitted == manifest_hash)
                        .then_some(raw_estimated_input_tokens.unwrap_or(*estimated_input_tokens))
                }) else {
                    return raw_current;
                };
                raw
            }
        };
        if raw_current >= raw_at_measurement {
            measured.saturating_add(raw_current - raw_at_measurement)
        } else {
            measured.saturating_sub(raw_at_measurement - raw_current)
        }
    }

    fn cumulative_token_usage(&self) -> CumulativeTokenUsage {
        let mut total = CumulativeTokenUsage::default();
        for event in &self.events {
            let EventKind::ProviderReceipt { provider_data, .. } = &event.kind else {
                continue;
            };
            let cached = provider_u64(provider_data, &["cachedInputTokens", "cached_input_tokens"]);
            let thinking = provider_u64(
                provider_data,
                &[
                    "thinkingTokens",
                    "thinking_tokens",
                    "reasoningOutputTokens",
                    "reasoning_output_tokens",
                ],
            );
            let normalized_delta = provider_data
                .get("usageIsDelta")
                .and_then(Value::as_bool)
                .unwrap_or(false);
            let non_cached = if normalized_delta {
                provider_u64(
                    provider_data,
                    &["nonCachedInputTokens", "non_cached_input_tokens"],
                )
            } else {
                provider_u64(provider_data, &["inputTokens", "input_tokens"]).saturating_sub(cached)
            };
            let output = if normalized_delta {
                provider_u64(provider_data, &["outputTokens", "output_tokens"])
            } else {
                provider_u64(provider_data, &["outputTokens", "output_tokens"])
                    .saturating_sub(thinking)
            };
            total.cached_input_tokens = total.cached_input_tokens.saturating_add(cached);
            total.non_cached_input_tokens =
                total.non_cached_input_tokens.saturating_add(non_cached);
            total.thinking_tokens = total.thinking_tokens.saturating_add(thinking);
            total.output_tokens = total.output_tokens.saturating_add(output);
        }
        total
    }

    fn cumulative_cost(&self) -> CumulativeCost {
        let mut total = CumulativeCost::default();
        for event in &self.events {
            let EventKind::ProviderReceipt { provider_data, .. } = &event.kind else {
                continue;
            };
            if let Some(cost) = provider_data
                .get("estimatedCostUsdNanos")
                .and_then(Value::as_u64)
            {
                total.estimated_cost_usd_nanos =
                    total.estimated_cost_usd_nanos.saturating_add(cost);
            } else {
                total.unpriced_provider_calls = total.unpriced_provider_calls.saturating_add(1);
            }
        }
        total
    }

    pub fn render(&self) -> String {
        self.projection().render()
    }

    fn apply_transition(&mut self, transition: &Transition) -> anyhow::Result<()> {
        ensure!(
            !transition.events.is_empty(),
            "a transition cannot be empty"
        );
        for event in &transition.events {
            self.apply_event(event)?;
        }
        Ok(())
    }

    fn apply_event(&mut self, event: &Event) -> anyhow::Result<()> {
        ensure!(
            event.id.0 >= self.next_id,
            "event {} reuses an allocated identity (next is {})",
            event.id,
            self.next_id
        );
        self.next_id = event.id.0.checked_add(1).context("event ID overflow")?;
        match &event.kind {
            EventKind::SessionConfigured {
                effective_context_tokens,
                kind,
            } => {
                ensure!(
                    *effective_context_tokens > 0,
                    "effective context window must be positive"
                );
                self.metadata.effective_context_tokens = *effective_context_tokens;
                self.metadata.kind = kind.clone();
            }
            EventKind::BoxCreated {
                box_id,
                name,
                owner,
                content,
            } => {
                ensure!(
                    box_id.0 == event.id.0,
                    "BoxId must equal its creation EventId"
                );
                ensure!(
                    !self.boxes.contains_key(box_id),
                    "box {} already exists",
                    box_id
                );
                self.boxes.insert(
                    *box_id,
                    BoxState {
                        id: *box_id,
                        name: name.clone(),
                        owner: owner.clone(),
                        created_at: event.id,
                        canonical: CanonicalRevision {
                            event_id: event.id,
                            content: content.clone(),
                        },
                        representation: Representation::Hydrated {
                            canonical_event: event.id,
                        },
                        occurrence_events: vec![event.id],
                        active: true,
                    },
                );
                if let BoxOwner::Tool {
                    tool_instance,
                    slot,
                } = owner
                {
                    self.tools
                        .entry(tool_instance.clone())
                        .or_default()
                        .slots
                        .push(ToolSlot {
                            slot: slot.clone(),
                            box_id: *box_id,
                            retired: false,
                        });
                }
            }
            EventKind::CanonicalUpdated { box_id, content } => {
                let state = active_box_mut(&mut self.boxes, *box_id)?;
                state.canonical = CanonicalRevision {
                    event_id: event.id,
                    content: content.clone(),
                };
                if matches!(state.representation, Representation::Hydrated { .. }) {
                    state.representation = Representation::Hydrated {
                        canonical_event: event.id,
                    };
                }
                state.occurrence_events.push(event.id);
            }
            EventKind::BoxRenamed { box_id, name } => {
                ensure!(!name.trim().is_empty(), "a box name cannot be empty");
                let state = active_box_mut(&mut self.boxes, *box_id)?;
                state.name = name.clone();
                state.occurrence_events.push(event.id);
            }
            EventKind::BoxDehydrated { box_id } => {
                let state = active_box_mut(&mut self.boxes, *box_id)?;
                state.representation = Representation::Dehydrated {
                    based_on: state.canonical.event_id,
                };
                state.occurrence_events.push(event.id);
            }
            EventKind::BoxSummarized { box_id, text } => {
                ensure!(!text.trim().is_empty(), "a box summary cannot be empty");
                let state = active_box_mut(&mut self.boxes, *box_id)?;
                state.representation = Representation::Summarized {
                    based_on: state.canonical.event_id,
                    text: text.clone(),
                };
                state.occurrence_events.push(event.id);
            }
            EventKind::BoxRehydrated { box_id } => {
                let state = active_box_mut(&mut self.boxes, *box_id)?;
                state.representation = Representation::Hydrated {
                    canonical_event: state.canonical.event_id,
                };
                state.occurrence_events.push(event.id);
            }
            EventKind::BoxRetired { box_id } => {
                let tool_instance = self.boxes.get(box_id).and_then(|state| {
                    let BoxOwner::Tool { tool_instance, .. } = &state.owner else {
                        return None;
                    };
                    Some(tool_instance.clone())
                });
                let state = active_box_mut(&mut self.boxes, *box_id)?;
                state.active = false;
                state.occurrence_events.push(event.id);
                if let Some(tool_instance) = tool_instance {
                    let slot = self
                        .tools
                        .get_mut(&tool_instance)
                        .and_then(|tool| tool.slots.iter_mut().find(|slot| slot.box_id == *box_id))
                        .with_context(|| {
                            format!("tool box {box_id} is missing from {tool_instance}")
                        })?;
                    slot.retired = true;
                }
            }
            EventKind::ToolLayoutChanged {
                tool_instance,
                box_ids,
            } => {
                let mut unique = std::collections::HashSet::new();
                for box_id in box_ids {
                    ensure!(
                        unique.insert(*box_id),
                        "tool layout contains duplicate box {box_id}"
                    );
                    let state = self
                        .boxes
                        .get(box_id)
                        .with_context(|| format!("tool layout references missing box {box_id}"))?;
                    ensure!(state.active, "tool layout references retired box {box_id}");
                    ensure!(
                        matches!(
                            &state.owner,
                            BoxOwner::Tool {
                                tool_instance: owner,
                                ..
                            } if owner == tool_instance
                        ),
                        "tool layout box {box_id} belongs to another tool"
                    );
                }
                self.tool_layouts
                    .insert(tool_instance.clone(), box_ids.clone());
            }
            EventKind::PendingAllocated {
                pending_id,
                resource,
            } => {
                ensure!(
                    pending_id.number() == event.id.0,
                    "pending identity must equal its allocation EventId"
                );
                ensure!(
                    self.pending
                        .insert(pending_id.clone(), resource.clone())
                        .is_none(),
                    "pending identity {} already exists",
                    pending_id
                );
            }
            EventKind::SourceTerminated { .. } => self.source_terminated = true,
            EventKind::HistoryIngressStarted => {
                ensure!(
                    self.source_terminated,
                    "history ingress requires source termination"
                );
                self.history_ingress_started = true;
            }
            EventKind::Note { label, .. } if label == HISTORY_INGRESS_ATTEMPT_RESET_NOTE => {
                self.restore_history_ingress_baseline()?;
            }
            EventKind::SessionCompleted { session_object_id } => {
                self.completed_session_object = Some(session_object_id.clone());
            }
            EventKind::ToolInvoked { .. }
            | EventKind::ToolCompleted { .. }
            | EventKind::InferenceSubmitted { .. }
            | EventKind::ProviderInputSubmitted { .. }
            | EventKind::ProjectionMarkersReset
            | EventKind::StaleBoxesMarked { .. }
            | EventKind::ContextSizeMarked { .. }
            | EventKind::ProviderReceipt { .. }
            | EventKind::CapacityError { .. }
            | EventKind::HistoryEventInspected { .. }
            | EventKind::HistoryEventReleased { .. }
            | EventKind::KwebPlanChanged { .. }
            | EventKind::KwebCommitted { .. }
            | EventKind::Note { .. } => {}
        }
        self.events.push(event.clone());
        Ok(())
    }
}

fn active_box_mut(
    boxes: &mut BTreeMap<BoxId, BoxState>,
    box_id: BoxId,
) -> anyhow::Result<&mut BoxState> {
    let state = boxes
        .get_mut(&box_id)
        .with_context(|| format!("box {box_id} does not exist"))?;
    ensure!(state.active, "box {box_id} is retired");
    Ok(state)
}

fn arrange_tool_projection(items: &mut Vec<ProjectionItem>, box_ids: &[BoxId]) {
    if box_ids.is_empty() {
        return;
    }
    let ranks = box_ids
        .iter()
        .enumerate()
        .map(|(rank, box_id)| (*box_id, rank))
        .collect::<HashMap<_, _>>();
    let insertion = items
        .iter()
        .position(|item| !item.marker && ranks.contains_key(&item.box_id));
    let Some(insertion) = insertion else {
        return;
    };
    let mut arranged = Vec::with_capacity(box_ids.len());
    let mut retained = Vec::with_capacity(items.len());
    for item in std::mem::take(items) {
        if !item.marker && ranks.contains_key(&item.box_id) {
            arranged.push(item);
        } else {
            retained.push(item);
        }
    }
    arranged.sort_by_key(|item| ranks[&item.box_id]);
    let insertion = insertion.min(retained.len());
    retained.splice(insertion..insertion, arranged);
    *items = retained;
}

fn event_box_id(kind: &EventKind) -> Option<BoxId> {
    match kind {
        EventKind::BoxCreated { box_id, .. }
        | EventKind::CanonicalUpdated { box_id, .. }
        | EventKind::BoxRenamed { box_id, .. }
        | EventKind::BoxDehydrated { box_id }
        | EventKind::BoxSummarized { box_id, .. }
        | EventKind::BoxRehydrated { box_id }
        | EventKind::BoxRetired { box_id } => Some(*box_id),
        _ => None,
    }
}

fn ingress_time_marker_text(value: &Value) -> Option<String> {
    let remaining_seconds = value.get("remainingSeconds")?.as_u64()?;
    let previous_attempt_timed_out = value
        .get("previousAttemptTimedOut")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let previous = if previous_attempt_timed_out {
        "Previous attempt ran out of time; please budget your ingress time carefully. "
    } else {
        ""
    };
    let label = if previous_attempt_timed_out {
        "Ingress"
    } else {
        "ingress"
    };
    let warning = if remaining_seconds <= 10 * 60 {
        ". If you do not call EndSession before time expires, the ingress will fail and all Kmap updates will be lost."
    } else {
        ""
    };
    Some(format!(
        "[{previous}{label} time remaining: {remaining_seconds} seconds{warning}]"
    ))
}

fn context_size_marker_text(estimated_tokens: u64, context_limit: u64) -> String {
    format!(
        "[current context size: approximately {estimated_tokens} tokens | session limit: {context_limit} tokens]"
    )
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProjectionItem {
    pub event_id: EventId,
    pub box_id: BoxId,
    pub marker: bool,
    pub stale: bool,
    pub approximate_tokens: u64,
    pub text: String,
}

impl ProjectionItem {
    fn marker(event_id: EventId, box_id: BoxId, text: String) -> Self {
        Self {
            event_id,
            box_id,
            marker: true,
            stale: false,
            approximate_tokens: estimate_tokens(&text),
            text,
        }
    }

    fn projection_marker(event_id: EventId, text: String) -> Self {
        Self::marker(event_id, BoxId(0), text)
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ContextProjection {
    pub items: Vec<ProjectionItem>,
    pub stale_boxes: Vec<BoxId>,
    pub footer: String,
    pub estimated_tokens: u64,
    pub raw_estimated_tokens: u64,
    pub context_bytes: u64,
    pub status: SessionStatus,
}

impl ContextProjection {
    pub fn render(&self) -> String {
        let mut blocks = self
            .items
            .iter()
            .map(|item| item.text.as_str())
            .collect::<Vec<_>>();
        if !self.footer.is_empty() {
            blocks.push(&self.footer);
        }
        blocks.join("\n\n")
    }
}

#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionStatus {
    pub current_context_tokens: u64,
    #[serde(default)]
    pub fully_hydrated_context_tokens: u64,
    pub context_limit_tokens: u64,
    pub current_context_bytes: u64,
    pub cached_input_tokens: u64,
    pub non_cached_input_tokens: u64,
    pub thinking_tokens: u64,
    pub output_tokens: u64,
    #[serde(default)]
    pub estimated_cost_usd_nanos: u64,
    #[serde(default)]
    pub unpriced_provider_calls: u64,
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
struct CumulativeTokenUsage {
    cached_input_tokens: u64,
    non_cached_input_tokens: u64,
    thinking_tokens: u64,
    output_tokens: u64,
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
struct CumulativeCost {
    estimated_cost_usd_nanos: u64,
    unpriced_provider_calls: u64,
}

pub fn estimate_tokens(text: &str) -> u64 {
    estimate_bytes(text.len() as u64)
}

fn estimate_bytes(bytes: u64) -> u64 {
    bytes.div_ceil(ESTIMATED_BYTES_PER_TOKEN)
}

fn projected_context_bytes(items: &[ProjectionItem], footer: &str) -> u64 {
    let blocks = items.len() as u64 + u64::from(!footer.is_empty());
    items
        .iter()
        .fold(footer.len() as u64, |total, item| {
            total.saturating_add(item.text.len() as u64)
        })
        .saturating_add(blocks.saturating_sub(1).saturating_mul(2))
}

fn provider_u64(value: &Value, keys: &[&str]) -> u64 {
    keys.iter()
        .find_map(|key| value.get(*key).and_then(Value::as_u64))
        .unwrap_or_default()
}

struct ActiveToolModel {
    invocation_id: Option<String>,
    tool_instance: String,
    tool_name: String,
    model: Option<String>,
}

fn normalize_requested_model(model: &str) -> String {
    model.strip_prefix("codex/").unwrap_or(model).to_owned()
}

fn provider_metering(provider_data: &Value) -> ProviderMetering {
    let metering = provider_data.get("metering");
    if metering
        .and_then(Value::as_str)
        .is_some_and(|kind| kind == "unavailable")
    {
        return ProviderMetering::Unavailable;
    }
    if let Some(metering_value) = metering
        && let Some(metering) = metering_value.as_object()
    {
        match metering.get("kind").and_then(Value::as_str) {
            Some("duration_seconds") => {
                return metering
                    .get("seconds")
                    .and_then(Value::as_f64)
                    .map(|seconds| ProviderMetering::DurationSeconds { seconds })
                    .unwrap_or(ProviderMetering::Unavailable);
            }
            Some("unavailable") => return ProviderMetering::Unavailable,
            Some("tokens") => {
                return ProviderMetering::Tokens(token_metering(metering_value));
            }
            _ => {}
        }
    }
    let has_tokens = metering.and_then(Value::as_str) == Some("tokens")
        || [
            "inputTokens",
            "input_tokens",
            "nonCachedInputTokens",
            "non_cached_input_tokens",
            "cachedInputTokens",
            "cached_input_tokens",
            "thinkingTokens",
            "thinking_tokens",
            "outputTokens",
            "output_tokens",
        ]
        .iter()
        .any(|key| provider_data.get(*key).and_then(Value::as_u64).is_some());
    if has_tokens {
        ProviderMetering::Tokens(token_metering(provider_data))
    } else {
        ProviderMetering::Unavailable
    }
}

fn token_metering(value: &Value) -> ProviderTokenUsage {
    let cached_input_tokens = provider_u64(value, &["cachedInputTokens", "cached_input_tokens"]);
    let thinking_tokens = provider_u64(
        value,
        &[
            "thinkingTokens",
            "thinking_tokens",
            "reasoningOutputTokens",
            "reasoning_output_tokens",
        ],
    );
    let normalized_delta = value
        .get("usageIsDelta")
        .or_else(|| value.get("usage_is_delta"))
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let input_tokens = if normalized_delta {
        provider_u64(value, &["nonCachedInputTokens", "non_cached_input_tokens"])
    } else {
        provider_u64(value, &["inputTokens", "input_tokens"]).saturating_sub(cached_input_tokens)
    };
    let output_tokens = if normalized_delta {
        provider_u64(value, &["outputTokens", "output_tokens"])
    } else {
        provider_u64(value, &["outputTokens", "output_tokens"]).saturating_sub(thinking_tokens)
    };
    ProviderTokenUsage {
        input_tokens,
        cached_input_tokens,
        thinking_tokens,
        output_tokens,
    }
}

fn context_event_role(kind: &EventKind) -> Role {
    match kind {
        EventKind::BoxCreated { owner, content, .. } => match owner {
            BoxOwner::System | BoxOwner::Controller => {
                if content
                    .metadata
                    .get("capacityError")
                    .and_then(Value::as_bool)
                    .unwrap_or(false)
                {
                    Role::SystemError
                } else {
                    Role::SystemMessage
                }
            }
            BoxOwner::User => Role::UserMessage,
            BoxOwner::Kennedy => Role::KennedyMessage,
            BoxOwner::Tool { .. } => Role::ToolResult,
        },
        EventKind::ToolInvoked { .. } => Role::KennedyToolCall,
        EventKind::ToolCompleted { outcome, .. } => {
            if outcome.get("ok").and_then(Value::as_bool).unwrap_or(true) {
                Role::ToolResult
            } else {
                Role::ToolError
            }
        }
        EventKind::CapacityError { .. } => Role::SystemError,
        EventKind::PendingAllocated {
            resource: PendingKind::Object,
            ..
        } => Role::PendingObject,
        EventKind::BoxDehydrated { .. }
        | EventKind::BoxSummarized { .. }
        | EventKind::BoxRehydrated { .. }
        | EventKind::BoxRetired { .. } => Role::KennedyToolCall,
        _ => Role::SystemMessage,
    }
}

fn encode_context_event(recorded_at: &str, kind: &EventKind) -> anyhow::Result<String> {
    let mut kind = serde_json::to_value(kind)?;
    let kind_object = kind
        .as_object_mut()
        .context("Kennedy context event kind must encode as an object")?;
    match kind_object.get("type").and_then(Value::as_str) {
        Some("box_created") => {
            kind_object.remove("box_id");
        }
        Some("pending_allocated") => {
            kind_object.remove("pending_id");
        }
        _ => {}
    }
    Ok(serde_json::to_string(&PersistedContextEventWire {
        context_event_version: FORMAT_VERSION,
        recorded_at: recorded_at.into(),
        kind,
    })?)
}

fn decode_context_event(
    event: &kcode_session_log::SessionEvent,
) -> anyhow::Result<PersistedContextEvent> {
    let wire: PersistedContextEventWire = match serde_json::from_str(&event.text) {
        Ok(persisted) => persisted,
        Err(_) if event.role == Role::PendingObject => {
            return Ok(PersistedContextEvent {
                context_event_version: FORMAT_VERSION,
                recorded_at: String::new(),
                kind: EventKind::PendingAllocated {
                    pending_id: PendingId::from_event(EventId(1)),
                    resource: PendingKind::Object,
                },
            });
        }
        Err(error) => {
            return Err(error).context("decoding Kennedy context event from session log");
        }
    };
    ensure!(
        wire.context_event_version == FORMAT_VERSION,
        "unsupported Kennedy context event version {}",
        wire.context_event_version
    );
    let mut kind = wire.kind;
    let kind_object = kind
        .as_object_mut()
        .context("Kennedy context event kind must be an object")?;
    match kind_object.get("type").and_then(Value::as_str) {
        Some("box_created") if !kind_object.contains_key("box_id") => {
            kind_object.insert("box_id".into(), Value::from(0));
        }
        Some("pending_allocated") if !kind_object.contains_key("pending_id") => {
            kind_object.insert("pending_id".into(), Value::String("pending:1".into()));
        }
        _ => {}
    }
    Ok(PersistedContextEvent {
        context_event_version: wire.context_event_version,
        recorded_at: wire.recorded_at,
        kind: serde_json::from_value(kind).context("decoding Kennedy context event kind")?,
    })
}

fn normalize_derived_identity(kind: &mut EventKind, id: EventId) -> anyhow::Result<()> {
    match kind {
        EventKind::BoxCreated { box_id, .. } => *box_id = BoxId(id.0),
        EventKind::PendingAllocated { pending_id, .. } => {
            *pending_id = PendingId::from_event(id);
        }
        _ => {}
    }
    Ok(())
}

pub struct Session {
    durable: DurableSession,
    chatend: Chatend,
    objects: BTreeMap<PendingId, ObjectLocation>,
}

struct UnfinishedToolInvocation {
    invocation_id: Option<String>,
    tool_instance: String,
    tool_name: String,
}

impl Session {
    pub(crate) fn create(
        path: impl AsRef<Path>,
        metadata: SessionMetadata,
    ) -> anyhow::Result<Self> {
        ensure!(
            metadata.effective_context_tokens > 0,
            "effective context window must be positive"
        );
        let requested = path.as_ref();
        let directory = requested
            .parent()
            .filter(|path| !path.as_os_str().is_empty())
            .unwrap_or_else(|| Path::new("."));
        let durable = DurableSessionStore::new(directory)
            .create_session(&metadata.session_id, &metadata.created_at)?;
        Ok(Self {
            durable,
            chatend: Chatend::opened(metadata),
            objects: BTreeMap::new(),
        })
    }

    pub(crate) fn open_with_metadata(
        path: impl AsRef<Path>,
        metadata: SessionMetadata,
    ) -> anyhow::Result<Self> {
        Self::open_with_metadata_and_provider_costs(path, metadata, None, None)
    }

    pub(crate) fn open_with_metadata_and_provider_costs(
        path: impl AsRef<Path>,
        metadata: SessionMetadata,
        default_provider_model: Option<&str>,
        estimator: Option<ProviderCostEstimator>,
    ) -> anyhow::Result<Self> {
        let requested = path.as_ref();
        ensure!(
            requested.extension().and_then(|value| value.to_str()) == Some("session-log"),
            "{} is not a session-log path",
            requested.display()
        );
        let directory = requested
            .parent()
            .filter(|path| !path.as_os_str().is_empty())
            .unwrap_or_else(|| Path::new("."));
        let session_id = requested
            .file_stem()
            .and_then(|value| value.to_str())
            .context("session-log filename is not valid UTF-8")?;
        let durable = DurableSessionStore::new(directory).open_session(session_id)?;
        let log = durable.list();
        let chatend =
            Chatend::replay_with_provider_costs(metadata, &log, default_provider_model, estimator)?;
        let mut objects = BTreeMap::new();
        for (position, stored) in log.events.iter().enumerate() {
            let persisted = decode_context_event(stored)?;
            let id = EventId(position as u64 + 1);
            let mut kind = persisted.kind;
            normalize_derived_identity(&mut kind, id)?;
            if let EventKind::PendingAllocated {
                pending_id,
                resource: PendingKind::Object,
            } = kind
            {
                let object = durable.read_pending_object(EventPosition(position as u64))?;
                let metadata = ObjectMetadata {
                    pending_id: pending_id.clone(),
                    event_id: id,
                    recorded_at: chatend
                        .event(id)
                        .map(|event| event.recorded_at.clone())
                        .unwrap_or_default(),
                    media_type: object.media_type,
                    file_name: Some(object.file_name),
                    transport: Value::Null,
                };
                objects.insert(
                    pending_id,
                    ObjectLocation {
                        metadata,
                        payload_offset: position as u64,
                        payload_len: object.bytes.len() as u64,
                    },
                );
            }
        }
        Ok(Self {
            durable,
            chatend,
            objects,
        })
    }

    pub fn id(&self) -> &str {
        &self.chatend.metadata.session_id
    }

    pub fn state(&self) -> &Chatend {
        &self.chatend
    }

    pub fn objects(&self) -> &BTreeMap<PendingId, ObjectLocation> {
        &self.objects
    }

    #[cfg(test)]
    fn session_log(&self) -> kcode_session_log::SessionLog {
        self.durable.list()
    }

    pub fn archive_bytes(&self) -> anyhow::Result<Vec<u8>> {
        let mut archive =
            serde_json::to_value(self.durable.list()).context("serializing the session log")?;
        let object = archive
            .as_object_mut()
            .context("serialized session log is not an object")?;
        object.insert(
            "metadata".into(),
            serde_json::to_value(&self.chatend.metadata)?,
        );
        object.insert("boxes".into(), serde_json::to_value(&self.chatend.boxes)?);
        let projection = self.chatend.projection();
        let submitted = self
            .chatend
            .current_ingress_attempt_events()
            .iter()
            .rev()
            .find_map(|event| {
                let EventKind::ProviderInputSubmitted { round, context, .. } = &event.kind else {
                    return None;
                };
                Some((event.recorded_at.as_str(), *round, context))
            });
        if let Some((submitted_at, round, submitted)) = submitted {
            object.insert("chatendText".into(), Value::String(submitted.input.clone()));
            object.insert(
                "chatendTextSource".into(),
                Value::String("submitted".into()),
            );
            object.insert(
                "structuredMaterial".into(),
                json!({
                    "provider":submitted.provider,
                    "model":submitted.model,
                    "reasoningEffort":submitted.reasoning_effort,
                    "baseInstructions":submitted.base_instructions,
                    "developerInstructions":submitted.developer_instructions,
                    "tools":submitted.tools,
                    "round":round,
                    "submittedAt":submitted_at,
                }),
            );
        } else {
            object.insert("chatendText".into(), Value::String(projection.render()));
            object.insert(
                "chatendTextSource".into(),
                Value::String("reconstructed".into()),
            );
            object.insert("structuredMaterial".into(), Value::Null);
        }
        object.insert("context".into(), serde_json::to_value(projection)?);
        serde_json::to_vec(&archive).context("serializing the session archive")
    }

    pub fn is_sealed(&self) -> bool {
        self.durable.is_sealed()
    }

    pub fn seal(&mut self) -> anyhow::Result<()> {
        let unfinished_tools = self.unfinished_tool_invocations()?;
        ensure!(
            unfinished_tools.is_empty(),
            "session ends with unfinished tools {}",
            unfinished_tools
                .iter()
                .map(|tool| tool.tool_name.as_str())
                .collect::<Vec<_>>()
                .join(", ")
        );
        self.durable.seal()?;
        Ok(())
    }

    pub fn repair_unfinished_tools(
        &mut self,
        recorded_at: impl Into<String>,
    ) -> anyhow::Result<Vec<EventId>> {
        let unfinished = self.unfinished_tool_invocations()?;
        if unfinished.is_empty() {
            return Ok(Vec::new());
        }
        let recorded_at = recorded_at.into();
        let mut repaired = Vec::with_capacity(unfinished.len());
        for tool in unfinished.iter().rev() {
            let message = format!(
                "{} was interrupted before a durable completion was recorded; the abandoned invocation was closed during session recovery.",
                tool.tool_name
            );
            let kind = if let Some(invocation_id) = &tool.invocation_id {
                EventKind::ToolCompleted {
                    tool_instance: tool.tool_instance.clone(),
                    tool_name: tool.tool_name.clone(),
                    outcome: serde_json::json!({"ok":false,"recovered":true,"result":message}),
                    invocation_id: Some(invocation_id.clone()),
                }
            } else {
                // Historical native Ktool completions used one generic
                // `call_ktool` event to close the most recent invocation.
                EventKind::ToolCompleted {
                    tool_instance: "call_ktool".into(),
                    tool_name: "call_ktool".into(),
                    outcome: serde_json::json!({"ok":false,"recovered":true,"result":message}),
                    invocation_id: None,
                }
            };
            repaired.push(self.record(recorded_at.clone(), kind)?);
        }
        ensure!(
            self.unfinished_tool_invocations()?.is_empty(),
            "session tool recovery left unfinished invocations"
        );
        Ok(repaired)
    }

    fn unfinished_tool_invocations(&self) -> anyhow::Result<Vec<UnfinishedToolInvocation>> {
        let mut identified = BTreeMap::<String, UnfinishedToolInvocation>::new();
        let mut legacy = Vec::<UnfinishedToolInvocation>::new();
        for event in self.chatend.current_ingress_attempt_events() {
            match &event.kind {
                EventKind::ToolInvoked {
                    tool_instance,
                    tool_name,
                    invocation_id,
                    ..
                } => {
                    let pending = UnfinishedToolInvocation {
                        invocation_id: invocation_id.clone(),
                        tool_instance: tool_instance.clone(),
                        tool_name: tool_name.clone(),
                    };
                    if let Some(invocation_id) = invocation_id {
                        ensure!(
                            identified.insert(invocation_id.clone(), pending).is_none(),
                            "duplicate tool invocation identity {invocation_id}"
                        );
                    } else {
                        legacy.push(pending);
                    }
                }
                EventKind::ToolCompleted {
                    tool_instance,
                    tool_name,
                    invocation_id,
                    ..
                } => {
                    if let Some(invocation_id) = invocation_id {
                        let invoked = identified.remove(invocation_id).with_context(|| {
                            format!(
                                "tool {tool_name} completed without matching invocation {invocation_id}"
                            )
                        })?;
                        ensure!(
                            invoked.tool_instance.as_str() == tool_instance
                                && invoked.tool_name.as_str() == tool_name,
                            "tool completion {invocation_id} does not match its invocation"
                        );
                    } else if tool_name == "call_ktool" {
                        // An invalid native call still produces an error result
                        // even when it could not be decoded into ToolInvoked.
                        legacy.pop();
                    } else {
                        let before = legacy.len();
                        legacy.retain(|unfinished| unfinished.tool_name.as_str() != tool_name);
                        ensure!(
                            legacy.len() != before,
                            "tool {tool_name} completed without a matching invocation"
                        );
                    }
                }
                _ => {}
            }
        }
        legacy.extend(identified.into_values());
        Ok(legacy)
    }

    pub fn mark_completed(&mut self, session_object_id: String) {
        self.chatend.completed_session_object = Some(session_object_id);
    }

    pub fn configure_context(&mut self, kind: SessionKind, effective_context_tokens: u64) {
        self.chatend.metadata.kind = kind;
        self.chatend.metadata.effective_context_tokens = effective_context_tokens;
    }

    pub fn create_box(
        &mut self,
        recorded_at: impl Into<String>,
        name: impl Into<String>,
        owner: BoxOwner,
        content: BoxContent,
    ) -> anyhow::Result<BoxId> {
        let recorded_at = recorded_at.into();
        let id = EventId(self.chatend.next_id);
        let box_id = BoxId(id.0);
        self.commit_events(
            recorded_at.clone(),
            vec![Event {
                id,
                recorded_at,
                kind: EventKind::BoxCreated {
                    box_id,
                    name: name.into(),
                    owner,
                    content,
                },
            }],
        )?;
        Ok(box_id)
    }

    pub fn update_box(
        &mut self,
        recorded_at: impl Into<String>,
        box_id: BoxId,
        content: BoxContent,
    ) -> anyhow::Result<Option<EventId>> {
        let state = self
            .chatend
            .boxes
            .get(&box_id)
            .with_context(|| format!("box {box_id} does not exist"))?;
        ensure!(state.active, "box {box_id} is retired");
        if state.canonical.content == content {
            return Ok(None);
        }
        let id = EventId(self.chatend.next_id);
        let recorded_at = recorded_at.into();
        self.commit_events(
            recorded_at.clone(),
            vec![Event {
                id,
                recorded_at,
                kind: EventKind::CanonicalUpdated { box_id, content },
            }],
        )?;
        Ok(Some(id))
    }

    pub fn dehydrate_boxes(
        &mut self,
        recorded_at: impl Into<String>,
        box_ids: &[BoxId],
    ) -> anyhow::Result<Vec<EventId>> {
        ensure!(
            !box_ids.is_empty(),
            "at least one box must be selected for dehydration"
        );
        ensure!(
            box_ids.iter().copied().collect::<HashSet<_>>().len() == box_ids.len(),
            "box dehydration cannot contain duplicate box IDs"
        );
        let recorded_at = recorded_at.into();
        let events = box_ids
            .iter()
            .enumerate()
            .map(|(offset, box_id)| {
                let offset = u64::try_from(offset).context("box dehydration batch is too large")?;
                let id = self
                    .chatend
                    .next_id
                    .checked_add(offset)
                    .context("event ID overflow")?;
                Ok(Event {
                    id: EventId(id),
                    recorded_at: recorded_at.clone(),
                    kind: EventKind::BoxDehydrated { box_id: *box_id },
                })
            })
            .collect::<anyhow::Result<Vec<_>>>()?;
        let ids = events.iter().map(|event| event.id).collect();
        self.commit_events(recorded_at, events)?;
        Ok(ids)
    }

    pub fn summarize_box(
        &mut self,
        recorded_at: impl Into<String>,
        box_id: BoxId,
        text: impl Into<String>,
    ) -> anyhow::Result<EventId> {
        self.box_operation(
            recorded_at,
            EventKind::BoxSummarized {
                box_id,
                text: text.into(),
            },
        )
    }

    pub fn rehydrate_box(
        &mut self,
        recorded_at: impl Into<String>,
        box_id: BoxId,
    ) -> anyhow::Result<EventId> {
        self.box_operation(recorded_at, EventKind::BoxRehydrated { box_id })
    }

    pub fn retire_box(
        &mut self,
        recorded_at: impl Into<String>,
        box_id: BoxId,
    ) -> anyhow::Result<EventId> {
        self.box_operation(recorded_at, EventKind::BoxRetired { box_id })
    }

    fn box_operation(
        &mut self,
        recorded_at: impl Into<String>,
        kind: EventKind,
    ) -> anyhow::Result<EventId> {
        let box_id = event_box_id(&kind).context("box operation has no box identity")?;
        let state = self
            .chatend
            .box_state(box_id)
            .with_context(|| format!("box {box_id} does not exist"))?;
        ensure!(state.active, "box {box_id} is retired");
        if let EventKind::BoxSummarized { text, .. } = &kind {
            ensure!(!text.trim().is_empty(), "a box summary cannot be empty");
        }
        if matches!(kind, EventKind::BoxRetired { .. })
            && let BoxOwner::Tool { tool_instance, .. } = &state.owner
        {
            ensure!(
                self.chatend
                    .tools
                    .get(tool_instance)
                    .is_some_and(|tool| tool.slots.iter().any(|slot| slot.box_id == box_id)),
                "tool box {box_id} is missing from {tool_instance}"
            );
        }
        let id = EventId(self.chatend.next_id);
        let recorded_at = recorded_at.into();
        self.commit_events(
            recorded_at.clone(),
            vec![Event {
                id,
                recorded_at,
                kind,
            }],
        )?;
        Ok(id)
    }

    pub fn allocate_pending_node(
        &mut self,
        recorded_at: impl Into<String>,
    ) -> anyhow::Result<PendingId> {
        let id = EventId(self.chatend.next_id);
        let pending_id = PendingId::from_event(id);
        let recorded_at = recorded_at.into();
        self.commit_events(
            recorded_at.clone(),
            vec![Event {
                id,
                recorded_at,
                kind: EventKind::PendingAllocated {
                    pending_id: pending_id.clone(),
                    resource: PendingKind::Node,
                },
            }],
        )?;
        Ok(pending_id)
    }

    pub fn stage_object(
        &mut self,
        recorded_at: impl Into<String>,
        media_type: impl Into<String>,
        file_name: Option<String>,
        transport: Value,
        bytes: &[u8],
    ) -> anyhow::Result<PendingId> {
        ensure!(
            bytes.len() as u64 <= MAX_OBJECT_BYTES,
            "object exceeds the 32 GiB V1 limit"
        );
        let aggregate = self
            .objects
            .values()
            .try_fold(bytes.len() as u64, |total, object| {
                total.checked_add(object.payload_len)
            })
            .context("staged object aggregate length overflow")?;
        ensure!(
            aggregate <= MAX_OBJECT_BYTES,
            "session object payload total exceeds the 32 GiB V1 limit"
        );
        let event_id = EventId(self.chatend.next_id);
        let pending_id = PendingId::from_event(event_id);
        let metadata = ObjectMetadata {
            pending_id: pending_id.clone(),
            event_id,
            recorded_at: recorded_at.into(),
            media_type: media_type.into(),
            file_name,
            transport,
        };
        let kind = EventKind::PendingAllocated {
            pending_id: pending_id.clone(),
            resource: PendingKind::Object,
        };
        let text = encode_context_event(&metadata.recorded_at, &kind)?;
        let object_file_name = metadata
            .file_name
            .clone()
            .unwrap_or_else(|| format!("object-{}", event_id.0));
        let position = self.durable.add_pending_object(
            text,
            object_file_name,
            metadata.media_type.clone(),
            bytes,
        )?;
        ensure!(
            position.0 + 1 == event_id.0,
            "session-log event position diverged from Kennedy context identity"
        );
        let allocation = Event {
            id: event_id,
            recorded_at: metadata.recorded_at.clone(),
            kind,
        };
        self.chatend.apply_transition(&Transition {
            recorded_at: metadata.recorded_at.clone(),
            events: vec![allocation],
        })?;
        self.objects.insert(
            pending_id.clone(),
            ObjectLocation {
                metadata,
                payload_offset: position.0,
                payload_len: bytes.len() as u64,
            },
        );
        Ok(pending_id)
    }

    pub fn read_object(&mut self, id: &PendingId) -> anyhow::Result<Vec<u8>> {
        let location = self
            .objects
            .get(id)
            .with_context(|| format!("staged object {id} does not exist"))?
            .clone();
        Ok(self
            .durable
            .read_pending_object(EventPosition(location.payload_offset))?
            .bytes)
    }

    /// Starts a fresh provider-visible ingress attempt while retaining audit history.
    pub fn reset_history_ingress_attempt(
        &mut self,
        recorded_at: impl Into<String>,
    ) -> anyhow::Result<EventId> {
        ensure!(
            !self.is_sealed(),
            "a sealed session cannot start another ingress attempt"
        );
        ensure!(
            self.chatend.history_ingress_started,
            "history ingress must have started before an attempt can be reset"
        );
        let recorded_at = recorded_at.into();
        self.record(
            recorded_at,
            EventKind::Note {
                label: HISTORY_INGRESS_ATTEMPT_RESET_NOTE.into(),
                value: json!({}),
            },
        )
    }

    /// Reconciles sparse cache-epoch markers and returns the exact projection
    /// to submit with caller-owned transient budget lines.
    pub fn prepare_provider_projection(
        &mut self,
        recorded_at: impl Into<String>,
        footer_lines: &[String],
        material_fingerprint: &str,
        resume_after: Option<EventId>,
    ) -> anyhow::Result<PreparedProviderProjection> {
        self.prepare_provider_projection_at(
            recorded_at,
            footer_lines,
            material_fingerprint,
            resume_after,
            None,
        )
    }

    /// Prepares a projection and sparsely appends the current ingress time.
    pub fn prepare_provider_projection_with_ingress_time(
        &mut self,
        recorded_at: impl Into<String>,
        footer_lines: &[String],
        material_fingerprint: &str,
        resume_after: Option<EventId>,
        remaining_seconds: u64,
        previous_attempt_timed_out: bool,
    ) -> anyhow::Result<PreparedProviderProjection> {
        self.prepare_provider_projection_at(
            recorded_at,
            footer_lines,
            material_fingerprint,
            resume_after,
            Some((remaining_seconds, previous_attempt_timed_out)),
        )
    }

    /// Commits every persistent marker due before a native provider resume and
    /// returns the marker lines with the exact projection-rewrite reason.
    pub fn prepare_provider_resume(
        &mut self,
        recorded_at: impl Into<String>,
        synchronized_after: EventId,
        ingress_time: Option<(u64, bool)>,
    ) -> anyhow::Result<PreparedProviderResume> {
        let rewrite_reason = self
            .chatend
            .projection_rewrite_reason_after(Some(synchronized_after));
        let thread_reset_reason = (!rewrite_reason.is_empty()).then_some(rewrite_reason);
        let expectation = match &thread_reset_reason {
            Some(reason) => CacheExpectation::PlannedInvalidation {
                reason: reason.clone(),
            },
            None => CacheExpectation::ExpectedWarm,
        };
        let marker_lines =
            self.append_provider_markers(recorded_at.into(), expectation, ingress_time)?;
        Ok(PreparedProviderResume {
            marker_lines,
            thread_reset_reason,
        })
    }

    /// Retained for source compatibility as a marker-lines-only resume wrapper.
    pub fn prepare_provider_resume_markers(
        &mut self,
        recorded_at: impl Into<String>,
        synchronized_after: EventId,
        ingress_time: Option<(u64, bool)>,
    ) -> anyhow::Result<Vec<String>> {
        self.prepare_provider_resume(recorded_at, synchronized_after, ingress_time)
            .map(|prepared| prepared.marker_lines)
    }

    fn prepare_provider_projection_at(
        &mut self,
        recorded_at: impl Into<String>,
        footer_lines: &[String],
        material_fingerprint: &str,
        resume_after: Option<EventId>,
        ingress_time: Option<(u64, bool)>,
    ) -> anyhow::Result<PreparedProviderProjection> {
        let cacheable_projection = self.chatend.projection().render();
        let previous = self.chatend.latest_provider_input();
        let previous_projection = previous
            .map(|(_, context, prefix_bytes, fingerprint)| {
                let prefix_bytes = usize::try_from(prefix_bytes)
                    .context("cacheable provider-prefix length does not fit usize")?;
                ensure!(
                    prefix_bytes <= context.input.len()
                        && context.input.is_char_boundary(prefix_bytes),
                    "cacheable provider-prefix length is outside the exact input"
                );
                Ok(PreviousProjection {
                    text: &context.input[..prefix_bytes],
                    material_fingerprint: fingerprint,
                })
            })
            .transpose()?;
        let expectation = classify(
            previous_projection,
            &cacheable_projection,
            material_fingerprint,
            &self
                .chatend
                .projection_rewrite_reason_after(previous.map(|(id, ..)| id)),
        );
        self.append_provider_markers(recorded_at.into(), expectation.clone(), ingress_time)?;
        let cacheable_projection = self.chatend.projection().render();
        let projection = self.chatend.projection_with_footer_lines(footer_lines);
        let rewrite_reason = self.chatend.projection_rewrite_reason_after(resume_after);
        let thread_reset_reason = resume_after
            .is_some()
            .then_some(rewrite_reason)
            .filter(|reason| !reason.is_empty());
        let provider_input = if let Some(resume_after) = resume_after
            && thread_reset_reason.is_none()
        {
            let mut blocks = projection
                .items
                .iter()
                .filter(|item| item.event_id > resume_after)
                .map(|item| item.text.as_str())
                .collect::<Vec<_>>();
            if !projection.footer.is_empty() {
                blocks.push(&projection.footer);
            }
            blocks.join("\n\n")
        } else {
            projection.render()
        };
        Ok(PreparedProviderProjection {
            projection,
            provider_input,
            thread_reset_reason,
            cacheable_prefix_bytes: cacheable_projection.len() as u64,
            expectation,
        })
    }

    fn append_provider_markers(
        &mut self,
        recorded_at: String,
        expectation: CacheExpectation,
        ingress_time: Option<(u64, bool)>,
    ) -> anyhow::Result<Vec<String>> {
        let projection = if expectation.expects_cache_hit() {
            self.chatend.projection()
        } else {
            let mut preview = self.chatend.clone();
            preview.apply_transition(&Transition {
                recorded_at: "marker-preview".into(),
                events: vec![Event {
                    id: EventId(preview.next_id),
                    recorded_at: "marker-preview".into(),
                    kind: EventKind::ProjectionMarkersReset,
                }],
            })?;
            preview.projection()
        };
        let decision = decide_markers(
            &self.chatend.marker_state(),
            MarkerObservation {
                stale_boxes: projection.stale_boxes.iter().map(|id| id.0).collect(),
                current_context_tokens: projection.estimated_tokens,
                context_limit_tokens: projection.status.context_limit_tokens,
                expectation: expectation.clone(),
            },
        );
        let mut next = self.chatend.next_id;
        let mut events = Vec::new();
        let mut marker_lines = Vec::new();
        let ingress_time_marker_state = self.chatend.ingress_time_marker_state();
        let append_ingress_time = ingress_time.is_some_and(|(remaining_seconds, _)| {
            should_append_ingress_time_marker(
                ingress_time_marker_state,
                remaining_seconds,
                decision.context_size_tokens.is_some(),
            )
        });
        {
            let mut push = |kind| -> anyhow::Result<()> {
                events.push(Event {
                    id: EventId(next),
                    recorded_at: recorded_at.clone(),
                    kind,
                });
                next = next.checked_add(1).context("event ID overflow")?;
                Ok(())
            };
            if decision.reset_epoch {
                push(EventKind::ProjectionMarkersReset)?;
            }
            if let Some(stale) = decision.stale {
                marker_lines.push(stale.render());
                let (box_ids, consolidated) = match stale {
                    StaleMarker::New(ids) => (ids, false),
                    StaleMarker::Consolidated(ids) => (ids, true),
                };
                push(EventKind::StaleBoxesMarked {
                    box_ids: box_ids.into_iter().map(BoxId).collect(),
                    consolidated,
                })?;
            }
            if let Some(estimated_tokens) = decision.context_size_tokens {
                marker_lines.push(context_size_marker_text(
                    estimated_tokens,
                    projection.status.context_limit_tokens,
                ));
                push(EventKind::ContextSizeMarked { estimated_tokens })?;
            }
            if append_ingress_time
                && let Some((remaining_seconds, previous_attempt_timed_out)) = ingress_time
            {
                let value = json!({
                    "remainingSeconds":remaining_seconds,
                    "previousAttemptTimedOut":previous_attempt_timed_out
                        && !ingress_time_marker_state.any,
                });
                let marker_line = ingress_time_marker_text(&value)
                    .expect("constructed ingress marker value is valid");
                push(EventKind::Note {
                    label: INGRESS_TIME_MARKER_NOTE.into(),
                    value,
                })?;
                marker_lines.push(marker_line);
            }
        }
        if !events.is_empty() {
            self.commit_events(recorded_at, events)?;
        }
        Ok(marker_lines)
    }

    pub fn record(
        &mut self,
        recorded_at: impl Into<String>,
        kind: EventKind,
    ) -> anyhow::Result<EventId> {
        let id = EventId(self.chatend.next_id);
        let recorded_at = recorded_at.into();
        self.commit_events(
            recorded_at.clone(),
            vec![Event {
                id,
                recorded_at,
                kind,
            }],
        )?;
        Ok(id)
    }

    pub fn commit_events(
        &mut self,
        recorded_at: impl Into<String>,
        events: Vec<Event>,
    ) -> anyhow::Result<()> {
        let transition = Transition {
            recorded_at: recorded_at.into(),
            events,
        };
        ensure!(
            !transition.events.is_empty(),
            "a transition cannot be empty"
        );
        ensure!(
            !transition.events.iter().any(|event| {
                matches!(
                    event.kind,
                    EventKind::PendingAllocated {
                        resource: PendingKind::Object,
                        ..
                    }
                )
            }),
            "pending objects must be added through stage_object"
        );
        let mut preview = self.chatend.clone();
        preview.apply_transition(&transition)?;
        for event in &transition.events {
            let expected = self.durable.list().events.len() as u64 + 1;
            ensure!(
                event.id.0 == expected,
                "Kennedy context event {} does not match session-log position {}",
                event.id,
                expected - 1
            );
            self.durable.add_event(
                context_event_role(&event.kind),
                encode_context_event(&event.recorded_at, &event.kind)?,
            )?;
        }
        self.chatend = preview;
        Ok(())
    }

    pub fn apply_tool_slots(
        &mut self,
        recorded_at: impl Into<String>,
        tool_instance: impl Into<String>,
        slots: Vec<ToolSlotInput>,
    ) -> anyhow::Result<Vec<EventId>> {
        self.apply_tool_slots_inner(recorded_at, tool_instance, slots, None)
    }

    pub fn apply_tool_slots_with_layout(
        &mut self,
        recorded_at: impl Into<String>,
        tool_instance: impl Into<String>,
        slots: Vec<ToolSlotInput>,
        layout_slots: &[String],
    ) -> anyhow::Result<Vec<EventId>> {
        self.apply_tool_slots_inner(recorded_at, tool_instance, slots, Some(layout_slots))
    }

    fn apply_tool_slots_inner(
        &mut self,
        recorded_at: impl Into<String>,
        tool_instance: impl Into<String>,
        slots: Vec<ToolSlotInput>,
        layout_slots: Option<&[String]>,
    ) -> anyhow::Result<Vec<EventId>> {
        let recorded_at = recorded_at.into();
        let tool_instance = tool_instance.into();
        let current = self
            .chatend
            .tools
            .get(&tool_instance)
            .cloned()
            .unwrap_or_default();
        ensure!(
            slots.len() >= current.slots.len(),
            "stateful tool slot sequence was truncated"
        );
        for (index, existing) in current.slots.iter().enumerate() {
            ensure!(
                slots[index].slot == existing.slot,
                "stateful tool slot sequence was reordered at index {index}"
            );
            ensure!(
                !existing.retired || slots[index].retired,
                "retired tool slot {} cannot be reactivated",
                existing.slot
            );
        }
        let mut events = Vec::new();
        let mut next = self.chatend.next_id;
        let mut next_state = current.clone();
        for (index, input) in slots.iter().enumerate() {
            if let Some(existing) = current.slots.get(index) {
                let state = self
                    .chatend
                    .boxes
                    .get(&existing.box_id)
                    .context("tool slot references a missing box")?;
                if input.retired && !existing.retired {
                    let id = EventId(next);
                    next += 1;
                    events.push(Event {
                        id,
                        recorded_at: recorded_at.clone(),
                        kind: EventKind::BoxRetired {
                            box_id: existing.box_id,
                        },
                    });
                    next_state.slots[index].retired = true;
                } else if !input.retired {
                    if state.name != input.name {
                        let id = EventId(next);
                        next += 1;
                        events.push(Event {
                            id,
                            recorded_at: recorded_at.clone(),
                            kind: EventKind::BoxRenamed {
                                box_id: existing.box_id,
                                name: input.name.clone(),
                            },
                        });
                    }
                    if state.canonical.content != input.content {
                        let id = EventId(next);
                        next += 1;
                        events.push(Event {
                            id,
                            recorded_at: recorded_at.clone(),
                            kind: EventKind::CanonicalUpdated {
                                box_id: existing.box_id,
                                content: input.content.clone(),
                            },
                        });
                    }
                }
            } else {
                ensure!(
                    !input.retired,
                    "a newly appended tool slot cannot start retired"
                );
                let id = EventId(next);
                next += 1;
                let box_id = BoxId(id.0);
                events.push(Event {
                    id,
                    recorded_at: recorded_at.clone(),
                    kind: EventKind::BoxCreated {
                        box_id,
                        name: input.name.clone(),
                        owner: BoxOwner::Tool {
                            tool_instance: tool_instance.clone(),
                            slot: input.slot.clone(),
                        },
                        content: input.content.clone(),
                    },
                });
                next_state.slots.push(ToolSlot {
                    slot: input.slot.clone(),
                    box_id,
                    retired: false,
                });
            }
        }
        if let Some(layout_slots) = layout_slots {
            let mut unique = std::collections::HashSet::new();
            let box_ids = layout_slots
                .iter()
                .map(|slot_name| {
                    ensure!(
                        unique.insert(slot_name),
                        "tool layout contains duplicate slot {slot_name}"
                    );
                    let slot = next_state
                        .slots
                        .iter()
                        .find(|slot| &slot.slot == slot_name)
                        .with_context(|| {
                            format!("tool layout references missing slot {slot_name}")
                        })?;
                    ensure!(
                        !slot.retired,
                        "tool layout references retired slot {slot_name}"
                    );
                    Ok(slot.box_id)
                })
                .collect::<anyhow::Result<Vec<_>>>()?;
            if self.chatend.tool_layouts.get(&tool_instance) != Some(&box_ids) {
                let id = EventId(next);
                events.push(Event {
                    id,
                    recorded_at: recorded_at.clone(),
                    kind: EventKind::ToolLayoutChanged {
                        tool_instance: tool_instance.clone(),
                        box_ids,
                    },
                });
            }
        }
        if events.is_empty() {
            return Ok(Vec::new());
        }
        let ids = events.iter().map(|event| event.id).collect::<Vec<_>>();
        self.commit_events(recorded_at, events)?;
        Ok(ids)
    }

    pub fn apply_box_representations(
        &mut self,
        recorded_at: impl Into<String>,
        desired: &BTreeMap<BoxId, BoxRepresentation>,
    ) -> anyhow::Result<Vec<EventId>> {
        let recorded_at = recorded_at.into();
        let events = self
            .chatend
            .box_representation_events(&recorded_at, desired)?;
        if events.is_empty() {
            return Ok(Vec::new());
        }
        let ids = events.iter().map(|event| event.id).collect::<Vec<_>>();
        self.commit_events(recorded_at, events)?;
        Ok(ids)
    }
}

/// Narrow integration surface used by the `kcode-session-history` facade.
///
/// Ordinary callers should create and reopen sessions through
/// `kcode_session_history::SessionHistory`. Keeping construction and raw-log
/// replay here lets that facade retain sole ownership of session lifecycle
/// while this crate owns Chatend's mechanical persistence and projection.
pub struct SessionHistoryIntegration;

impl SessionHistoryIntegration {
    /// Creates a durable Chatend session for the history facade.
    pub fn create_session(
        path: impl AsRef<Path>,
        metadata: SessionMetadata,
    ) -> anyhow::Result<Session> {
        Session::create(path, metadata)
    }

    /// Reopens a durable Chatend session for the history facade.
    ///
    /// When `estimator` is present, historical provider receipts that predate
    /// stored cost fields are projected using `default_provider_model` and the
    /// supplied estimator without rewriting durable history.
    pub fn open_session(
        path: impl AsRef<Path>,
        metadata: SessionMetadata,
        default_provider_model: Option<&str>,
        estimator: Option<ProviderCostEstimator>,
    ) -> anyhow::Result<Session> {
        match (default_provider_model, estimator) {
            (None, None) => Session::open_with_metadata(path, metadata),
            (default_provider_model, estimator) => Session::open_with_metadata_and_provider_costs(
                path,
                metadata,
                default_provider_model,
                estimator,
            ),
        }
    }

    /// Replays a raw session log into exact Chatend state for presentation.
    ///
    /// Optional compatibility pricing affects only the returned projection and
    /// never mutates the supplied log.
    pub fn replay(
        metadata: SessionMetadata,
        log: &kcode_session_log::SessionLog,
        default_provider_model: Option<&str>,
        estimator: Option<ProviderCostEstimator>,
    ) -> anyhow::Result<Chatend> {
        match (default_provider_model, estimator) {
            (None, None) => Chatend::replay(metadata, log),
            (default_provider_model, estimator) => Chatend::replay_with_provider_costs(
                metadata,
                log,
                default_provider_model,
                estimator,
            ),
        }
    }

    /// Reconstructs compatible cost fields from an immutable session archive.
    ///
    /// The archive is not modified.
    pub fn legacy_provider_cost_summary_for_archive(
        archive: &Value,
        default_provider_model: Option<&str>,
        estimator: ProviderCostEstimator,
    ) -> anyhow::Result<ProviderCostSummary> {
        legacy_provider_cost_summary_for_archive(archive, default_provider_model, estimator)
    }
}

#[cfg(test)]
type SessionJournal = Session;

#[cfg(test)]
mod tests {
    use std::path::PathBuf;
    use std::time::{SystemTime, UNIX_EPOCH};

    use serde_json::json;

    use super::*;

    fn path(label: &str) -> PathBuf {
        std::env::temp_dir()
            .join(format!(
                "kennedy-chatend-{label}-{}-{}",
                std::process::id(),
                SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap()
                    .as_nanos()
            ))
            .join("session-1.session-log")
    }

    fn metadata() -> SessionMetadata {
        SessionMetadata {
            session_id: "session-1".into(),
            kind: SessionKind::Conversation,
            created_at: "2026-07-23T00:00:00Z".into(),
            effective_context_tokens: 1_000,
            channel: json!({"kind":"test"}),
        }
    }

    #[test]
    fn box_identity_continuations_staleness_and_replay_are_exact() {
        let path = path("boxes");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        let id = journal
            .create_box(
                "t1",
                "message",
                BoxOwner::User,
                BoxContent::text("original"),
            )
            .unwrap();
        assert_eq!(id, BoxId(1));
        journal.summarize_box("t2", id, "summary").unwrap();
        journal
            .update_box("t3", id, BoxContent::text("changed"))
            .unwrap();
        let state = journal.state().box_state(id).unwrap().clone();
        assert!(state.stale());
        assert_eq!(
            state.representation,
            Representation::Summarized {
                based_on: EventId(1),
                text: "summary".into()
            }
        );
        let projection = journal.state().projection();
        assert_eq!(projection.items[0].text, "[box updated]");
        assert_eq!(projection.items[1].text, "[box updated]");
        assert!(projection.items[2].text.contains("summary"));
        assert!(projection.items[2].stale);
        assert!(projection.footer.is_empty());
        assert!(projection.render().ends_with("summary"));

        let transient = vec![
            "[provider calls remaining after this call: 249 | max provider calls: 250]".into(),
            "[provider call time remaining: 3h 45m]".into(),
        ];
        let with_limits = journal.state().projection_with_footer_lines(&transient);
        assert_eq!(with_limits.footer, transient.join("\n"));
        assert!(with_limits.render().ends_with(&transient.join("\n")));
        assert!(with_limits.context_bytes > projection.context_bytes);
        drop(journal);
        let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
        assert_eq!(reopened.state().box_state(id), Some(&state));
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn box_headers_hide_internal_ownership_and_timestamp_messages() {
        let path = path("box-headers");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        journal
            .create_box(
                "2026-07-23T00:00:01Z",
                "User message",
                BoxOwner::User,
                BoxContent::text("Hello"),
            )
            .unwrap();
        journal
            .create_box(
                "2026-07-23T00:00:02Z",
                "Kennedy message",
                BoxOwner::Kennedy,
                BoxContent::text("Hi"),
            )
            .unwrap();
        let mut content = BoxContent::text("Node ID: AAAAAAAB");
        content.use_concise_header();
        journal
            .create_box(
                "2026-07-23T00:00:03Z",
                "Kweb loaded node",
                BoxOwner::Tool {
                    tool_instance: "kweb".into(),
                    slot: "loaded".into(),
                },
                content,
            )
            .unwrap();

        let projection = journal.state().projection();
        assert_eq!(
            projection.items[0].text,
            "[box 1 | User message | timestamp=2026-07-23T00:00:01Z | hydrated]\nHello"
        );
        assert_eq!(
            projection.items[1].text,
            "[box 2 | Kennedy message | timestamp=2026-07-23T00:00:02Z | hydrated]\nHi"
        );
        assert_eq!(
            projection.items[2].text,
            "[box 3 | Kweb loaded node | hydrated]\nNode ID: AAAAAAAB"
        );
        assert!(
            projection
                .items
                .iter()
                .all(|item| !item.text.contains("owner="))
        );
        assert!(projection.footer.is_empty());
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn provider_preparation_anchors_sparse_markers_and_resets_rewritten_epochs() {
        let path = path("provider-markers");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        let first_box = journal
            .create_box(
                "t1",
                "System prompt",
                BoxOwner::System,
                BoxContent::text("x".repeat(1_000)),
            )
            .unwrap();
        let first = journal
            .prepare_provider_projection(
                "t2",
                &["[provider calls remaining: 9]".into()],
                "m1",
                None,
            )
            .unwrap();
        assert_eq!(first.expectation, CacheExpectation::ColdStart);
        assert!(first.projection.render().contains("[current context size:"));
        assert!(
            first
                .projection
                .render()
                .ends_with("[provider calls remaining: 9]")
        );
        let first_input = first.projection.render();
        let synchronized_event_id = journal
            .record(
                "t3",
                EventKind::ProviderInputSubmitted {
                    round: 1,
                    context: provider_context(first_input.clone()),
                    transport_input_hash: None,
                    transport_input_bytes: None,
                    thread_action: None,
                    thread_reset_reason: None,
                    cacheable_prefix_bytes: first.cacheable_prefix_bytes,
                    material_fingerprint: "m1".into(),
                    cache_expectation: first.expectation.label().into(),
                    planned_invalidation_reason: None,
                },
            )
            .unwrap();
        let archive: Value = serde_json::from_slice(&journal.archive_bytes().unwrap()).unwrap();
        assert_eq!(archive["chatendText"], first_input);
        assert_eq!(archive["chatendTextSource"], "submitted");
        assert_eq!(archive["structuredMaterial"]["provider"], "codex");
        assert_eq!(archive["structuredMaterial"]["model"], "test");
        journal
            .create_box(
                "t4",
                "User message",
                BoxOwner::User,
                BoxContent::text("next"),
            )
            .unwrap();
        let warm = journal
            .prepare_provider_projection("t5", &[], "m1", Some(synchronized_event_id))
            .unwrap();
        assert_eq!(warm.expectation, CacheExpectation::ExpectedWarm);
        assert!(warm.provider_input.contains("next"));
        assert!(!warm.provider_input.contains(&"x".repeat(1_000)));
        assert!(warm.thread_reset_reason.is_none());

        journal.summarize_box("t6", first_box, "short").unwrap();
        let reset = journal
            .prepare_provider_projection("t7", &[], "m1", Some(synchronized_event_id))
            .unwrap();
        assert_eq!(
            reset.expectation,
            CacheExpectation::PlannedInvalidation {
                reason: "summarization".into()
            }
        );
        assert_eq!(reset.thread_reset_reason.as_deref(), Some("summarization"));
        assert_eq!(reset.provider_input, reset.projection.render());
        assert!(!reset.projection.render().contains("[current context size:"));
        assert_eq!(
            journal
                .state()
                .events
                .iter()
                .filter(|event| matches!(event.kind, EventKind::ProjectionMarkersReset))
                .count(),
            2
        );
        std::fs::remove_file(path).unwrap();
    }

    fn prepared_resume_after<T>(
        label: &str,
        setup: impl FnOnce(&mut SessionJournal) -> T,
        mutate: impl FnOnce(&mut SessionJournal, T),
    ) -> PreparedProviderResume {
        let path = path(label);
        let mut journal = SessionJournal::create(
            &path,
            SessionMetadata {
                effective_context_tokens: 100_000,
                ..metadata()
            },
        )
        .unwrap();
        let value = setup(&mut journal);
        let synchronized_after = journal.state().events.last().unwrap().id;
        mutate(&mut journal, value);
        let prepared = journal
            .prepare_provider_resume("resume", synchronized_after, None)
            .unwrap();
        std::fs::remove_file(path).unwrap();
        prepared
    }

    #[test]
    fn structured_provider_resume_reports_append_only_and_exact_rewrite_reasons() {
        let append_only = prepared_resume_after(
            "resume-append-only",
            |journal| {
                journal
                    .create_box("t1", "first", BoxOwner::User, BoxContent::text("first"))
                    .unwrap()
            },
            |journal, _| {
                journal
                    .create_box("t2", "second", BoxOwner::User, BoxContent::text("second"))
                    .unwrap();
            },
        );
        assert!(append_only.marker_lines.is_empty());
        assert_eq!(append_only.thread_reset_reason, None);

        let dehydration = prepared_resume_after(
            "resume-dehydration",
            |journal| {
                journal
                    .create_box("t1", "box", BoxOwner::Controller, BoxContent::text("body"))
                    .unwrap()
            },
            |journal, box_id| {
                journal.dehydrate_boxes("t2", &[box_id]).unwrap();
            },
        );
        assert_eq!(
            dehydration.thread_reset_reason.as_deref(),
            Some("dehydration")
        );

        let summarization = prepared_resume_after(
            "resume-summarization",
            |journal| {
                journal
                    .create_box("t1", "box", BoxOwner::Controller, BoxContent::text("body"))
                    .unwrap()
            },
            |journal, box_id| {
                journal.summarize_box("t2", box_id, "summary").unwrap();
            },
        );
        assert_eq!(
            summarization.thread_reset_reason.as_deref(),
            Some("summarization")
        );

        let rehydration = prepared_resume_after(
            "resume-rehydration",
            |journal| {
                let box_id = journal
                    .create_box("t1", "box", BoxOwner::Controller, BoxContent::text("body"))
                    .unwrap();
                journal.dehydrate_boxes("t2", &[box_id]).unwrap();
                box_id
            },
            |journal, box_id| {
                journal.rehydrate_box("t3", box_id).unwrap();
            },
        );
        assert_eq!(
            rehydration.thread_reset_reason.as_deref(),
            Some("rehydration")
        );

        let canonical_replacement = prepared_resume_after(
            "resume-canonical-replacement",
            |journal| {
                journal
                    .create_box(
                        "t1",
                        "box",
                        BoxOwner::Controller,
                        BoxContent::text("before"),
                    )
                    .unwrap()
            },
            |journal, box_id| {
                journal
                    .update_box("t2", box_id, BoxContent::text("after"))
                    .unwrap();
            },
        );
        assert_eq!(
            canonical_replacement.thread_reset_reason.as_deref(),
            Some("canonical_state_replacement")
        );

        let projection_reordered = prepared_resume_after(
            "resume-projection-reordered",
            |journal| {
                journal
                    .apply_tool_slots_with_layout(
                        "t1",
                        "tool",
                        vec![
                            ToolSlotInput {
                                slot: "a".into(),
                                name: "a".into(),
                                content: BoxContent::text("a"),
                                retired: false,
                            },
                            ToolSlotInput {
                                slot: "b".into(),
                                name: "b".into(),
                                content: BoxContent::text("b"),
                                retired: false,
                            },
                        ],
                        &["a".into(), "b".into()],
                    )
                    .unwrap();
            },
            |journal, ()| {
                journal
                    .apply_tool_slots_with_layout(
                        "t2",
                        "tool",
                        vec![
                            ToolSlotInput {
                                slot: "a".into(),
                                name: "a".into(),
                                content: BoxContent::text("a"),
                                retired: false,
                            },
                            ToolSlotInput {
                                slot: "b".into(),
                                name: "b".into(),
                                content: BoxContent::text("b"),
                                retired: false,
                            },
                        ],
                        &["b".into(), "a".into()],
                    )
                    .unwrap();
            },
        );
        assert_eq!(
            projection_reordered.thread_reset_reason.as_deref(),
            Some("projection_reordered")
        );
    }

    #[test]
    fn provider_resume_marker_wrapper_preserves_growth_and_rewritten_epochs() {
        let path = path("provider-resume-markers");
        let mut journal = SessionJournal::create(
            &path,
            SessionMetadata {
                effective_context_tokens: 100_000,
                ..metadata()
            },
        )
        .unwrap();
        let first_box = journal
            .create_box(
                "t1",
                "System prompt",
                BoxOwner::System,
                BoxContent::text("small"),
            )
            .unwrap();
        let initial = journal
            .prepare_provider_projection("t2", &[], "m1", None)
            .unwrap();
        assert!(
            !initial
                .projection
                .render()
                .contains("[current context size:")
        );
        let mut synchronized_after = journal
            .record(
                "t3",
                EventKind::ProviderInputSubmitted {
                    round: 1,
                    context: provider_context(initial.projection.render()),
                    transport_input_hash: None,
                    transport_input_bytes: None,
                    thread_action: None,
                    thread_reset_reason: None,
                    cacheable_prefix_bytes: initial.cacheable_prefix_bytes,
                    material_fingerprint: "m1".into(),
                    cache_expectation: initial.expectation.label().into(),
                    planned_invalidation_reason: None,
                },
            )
            .unwrap();

        journal
            .create_box(
                "t4",
                "large result",
                BoxOwner::Controller,
                BoxContent::text("x".repeat(130_000)),
            )
            .unwrap();
        let first = journal
            .prepare_provider_resume_markers("t5", synchronized_after, None)
            .unwrap();
        let (marker_event, marked_tokens) = journal
            .state()
            .events
            .iter()
            .rev()
            .find_map(|event| match &event.kind {
                EventKind::ContextSizeMarked { estimated_tokens } => {
                    Some((event.id, *estimated_tokens))
                }
                _ => None,
            })
            .unwrap();
        let expected = format!(
            "[current context size: approximately {marked_tokens} tokens | session limit: 70000 tokens]"
        );
        assert_eq!(first.as_slice(), std::slice::from_ref(&expected));
        assert_eq!(
            journal
                .state()
                .projection()
                .items
                .into_iter()
                .find(|item| item.event_id == marker_event)
                .unwrap()
                .text,
            expected
        );

        synchronized_after = journal.state().events.last().unwrap().id;
        journal
            .create_box(
                "t6",
                "small growth",
                BoxOwner::Controller,
                BoxContent::text("y".repeat(20_000)),
            )
            .unwrap();
        assert!(
            journal
                .prepare_provider_resume_markers("t7", synchronized_after, None)
                .unwrap()
                .is_empty()
        );

        synchronized_after = journal.state().events.last().unwrap().id;
        journal
            .create_box(
                "t8",
                "large growth",
                BoxOwner::Controller,
                BoxContent::text("z".repeat(40_000)),
            )
            .unwrap();
        let second = journal
            .prepare_provider_resume_markers("t9", synchronized_after, None)
            .unwrap();
        assert_eq!(second.len(), 1);
        assert!(second[0].starts_with("[current context size: approximately "));

        synchronized_after = journal.state().events.last().unwrap().id;
        journal
            .update_box(
                "t10",
                first_box,
                BoxContent::text("rewritten system prompt"),
            )
            .unwrap();
        let reset = journal
            .prepare_provider_resume_markers("t11", synchronized_after, None)
            .unwrap();
        assert_eq!(reset.len(), 1);
        assert!(reset[0].starts_with("[current context size: approximately "));
        assert_eq!(
            journal
                .state()
                .events
                .iter()
                .filter(|event| matches!(event.kind, EventKind::ProjectionMarkersReset))
                .count(),
            2,
        );
        std::fs::remove_file(path).unwrap();
    }

    fn provider_context(input: String) -> ProviderContext {
        ProviderContext {
            input,
            provider: "codex".into(),
            model: "test".into(),
            reasoning_effort: "xhigh".into(),
            base_instructions: Some("base".into()),
            developer_instructions: Some(String::new()),
            tools: Vec::new(),
        }
    }

    #[test]
    fn shared_pending_and_box_identity_space_never_overlaps() {
        let path = path("pending");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        let first = journal.allocate_pending_node("t1").unwrap();
        let box_id = journal
            .create_box("t2", "box", BoxOwner::Kennedy, BoxContent::text("hello"))
            .unwrap();
        let object = journal
            .stage_object(
                "t3",
                "application/octet-stream",
                None,
                Value::Null,
                b"\0binary\xff",
            )
            .unwrap();
        assert_eq!(first.to_string(), "pending:1");
        assert_eq!(box_id, BoxId(2));
        assert_eq!(object.to_string(), "pending:3");
        assert_eq!(journal.read_object(&object).unwrap(), b"\0binary\xff");
        let stored = journal.session_log();
        assert!(!stored.events[0].text.contains("pending_id"));
        assert!(!stored.events[1].text.contains("box_id"));
        assert!(!stored.events[2].text.contains("pending_id"));
        drop(journal);
        let mut reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
        assert_eq!(reopened.read_object(&object).unwrap(), b"\0binary\xff");
        assert_eq!(reopened.state().next_id, 4);
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn kennedy_tool_completion_is_validated_before_storage_seals() {
        let path = path("seal-tool");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        journal
            .record(
                "t1",
                EventKind::ToolInvoked {
                    tool_instance: "CreateNode:1".into(),
                    tool_name: "CreateNode".into(),
                    arguments: json!({}),
                    invocation_id: None,
                },
            )
            .unwrap();
        assert!(journal.seal().is_err());
        journal
            .record(
                "t2",
                EventKind::ToolCompleted {
                    tool_instance: "call_ktool".into(),
                    tool_name: "call_ktool".into(),
                    outcome: json!({"ok":true}),
                    invocation_id: None,
                },
            )
            .unwrap();
        journal.seal().unwrap();
        assert!(journal.is_sealed());
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn interrupted_tools_are_recovered_by_identity_without_losing_legacy_journals() {
        let path = path("repair-tools");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        journal
            .record(
                "t1",
                EventKind::ToolInvoked {
                    tool_instance: "WebSearch:legacy".into(),
                    tool_name: "WebSearch".into(),
                    arguments: json!({"question":"legacy"}),
                    invocation_id: None,
                },
            )
            .unwrap();
        journal
            .record(
                "t2",
                EventKind::ToolInvoked {
                    tool_instance: "WebSearch:new".into(),
                    tool_name: "WebSearch".into(),
                    arguments: json!({"question":"identified"}),
                    invocation_id: Some("call-1".into()),
                },
            )
            .unwrap();
        assert!(journal.seal().is_err());

        let repaired = journal.repair_unfinished_tools("recovery").unwrap();
        assert_eq!(repaired.len(), 2);
        assert!(
            journal
                .state()
                .events
                .iter()
                .rev()
                .take(2)
                .all(|event| matches!(
                    &event.kind,
                    EventKind::ToolCompleted { outcome, .. }
                        if outcome.get("recovered").and_then(Value::as_bool) == Some(true)
                ))
        );
        journal.seal().unwrap();
        assert!(journal.is_sealed());
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn identified_tool_completions_can_arrive_out_of_order() {
        let path = path("tool-identity");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        for id in ["call-1", "call-2"] {
            journal
                .record(
                    id,
                    EventKind::ToolInvoked {
                        tool_instance: format!("WebSearch:{id}"),
                        tool_name: "WebSearch".into(),
                        arguments: json!({"question":id}),
                        invocation_id: Some(id.into()),
                    },
                )
                .unwrap();
        }
        for id in ["call-1", "call-2"] {
            journal
                .record(
                    format!("{id}-complete"),
                    EventKind::ToolCompleted {
                        tool_instance: format!("WebSearch:{id}"),
                        tool_name: "WebSearch".into(),
                        outcome: json!({"ok":true}),
                        invocation_id: Some(id.into()),
                    },
                )
                .unwrap();
        }
        journal.seal().unwrap();
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn invalid_box_operations_do_not_poison_the_append_only_journal() {
        let path = path("invalid-box-operation");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        let box_id = journal
            .create_box(
                "t1",
                "valid",
                BoxOwner::Controller,
                BoxContent::text("canonical"),
            )
            .unwrap();
        let valid_length = std::fs::metadata(&path).unwrap().len();

        let result = journal.dehydrate_boxes("t2", &[box_id, BoxId(97)]);
        assert_eq!(result.unwrap_err().to_string(), "box 97 does not exist");
        assert_eq!(std::fs::metadata(&path).unwrap().len(), valid_length);
        assert!(matches!(
            journal.state().box_state(box_id).unwrap().representation,
            Representation::Hydrated { .. }
        ));

        for result in [
            journal.summarize_box("t3", BoxId(97), "summary"),
            journal.rehydrate_box("t4", BoxId(97)),
            journal.retire_box("t5", BoxId(97)),
        ] {
            assert_eq!(result.unwrap_err().to_string(), "box 97 does not exist");
            assert_eq!(std::fs::metadata(&path).unwrap().len(), valid_length);
        }

        drop(journal);
        let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
        assert_eq!(
            reopened
                .state()
                .box_state(box_id)
                .unwrap()
                .canonical
                .content,
            BoxContent::text("canonical")
        );
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn multiple_boxes_dehydrate_in_one_replayable_batch() {
        let path = path("dehydrate-boxes");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        let boxes = ["first", "second", "third"]
            .into_iter()
            .map(|text| {
                journal
                    .create_box("t1", text, BoxOwner::Controller, BoxContent::text(text))
                    .unwrap()
            })
            .collect::<Vec<_>>();

        let event_ids = journal
            .dehydrate_boxes("t2", &[boxes[0], boxes[2]])
            .unwrap();
        assert_eq!(event_ids, [EventId(4), EventId(5)]);
        assert!(matches!(
            journal.state().box_state(boxes[0]).unwrap().representation,
            Representation::Dehydrated { .. }
        ));
        assert!(matches!(
            journal.state().box_state(boxes[1]).unwrap().representation,
            Representation::Hydrated { .. }
        ));
        assert!(matches!(
            journal.state().box_state(boxes[2]).unwrap().representation,
            Representation::Dehydrated { .. }
        ));

        let valid_length = std::fs::metadata(&path).unwrap().len();
        assert_eq!(
            journal.dehydrate_boxes("t3", &[]).unwrap_err().to_string(),
            "at least one box must be selected for dehydration"
        );
        assert_eq!(
            journal
                .dehydrate_boxes("t3", &[boxes[1], boxes[1]])
                .unwrap_err()
                .to_string(),
            "box dehydration cannot contain duplicate box IDs"
        );
        assert_eq!(std::fs::metadata(&path).unwrap().len(), valid_length);

        drop(journal);
        let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
        assert!(matches!(
            reopened.state().box_state(boxes[0]).unwrap().representation,
            Representation::Dehydrated { .. }
        ));
        assert!(matches!(
            reopened.state().box_state(boxes[2]).unwrap().representation,
            Representation::Dehydrated { .. }
        ));
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn status_estimates_the_context_with_every_active_box_hydrated() {
        let path = path("fully-hydrated-estimate");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        let box_id = journal
            .create_box(
                "t1",
                "large",
                BoxOwner::Controller,
                BoxContent::text("canonical material ".repeat(1_000)),
            )
            .unwrap();
        journal.dehydrate_boxes("t2", &[box_id]).unwrap();

        let status = journal.state().projection().status;
        assert!(
            status.fully_hydrated_context_tokens > status.current_context_tokens,
            "the hydrated estimate should include the canonical body"
        );
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn provider_measurements_recalibrate_the_matching_manifest_then_track_deltas() {
        let path = path("calibration");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        journal
            .create_box(
                "t1",
                "system",
                BoxOwner::System,
                BoxContent::text("baseline provider content"),
            )
            .unwrap();
        let submitted = journal.state().projection();
        journal
            .record(
                "t2",
                EventKind::InferenceSubmitted {
                    manifest_hash: "manifest-1".into(),
                    estimated_input_tokens: submitted.estimated_tokens,
                    raw_estimated_input_tokens: Some(submitted.raw_estimated_tokens),
                },
            )
            .unwrap();
        let tool_box = journal
            .create_box(
                "t3",
                "tool result",
                BoxOwner::Controller,
                BoxContent::text("x".repeat(3_000)),
            )
            .unwrap();
        let measured_context = journal.state().projection();
        journal
            .record(
                "t4",
                EventKind::ProviderReceipt {
                    manifest_hash: "manifest-1".into(),
                    input_tokens: Some(777),
                    output_tokens: Some(3),
                    context_bytes: Some(measured_context.context_bytes),
                    raw_context_tokens: Some(measured_context.raw_estimated_tokens),
                    provider_data: Value::Null,
                },
            )
            .unwrap();
        assert_eq!(journal.state().projection().estimated_tokens, 777);
        journal
            .create_box(
                "t5",
                "new material",
                BoxOwner::User,
                BoxContent::text("x".repeat(300)),
            )
            .unwrap();
        let expanded = journal.state().projection();
        assert_eq!(
            expanded.estimated_tokens,
            777 + expanded
                .context_bytes
                .abs_diff(measured_context.context_bytes)
                / ESTIMATED_BYTES_PER_TOKEN
        );
        assert!(expanded.raw_estimated_tokens > submitted.raw_estimated_tokens);
        journal.dehydrate_boxes("t6", &[tool_box]).unwrap();
        let shrunken = journal.state().projection();
        assert_eq!(
            shrunken.estimated_tokens,
            777_u64.saturating_sub(
                measured_context
                    .context_bytes
                    .abs_diff(shrunken.context_bytes)
                    / ESTIMATED_BYTES_PER_TOKEN
            )
        );
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn session_status_keeps_provider_usage_categories_exact_and_exclusive() {
        let path = path("session-status");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        journal
            .create_box("t1", "system", BoxOwner::System, BoxContent::text("é"))
            .unwrap();
        let before_usage = journal.state().projection();
        assert_eq!(
            before_usage.raw_estimated_tokens,
            before_usage
                .context_bytes
                .div_ceil(ESTIMATED_BYTES_PER_TOKEN)
        );
        journal
            .record(
                "t2",
                EventKind::ProviderReceipt {
                    manifest_hash: "manifest-1".into(),
                    input_tokens: Some(100),
                    output_tokens: Some(20),
                    context_bytes: Some(before_usage.context_bytes),
                    raw_context_tokens: Some(before_usage.raw_estimated_tokens),
                    provider_data: json!({
                        "usageIsDelta":true,
                        "cachedInputTokens":40,
                        "nonCachedInputTokens":60,
                        "thinkingTokens":8,
                        "outputTokens":12,
                        "estimatedCostUsdNanos":12345
                    }),
                },
            )
            .unwrap();
        let second_anchor = journal.state().projection();
        journal
            .record(
                "t3",
                EventKind::ProviderReceipt {
                    manifest_hash: "manifest-1".into(),
                    input_tokens: Some(120),
                    output_tokens: Some(5),
                    context_bytes: Some(second_anchor.context_bytes),
                    raw_context_tokens: Some(second_anchor.raw_estimated_tokens),
                    provider_data: json!({
                        "usageIsDelta":true,
                        "cachedInputTokens":10,
                        "nonCachedInputTokens":10,
                        "thinkingTokens":2,
                        "outputTokens":3,
                        "estimatedCostUsdNanos":6789
                    }),
                },
            )
            .unwrap();
        let projection = journal.state().projection();
        assert_eq!(
            projection.status,
            SessionStatus {
                current_context_tokens: 120,
                fully_hydrated_context_tokens: 120,
                context_limit_tokens: 700,
                current_context_bytes: projection.context_bytes,
                cached_input_tokens: 50,
                non_cached_input_tokens: 70,
                thinking_tokens: 10,
                output_tokens: 15,
                estimated_cost_usd_nanos: 19_134,
                unpriced_provider_calls: 0,
            }
        );
        let archive: Value = serde_json::from_slice(&journal.archive_bytes().unwrap()).unwrap();
        assert_eq!(archive["context"]["status"]["cachedInputTokens"], 50);
        assert!(
            archive["chatendText"]
                .as_str()
                .unwrap()
                .ends_with(archive["context"]["footer"].as_str().unwrap())
        );
        std::fs::remove_file(path).unwrap();
    }

    fn compatibility_cost(
        model: &str,
        metering: &ProviderMetering,
    ) -> Option<ProviderCostEstimate> {
        let ProviderMetering::Tokens(usage) = metering else {
            return None;
        };
        let base = match model {
            "gpt-5.6-sol" => 1_000,
            "gemini-3.1-pro-preview" => 2_000,
            _ => return None,
        };
        Some(ProviderCostEstimate {
            usd_nanos: base + usage.input_tokens,
            accuracy: json!("exact"),
            pricing_version: "test-prices".into(),
        })
    }

    #[test]
    fn legacy_provider_costs_are_reconstructed_without_rewriting_history() {
        let path = path("legacy-provider-costs");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        journal
            .record(
                "t1",
                EventKind::ProviderReceipt {
                    manifest_hash: "top".into(),
                    input_tokens: Some(10),
                    output_tokens: Some(1),
                    context_bytes: None,
                    raw_context_tokens: None,
                    provider_data: json!({
                        "usageIsDelta":true,
                        "nonCachedInputTokens":10,
                        "cachedInputTokens":0,
                        "thinkingTokens":0,
                        "outputTokens":1
                    }),
                },
            )
            .unwrap();
        journal
            .record(
                "t2",
                EventKind::ToolInvoked {
                    tool_instance: "RunSubagent:1".into(),
                    tool_name: "RunSubagent".into(),
                    arguments: json!({"model":"codex/gpt-5.6-sol"}),
                    invocation_id: Some("subagent-1".into()),
                },
            )
            .unwrap();
        journal
            .record(
                "t3",
                EventKind::Note {
                    label: "subagent_started".into(),
                    value: json!({"providerModel":"gpt-5.6-sol"}),
                },
            )
            .unwrap();
        journal
            .record(
                "t4",
                EventKind::ProviderReceipt {
                    manifest_hash: "subagent".into(),
                    input_tokens: None,
                    output_tokens: None,
                    context_bytes: None,
                    raw_context_tokens: None,
                    provider_data: json!({
                        "source":"subagent",
                        "usageIsDelta":true,
                        "nonCachedInputTokens":20,
                        "cachedInputTokens":0,
                        "thinkingTokens":0,
                        "outputTokens":1
                    }),
                },
            )
            .unwrap();
        journal
            .record(
                "t5",
                EventKind::ToolInvoked {
                    tool_instance: "AnnotateMedia:1".into(),
                    tool_name: "AnnotateMedia".into(),
                    arguments: json!({"model":"gemini-3.1-pro-preview"}),
                    invocation_id: Some("media-1".into()),
                },
            )
            .unwrap();
        journal
            .record(
                "t6",
                EventKind::ProviderReceipt {
                    manifest_hash: "media".into(),
                    input_tokens: None,
                    output_tokens: None,
                    context_bytes: None,
                    raw_context_tokens: None,
                    provider_data: json!({
                        "source":"media_annotation",
                        "usageIsDelta":true,
                        "nonCachedInputTokens":30,
                        "cachedInputTokens":0,
                        "thinkingTokens":0,
                        "outputTokens":1
                    }),
                },
            )
            .unwrap();
        journal
            .record(
                "t7",
                EventKind::ToolCompleted {
                    tool_instance: "AnnotateMedia:1".into(),
                    tool_name: "AnnotateMedia".into(),
                    outcome: json!({"ok":true}),
                    invocation_id: Some("media-1".into()),
                },
            )
            .unwrap();
        journal
            .record(
                "t8",
                EventKind::ToolCompleted {
                    tool_instance: "RunSubagent:1".into(),
                    tool_name: "RunSubagent".into(),
                    outcome: json!({"ok":true}),
                    invocation_id: Some("subagent-1".into()),
                },
            )
            .unwrap();
        journal
            .record(
                "t9",
                EventKind::ProviderReceipt {
                    manifest_hash: "unknown".into(),
                    input_tokens: None,
                    output_tokens: None,
                    context_bytes: None,
                    raw_context_tokens: None,
                    provider_data: json!({
                        "source":"web_search",
                        "usageIsDelta":true,
                        "nonCachedInputTokens":40,
                        "cachedInputTokens":0,
                        "thinkingTokens":0,
                        "outputTokens":1
                    }),
                },
            )
            .unwrap();
        drop(journal);

        let compatible = SessionJournal::open_with_metadata_and_provider_costs(
            &path,
            metadata(),
            Some("gpt-5.6-sol"),
            Some(compatibility_cost),
        )
        .unwrap();
        let status = &compatible.state().projection().status;
        assert_eq!(status.estimated_cost_usd_nanos, 4_060);
        assert_eq!(status.unpriced_provider_calls, 1);
        let inferred_models = compatible
            .state()
            .events
            .iter()
            .filter_map(|event| match &event.kind {
                EventKind::ProviderReceipt { provider_data, .. } => provider_data
                    .get("providerModel")
                    .and_then(Value::as_str)
                    .map(str::to_owned),
                _ => None,
            })
            .collect::<Vec<_>>();
        assert_eq!(
            inferred_models,
            vec!["gpt-5.6-sol", "gpt-5.6-sol", "gemini-3.1-pro-preview"]
        );
        drop(compatible);

        let unchanged = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
        assert_eq!(
            unchanged
                .state()
                .projection()
                .status
                .estimated_cost_usd_nanos,
            0
        );
        assert_eq!(
            unchanged
                .state()
                .projection()
                .status
                .unpriced_provider_calls,
            4
        );
        let archive: Value = serde_json::from_slice(&unchanged.archive_bytes().unwrap()).unwrap();
        let archive_summary = legacy_provider_cost_summary_for_archive(
            &archive,
            Some("gpt-5.6-sol"),
            compatibility_cost,
        )
        .unwrap();
        assert_eq!(archive_summary.estimated_cost_usd_nanos, 4_060);
        assert_eq!(archive_summary.unpriced_provider_calls, 1);
        assert_eq!(archive["context"]["status"]["estimatedCostUsdNanos"], 0);
        assert_eq!(archive["context"]["status"]["unpricedProviderCalls"], 4);
        drop(unchanged);
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn legacy_provider_receipts_without_a_raw_context_anchor_remain_readable() {
        let receipt: EventKind = serde_json::from_value(json!({
            "type":"provider_receipt",
            "manifest_hash":"legacy-manifest",
            "input_tokens":123,
            "output_tokens":4,
            "provider_data":null
        }))
        .unwrap();
        assert!(matches!(
            receipt,
            EventKind::ProviderReceipt {
                context_bytes: None,
                raw_context_tokens: None,
                ..
            }
        ));
    }

    #[test]
    fn stateful_tool_slots_are_batched_append_only_and_do_not_see_summaries() {
        let path = path("slots");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        journal
            .apply_tool_slots(
                "t1",
                "rust-1",
                vec![
                    ToolSlotInput {
                        slot: "a.rs".into(),
                        name: "a.rs".into(),
                        content: BoxContent::text("a"),
                        retired: false,
                    },
                    ToolSlotInput {
                        slot: "b.rs".into(),
                        name: "b.rs".into(),
                        content: BoxContent::text("b"),
                        retired: false,
                    },
                ],
            )
            .unwrap();
        let a = journal.state().tools["rust-1"].slots[0].box_id;
        journal.summarize_box("t2", a, "Kennedy summary").unwrap();
        let before = journal.state().clone();
        assert!(
            journal
                .apply_tool_slots(
                    "t3",
                    "rust-1",
                    vec![ToolSlotInput {
                        slot: "b.rs".into(),
                        name: "b.rs".into(),
                        content: BoxContent::text("b"),
                        retired: false,
                    }]
                )
                .is_err()
        );
        assert_eq!(journal.state(), &before);
        journal
            .apply_tool_slots(
                "t4",
                "rust-1",
                vec![
                    ToolSlotInput {
                        slot: "a.rs".into(),
                        name: "a.rs".into(),
                        content: BoxContent::text("a2"),
                        retired: false,
                    },
                    ToolSlotInput {
                        slot: "b.rs".into(),
                        name: "b.rs".into(),
                        content: BoxContent::text("b"),
                        retired: true,
                    },
                    ToolSlotInput {
                        slot: "c.rs".into(),
                        name: "c.rs".into(),
                        content: BoxContent::text("c"),
                        retired: false,
                    },
                ],
            )
            .unwrap();
        let a_state = journal.state().box_state(a).unwrap();
        assert_eq!(a_state.canonical.content.text, "a2");
        assert!(matches!(
            a_state.representation,
            Representation::Summarized { .. }
        ));
        drop(journal);
        let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
        assert_eq!(reopened.state().tools["rust-1"].slots.len(), 3);
        assert!(reopened.state().tools["rust-1"].slots[1].retired);
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn tool_layout_orders_current_boxes_without_changing_their_identities() {
        let path = path("tool-layout");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        journal
            .apply_tool_slots_with_layout(
                "t1",
                "kweb",
                vec![
                    ToolSlotInput {
                        slot: "active".into(),
                        name: "Active node".into(),
                        content: BoxContent::text("ACTIVE NODE"),
                        retired: false,
                    },
                    ToolSlotInput {
                        slot: "direct".into(),
                        name: "Direct node".into(),
                        content: BoxContent::text("DIRECT NODE"),
                        retired: false,
                    },
                ],
                &["direct".into(), "active".into()],
            )
            .unwrap();
        let tool = &journal.state().tools["kweb"];
        let active_id = tool.slots[0].box_id;
        let direct_id = tool.slots[1].box_id;
        let projected_items = journal.state().projection().items;
        let rendered = projected_items
            .iter()
            .map(|item| item.text.as_str())
            .collect::<Vec<_>>()
            .join("\n\n");
        assert!(rendered.find("DIRECT NODE") < rendered.find("ACTIVE NODE"));
        assert_eq!(
            journal.state().tool_layouts["kweb"],
            vec![direct_id, active_id]
        );
        drop(journal);
        let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
        assert_eq!(reopened.state().projection().items, projected_items);
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn ingress_attempt_reset_restores_the_initial_provider_visible_state() {
        let path = path("ingress-attempt-reset");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        let baseline_box = journal
            .create_box(
                "t1",
                "source",
                BoxOwner::User,
                BoxContent::text("original source"),
            )
            .unwrap();
        journal
            .record(
                "t2",
                EventKind::SourceTerminated {
                    reason: "history_ingress".into(),
                },
            )
            .unwrap();
        journal
            .record("t3", EventKind::HistoryIngressStarted)
            .unwrap();
        let baseline = journal.state().projection().render();

        journal
            .update_box(
                "t4",
                baseline_box,
                BoxContent::text("failed attempt update"),
            )
            .unwrap();
        journal
            .create_box(
                "t5",
                "failed attempt",
                BoxOwner::Kennedy,
                BoxContent::text("discard me"),
            )
            .unwrap();
        journal.reset_history_ingress_attempt("t6").unwrap();

        assert_eq!(journal.state().projection().render(), baseline);
        assert!(journal.state().current_ingress_attempt_events().is_empty());
        drop(journal);
        let reopened = SessionJournal::open_with_metadata(&path, metadata()).unwrap();
        assert_eq!(reopened.state().projection().render(), baseline);
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn ingress_time_markers_are_sparse_resettable_and_system_bracketed() {
        let path = path("ingress-time-markers");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        journal
            .create_box(
                "t1",
                "source",
                BoxOwner::User,
                BoxContent::text("short source"),
            )
            .unwrap();
        journal
            .record(
                "t2",
                EventKind::SourceTerminated {
                    reason: "history_ingress".into(),
                },
            )
            .unwrap();
        journal
            .record("t3", EventKind::HistoryIngressStarted)
            .unwrap();

        journal
            .prepare_provider_projection_with_ingress_time("t4", &[], "m1", None, 2_700, false)
            .unwrap();
        let mut synchronized_after = journal.state().events.last().unwrap().id;
        let twenty = journal
            .prepare_provider_resume_markers("t5", synchronized_after, Some((1_200, false)))
            .unwrap();
        assert_eq!(twenty, ["[ingress time remaining: 1200 seconds]"]);
        synchronized_after = journal.state().events.last().unwrap().id;
        assert!(
            journal
                .prepare_provider_resume_markers("t6", synchronized_after, Some((1_199, false)),)
                .unwrap()
                .is_empty()
        );
        let ten = journal
            .prepare_provider_resume_markers("t7", synchronized_after, Some((600, false)))
            .unwrap();
        assert_eq!(
            ten,
            [
                "[ingress time remaining: 600 seconds. If you do not call EndSession before time expires, the ingress will fail and all Kmap updates will be lost.]"
            ],
        );
        let markers = journal
            .state()
            .projection()
            .items
            .into_iter()
            .filter(|item| item.text.contains("ingress time remaining"))
            .map(|item| item.text)
            .collect::<Vec<_>>();
        assert_eq!(markers.len(), 3);
        assert_eq!(markers[0], "[ingress time remaining: 2700 seconds]");
        assert_eq!(markers[1], "[ingress time remaining: 1200 seconds]");
        assert_eq!(
            markers[2],
            "[ingress time remaining: 600 seconds. If you do not call EndSession before time expires, the ingress will fail and all Kmap updates will be lost.]"
        );
        assert!(should_append_ingress_time_marker(
            IngressTimeMarkerState {
                any: true,
                at_or_below_thirty_minutes: true,
                at_or_below_ten_minutes: true,
            },
            500,
            true,
        ));

        journal.reset_history_ingress_attempt("t8").unwrap();
        let retry = journal
            .prepare_provider_projection_with_ingress_time("t9", &[], "m1", None, 2_700, true)
            .unwrap();
        assert!(retry.projection.render().contains(
            "[Previous attempt ran out of time; please budget your ingress time carefully. Ingress time remaining: 2700 seconds]"
        ));
        assert!(
            !retry
                .projection
                .render()
                .contains("[ingress time remaining: 600 seconds. If you do not call EndSession")
        );
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn limits_use_exact_floor_percentages() {
        let state = Chatend::opened(SessionMetadata {
            effective_context_tokens: 101,
            ..metadata()
        });
        assert_eq!(state.live_context_limit(), 70);
        assert_eq!(state.forced_ingress_context_limit(), 75);
        assert_eq!(state.ingress_initial_context_limit(), 75);
        assert_eq!(state.ingress_context_limit(), 90);
        assert_eq!(state.active_context_limit(), 70);
        for kind in [
            SessionKind::SelfTime,
            SessionKind::HistoryIngress,
            SessionKind::AudioIngress,
        ] {
            let writer = Chatend::opened(SessionMetadata {
                kind,
                effective_context_tokens: 101,
                ..metadata()
            });
            assert_eq!(writer.active_context_limit(), 90);
        }
        for kind in [SessionKind::Telegram, SessionKind::Other("test".into())] {
            let ordinary = Chatend::opened(SessionMetadata {
                kind,
                effective_context_tokens: 101,
                ..metadata()
            });
            assert_eq!(ordinary.active_context_limit(), 70);
        }
        assert_eq!(estimate_tokens("1234"), 1);
    }

    #[test]
    fn new_box_projection_preview_matches_the_committed_projection() {
        let path = path("new-box-preview");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        let boxes = vec![
            (
                "User message".into(),
                BoxOwner::User,
                BoxContent::text("prospective user text"),
            ),
            (
                "attachment".into(),
                BoxOwner::User,
                BoxContent::text("prospective attachment text"),
            ),
        ];
        let preview = journal
            .state()
            .projection_with_new_boxes_at("t1", &boxes)
            .unwrap();
        for (name, owner, content) in boxes {
            journal.create_box("t1", name, owner, content).unwrap();
        }
        assert_eq!(journal.state().projection(), preview);
        std::fs::remove_file(path).unwrap();
    }

    #[test]
    fn box_representation_preview_matches_the_batched_append() {
        let path = path("representation-plan");
        let mut journal = SessionJournal::create(&path, metadata()).unwrap();
        let hydrated = journal
            .create_box(
                "t1",
                "large result",
                BoxOwner::Controller,
                BoxContent::text("x".repeat(3_000)),
            )
            .unwrap();
        let summarized = journal
            .create_box(
                "t2",
                "Kennedy summary",
                BoxOwner::Kennedy,
                BoxContent::text("y".repeat(3_000)),
            )
            .unwrap();
        journal
            .summarize_box("t3", summarized, "important points")
            .unwrap();
        let desired = BTreeMap::from([
            (hydrated, BoxRepresentation::Dehydrated),
            (
                summarized,
                BoxRepresentation::Summarized("important points".into()),
            ),
        ]);
        let preview = journal
            .state()
            .projection_with_box_representations(&desired)
            .unwrap();
        let next_id = journal.state().next_id;
        let ids = journal.apply_box_representations("t4", &desired).unwrap();
        assert_eq!(ids, vec![EventId(next_id)]);
        assert_eq!(journal.state().projection(), preview);
        assert!(matches!(
            journal.state().box_state(hydrated).unwrap().representation,
            Representation::Dehydrated { .. }
        ));
        assert_eq!(
            journal
                .state()
                .box_state(summarized)
                .unwrap()
                .representation,
            Representation::Summarized {
                based_on: EventId(2),
                text: "important points".into(),
            }
        );
        std::fs::remove_file(path).unwrap();
    }
}