kcode_session_history_catalog/

lib.rs

//! Append-only completion receipts and Session History read projections.

use std::{
    collections::{BTreeMap, HashSet},
    fs::{File, OpenOptions},
    io::{BufRead, BufReader, Write},
    path::{Path, PathBuf},
    sync::{Arc, Mutex, MutexGuard},
};

use anyhow::Context as _;
use kcode_chatend::SessionHistoryIntegration;
pub use kcode_session_control_state::SessionRecord;
use kcode_session_log::Role;
pub use kcode_session_log::SessionLog;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

/// Input used to persist one completed session receipt.
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct RecordCompletion {
    pub session_object_id: String,
    #[serde(default)]
    pub commit_receipt: Option<CompletionReceipt>,
    #[serde(default)]
    pub session_id: Option<String>,
    #[serde(default)]
    pub session_type: Option<String>,
    #[serde(default)]
    pub created_at: Option<String>,
}

/// The durable, camel-case completion receipt stored in the catalog.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CompletionReceipt {
    #[serde(default)]
    pub transaction_id: Option<String>,
    pub session_object_id: String,
    #[serde(default)]
    pub session_id: Option<String>,
    #[serde(default)]
    pub session_type: Option<String>,
    #[serde(default)]
    pub created_at: Option<String>,
    #[serde(default)]
    pub committed_at: Option<String>,
    #[serde(default)]
    pub ingress_source: Option<Value>,
    #[serde(default)]
    pub node_ids: BTreeMap<String, String>,
    #[serde(default)]
    pub object_ids: BTreeMap<String, String>,
}

/// The effective first receipt and whether this call appended it.
#[derive(Clone, Debug, PartialEq)]
pub struct RecordedCompletion {
    pub receipt: CompletionReceipt,
    pub appended: bool,
}

/// Catalog identity conflicts and local storage failures.
#[derive(Debug)]
pub enum Error {
    Conflict(String),
    Storage(anyhow::Error),
}

impl std::fmt::Display for Error {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Conflict(message) => formatter.write_str(message),
            Self::Storage(error) => write!(formatter, "{error:#}"),
        }
    }
}

impl std::error::Error for Error {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Conflict(_) => None,
            Self::Storage(error) => Some(error.as_ref()),
        }
    }
}

impl From<anyhow::Error> for Error {
    fn from(error: anyhow::Error) -> Self {
        Self::Storage(error)
    }
}

/// A cheaply cloneable handle to one completion catalog.
#[derive(Clone, Debug)]
pub struct Catalog {
    path: PathBuf,
    mutation: Arc<Mutex<()>>,
}

impl Catalog {
    pub fn open(path: PathBuf) -> Result<Self, Error> {
        let parent = path
            .parent()
            .filter(|parent| !parent.as_os_str().is_empty())
            .unwrap_or_else(|| Path::new("."));
        create_private_directory(parent)?;
        if !path.exists() {
            let file = OpenOptions::new()
                .create_new(true)
                .write(true)
                .open(&path)
                .with_context(|| format!("creating {}", path.display()))?;
            file.sync_all()
                .with_context(|| format!("synchronizing {}", path.display()))?;
            sync_directory(parent)?;
        }
        Ok(Self {
            path,
            mutation: Arc::new(Mutex::new(())),
        })
    }

    pub fn receipts(&self) -> Result<Vec<CompletionReceipt>, Error> {
        let _guard = self.lock()?;
        read_completion_receipts(&self.path).map_err(Into::into)
    }

