kcode_session_history/

lib.rs

//! Session lifecycle and local Session History.
//!
//! In-progress transcript entries live in `kcode-session-log`. Typed lifecycle,
//! command, and stop records live in a separate `.session-control` journal.
//! Successfully committed sessions leave a synchronized local receipt for
//! their immutable Kweb object.

/// Provider-independent Chatend state and durable session API.
///
/// This re-export preserves the original `kcode_session_history::chatend`
/// boundary while `kcode-chatend` owns its cohesive implementation.
pub mod chatend {
    pub use kcode_chatend::{
        BoxContent, BoxId, BoxOwner, BoxRepresentation, BoxState, CanonicalRevision, Chatend,
        ContextProjection, ESTIMATED_BYTES_PER_TOKEN, Event, EventId, EventKind, FORMAT_VERSION,
        MAX_OBJECT_BYTES, ObjectLocation, ObjectMetadata, PendingId, PendingKind, ProjectionItem,
        ProviderCostEstimate, ProviderCostEstimator, ProviderCostSummary, ProviderMetering,
        ProviderTokenUsage, Representation, Session, SessionKind, SessionMetadata, SessionStatus,
        ToolSlot, ToolSlotInput, ToolState, Transition, estimate_tokens,
    };
}
pub use chatend::Session;
pub use kcode_session_control_state::{SessionCommand, SessionRecord, SessionStopRequest};

use std::{
    collections::{BTreeMap, HashMap, HashSet},
    fs::{File, OpenOptions},
    io::{BufRead, BufReader, Write},
    path::{Path as FilePath, PathBuf},
    sync::{
        Arc, Mutex, Weak,
        atomic::{AtomicBool, Ordering},
    },
};

use anyhow::{Context as _, ensure};
use chrono::{DateTime, Duration, Utc};
use kcode_chatend::SessionHistoryIntegration;
use kcode_session_control_state::{ControlProjection, ControlUpdate, OpenMode, SessionControl};
use kcode_session_log::{EventPosition, Role, Session as DurableSession, SessionLog, SessionStore};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use tokio::sync::Notify;
use uuid::Uuid;

const INGRESS_FAILURE_LIMIT: i64 = 5;
const INGRESS_RETRY_DELAY_SECONDS: i64 = 15;
const RETAINED_INGRESS_FAILURES: usize = 5;

struct SessionJournal {
    log: DurableSession,
    control: SessionControl,
}

impl SessionJournal {
    fn create(directory: &FilePath, id: &str, created_at: &str) -> anyhow::Result<Self> {
        let log = SessionStore::new(directory).create_session(id, created_at)?;
        let control = SessionControl::open(directory, id, OpenMode::CreateNew)?
            .context("created session-control journal was unexpectedly absent")?;
        Ok(Self { log, control })
    }

    fn open(path: impl AsRef<FilePath>) -> anyhow::Result<Self> {
        let path = path.as_ref();
        ensure!(
            path.extension().and_then(|value| value.to_str()) == Some("session-log"),
            "{} is not a session-log path",
            path.display()
        );
        let directory = path.parent().unwrap_or_else(|| FilePath::new("."));
        let id = path
            .file_stem()
            .and_then(|value| value.to_str())
            .context("session-log filename is not valid UTF-8")?;
        let log = SessionStore::new(directory).open_session(id)?;
        let control = SessionControl::open(directory, id, OpenMode::OpenOrCreate)?
            .context("opened session-control journal was unexpectedly absent")?;
        Ok(Self { log, control })
    }

    fn open_existing(path: impl AsRef<FilePath>) -> anyhow::Result<Option<Self>> {
        let path = path.as_ref();
        let Some(control) = open_listed_control(path)? else {
            return Ok(None);
        };
        if control.projection().lifecycle.is_none() {
            return Ok(None);
        }
        let (directory, id) = listed_session_parts(path)?;
        let log = match SessionStore::new(directory).open_session(id) {
            Ok(log) => log,
            Err(_) if !path.exists() => return Ok(None),
            Err(error) => return Err(error),
        };
        Ok(Some(Self { log, control }))
    }

    fn list(&self) -> SessionLog {
        self.log.list()
    }

    fn projection(&self) -> ControlProjection {
        self.control.projection()
    }

    fn append_control(&mut self, update: ControlUpdate) -> anyhow::Result<ControlUpdate> {
        self.control.append(now(), update)
    }

    fn stage_object(
        &mut self,
        media_type: String,
        file_name: Option<String>,
        bytes: &[u8],
    ) -> anyhow::Result<String> {
        let file_name = file_name.unwrap_or_else(|| "uploaded-object".into());
        let position =
            self.log
                .add_pending_object(file_name.clone(), file_name, media_type, bytes)?;
        Ok(format!("pending:{}", position.index() + 1))
    }
}

#[derive(Clone, Debug)]
pub struct Config {
    pub directory: PathBuf,
    pub completed_list: PathBuf,
    pub provider_cost_compatibility: Option<ProviderCostCompatibility>,
}

#[derive(Clone, Copy)]
pub struct ProviderCostCompatibility {
    pub session_model: fn(&Value) -> Option<String>,
    pub estimator: 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()
    }
}

#[derive(Clone, Debug)]
pub struct NewSession {
    pub kind: chatend::SessionKind,
    pub created_at: String,
    pub effective_context_tokens: u64,
    pub channel: Value,
}

#[derive(Clone)]
struct AppState {
    config: Config,
    catalog_mutation: Arc<Mutex<()>>,
    session_mutations: Arc<Mutex<HashMap<String, Weak<Mutex<()>>>>>,
    stop_listeners: Arc<Mutex<HashMap<String, Weak<StopSignal>>>>,
}

#[derive(Clone)]
pub struct SessionHistory {
    state: AppState,
}

struct StopSignal {
    requested: AtomicBool,
    notification: Notify,
}

#[derive(Clone)]
pub struct StopListener {
    signal: Arc<StopSignal>,
}

impl StopListener {
    pub async fn requested(&self) {
        loop {
            let notified = self.signal.notification.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            if self.signal.requested.load(Ordering::Acquire) {
                return;
            }
            notified.await;
        }
    }
}

#[derive(Debug)]
pub struct Error {
    pub kind: ErrorKind,
    pub message: String,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ErrorKind {
    InvalidInput,
    NotFound,
    Conflict,
    Storage,
}

impl ErrorKind {
    pub fn code(self) -> &'static str {
        match self {
            Self::InvalidInput => "invalid_request",
            Self::NotFound => "not_found",
            Self::Conflict => "state_conflict",
            Self::Storage => "internal_error",
        }
    }
}

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

impl std::error::Error for Error {}

impl From<ApiError> for Error {
    fn from(error: ApiError) -> Self {
        Self {
            kind: error.kind,
            message: error.message,
        }
    }
}

#[derive(Debug)]
struct ApiError {
    kind: ErrorKind,
    message: String,
}

impl ApiError {
    fn new(kind: ErrorKind, message: impl Into<String>) -> Self {
        Self {
            kind,
            message: message.into(),
        }
    }

    fn bad(message: impl Into<String>) -> Self {
        Self::new(ErrorKind::InvalidInput, message)
    }

    fn not_found() -> Self {
        Self::new(ErrorKind::NotFound, "Session not found.")
    }

    fn conflict(message: impl Into<String>) -> Self {
        Self::new(ErrorKind::Conflict, message)
    }