    pub fn record(
        &self,
        input: RecordCompletion,
        committed_at: String,
    ) -> Result<RecordedCompletion, Error> {
        let _guard = self.lock()?;
        let mut receipt = input.commit_receipt.unwrap_or(CompletionReceipt {
            transaction_id: None,
            session_object_id: input.session_object_id.clone(),
            session_id: None,
            session_type: None,
            created_at: None,
            committed_at: None,
            ingress_source: None,
            node_ids: BTreeMap::new(),
            object_ids: BTreeMap::new(),
        });
        if receipt.session_object_id != input.session_object_id {
            return Err(Error::Conflict(
                "completion receipt and requested session object differ".into(),
            ));
        }
        receipt.session_id = receipt.session_id.or(input.session_id);
        receipt.session_type = receipt.session_type.or(input.session_type);
        receipt.created_at = receipt.created_at.or(input.created_at);
        receipt.committed_at.get_or_insert(committed_at);

        if let Some(existing) = read_completion_receipts(&self.path)?
            .into_iter()
            .find(|existing| existing.session_object_id == receipt.session_object_id)
        {
            return Ok(RecordedCompletion {
                receipt: existing,
                appended: false,
            });
        }

        let mut file = OpenOptions::new()
            .append(true)
            .open(&self.path)
            .with_context(|| format!("opening {} for append", self.path.display()))?;
        let encoded = serde_json::to_string(&receipt)
            .context("encoding Session History completion receipt")?;
        writeln!(file, "{encoded}")
            .with_context(|| format!("appending completion receipt to {}", self.path.display()))?;
        file.flush()
            .with_context(|| format!("flushing {}", self.path.display()))?;
        file.sync_data()
            .with_context(|| format!("synchronizing {}", self.path.display()))?;
        Ok(RecordedCompletion {
            receipt,
            appended: true,
        })
    }

    fn lock(&self) -> Result<MutexGuard<'_, ()>, Error> {
        self.mutation
            .lock()
            .map_err(|error| anyhow::anyhow!("completion catalog lock is poisoned: {error}"))
            .map_err(Into::into)
    }
}

fn read_completion_receipts(path: &Path) -> anyhow::Result<Vec<CompletionReceipt>> {
    let file = File::open(path).with_context(|| format!("opening {}", path.display()))?;
    let mut receipts = Vec::new();
    let mut seen = HashSet::new();
    for (line_index, line) in BufReader::new(file).lines().enumerate() {
        let line =
            line.with_context(|| format!("reading completion receipt line {}", line_index + 1))?;
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let receipt = if line.starts_with('{') {
            serde_json::from_str::<CompletionReceipt>(line).with_context(|| {
                format!(
                    "decoding Session History completion receipt line {}",
                    line_index + 1
                )
            })?
        } else {
            CompletionReceipt {
                transaction_id: None,
                session_object_id: line.to_owned(),
                session_id: None,
                session_type: None,
                created_at: None,
                committed_at: None,
                ingress_source: None,
                node_ids: BTreeMap::new(),
                object_ids: BTreeMap::new(),
            }
        };
        if seen.insert(receipt.session_object_id.clone()) {
            receipts.push(receipt);
        }
    }
    Ok(receipts)
}

fn create_private_directory(path: &Path) -> anyhow::Result<()> {
    if path.is_dir() {
        return Ok(());
    }
    let mut builder = std::fs::DirBuilder::new();
    builder.recursive(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::DirBuilderExt as _;
        builder.mode(0o700);
    }
    builder
        .create(path)
        .with_context(|| format!("creating {}", path.display()))?;
    let parent = path
        .parent()
        .filter(|parent| !parent.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."));
    sync_directory(parent)
}

fn sync_directory(path: &Path) -> anyhow::Result<()> {
    File::open(path)
        .with_context(|| format!("opening directory {} for sync", path.display()))?
        .sync_all()
        .with_context(|| format!("syncing directory {}", path.display()))
}

/// Optional compatibility callbacks for legacy provider-cost projection.
#[derive(Clone, Copy)]
pub struct ProviderCostCompatibility {
    pub session_model: fn(&Value) -> Option<String>,
    pub estimator: kcode_chatend::ProviderCostEstimator,
}

impl std::fmt::Debug for ProviderCostCompatibility {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ProviderCostCompatibility")
            .finish_non_exhaustive()
    }
}

/// Pure projections over caller-supplied control and transcript values.
#[derive(Clone, Debug)]
pub struct ReadModel {
    pub provider_cost_compatibility: Option<ProviderCostCompatibility>,
}

impl ReadModel {
    pub fn prepare_control_state(&self, state: &mut Value) {
        retain_summary_fields(state);
    }

    pub fn active_summary(&self, mut record: SessionRecord) -> SessionRecord {
        record.summary = true;
        record.state = summary_state(&record.state);
        record
    }

    pub fn active(&self, mut record: SessionRecord, log: &SessionLog) -> SessionRecord {
        record.state["sessionId"] = json!(log.header.session_id);
        if !record.state.get("transcript").is_some_and(Value::is_array) {
            record.state["transcript"] = Value::Array(
                log.events
                    .iter()
                    .enumerate()
                    .filter_map(|(position, event)| transcript_entry(position, event))
                    .collect(),
            );
        }
        if !record.state.get("events").is_some_and(Value::is_array) {
            record.state["events"] = serde_json::to_value(&log.events).unwrap_or(Value::Null);
        }
        clear_chatend_projection(&mut record.state);

        let context_state = record
            .state
            .get("historyIngress")
            .filter(|state| state.get("chatendMetadata").is_some())
            .unwrap_or(&record.state);
        let default_provider_model = self
            .provider_cost_compatibility
            .and_then(|compatibility| (compatibility.session_model)(context_state))
            .or_else(|| {
                self.provider_cost_compatibility
                    .and_then(|compatibility| (compatibility.session_model)(&record.state))
            });
        let exact_chatend = context_state
            .get("chatendMetadata")
            .cloned()
            .and_then(|value| serde_json::from_value::<kcode_chatend::SessionMetadata>(value).ok())
            .and_then(|metadata| match self.provider_cost_compatibility {
                Some(compatibility) => SessionHistoryIntegration::replay(
                    metadata,
                    log,
                    default_provider_model.as_deref(),
                    Some(compatibility.estimator),
                )
                .ok(),
                None => SessionHistoryIntegration::replay(metadata, log, None, None).ok(),
            });
        if let Some(chatend) = exact_chatend {
            let boxes = serde_json::to_value(&chatend.boxes).unwrap_or(Value::Null);
            let projection = chatend.projection();
            let submitted = chatend.events.iter().rev().find_map(|event| {
                let kcode_chatend::EventKind::ProviderInputSubmitted { round, context, .. } =
                    &event.kind
                else {
                    return None;
                };
                Some((event.recorded_at.as_str(), *round, context))
            });
            let (chatend_text, chatend_text_source, structured_material) = match submitted {
                Some((submitted_at, round, submitted)) => (
                    Value::String(submitted.input.clone()),
                    Value::String("submitted".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,
                    }),
                ),
                None => (
                    Value::String(projection.render()),
                    Value::String("reconstructed".into()),
                    Value::Null,
                ),
            };
            let context = serde_json::to_value(projection).unwrap_or(Value::Null);
            record.state["boxes"] = boxes.clone();
            record.state["context"] = context.clone();
            record.state["chatendText"] = chatend_text.clone();
            record.state["chatendTextSource"] = chatend_text_source.clone();
            record.state["structuredMaterial"] = structured_material.clone();
            if let Some(ingress) = record
                .state
                .get_mut("historyIngress")
                .and_then(Value::as_object_mut)
            {
                ingress.insert("boxes".into(), boxes);
                ingress.insert("context".into(), context);
                ingress.insert("chatendText".into(), chatend_text);
                ingress.insert("chatendTextSource".into(), chatend_text_source);
                ingress.insert("structuredMaterial".into(), structured_material);
            }
        }
        record
    }

    pub fn completed(&self, receipt: CompletionReceipt, summary: bool) -> SessionRecord {
        let object_id = receipt.session_object_id.clone();
        let started_at = receipt.created_at.clone().unwrap_or_default();
        let updated_at = receipt.committed_at.clone().unwrap_or_default();
        SessionRecord {
            id: object_id.clone(),
            phase: "complete".into(),
            started_at,
            updated_at,
            state: json!({
                "sessionObjectId":object_id,
                "sessionId":receipt.session_id.clone(),
                "sessionType":receipt.session_type.clone(),
                "ingressSource":receipt.ingress_source.clone(),
                "commitReceipt":receipt,
            }),
            provenance_id: None,
            version: 1,
            last_user_message_at: None,
            ended_at: None,
            ingress_failure_count: 0,
            ingress_failures: json!([]),
            ingress_next_attempt_at: None,
            summary,
        }
    }

    pub fn legacy_provider_cost_summary_for_archive(
        &self,
        archive: &Value,
        session_state: Option<&Value>,
    ) -> anyhow::Result<Option<kcode_chatend::ProviderCostSummary>> {
        let Some(compatibility) = self.provider_cost_compatibility else {
            return Ok(None);
        };
        if is_metadata_free_session_log_archive(archive) {
            return Ok(None);
        }
        let default_provider_model =
            session_state.and_then(|state| (compatibility.session_model)(state));
        SessionHistoryIntegration::legacy_provider_cost_summary_for_archive(
            archive,
            default_provider_model.as_deref(),
            compatibility.estimator,
        )
        .map(Some)
    }
}