    fn internal(error: impl std::fmt::Display) -> Self {
        tracing::warn!(error=%format!("{error:#}"), "Session History request failed");
        Self::new(
            ErrorKind::Storage,
            "An unexpected Session History storage error occurred.",
        )
    }
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct RegisterSession {
    pub id: String,
    pub started_at: String,
    pub state: Value,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct StartSession {
    pub idempotency_id: String,
    pub started_at: String,
    pub session_type: String,
    #[serde(default)]
    pub duration_minutes: Option<f64>,
    #[serde(default)]
    pub custom_prompt: Option<String>,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct NewIngressSession {
    pub idempotency_id: String,
    pub started_at: String,
    pub source_session_type: String,
    pub kind: chatend::SessionKind,
    pub effective_context_tokens: u64,
    pub text: String,
    #[serde(default)]
    pub metadata: Value,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct NewCommand {
    pub idempotency_id: String,
    pub kind: String,
    #[serde(default = "empty_object")]
    pub payload: Value,
}

fn empty_object() -> Value {
    json!({})
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct CommandOutcome {
    #[serde(default = "empty_object")]
    pub outcome: Value,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct NewStopRequest {
    pub idempotency_id: String,
    pub scope: String,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct NewCurrentWorkStop {
    pub idempotency_id: String,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct StopOutcome {
    #[serde(default = "empty_object")]
    pub outcome: Value,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct Checkpoint {
    pub expected_version: i64,
    pub state: Value,
    #[serde(default)]
    pub user_activity: bool,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct ExpectedVersion {
    pub expected_version: i64,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct RetryIngress {
    pub expected_version: i64,
    pub state: Value,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct StartIngress {
    pub expected_version: i64,
    pub provenance_id: String,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct IngressFailure {
    pub expected_version: i64,
    pub stage: String,
    #[serde(default)]
    pub code: Option<String>,
    pub message: String,
    #[serde(default)]
    pub rounds_used: Option<u64>,
    #[serde(default)]
    pub context_tokens: Option<u64>,
    #[serde(default)]
    pub context_window_tokens: Option<u64>,
}

#[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>,
}

#[derive(Clone, Debug, Deserialize, 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>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Created<T> {
    pub value: T,
    pub created: bool,
}

#[derive(Clone, Debug)]
pub struct NewObject {
    pub file_name: Option<String>,
    pub media_type: String,
    pub bytes: Vec<u8>,
}

#[derive(Clone, Debug)]
pub struct StoredObject {
    pub file_name: String,
    pub media_type: String,
    pub bytes: Vec<u8>,
}

impl SessionHistory {
    pub fn open(config: Config) -> anyhow::Result<Self> {
        create_private_directory(&config.directory)?;
        SessionControl::compact_directory(&config.directory)?;
        if let Some(parent) = config
            .completed_list
            .parent()
            .filter(|path| !path.as_os_str().is_empty())
        {
            create_private_directory(parent)?;
        }
        if !config.completed_list.exists() {
            let file = OpenOptions::new()
                .create_new(true)
                .write(true)
                .open(&config.completed_list)
                .with_context(|| format!("creating {}", config.completed_list.display()))?;
            file.sync_all()?;
            sync_directory(
                config
                    .completed_list
                    .parent()
                    .filter(|path| !path.as_os_str().is_empty())
                    .unwrap_or_else(|| FilePath::new(".")),
            )?;
        }
        Ok(Self {
            state: AppState {
                config,
                catalog_mutation: Arc::new(Mutex::new(())),
                session_mutations: Arc::new(Mutex::new(HashMap::new())),
                stop_listeners: Arc::new(Mutex::new(HashMap::new())),
            },
        })
    }

    pub fn health(&self) -> Result<(), Error> {
        read_completed_ids(&self.state.config.completed_list).map_err(ApiError::internal)?;
        Ok(())
    }

    pub fn create_session(&self, input: NewSession) -> anyhow::Result<Session> {
        let metadata = chatend::SessionMetadata {
            session_id: Uuid::new_v4().to_string(),
            kind: input.kind,
            created_at: input.created_at,
            effective_context_tokens: input.effective_context_tokens,
            channel: input.channel,
        };
        SessionHistoryIntegration::create_session(
            self.state
                .config
                .directory
                .join(format!("{}.session-log", metadata.session_id)),
            metadata,
        )
    }

    pub fn open_session(&self, metadata: chatend::SessionMetadata) -> anyhow::Result<Session> {
        self.open_session_with_provider_model(metadata, None)
    }

    pub fn open_session_with_provider_model(
        &self,
        metadata: chatend::SessionMetadata,
        provider_model: Option<&str>,
    ) -> anyhow::Result<Session> {
        validate_session_id(&metadata.session_id)
            .map_err(|error| anyhow::anyhow!(error.message))?;
        let path = self
            .state
            .config
            .directory
            .join(format!("{}.session-log", metadata.session_id));
        match self.state.config.provider_cost_compatibility {
            Some(compatibility) => SessionHistoryIntegration::open_session(
                path,
                metadata,
                provider_model,
                Some(compatibility.estimator),
            ),
            None => SessionHistoryIntegration::open_session(path, metadata, None, None),
        }
    }

    pub fn legacy_provider_cost_summary_for_archive(
        &self,
        archive: &Value,
        session_state: Option<&Value>,
    ) -> anyhow::Result<Option<chatend::ProviderCostSummary>> {
        let Some(compatibility) = self.state.config.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)
    }

    pub async fn register(&self, input: RegisterSession) -> Result<SessionRecord, Error> {
        create_session(self.state.clone(), input)
            .await
            .map_err(Into::into)
    }

    pub async fn start(&self, input: StartSession) -> Result<Created<SessionRecord>, Error> {
        let (created, record) = start_managed_session(self.state.clone(), input).await?;
        Ok(Created {
            value: record,
            created,
        })
    }

    pub async fn enqueue_ingress(
        &self,
        input: NewIngressSession,
    ) -> Result<Created<SessionRecord>, Error> {
        let (created, record) = enqueue_ingress_session(self.state.clone(), input).await?;
        Ok(Created {
            value: record,
            created,
        })
    }

    pub async fn list(&self) -> Result<Vec<SessionRecord>, Error> {
        list_session_summaries(self.state.clone())
            .await
            .map_err(Into::into)
    }

    pub async fn get(&self, id: &str) -> Result<SessionRecord, Error> {
        get_session(self.state.clone(), id.to_owned())
            .await
            .map_err(Into::into)
    }

    pub async fn enqueue(
        &self,
        id: &str,
        input: NewCommand,
    ) -> Result<Created<SessionCommand>, Error> {
        let (created, command) =
            queue_session_command(self.state.clone(), id.to_owned(), input).await?;
        Ok(Created {
            value: command,
            created,
        })
    }

    pub async fn command_heads(&self) -> Result<Vec<SessionCommand>, Error> {
        list_command_heads(self.state.clone())
            .await
            .map_err(Into::into)
    }

    pub async fn claim_command(&self, id: &str) -> Result<SessionCommand, Error> {
        claim_command(self.state.clone(), id.to_owned())
            .await
            .map_err(Into::into)
    }

    pub async fn complete_command(
        &self,
        id: &str,
        outcome: CommandOutcome,
    ) -> Result<SessionCommand, Error> {
        complete_command(self.state.clone(), id.to_owned(), outcome)
            .await
            .map_err(Into::into)
    }

    pub async fn request_stop(
        &self,
        id: &str,
        input: NewStopRequest,
    ) -> Result<Created<SessionStopRequest>, Error> {
        let (created, request) = request_session_stop(
            self.state.clone(),
            id.to_owned(),
            input.idempotency_id,
            Some(input.scope),
        )
        .await?;
        signal_stop_listener(&self.state, id).map_err(Error::from)?;
        Ok(Created {
            value: request,
            created,
        })
    }

    pub async fn request_current_work_stop(
        &self,
        id: &str,
        input: NewCurrentWorkStop,
    ) -> Result<Created<SessionStopRequest>, Error> {
        let (created, request) = request_session_stop(
            self.state.clone(),
            id.to_owned(),
            input.idempotency_id,
            None,
        )
        .await?;
        signal_stop_listener(&self.state, id).map_err(Error::from)?;
        Ok(Created {
            value: request,
            created,
        })
    }

    pub fn listen_for_stop(&self, id: &str) -> Result<StopListener, Error> {
        listen_for_stop(&self.state, id).map_err(Into::into)
    }

    pub async fn stop_heads(&self) -> Result<Vec<SessionStopRequest>, Error> {
        list_stop_heads(self.state.clone())
            .await
            .map_err(Into::into)
    }

    pub async fn complete_stop(
        &self,
        id: &str,
        outcome: StopOutcome,
    ) -> Result<SessionStopRequest, Error> {
        complete_stop_request(self.state.clone(), id.to_owned(), outcome)
            .await
            .map_err(Into::into)
    }

    pub async fn stage_object(&self, id: &str, object: NewObject) -> Result<String, Error> {
        stage_session_object(&self.state, id, object).map_err(Into::into)
    }

    pub fn object(&self, id: &str, pending_id: &str) -> Result<StoredObject, Error> {
        get_session_object(&self.state, id, pending_id).map_err(Into::into)
    }

    pub async fn checkpoint(&self, id: &str, input: Checkpoint) -> Result<SessionRecord, Error> {
        checkpoint(
            self.state.clone(),
            id,
            input.expected_version,
            input.state,
            input.user_activity,
        )
        .await
        .map_err(Into::into)
    }

    pub async fn request_ingress(
        &self,
        id: &str,
        input: Checkpoint,
    ) -> Result<SessionRecord, Error> {
        transition_with_checkpoint(self.state.clone(), id, input, "ingress_pending")
            .await
            .map_err(Into::into)
    }

    pub async fn start_ingress(
        &self,
        id: &str,
        input: StartIngress,
    ) -> Result<SessionRecord, Error> {
        transition(
            self.state.clone(),
            id,
            input.expected_version,
            "ingress_in_progress",
            Some(input.provenance_id),
        )
        .await
        .map_err(Into::into)
    }

    pub async fn complete_ingress(
        &self,
        id: &str,
        input: ExpectedVersion,
    ) -> Result<SessionRecord, Error> {
        let current = fetch_active(&self.state, id)?;
        complete_session(
            self.state.clone(),
            id,
            input.expected_version,
            current.state,
        )
        .await
        .map_err(Into::into)
    }

    pub async fn fail_ingress(
        &self,
        id: &str,
        input: IngressFailure,
    ) -> Result<SessionRecord, Error> {
        record_ingress_failure(self.state.clone(), id, input)
            .await
            .map_err(Into::into)
    }

    pub async fn retry_ingress(
        &self,
        id: &str,
        input: RetryIngress,
    ) -> Result<SessionRecord, Error> {
        retry_ingress(self.state.clone(), id.to_owned(), input)
            .await
            .map_err(Into::into)
    }

    pub async fn release_interrupted_ingress(&self) -> Result<Vec<String>, Error> {
        release_ingress_repairs(self.state.clone())
            .await
            .map_err(Into::into)
    }

    pub async fn complete(&self, id: &str, input: Checkpoint) -> Result<SessionRecord, Error> {
        complete_session(self.state.clone(), id, input.expected_version, input.state)
            .await
            .map_err(Into::into)
    }

    pub async fn record_completion(&self, input: RecordCompletion) -> Result<(), Error> {
        record_completed_session(self.state.clone(), input).await?;
        Ok(())
    }
}

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"
                            )
                        })
            })
        })
}

async fn record_completed_session(
    state: AppState,
    input: RecordCompletion,
) -> Result<Value, ApiError> {
    let session_guard = input
        .session_id
        .as_deref()
        .map(|id| session_mutation(&state, id))
        .transpose()?;
    let _session_guard = session_guard
        .as_ref()
        .map(|guard| guard.lock().map_err(ApiError::internal))
        .transpose()?;
    let _catalog_guard = state.catalog_mutation.lock().map_err(ApiError::internal)?;
    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(ApiError::conflict(
            "completion receipt and requested session object differ",
        ));
    }
    receipt.session_id = receipt.session_id.or(input.session_id.clone());
    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_with(now);
    append_completion_receipt(&state.config.completed_list, &receipt)
        .map_err(ApiError::internal)?;
    if let Some(id) = input.session_id {
        validate_session_id(&id)?;
        let path = state.config.directory.join(format!("{id}.session-log"));
        if path.exists() {
            let SessionJournal { log, control } =
                SessionJournal::open(&path).map_err(ApiError::internal)?;
            log.delete_committed().map_err(ApiError::internal)?;
            control.delete().map_err(ApiError::internal)?;
        }
    }
    Ok(json!({
        "sessionObjectId":input.session_object_id,
        "recorded":true
    }))
}

async fn create_session(
    state: AppState,
    input: RegisterSession,
) -> Result<SessionRecord, ApiError> {
    validate_started_at(&input.started_at)?;
    validate_session_id(&input.id)?;
    let path = state
        .config
        .directory
        .join(format!("{}.session-log", input.id));
    let journal = SessionJournal::open(&path).map_err(ApiError::internal)?;
    let id = journal.list().header.session_id;
    if id != input.id {
        return Err(ApiError::bad(
            "registered session ID does not match its durable session log",
        ));
    }
    drop(journal);
    let session_guard = session_mutation(&state, &id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = SessionJournal::open(&path).map_err(ApiError::internal)?;
    if latest_lifecycle(&journal).is_some() {
        return Err(ApiError::conflict("Session is already registered."));
    }
    let mut record = SessionRecord {
        id,
        phase: "active".into(),
        started_at: input.started_at.clone(),
        updated_at: input.started_at,
        state: input.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,
    };
    append_lifecycle(&mut journal, &mut record)?;
    Ok(materialize(
        record,
        &journal,
        state.config.provider_cost_compatibility,
    ))
}

async fn start_managed_session(
    state: AppState,
    input: StartSession,
) -> Result<(bool, SessionRecord), ApiError> {
    validate_started_at(&input.started_at)?;
    validate_idempotency(&input.idempotency_id)?;
    let _catalog_guard = state.catalog_mutation.lock().map_err(ApiError::internal)?;
    for path in journal_paths(&state.config.directory)? {
        let journal = SessionJournal::open(&path).map_err(ApiError::internal)?;
        if let Some(record) = latest_lifecycle(&journal)
            && record
                .state
                .get("startIdempotencyId")
                .and_then(Value::as_str)
                == Some(&input.idempotency_id)
        {
            return Ok((
                false,
                materialize(record, &journal, state.config.provider_cost_compatibility),
            ));
        }
    }
    let id = Uuid::new_v4().to_string();
    let mut journal = SessionJournal::create(&state.config.directory, &id, &input.started_at)
        .map_err(ApiError::internal)?;
    let mut session_state = json!({
        "stateVersion":3,
        "sessionId":id,
        "sessionType":input.session_type,
        "startedAt":input.started_at,
        "startIdempotencyId":input.idempotency_id,
        "orchestration":{"owner":"backend","status":"idle"},
    });
    if input.session_type == "free-time" {
        session_state["selfTimeIntent"] = json!({
            "requestedAt":input.started_at,
            "durationMinutes":input.duration_minutes,
            "customPrompt":input.custom_prompt.unwrap_or_default(),
        });
    }
    let mut record = SessionRecord {
        id,
        phase: "active".into(),
        started_at: input.started_at.clone(),
        updated_at: input.started_at,
        state: session_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,
    };
    append_lifecycle(&mut journal, &mut record)?;
    Ok((
        true,
        materialize(record, &journal, state.config.provider_cost_compatibility),
    ))
}

async fn enqueue_ingress_session(
    state: AppState,
    input: NewIngressSession,
) -> Result<(bool, SessionRecord), ApiError> {
    validate_started_at(&input.started_at)?;
    validate_idempotency(&input.idempotency_id)?;
    if input.source_session_type.trim().is_empty() {
        return Err(ApiError::bad("Source session type must not be empty."));
    }
    if input.text.trim().is_empty() {
        return Err(ApiError::bad("Ingress source text must not be empty."));
    }
    let _catalog_guard = state.catalog_mutation.lock().map_err(ApiError::internal)?;
    for path in journal_paths(&state.config.directory)? {
        let journal = SessionJournal::open(&path).map_err(ApiError::internal)?;
        if let Some(record) = latest_lifecycle(&journal)
            && ingress_idempotency_id(record.state.get("ingressSource"))
                == Some(input.idempotency_id.as_str())
        {
            return Ok((
                false,
                materialize(record, &journal, state.config.provider_cost_compatibility),
            ));
        }
    }
    for receipt in
        read_completion_receipts(&state.config.completed_list).map_err(ApiError::internal)?
    {
        if ingress_idempotency_id(receipt.ingress_source.as_ref())
            == Some(input.idempotency_id.as_str())
        {
            return Ok((false, completed_session_record(receipt, false)));
        }
    }

    let id = Uuid::new_v4().to_string();
    let path = state.config.directory.join(format!("{id}.session-log"));
    let mut source = SessionHistoryIntegration::create_session(
        &path,
        chatend::SessionMetadata {
            session_id: id.clone(),
            kind: input.kind,
            created_at: input.started_at.clone(),
            effective_context_tokens: input.effective_context_tokens,
            channel: Value::Null,
        },
    )
    .map_err(ApiError::internal)?;
    source
        .create_box(
            input.started_at.clone(),
            "Ingress source",
            chatend::BoxOwner::User,
            chatend::BoxContent {
                text: input.text.trim().to_owned(),
                objects: Vec::new(),
                metadata: input.metadata.clone(),
            },
        )
        .map_err(ApiError::internal)?;
    let chatend_metadata = source.state().metadata.clone();
    drop(source);

    let ingress_source = json!({
        "idempotencyId":input.idempotency_id,
        "metadata":input.metadata,
    });
    let mut journal = SessionJournal::open(&path).map_err(ApiError::internal)?;
    let mut record = SessionRecord {
        id: id.clone(),
        phase: "ingress_pending".into(),
        started_at: input.started_at.clone(),
        updated_at: input.started_at.clone(),
        state: json!({
            "stateVersion":3,
            "sessionId":id,
            "sessionType":input.source_session_type,
            "startedAt":input.started_at,
            "ingressSource":ingress_source,
            "chatendMetadata":chatend_metadata,
        }),
        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,
    };
    append_lifecycle(&mut journal, &mut record)?;
    Ok((
        true,
        materialize(record, &journal, state.config.provider_cost_compatibility),
    ))
}

fn ingress_idempotency_id(source: Option<&Value>) -> Option<&str> {
    source?.get("idempotencyId").and_then(Value::as_str)
}

async fn list_session_summaries(state: AppState) -> Result<Vec<SessionRecord>, ApiError> {
    let mut sessions = Vec::new();
    for journal in open_listed_journals(journal_paths(&state.config.directory)?)? {
        if let Some(mut record) = latest_lifecycle(&journal) {
            record.summary = true;
            record.state = summary_state(&record.state, &journal);
            sessions.push(record);
        }
    }
    let receipts =
        read_completion_receipts(&state.config.completed_list).map_err(ApiError::internal)?;
    let completed_session_ids = receipts
        .iter()
        .filter_map(|receipt| receipt.session_id.as_ref())
        .collect::<HashSet<_>>();
    sessions.retain(|session| !completed_session_ids.contains(&session.id));
    for receipt in receipts {
        sessions.push(completed_session_record(receipt, true));
    }
    sessions.sort_by(|a, b| b.updated_at.cmp(&a.updated_at));
    Ok(sessions)
}

async fn get_session(state: AppState, id: String) -> Result<SessionRecord, ApiError> {
    if let Some(receipt) = read_completion_receipts(&state.config.completed_list)
        .map_err(ApiError::internal)?
        .into_iter()
        .find(|receipt| receipt.session_object_id == id)
    {
        return Ok(completed_session_record(receipt, false));
    }
    let journal = open_by_id(&state, &id)?;
    let record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    Ok(materialize(
        record,
        &journal,
        state.config.provider_cost_compatibility,
    ))
}

fn completed_session_record(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,
    }
}

fn get_session_object(
    state: &AppState,
    id: &str,
    pending_id: &str,
) -> Result<StoredObject, ApiError> {
    let number = pending_id
        .strip_prefix("pending:")
        .and_then(|value| value.parse::<u64>().ok())
        .filter(|value| *value > 0)
        .ok_or_else(|| ApiError::bad("Object ID must have the form pending:N."))?;
    let journal = open_by_id(state, id)?;
    let object = journal
        .log
        .read_pending_object(EventPosition(number - 1))
        .map_err(|error| {
            tracing::warn!(session_id=id, pending_id, %error, "pending session object is unavailable");
            ApiError::not_found()
        })?;
    let media_type = if object.media_type.trim().is_empty()
        || object
            .media_type
            .chars()
            .any(|character| character.is_control() || character.is_whitespace())
        || !object.media_type.contains('/')
    {
        "application/octet-stream"
    } else {
        &object.media_type
    };
    let mut file_name = object
        .file_name
        .chars()
        .map(|character| {
            if character.is_ascii_graphic() && !matches!(character, '"' | '\\' | '/' | ';') {
                character
            } else {
                '_'
            }
        })
        .take(255)
        .collect::<String>();
    if file_name.is_empty() {
        file_name = "uploaded-object".into();
    }
    Ok(StoredObject {
        file_name,
        media_type: media_type.to_owned(),
        bytes: object.bytes,
    })
}

async fn queue_session_command(
    state: AppState,
    id: String,
    input: NewCommand,
) -> Result<(bool, SessionCommand), ApiError> {
    validate_idempotency(&input.idempotency_id)?;
    let session_guard = session_mutation(&state, &id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = open_by_id(&state, &id)?;
    let record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    if record.phase != "active" {
        return Err(ApiError::conflict("Session is no longer active."));
    }
    let commands = commands(&journal);
    if let Some(command) = commands
        .values()
        .find(|command| command.idempotency_id == input.idempotency_id)
    {
        return Ok((false, command.clone()));
    }
    let command = SessionCommand {
        id: Uuid::new_v4().to_string(),
        conversation_id: id,
        sequence: commands
            .values()
            .map(|command| command.sequence)
            .max()
            .unwrap_or(0)
            + 1,
        kind: input.kind,
        payload: input.payload,
        status: "pending".into(),
        cancel_requested: false,
        outcome: None,
        created_at: now(),
        processing_started_at: None,
        completed_at: None,
        idempotency_id: input.idempotency_id,
    };
    append_command(&mut journal, &command)?;
    Ok((true, command))
}

fn stage_session_object(state: &AppState, id: &str, object: NewObject) -> Result<String, ApiError> {
    let NewObject {
        file_name,
        media_type,
        bytes,
    } = object;
    let session_guard = session_mutation(state, id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = open_by_id(state, id)?;
    let record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    if record.phase != "active" {
        return Err(ApiError::conflict(
            "Objects can only be supplied to an active session.",
        ));
    }
    if commands(&journal)
        .values()
        .any(|command| matches!(command.status.as_str(), "pending" | "processing"))
    {
        return Err(ApiError::conflict(
            "Objects cannot be supplied while the session is processing a command.",
        ));
    }
    journal
        .stage_object(media_type, file_name, &bytes)
        .map_err(|error| ApiError::bad(error.to_string()))
}

async fn list_command_heads(state: AppState) -> Result<Vec<SessionCommand>, ApiError> {
    let mut heads = Vec::new();
    for projection in open_listed_control_projections(journal_paths(&state.config.directory)?)? {
        let mut active = projection
            .commands
            .into_values()
            .filter(|command| matches!(command.status.as_str(), "pending" | "processing"))
            .collect::<Vec<_>>();
        active.sort_by_key(|command| command.sequence);
        if let Some(head) = active.into_iter().next() {
            heads.push(head);
        }
    }
    heads.sort_by(|a, b| a.created_at.cmp(&b.created_at));
    Ok(heads)
}

async fn claim_command(state: AppState, command_id: String) -> Result<SessionCommand, ApiError> {
    mutate_command(&state, &command_id, |command| {
        if command.status == "pending" {
            command.status = "processing".into();
            command.processing_started_at = Some(now());
        } else if command.status != "processing" {
            return Err(ApiError::conflict("Command is already complete."));
        }
        Ok(())
    })
}

async fn complete_command(
    state: AppState,
    command_id: String,
    input: CommandOutcome,
) -> Result<SessionCommand, ApiError> {
    mutate_command(&state, &command_id, |command| {
        if command.status == "complete" {
            return Ok(());
        }
        if command.status != "processing" {
            return Err(ApiError::conflict("Command was not claimed."));
        }
        command.status = "complete".into();
        command.outcome = Some(input.outcome);
        command.completed_at = Some(now());
        Ok(())
    })
}

async fn request_session_stop(
    state: AppState,
    id: String,
    idempotency_id: String,
    requested_scope: Option<String>,
) -> Result<(bool, SessionStopRequest), ApiError> {
    validate_idempotency(&idempotency_id)?;
    if requested_scope
        .as_deref()
        .is_some_and(|scope| !matches!(scope, "turn" | "session" | "self-time-run"))
    {
        return Err(ApiError::bad(
            "Stop scope must be turn, session, or self-time-run.",
        ));
    }
    let session_guard = session_mutation(&state, &id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = open_by_id(&state, &id)?;
    let lifecycle = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    if !matches!(
        lifecycle.phase.as_str(),
        "active" | "ingress_pending" | "ingress_in_progress"
    ) {
        return Err(ApiError::conflict("Session has no work in progress."));
    }
    let existing = stop_requests(&journal);
    if let Some(request) = existing
        .values()
        .find(|request| request.idempotency_id == idempotency_id)
    {
        return Ok((false, request.clone()));
    }
    if let Some(request) = existing
        .values()
        .find(|request| request.status == "pending")
    {
        return Ok((false, request.clone()));
    }
    let scope = match requested_scope {
        Some(scope) => scope,
        None => current_work_stop_scope(&lifecycle)?.into(),
    };
    let request = SessionStopRequest {
        id: Uuid::new_v4().to_string(),
        session_id: id,
        scope,
        status: "pending".into(),
        outcome: None,
        requested_at: now(),
        completed_at: None,
        idempotency_id,
    };
    append_stop_request(&mut journal, &request)?;

    if request.scope == "turn" {
        let mut active = commands(&journal)
            .into_values()
            .filter(|command| {
                matches!(command.status.as_str(), "pending" | "processing")
                    && matches!(command.kind.as_str(), "message" | "retry")
            })
            .collect::<Vec<_>>();
        active.sort_by_key(|command| command.sequence);
        if let Some(mut command) = active.into_iter().next() {
            command.cancel_requested = true;
            append_command(&mut journal, &command)?;
        }
    }
    Ok((true, request))
}

fn current_work_stop_scope(record: &SessionRecord) -> Result<&'static str, ApiError> {
    if record.phase != "active" {
        return Ok("session");
    }
    let kind = record
        .state
        .get("chatendMetadata")
        .cloned()
        .and_then(|value| serde_json::from_value::<chatend::SessionMetadata>(value).ok())
        .map(|metadata| metadata.kind);
    match kind {
        Some(
            chatend::SessionKind::Conversation
            | chatend::SessionKind::Telegram
            | chatend::SessionKind::TelegramGroup,
        ) => Ok("turn"),
        Some(chatend::SessionKind::SelfTime) => Ok("self-time-run"),
        Some(_) => Ok("session"),
        None => match record.state.get("sessionType").and_then(Value::as_str) {
            Some("conversation" | "telegram" | "telegram-group") => Ok("turn"),
            Some("free-time") => Ok("self-time-run"),
            Some(_) => Ok("session"),
            None => Err(ApiError::conflict(
                "Session does not identify the work that should stop.",
            )),
        },
    }
}

fn listen_for_stop(state: &AppState, id: &str) -> Result<StopListener, ApiError> {
    let session_guard = session_mutation(state, id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let journal = open_by_id(state, id)?;
    let signal = {
        let mut listeners = state.stop_listeners.lock().map_err(ApiError::internal)?;
        listeners.retain(|_, listener| listener.strong_count() > 0);
        if let Some(signal) = listeners.get(id).and_then(Weak::upgrade) {
            signal
        } else {
            let signal = Arc::new(StopSignal {
                requested: AtomicBool::new(false),
                notification: Notify::new(),
            });
            listeners.insert(id.to_owned(), Arc::downgrade(&signal));
            signal
        }
    };
    if stop_requests(&journal)
        .values()
        .any(|request| request.status == "pending")
    {
        signal_stop(&signal);
    }
    Ok(StopListener { signal })
}

fn signal_stop_listener(state: &AppState, id: &str) -> Result<(), ApiError> {
    let signal = state
        .stop_listeners
        .lock()
        .map_err(ApiError::internal)?
        .get(id)
        .and_then(Weak::upgrade);
    if let Some(signal) = signal {
        signal_stop(&signal);
    }
    Ok(())
}

fn signal_stop(signal: &StopSignal) {
    signal.requested.store(true, Ordering::Release);
    signal.notification.notify_waiters();
}

async fn list_stop_heads(state: AppState) -> Result<Vec<SessionStopRequest>, ApiError> {
    let mut heads = Vec::new();
    for projection in open_listed_control_projections(journal_paths(&state.config.directory)?)? {
        if let Some(request) = projection
            .stop_requests
            .into_values()
            .find(|request| request.status == "pending")
        {
            heads.push(request);
        }
    }
    heads.sort_by(|left, right| left.requested_at.cmp(&right.requested_at));
    Ok(heads)
}

async fn complete_stop_request(
    state: AppState,
    request_id: String,
    input: StopOutcome,
) -> Result<SessionStopRequest, ApiError> {
    mutate_stop_request(&state, &request_id, |request| {
        if request.status == "complete" {
            return Ok(());
        }
        request.status = "complete".into();
        request.outcome = Some(input.outcome);
        request.completed_at = Some(now());
        Ok(())
    })
}

fn mutate_command(
    state: &AppState,
    command_id: &str,
    mutation: impl FnOnce(&mut SessionCommand) -> Result<(), ApiError>,
) -> Result<SessionCommand, ApiError> {
    let mut target = None;
    for path in journal_paths(&state.config.directory)? {
        let journal = SessionJournal::open(&path).map_err(ApiError::internal)?;
        if let Some(command) = commands(&journal).remove(command_id) {
            target = Some((path, command.conversation_id));
            break;
        }
    }
    let (path, conversation_id) = target.ok_or_else(ApiError::not_found)?;
    let session_guard = session_mutation(state, &conversation_id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = SessionJournal::open(path).map_err(ApiError::internal)?;
    let mut command = commands(&journal)
        .remove(command_id)
        .ok_or_else(ApiError::not_found)?;
    mutation(&mut command)?;
    append_command(&mut journal, &command)?;
    Ok(command)
}

fn mutate_stop_request(
    state: &AppState,
    request_id: &str,
    mutation: impl FnOnce(&mut SessionStopRequest) -> Result<(), ApiError>,
) -> Result<SessionStopRequest, ApiError> {
    let mut target = None;
    for path in journal_paths(&state.config.directory)? {
        let journal = SessionJournal::open(&path).map_err(ApiError::internal)?;
        if let Some(request) = stop_requests(&journal).remove(request_id) {
            target = Some((path, request.session_id));
            break;
        }
    }
    let (path, session_id) = target.ok_or_else(ApiError::not_found)?;
    let session_guard = session_mutation(state, &session_id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = SessionJournal::open(path).map_err(ApiError::internal)?;
    let mut request = stop_requests(&journal)
        .remove(request_id)
        .ok_or_else(ApiError::not_found)?;
    mutation(&mut request)?;
    append_stop_request(&mut journal, &request)?;
    Ok(request)
}

async fn checkpoint(
    state: AppState,
    id: &str,
    expected_version: i64,
    new_state: Value,
    user_activity: bool,
) -> Result<SessionRecord, ApiError> {
    let session_guard = session_mutation(&state, id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = open_by_id(&state, id)?;
    let mut record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    require_version(&record, expected_version)?;
    record.state = new_state;
    record.version += 1;
    record.updated_at = now();
    if user_activity {
        record.last_user_message_at = Some(record.updated_at.clone());
    }
    append_lifecycle(&mut journal, &mut record)?;
    Ok(materialize(
        record,
        &journal,
        state.config.provider_cost_compatibility,
    ))
}

async fn transition_with_checkpoint(
    state: AppState,
    id: &str,
    input: Checkpoint,
    phase: &str,
) -> Result<SessionRecord, ApiError> {
    let session_guard = session_mutation(&state, id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = open_by_id(&state, id)?;
    let mut record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    require_version(&record, input.expected_version)?;
    record.state = input.state;
    record.phase = phase.into();
    if phase == "ingress_pending" {
        record.ingress_next_attempt_at = None;
    }
    record.version += 1;
    record.updated_at = now();
    append_lifecycle(&mut journal, &mut record)?;
    Ok(materialize(
        record,
        &journal,
        state.config.provider_cost_compatibility,
    ))
}

async fn transition(
    state: AppState,
    id: &str,
    expected_version: i64,
    phase: &str,
    provenance_id: Option<String>,
) -> Result<SessionRecord, ApiError> {
    let session_guard = session_mutation(&state, id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = open_by_id(&state, id)?;
    let mut record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    require_version(&record, expected_version)?;
    record.phase = phase.into();
    record.provenance_id = provenance_id;
    if phase == "ingress_in_progress" {
        record.ingress_next_attempt_at = None;
    }
    record.version += 1;
    record.updated_at = now();
    append_lifecycle(&mut journal, &mut record)?;
    Ok(materialize(
        record,
        &journal,
        state.config.provider_cost_compatibility,
    ))
}

async fn record_ingress_failure(
    state: AppState,
    id: &str,
    input: IngressFailure,
) -> Result<SessionRecord, ApiError> {
    let session_guard = session_mutation(&state, id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = open_by_id(&state, id)?;
    let mut record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    require_version(&record, input.expected_version)?;
    if !matches!(
        record.phase.as_str(),
        "ingress_pending" | "ingress_in_progress"
    ) {
        return Err(ApiError::conflict(
            "Session History ingress is not in an active attempt.",
        ));
    }
    let attempt = record.ingress_failure_count.saturating_add(1);
    let terminal =
        input.code.as_deref() == Some("input_too_large") || attempt >= INGRESS_FAILURE_LIMIT;
    let mut failures = record
        .ingress_failures
        .as_array()
        .cloned()
        .unwrap_or_default();
    failures.push(json!({
        "attempt":attempt,
        "at":now(),
        "stage":input.stage,
        "code":input.code,
        "message":input.message,
        "roundsUsed":input.rounds_used,
        "contextTokens":input.context_tokens,
        "contextWindowTokens":input.context_window_tokens,
    }));
    if failures.len() > RETAINED_INGRESS_FAILURES {
        failures.drain(..failures.len() - RETAINED_INGRESS_FAILURES);
    }
    record.ingress_failures = Value::Array(failures);
    record.ingress_failure_count = attempt;
    record.phase = if terminal {
        "ingress_failed".into()
    } else {
        "ingress_pending".into()
    };
    let updated_at = now();
    record.ingress_next_attempt_at = (!terminal)
        .then(|| (Utc::now() + Duration::seconds(INGRESS_RETRY_DELAY_SECONDS)).to_rfc3339());
    record.version += 1;
    record.updated_at = updated_at;
    append_lifecycle(&mut journal, &mut record)?;
    if terminal {
        tracing::error!(
            session_id = id,
            attempt,
            stage = %input.stage,
            code = input.code.as_deref().unwrap_or("ingress_error"),
            terminal_reason = if input.code.as_deref() == Some("input_too_large") {
                "non_retryable"
            } else {
                "retry_limit"
            },
            "Session History ingress stopped after a terminal failure"
        );
    }
    Ok(materialize(
        record,
        &journal,
        state.config.provider_cost_compatibility,
    ))
}

async fn retry_ingress(
    state: AppState,
    id: String,
    input: RetryIngress,
) -> Result<SessionRecord, ApiError> {
    let session_guard = session_mutation(&state, &id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let mut journal = open_by_id(&state, &id)?;
    let mut record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    require_version(&record, input.expected_version)?;
    if record.phase != "ingress_failed" {
        return Err(ApiError::conflict(
            "Session History ingress is not in the failed state.",
        ));
    }
    record.state = input.state;
    record.phase = "ingress_pending".into();
    record.ingress_failure_count = 0;
    record.ingress_next_attempt_at = None;
    record.version += 1;
    record.updated_at = now();
    append_lifecycle(&mut journal, &mut record)?;
    Ok(materialize(
        record,
        &journal,
        state.config.provider_cost_compatibility,
    ))
}

async fn release_ingress_repairs(state: AppState) -> Result<Vec<String>, ApiError> {
    let mut released = Vec::new();
    for path in journal_paths(&state.config.directory)? {
        let Some(id) = path
            .file_stem()
            .and_then(|value| value.to_str())
            .map(str::to_owned)
        else {
            continue;
        };
        let session_guard = session_mutation(&state, &id)?;
        let _guard = session_guard.lock().map_err(ApiError::internal)?;
        let mut journal = open_by_id(&state, &id)?;
        let Some(mut record) = latest_lifecycle(&journal) else {
            continue;
        };
        if record.phase != "ingress_in_progress" {
            continue;
        }
        record.phase = "ingress_pending".into();
        record.ingress_next_attempt_at = None;
        if record.ingress_failure_count > INGRESS_FAILURE_LIMIT {
            record.ingress_failure_count = 0;
        }
        if let Some(failures) = record.ingress_failures.as_array_mut()
            && failures.len() > RETAINED_INGRESS_FAILURES
        {
            failures.drain(..failures.len() - RETAINED_INGRESS_FAILURES);
        }
        record.version += 1;
        record.updated_at = now();
        append_lifecycle(&mut journal, &mut record)?;
        released.push(id);
    }
    Ok(released)
}

async fn complete_session(
    state: AppState,
    id: &str,
    expected_version: i64,
    new_state: Value,
) -> Result<SessionRecord, ApiError> {
    let session_guard = session_mutation(&state, id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;
    let journal = open_by_id(&state, id)?;
    let mut record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    require_version(&record, expected_version)?;
    let completion_state = new_state
        .get("historyIngress")
        .filter(|state| state.get("sessionObjectId").is_some_and(Value::is_string))
        .cloned()
        .unwrap_or_else(|| new_state.clone());
    record.state = new_state;
    record = projected_lifecycle(record);
    let object_id = completion_state
        .get("sessionObjectId")
        .and_then(Value::as_str)
        .ok_or_else(|| {
            ApiError::conflict("completed session has no permanent Kweb session object")
        })?
        .to_owned();
    let _catalog_guard = state.catalog_mutation.lock().map_err(ApiError::internal)?;
    let mut receipt = completion_state
        .get("commitReceipt")
        .filter(|receipt| !receipt.is_null())
        .cloned()
        .map(serde_json::from_value::<CompletionReceipt>)
        .transpose()
        .map_err(|error| ApiError::conflict(format!("invalid session commit receipt: {error}")))?
        .unwrap_or(CompletionReceipt {
            transaction_id: None,
            session_object_id: 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 != object_id {
        return Err(ApiError::conflict(
            "session commit receipt names a different archive object",
        ));
    }
    let completed_at = now();
    receipt.session_id = Some(record.id.clone());
    receipt.session_type = record
        .state
        .get("sessionType")
        .and_then(Value::as_str)
        .map(str::to_owned);
    receipt.created_at = Some(record.started_at.clone());
    receipt.committed_at = Some(completed_at.clone());
    receipt.ingress_source = record.state.get("ingressSource").cloned();
    append_completion_receipt(&state.config.completed_list, &receipt)
        .map_err(ApiError::internal)?;
    record.phase = "complete".into();
    record.version += 1;
    record.updated_at = completed_at;
    record.ended_at = Some(record.updated_at.clone());
    record.state["sessionObjectId"] = json!(object_id);
    record.state["commitReceipt"] = completion_state
        .get("commitReceipt")
        .cloned()
        .unwrap_or(Value::Null);
    let output = materialize(record, &journal, state.config.provider_cost_compatibility);
    let SessionJournal { log, control } = journal;
    log.delete_committed().map_err(ApiError::internal)?;
    control.delete().map_err(ApiError::internal)?;
    Ok(output)
}

fn fetch_active(state: &AppState, id: &str) -> Result<SessionRecord, ApiError> {
    let journal = open_by_id(state, id)?;
    latest_lifecycle(&journal).ok_or_else(ApiError::not_found)
}

fn session_mutation(state: &AppState, id: &str) -> Result<Arc<Mutex<()>>, ApiError> {
    let mut sessions = state.session_mutations.lock().map_err(ApiError::internal)?;
    sessions.retain(|_, lock| lock.strong_count() > 0);
    if let Some(lock) = sessions.get(id).and_then(Weak::upgrade) {
        return Ok(lock);
    }
    let lock = Arc::new(Mutex::new(()));
    sessions.insert(id.to_owned(), Arc::downgrade(&lock));
    Ok(lock)
}

fn open_by_id(state: &AppState, id: &str) -> Result<SessionJournal, ApiError> {
    validate_session_id(id)?;
    let path = state.config.directory.join(format!("{id}.session-log"));
    SessionJournal::open(&path).map_err(|error| {
        if !path.exists() {
            ApiError::not_found()
        } else {
            ApiError::internal(error)
        }
    })
}

fn latest_lifecycle(journal: &SessionJournal) -> Option<SessionRecord> {
    journal.projection().lifecycle
}

fn projected_lifecycle(record: SessionRecord) -> SessionRecord {
    let ControlUpdate::Lifecycle(record) = ControlUpdate::Lifecycle(record).projected() else {
        unreachable!("lifecycle projection changed update kind");
    };
    record
}

fn append_lifecycle(
    journal: &mut SessionJournal,
    record: &mut SessionRecord,
) -> Result<(), ApiError> {
    let update = journal
        .append_control(ControlUpdate::Lifecycle(record.clone()))
        .map_err(ApiError::internal)?;
    let ControlUpdate::Lifecycle(projected) = update else {
        unreachable!("lifecycle append changed update kind");
    };
    *record = projected;
    Ok(())
}

fn commands(journal: &SessionJournal) -> BTreeMap<String, SessionCommand> {
    journal.projection().commands
}

fn append_command(journal: &mut SessionJournal, command: &SessionCommand) -> Result<(), ApiError> {
    journal
        .append_control(ControlUpdate::Command(command.clone()))
        .map(|_| ())
        .map_err(ApiError::internal)
}

fn stop_requests(journal: &SessionJournal) -> BTreeMap<String, SessionStopRequest> {
    journal.projection().stop_requests
}

fn append_stop_request(
    journal: &mut SessionJournal,
    request: &SessionStopRequest,
) -> Result<(), ApiError> {
    journal
        .append_control(ControlUpdate::StopRequest(request.clone()))
        .map(|_| ())
        .map_err(ApiError::internal)
}

fn materialize(
    mut record: SessionRecord,
    journal: &SessionJournal,
    provider_cost_compatibility: Option<ProviderCostCompatibility>,
) -> SessionRecord {
    let log = journal.list();
    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);
    }
    let context_state = record
        .state
        .get("historyIngress")
        .filter(|state| state.get("chatendMetadata").is_some())
        .unwrap_or(&record.state);
    let default_provider_model = provider_cost_compatibility
        .and_then(|compatibility| (compatibility.session_model)(context_state))
        .or_else(|| {
            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::<chatend::SessionMetadata>(value).ok())
        .and_then(|metadata| match 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 chatend_text = Value::String(projection.render());
        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();
        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);
        }
    }
    record
}

fn summary_state(control: &Value, journal: &SessionJournal) -> Value {
    let log = journal.list();
    let first_user = log
        .events
        .iter()
        .find(|event| event.role == Role::UserMessage)
        .map(display_text)
        .map(|text| text.chars().take(512).collect::<String>());
    json!({
        "sessionType":control.get("sessionType"),
        "channel":control.get("channel"),
        "freeTime":control.get("freeTime"),
        "orchestration":control.get("orchestration"),
        "ingressSource":control.get("ingressSource"),
        "firstUserMessage":first_user,
        "boxCount":log.events.len(),
        "eventCount":log.events.len(),
        "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 journal_paths(directory: &FilePath) -> Result<Vec<PathBuf>, ApiError> {
    let mut paths = std::fs::read_dir(directory)
        .map_err(ApiError::internal)?
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .filter(|path| path.extension().and_then(|value| value.to_str()) == Some("session-log"))
        .collect::<Vec<_>>();
    paths.sort();
    Ok(paths)
}

fn listed_session_parts(path: &FilePath) -> anyhow::Result<(&FilePath, &str)> {
    ensure!(
        path.extension().and_then(|value| value.to_str()) == Some("session-log"),
        "{} is not a session-log path",
        path.display()
    );
    let directory = path.parent().unwrap_or_else(|| FilePath::new("."));
    let id = path
        .file_stem()
        .and_then(|value| value.to_str())
        .context("session-log filename is not valid UTF-8")?;
    Ok((directory, id))
}

fn open_listed_control(path: &FilePath) -> anyhow::Result<Option<SessionControl>> {
    let (directory, id) = listed_session_parts(path)?;
    let control = SessionControl::open(directory, id, OpenMode::ExistingOnly)?;
    if !path.exists() {
        return Ok(None);
    }
    Ok(control)
}

fn open_listed_control_projections(
    paths: Vec<PathBuf>,
) -> Result<Vec<ControlProjection>, ApiError> {
    let mut projections = Vec::with_capacity(paths.len());
    for path in paths {
        let Some(control) = open_listed_control(&path).map_err(ApiError::internal)? else {
            continue;
        };
        let projection = control.projection();
        if projection.lifecycle.is_some() {
            projections.push(projection);
        }
    }
    Ok(projections)
}

fn open_listed_journals(paths: Vec<PathBuf>) -> Result<Vec<SessionJournal>, ApiError> {
    let mut journals = Vec::with_capacity(paths.len());
    for path in paths {
        match SessionJournal::open_existing(&path) {
            Ok(Some(journal)) => journals.push(journal),
            Ok(None) => {}
            Err(_) if !path.exists() => {}
            Err(error) => return Err(ApiError::internal(error)),
        }
    }
    Ok(journals)
}

fn read_completed_ids(path: &FilePath) -> anyhow::Result<Vec<String>> {
    Ok(read_completion_receipts(path)?
        .into_iter()
        .map(|receipt| receipt.session_object_id)
        .collect())
}

fn read_completion_receipts(path: &FilePath) -> 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 in BufReader::new(file).lines() {
        let line = line?;
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let receipt = if line.starts_with('{') {
            serde_json::from_str::<CompletionReceipt>(line)
                .context("decoding Session History completion receipt")?
        } 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)
}

#[cfg(test)]
fn append_completed_id(path: &FilePath, id: &str) -> anyhow::Result<()> {
    append_completion_receipt(
        path,
        &CompletionReceipt {
            transaction_id: None,
            session_object_id: id.into(),
            session_id: None,
            session_type: None,
            created_at: None,
            committed_at: None,
            ingress_source: None,
            node_ids: BTreeMap::new(),
            object_ids: BTreeMap::new(),
        },
    )
}

fn append_completion_receipt(path: &FilePath, receipt: &CompletionReceipt) -> anyhow::Result<()> {
    if read_completed_ids(path)?
        .iter()
        .any(|existing| existing == &receipt.session_object_id)
    {
        return Ok(());
    }
    let mut file = OpenOptions::new().append(true).open(path)?;
    writeln!(file, "{}", serde_json::to_string(receipt)?)?;
    file.flush()?;
    file.sync_data()?;
    Ok(())
}

fn create_private_directory(path: &FilePath) -> 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(|| FilePath::new("."));
    sync_directory(parent)
}

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

fn validate_started_at(value: &str) -> Result<(), ApiError> {
    DateTime::parse_from_rfc3339(value)
        .map(|_| ())
        .map_err(|_| ApiError::bad("started_at must be an RFC 3339 timestamp"))
}

fn validate_idempotency(value: &str) -> Result<(), ApiError> {
    if value.is_empty() || value.len() > 255 {
        return Err(ApiError::bad(
            "idempotency_id must contain between 1 and 255 bytes",
        ));
    }
    Ok(())
}

fn validate_session_id(value: &str) -> Result<(), ApiError> {
    Uuid::parse_str(value)
        .map(|_| ())
        .map_err(|_| ApiError::bad("invalid session ID"))
}

fn require_version(record: &SessionRecord, expected: i64) -> Result<(), ApiError> {
    if record.version != expected {
        return Err(ApiError::conflict(format!(
            "Expected session version {expected}, found {}.",
            record.version
        )));
    }
    Ok(())
}

fn now() -> String {
    Utc::now().to_rfc3339()
}

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

    use super::*;

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

    fn service(label: &str) -> SessionHistory {
        let root = root(label);
        std::fs::create_dir_all(&root).unwrap();
        SessionHistory::open(Config {
            directory: root.join("sessions"),
            completed_list: root.join("session-history.txt"),
            provider_cost_compatibility: None,
        })
        .unwrap()
    }

    async fn start(service: &SessionHistory, idempotency_id: &str) -> SessionRecord {
        service
            .start(StartSession {
                idempotency_id: idempotency_id.into(),
                started_at: "2026-07-23T00:00:00Z".into(),
                session_type: "conversation".into(),
                duration_minutes: None,
                custom_prompt: None,
            })
            .await
            .unwrap()
            .value
    }

    #[tokio::test]
    async fn managed_log_and_control_creation_remain_coordinated() {
        let service = service("coordinated");
        let record = start(&service, "start-1").await;
        let mut journal = open_by_id(&service.state, &record.id).unwrap();
        journal
            .log
            .add_event(Role::SystemError, "The message exceeded capacity.")
            .unwrap();
        drop(journal);

        let command = service
            .enqueue(
                &record.id,
                NewCommand {
                    idempotency_id: "message-1".into(),
                    kind: "message".into(),
                    payload: json!({"text":"hello"}),
                },
            )
            .await
            .unwrap()
            .value;
        assert_eq!(command.status, "pending");
        assert_eq!(
            service.get(&record.id).await.unwrap().state["transcript"][0]["content"],
            "The message exceeded capacity."
        );
        assert_eq!(
            std::fs::read_dir(&service.state.config.directory)
                .unwrap()
                .count(),
            2
        );
    }

    #[tokio::test]
    async fn projected_state_is_rebuilt_from_the_preserved_transcript() {
        let service = service("transcript-preservation");
        let record = start(&service, "start-1").await;
        let mut journal = open_by_id(&service.state, &record.id).unwrap();
        journal
            .log
            .add_event(Role::UserMessage, "hello from the log")
            .unwrap();
        drop(journal);

        let checkpointed = service
            .checkpoint(
                &record.id,
                Checkpoint {
                    expected_version: record.version,
                    state: json!({
                        "sessionId":record.id,
                        "sessionType":"conversation",
                        "boxes":{"1":{"presentation":"discard"}},
                        "context":{"presentation":"discard"},
                        "chatendText":"discard",
                        "historyIngress":{"format":"kennedy-chatend","version":1}
                    }),
                    user_activity: false,
                },
            )
            .await
            .unwrap();

        assert!(checkpointed.state.get("boxes").is_none());
        assert!(checkpointed.state.get("context").is_none());
        assert!(checkpointed.state.get("chatendText").is_none());
        assert_eq!(
            checkpointed.state["transcript"][0]["content"],
            "hello from the log"
        );
        assert_eq!(
            checkpointed.state["historyIngress"]["format"],
            "kennedy-chatend"
        );
    }

    #[tokio::test]
    async fn command_and_stop_policy_remain_in_session_history() {
        let service = service("command-stop");
        let record = start(&service, "start-1").await;
        let command = service
            .enqueue(
                &record.id,
                NewCommand {
                    idempotency_id: "message-1".into(),
                    kind: "message".into(),
                    payload: json!({"text":"hello"}),
                },
            )
            .await
            .unwrap()
            .value;
        service.claim_command(&command.id).await.unwrap();

        let stop = service
            .request_stop(
                &record.id,
                NewStopRequest {
                    idempotency_id: "stop-1".into(),
                    scope: "turn".into(),
                },
            )
            .await
            .unwrap()
            .value;
        assert_eq!(stop.scope, "turn");
        assert_eq!(
            service.get(&record.id).await.unwrap().version,
            record.version
        );
        assert!(service.command_heads().await.unwrap()[0].cancel_requested);

        let completed = service
            .complete_stop(
                &stop.id,
                StopOutcome {
                    outcome: json!({"status":"stopped"}),
                },
            )
            .await
            .unwrap();
        assert_eq!(completed.status, "complete");
        assert!(service.stop_heads().await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn enumeration_skips_logs_without_lifecycle_without_replay() {
        let service = service("control-first-enumeration");
        let id = Uuid::new_v4().to_string();
        let directory = &service.state.config.directory;
        std::fs::write(
            directory.join(format!("{id}.session-log")),
            b"not a valid session log",
        )
        .unwrap();
        SessionControl::open(directory, &id, OpenMode::CreateNew)
            .unwrap()
            .unwrap();

        assert!(service.list().await.unwrap().is_empty());
        assert!(service.command_heads().await.unwrap().is_empty());
        assert!(service.stop_heads().await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn queue_heads_do_not_replay_session_logs() {
        let service = service("control-only-heads");
        let record = start(&service, "start-1").await;
        let command = service
            .enqueue(
                &record.id,
                NewCommand {
                    idempotency_id: "message-1".into(),
                    kind: "message".into(),
                    payload: json!({"text":"hello"}),
                },
            )
            .await
            .unwrap()
            .value;
        let stop = service
            .request_stop(
                &record.id,
                NewStopRequest {
                    idempotency_id: "stop-1".into(),
                    scope: "session".into(),
                },
            )
            .await
            .unwrap()
            .value;
        std::fs::write(
            service
                .state
                .config
                .directory
                .join(format!("{}.session-log", record.id)),
            b"not a valid session log",
        )
        .unwrap();

        let command_heads = service.command_heads().await.unwrap();
        assert_eq!(command_heads.len(), 1);
        assert_eq!(command_heads[0].id, command.id);
        let stop_heads = service.stop_heads().await.unwrap();
        assert_eq!(stop_heads.len(), 1);
        assert_eq!(stop_heads[0].id, stop.id);
        assert!(service.list().await.is_err());
    }

    #[tokio::test]
    async fn completion_synchronizes_receipt_before_live_file_cleanup() {
        let service = service("completion-order");
        let record = start(&service, "start-1").await;
        let id = record.id.clone();
        let mut state = record.state;
        state["historyIngress"] = json!({
            "completed":true,
            "sessionObjectId":"A1234567",
            "commitReceipt":{
                "transactionId":"T1234567",
                "sessionObjectId":"A1234567",
                "nodeIds":{},
                "objectIds":{}
            }
        });

        service
            .complete(
                &id,
                Checkpoint {
                    expected_version: record.version,
                    state,
                    user_activity: false,
                },
            )
            .await
            .unwrap();

        let receipts = read_completion_receipts(&service.state.config.completed_list).unwrap();
        assert_eq!(receipts[0].session_object_id, "A1234567");
        assert_eq!(
            std::fs::read_dir(&service.state.config.directory)
                .unwrap()
                .count(),
            0
        );
    }

    #[tokio::test]
    async fn listing_tolerates_completion_after_path_enumeration() {
        let service = service("concurrent-list");
        let record = start(&service, "start-1").await;
        let id = record.id.clone();
        let listed_paths = journal_paths(&service.state.config.directory).unwrap();

        let mut state = record.state;
        state["sessionObjectId"] = json!("A1234567");
        service
            .complete(
                &id,
                Checkpoint {
                    expected_version: record.version,
                    state,
                    user_activity: false,
                },
            )
            .await
            .unwrap();

        assert!(open_listed_journals(listed_paths).unwrap().is_empty());
        let listed = service.list().await.unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].phase, "complete");
        assert_eq!(listed[0].state["sessionObjectId"], "A1234567");
    }

    #[tokio::test]
    async fn ingress_retry_policy_remains_in_session_history() {
        let service = service("ingress-retry");
        let created = start(&service, "start-1").await;
        let mut record = service
            .request_ingress(
                &created.id,
                Checkpoint {
                    expected_version: created.version,
                    state: created.state,
                    user_activity: false,
                },
            )
            .await
            .unwrap();

        for attempt in 1..=INGRESS_FAILURE_LIMIT {
            record = service
                .start_ingress(
                    &record.id,
                    StartIngress {
                        expected_version: record.version,
                        provenance_id: "session:test".into(),
                    },
                )
                .await
                .unwrap();
            record = service
                .fail_ingress(
                    &record.id,
                    IngressFailure {
                        expected_version: record.version,
                        stage: "model_loop".into(),
                        code: Some("ingress_error".into()),
                        message: "transient failure".into(),
                        rounds_used: None,
                        context_tokens: None,
                        context_window_tokens: None,
                    },
                )
                .await
                .unwrap();
            assert_eq!(record.ingress_failure_count, attempt);
        }
        assert_eq!(record.phase, "ingress_failed");
        assert_eq!(
            record.ingress_failures.as_array().unwrap().len(),
            RETAINED_INGRESS_FAILURES
        );
    }

    #[tokio::test]
    async fn historical_completion_lines_remain_readable() {
        let service = service("completed");
        append_completed_id(&service.state.config.completed_list, "A1234567").unwrap();
        let listed = service.list().await.unwrap();
        assert_eq!(listed[0].state["sessionObjectId"], "A1234567");
        assert_eq!(
            listed[0].state["commitReceipt"]["sessionObjectId"],
            "A1234567"
        );
    }
}