fn clear_chatend_projection(state: &mut Value) {
    const FIELDS: [&str; 5] = [
        "boxes",
        "context",
        "chatendText",
        "chatendTextSource",
        "structuredMaterial",
    ];
    let Some(state) = state.as_object_mut() else {
        return;
    };
    for field in FIELDS {
        state.remove(field);
    }
    if let Some(ingress) = state
        .get_mut("historyIngress")
        .and_then(Value::as_object_mut)
    {
        for field in FIELDS {
            ingress.remove(field);
        }
    }
}

fn retain_summary_fields(state: &mut Value) {
    if state
        .get("firstUserMessage")
        .and_then(Value::as_str)
        .is_some()
    {
        return;
    }
    let Some(first_user) = state
        .get("transcript")
        .and_then(Value::as_array)
        .and_then(|transcript| {
            transcript
                .iter()
                .find(|entry| entry.get("role").and_then(Value::as_str) == Some("user"))
        })
        .and_then(|entry| entry.get("content"))
        .and_then(Value::as_str)
    else {
        return;
    };
    state["firstUserMessage"] = Value::String(first_user.chars().take(512).collect());
}

fn summary_state(control: &Value) -> Value {
    json!({
        "sessionType":control.get("sessionType"),
        "channel":control.get("channel"),
        "freeTime":control.get("freeTime"),
        "orchestration":control.get("orchestration"),
        "ingressSource":control.get("ingressSource"),
        "firstUserMessage":control.get("firstUserMessage"),
        "boxCount":control.get("boxCount"),
        "eventCount":control.get("eventCount"),
        "pendingTurn":control.get("pendingTurn").cloned().unwrap_or(Value::Bool(false)),
    })
}

fn persisted_context_kind(event: &kcode_session_log::SessionEvent) -> Option<Value> {
    serde_json::from_str::<Value>(&event.text)
        .ok()?
        .get("kind")
        .cloned()
}

fn display_text(event: &kcode_session_log::SessionEvent) -> String {
    persisted_context_kind(event)
        .and_then(|kind| {
            (kind.get("type").and_then(Value::as_str) == Some("box_created"))
                .then(|| {
                    kind.get("content")
                        .and_then(|content| content.get("text"))
                        .and_then(Value::as_str)
                        .map(str::to_owned)
                })
                .flatten()
        })
        .unwrap_or_else(|| event.text.clone())
}

fn transcript_entry(position: usize, event: &kcode_session_log::SessionEvent) -> Option<Value> {
    let kind = persisted_context_kind(event);
    let box_content = kind
        .as_ref()
        .filter(|kind| kind.get("type").and_then(Value::as_str) == Some("box_created"))
        .and_then(|kind| kind.get("content"));
    let metadata = box_content
        .and_then(|content| content.get("metadata"))
        .filter(|value| value.is_object());
    let role = match event.role {
        Role::UserMessage => "user",
        Role::KennedyMessage => "kennedy",
        Role::SystemError => "system",
        Role::SystemMessage => (box_content?
            .get("metadata")
            .and_then(|metadata| metadata.get("transcriptRole"))
            .and_then(Value::as_str)
            == Some("system"))
        .then_some("system")?,
        _ => return None,
    };
    let mut item = json!({
        "role":role,
        "content":display_text(event),
        "boxId":position + 1,
    });
    if let Some(objects) = box_content
        .and_then(|content| content.get("objects"))
        .filter(|value| value.is_array())
    {
        item["objects"] = objects.clone();
    }
    if let Some(metadata) = metadata {
        for key in ["inputKind", "externalEventId"] {
            if let Some(value) = metadata.get(key) {
                item[key] = value.clone();
            }
        }
        if let Some(attachments) = metadata.get("attachments").filter(|value| value.is_array()) {
            item["attachments"] = attachments.clone();
        } else if let Some(media) = metadata.get("media").filter(|value| value.is_object()) {
            item["attachments"] = json!([media]);
        }
    }
    Some(item)
}

fn is_metadata_free_session_log_archive(archive: &Value) -> bool {
    if archive.get("metadata").is_some() {
        return false;
    }
    let Some(header) = archive.get("header") else {
        return false;
    };
    if header.get("formatVersion").and_then(Value::as_str)
        != Some(kcode_session_log::FORMAT_VERSION)
        || !header
            .get("sessionId")
            .and_then(Value::as_str)
            .is_some_and(|value| !value.trim().is_empty())
        || !header
            .get("createdAt")
            .and_then(Value::as_str)
            .is_some_and(|value| !value.trim().is_empty())
    {
        return false;
    }
    archive
        .get("events")
        .and_then(Value::as_array)
        .is_some_and(|events| {
            events.iter().all(|event| {
                event.get("text").and_then(Value::as_str).is_some()
                    && event
                        .get("role")
                        .and_then(Value::as_str)
                        .is_some_and(|role| {
                            matches!(
                                role,
                                "system-message"
                                    | "system-error"
                                    | "user-message"
                                    | "kennedy-message"
                                    | "kennedy-tool-call"
                                    | "tool-result"
                                    | "tool-error"
                                    | "object"
                                    | "pending-object"
                            )
                        })
            })
        })
}

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

    use super::*;
    use kcode_chatend::{
        BoxContent, BoxOwner, CacheExpectation, EventKind, ProviderContext, SessionKind,
        SessionMetadata,
    };
    use kcode_session_log::SessionStore;

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

    fn receipt(id: &str, transaction: &str) -> CompletionReceipt {
        CompletionReceipt {
            transaction_id: Some(transaction.into()),
            session_object_id: id.into(),
            session_id: Some("session-42".into()),
            session_type: Some("conversation".into()),
            created_at: Some("2026-08-14T00:00:00Z".into()),
            committed_at: None,
            ingress_source: Some(json!({"idempotencyId":"ingress-1"})),
            node_ids: BTreeMap::new(),
            object_ids: BTreeMap::new(),
        }
    }

    fn input(id: &str, receipt: Option<CompletionReceipt>) -> RecordCompletion {
        RecordCompletion {
            session_object_id: id.into(),
            commit_receipt: receipt,
            session_id: Some("fallback-session".into()),
            session_type: Some("fallback-type".into()),
            created_at: Some("fallback-created".into()),
        }
    }

    fn session_record(state: Value) -> SessionRecord {
        SessionRecord {
            id: "session-42".into(),
            phase: "active".into(),
            started_at: "2026-08-14T00:00:00Z".into(),
            updated_at: "2026-08-14T00:00:00Z".into(),
            state,
            provenance_id: None,
            version: 1,
            last_user_message_at: None,
            ended_at: None,
            ingress_failure_count: 0,
            ingress_failures: json!([]),
            ingress_next_attempt_at: None,
            summary: false,
        }
    }

    #[test]
    fn catalog_preserves_legacy_first_writer_and_conflict_behavior() {
        let root = root("compatibility");
        std::fs::create_dir_all(&root).unwrap();
        let path = root.join("completed.txt");
        std::fs::write(&path, "legacy-object\n\n").unwrap();
        let catalog = Catalog::open(path.clone()).unwrap();

        let first = catalog
            .record(
                input("archive-1", Some(receipt("archive-1", "transaction-1"))),
                "2026-08-14T01:00:00Z".into(),
            )
            .unwrap();
        assert!(first.appended);
        assert_eq!(
            first.receipt.committed_at.as_deref(),
            Some("2026-08-14T01:00:00Z")
        );
        let duplicate = catalog
            .record(
                input("archive-1", Some(receipt("archive-1", "replacement"))),
                "later".into(),
            )
            .unwrap();
        assert!(!duplicate.appended);
        assert_eq!(
            duplicate.receipt.transaction_id.as_deref(),
            Some("transaction-1")
        );

        let receipts = catalog.receipts().unwrap();
        assert_eq!(receipts.len(), 2);
        assert_eq!(receipts[0].session_object_id, "legacy-object");
        assert_eq!(receipts[1].session_object_id, "archive-1");
        let persisted = std::fs::read_to_string(&path).unwrap();
        assert!(persisted.contains("\"sessionObjectId\":\"archive-1\""));
        assert!(!persisted.contains("replacement"));

        let conflict = catalog
            .record(
                input("archive-2", Some(receipt("other-archive", "transaction-2"))),
                "now".into(),
            )
            .unwrap_err();
        assert!(matches!(conflict, Error::Conflict(_)));

        std::fs::write(&path, "legacy-object\n{broken}\n").unwrap();
        let malformed = catalog.receipts().unwrap_err().to_string();
        assert!(malformed.contains("line 2"));
        let _ = std::fs::remove_dir_all(root);
    }

    #[test]
    fn cloned_catalog_serializes_duplicate_append() {
        let root = root("synchronized");
        let catalog = Catalog::open(root.join("nested/completed.txt")).unwrap();
        let handles = (0..8)
            .map(|index| {
                let catalog = catalog.clone();
                std::thread::spawn(move || {
                    catalog
                        .record(input("archive-1", None), format!("committed-{index}"))
                        .unwrap()
                        .appended
                })
            })
            .collect::<Vec<_>>();
        assert_eq!(
            handles
                .into_iter()
                .map(|handle| handle.join().unwrap())
                .filter(|appended| *appended)
                .count(),
            1
        );
        assert_eq!(catalog.receipts().unwrap().len(), 1);
        let _ = std::fs::remove_dir_all(root);
    }

    #[test]
    fn summaries_preserve_the_existing_unicode_bound() {
        let model = ReadModel {
            provider_cost_compatibility: None,
        };
        let mut state = json!({
            "sessionType":"conversation",
            "transcript":[{"role":"user","content":"é".repeat(600)}],
            "boxCount":2,
            "eventCount":3,
        });
        model.prepare_control_state(&mut state);
        assert_eq!(
            state["firstUserMessage"].as_str().unwrap().chars().count(),
            512
        );
        let summary = model.active_summary(session_record(state));
        assert!(summary.summary);
        assert!(summary.state.get("transcript").is_none());
        assert_eq!(summary.state["boxCount"], 2);
        assert_eq!(summary.state["pendingTurn"], false);
    }

    #[test]
    fn active_projection_reconstructs_chatend_and_discards_failed_replay_fields() {
        let root = root("active");
        std::fs::create_dir_all(&root).unwrap();
        let metadata = SessionMetadata {
            session_id: "session-42".into(),
            kind: SessionKind::Conversation,
            created_at: "2026-08-14T00:00:00Z".into(),
            effective_context_tokens: 100_000,
            channel: Value::Null,
        };
        let mut source = SessionHistoryIntegration::create_session(
            root.join("session-42.session-log"),
            metadata.clone(),
        )
        .unwrap();
        source
            .create_box(
                "2026-08-14T00:00:01Z",
                "User message",
                BoxOwner::User,
                BoxContent::text("hello from the log"),
            )
            .unwrap();
        source
            .record(
                "2026-08-14T00:00:02Z",
                EventKind::ProviderInputSubmitted {
                    round: 1,
                    context: ProviderContext {
                        input: "exact provider input".into(),
                        provider: "provider".into(),
                        model: "model".into(),
                        reasoning_effort: "medium".into(),
                        base_instructions: Some("base".into()),
                        developer_instructions: None,
                        tools: Vec::new(),
                    },
                    transport_input_hash: None,
                    transport_input_bytes: None,
                    thread_action: None,
                    thread_reset_reason: None,
                    cacheable_prefix_bytes: 0,
                    material_fingerprint: "fixture".into(),
                    cache_expectation: CacheExpectation::ColdStart.label().into(),
                    planned_invalidation_reason: None,
                },
            )
            .unwrap();
        drop(source);
        let log = SessionStore::new(&root)
            .open_session("session-42")
            .unwrap()
            .list();
        let model = ReadModel {
            provider_cost_compatibility: None,
        };
        let active = model.active(session_record(json!({"chatendMetadata":metadata})), &log);
        assert_eq!(
            active.state["transcript"][0]["content"],
            "hello from the log"
        );
        assert!(active.state["events"].is_array());
        assert!(active.state["boxes"].is_object());
        assert_eq!(active.state["chatendText"], "exact provider input");
        assert_eq!(active.state["chatendTextSource"], "submitted");
        assert_eq!(active.state["structuredMaterial"]["round"], 1);

        let mut bad_metadata = metadata;
        bad_metadata.session_id = "different-session".into();
        let failed = model.active(
            session_record(json!({
                "chatendMetadata":bad_metadata,
                "boxes":{"stale":true},
                "context":{"stale":true},
                "chatendText":"stale",
                "chatendTextSource":"stale",
                "structuredMaterial":{"stale":true},
            })),
            &log,
        );
        for field in [
            "boxes",
            "context",
            "chatendText",
            "chatendTextSource",
            "structuredMaterial",
        ] {
            assert!(failed.state.get(field).is_none());
        }
        let _ = std::fs::remove_dir_all(root);
    }

    fn no_session_model(_: &Value) -> Option<String> {
        None
    }

    fn no_cost(
        _: &str,
        _: &kcode_chatend::ProviderMetering,
    ) -> Option<kcode_chatend::ProviderCostEstimate> {
        None
    }

    #[test]
    fn completed_and_metadata_free_cost_compatibility_match_legacy_behavior() {
        let model = ReadModel {
            provider_cost_compatibility: Some(ProviderCostCompatibility {
                session_model: no_session_model,
                estimator: no_cost,
            }),
        };
        let completed = model.completed(receipt("archive-1", "transaction-1"), false);
        assert_eq!(completed.id, "archive-1");
        assert_eq!(completed.phase, "complete");
        assert_eq!(completed.state["sessionObjectId"], "archive-1");
        assert_eq!(
            completed.state["commitReceipt"]["transactionId"],
            "transaction-1"
        );

        let archive = json!({
            "header":{
                "formatVersion":kcode_session_log::FORMAT_VERSION,
                "sessionId":"session-42",
                "createdAt":"2026-08-14T00:00:00Z"
            },
            "events":[{"role":"user-message","text":"hello"}]
        });
        assert!(
            model
                .legacy_provider_cost_summary_for_archive(&archive, None)
                .unwrap()
                .is_none()
        );
    }
}