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.

pub mod chatend {
    pub use kcode_chatend::{
        BoxContent, BoxId, BoxOwner, BoxRepresentation, BoxState, CacheExpectation,
        CanonicalRevision, Chatend, ContextProjection, ESTIMATED_BYTES_PER_TOKEN, Event, EventId,
        EventKind, FORMAT_VERSION, MAX_OBJECT_BYTES, ObjectLocation, ObjectMetadata, PendingId,
        PendingKind, PreparedProviderProjection, PreparedProviderResume, ProjectionItem,
        ProviderContext, ProviderCostEstimate, ProviderCostEstimator, ProviderCostSummary,
        ProviderMetering, ProviderTokenUsage, ProviderToolDefinition, Representation, Session,
        SessionKind, SessionMetadata, SessionStatus, ToolSlot, ToolSlotInput, ToolState,
        Transition, estimate_tokens,
    };
}

pub use chatend::Session;
pub use kcode_session_control_state::{SessionCommand, SessionRecord, SessionStopRequest};
pub use kcode_session_history_catalog::{
    CompletionReceipt, ProviderCostCompatibility, RecordCompletion,
};

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

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

const RETAINED_INGRESS_FAILURES: usize = 5;
const LAUNCH_COMMAND_NAME: &[u8] = b"kennedy.launch-session.initial-command.v1";

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

    #[cfg(test)]
    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, 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: Catalog,
    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 LaunchSession {
    pub session_id: String,
    pub started_at: String,
    pub effective_context_tokens: u64,
    pub channel: Value,
    pub state: Value,
    pub initial_message: Value,
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
pub struct SessionLaunch {
    pub session_id: String,
    pub command_id: String,
}

#[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, 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)?;
        let catalog = Catalog::open(config.completed_list.clone()).map_err(catalog_anyhow)?;
        Ok(Self {
            state: AppState {
                config,
                catalog,
                session_mutations: Arc::new(Mutex::new(HashMap::new())),
                stop_listeners: Arc::new(Mutex::new(HashMap::new())),
            },
        })
    }

    pub fn health(&self) -> Result<(), Error> {
        self.state.catalog.receipts().map_err(catalog_api)?;
        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>> {
        read_model(&self.state).legacy_provider_cost_summary_for_archive(archive, session_state)
    }

    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 launch_session(&self, input: LaunchSession) -> Result<SessionLaunch, Error> {
        launch_managed_session(self.state.clone(), input)
            .await
            .map_err(Into::into)
    }

    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 read_model(state: &AppState) -> ReadModel {
    ReadModel {
        provider_cost_compatibility: state.config.provider_cost_compatibility,
    }
}

fn catalog_anyhow(error: CatalogError) -> anyhow::Error {
    match error {
        CatalogError::Conflict(message) => anyhow::anyhow!(message),
        CatalogError::Storage(error) => error,
    }
}

fn catalog_api(error: CatalogError) -> ApiError {
    match error {
        CatalogError::Conflict(message) => ApiError::conflict(message),
        CatalogError::Storage(error) => ApiError::internal(error),
    }
}

async fn record_completed_session(
    state: AppState,
    input: RecordCompletion,
) -> Result<Value, ApiError> {
    let session_id = input.session_id.clone();
    let session_object_id = input.session_object_id.clone();
    let session_guard = 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()?;

    state.catalog.record(input, now()).map_err(catalog_api)?;

    if let Some(id) = 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":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(&state, &mut journal, &mut record)?;
    Ok(materialize(&state, record, &journal))
}

async fn start_managed_session(
    state: AppState,
    input: StartSession,
) -> Result<(bool, SessionRecord), ApiError> {
    validate_started_at(&input.started_at)?;
    validate_idempotency(&input.idempotency_id)?;
    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(&state, record, &journal)));
        }
    }

    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(&state, &mut journal, &mut record)?;
    Ok((true, materialize(&state, record, &journal)))
}

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

    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(&state, record, &journal)));
        }
    }
    for receipt in state.catalog.receipts().map_err(catalog_api)? {
        if ingress_idempotency_id(receipt.ingress_source.as_ref())
            == Some(input.idempotency_id.as_str())
        {
            return Ok((false, read_model(&state).completed(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(&state, &mut journal, &mut record)?;
    Ok((true, materialize(&state, record, &journal)))
}

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

async fn launch_managed_session(
    state: AppState,
    input: LaunchSession,
) -> Result<SessionLaunch, ApiError> {
    let prepared = prepare_launch(&input)?;
    if let Some(result) = completed_launch(&state, &input, &prepared.result)? {
        return Ok(result);
    }

    let session_guard = session_mutation(&state, &input.session_id)?;
    let _guard = session_guard.lock().map_err(ApiError::internal)?;

    if let Some(result) = completed_launch(&state, &input, &prepared.result)? {
        return Ok(result);
    }

    let path = state
        .config
        .directory
        .join(format!("{}.session-log", input.session_id));
    let existing_control = SessionControl::open(
        &state.config.directory,
        &input.session_id,
        OpenMode::ExistingOnly,
    )
    .map_err(ApiError::internal)?;
    if !path.exists() && existing_control.is_some() {
        return Err(ApiError::conflict(
            "The launch control journal exists without its session log.",
        ));
    }

    if path.exists() {
        match state.config.provider_cost_compatibility {
            Some(compatibility) => SessionHistoryIntegration::open_session(
                &path,
                prepared.metadata.clone(),
                None,
                Some(compatibility.estimator),
            ),
            None => SessionHistoryIntegration::open_session(
                &path,
                prepared.metadata.clone(),
                None,
                None,
            ),
        }
        .map_err(|error| {
            tracing::warn!(
                session_id=%input.session_id,
                %error,
                "LaunchSession found a nonmatching or unreadable session log"
            );
            ApiError::conflict("The launch session log does not match the requested launch.")
        })?;
    } else {
        SessionHistoryIntegration::create_session(&path, prepared.metadata.clone())
            .map_err(ApiError::internal)?;
    }

    let control = match existing_control {
        Some(control) => control,
        None => SessionControl::open(
            &state.config.directory,
            &input.session_id,
            OpenMode::OpenOrCreate,
        )
        .map_err(ApiError::internal)?
        .context("launch session-control journal was unexpectedly absent")
        .map_err(ApiError::internal)?,
    };
    let log = SessionStore::new(&state.config.directory)
        .open_session(&input.session_id)
        .map_err(ApiError::internal)?;
    let mut journal = SessionJournal { log, control };

    let durable_log = journal.list();
    if durable_log.header.session_id != input.session_id
        || durable_log.header.created_at != input.started_at
    {
        return Err(ApiError::conflict(
            "The launch session log identity does not match the requested launch.",
        ));
    }

    let projection = journal.projection();
    if projection.commands.len() > 1 {
        return Err(ApiError::conflict(
            "The launch session already contains additional commands.",
        ));
    }
    if let Some(existing) = projection.commands.values().next() {
        if !launch_command_matches(existing, &prepared.command) {
            return Err(ApiError::conflict(
                "The launch session initial command does not match the requested launch.",
            ));
        }
    } else {
        if projection.lifecycle.is_some() {
            return Err(ApiError::conflict(
                "The launch lifecycle exists without its initial command.",
            ));
        }
        append_command(&mut journal, &prepared.command)?;
    }

    match latest_lifecycle(&journal) {
        Some(existing) => {
            if !launch_lifecycle_matches(&existing, &prepared.record) {
                return Err(ApiError::conflict(
                    "The launch lifecycle does not match the requested launch.",
                ));
            }
        }
        None => {
            let mut record = prepared.record;
            append_lifecycle(&state, &mut journal, &mut record)?;
        }
    }

    Ok(prepared.result)
}

fn completed_launch(
    state: &AppState,
    input: &LaunchSession,
    result: &SessionLaunch,
) -> Result<Option<SessionLaunch>, ApiError> {
    let matching = state
        .catalog
        .receipts()
        .map_err(catalog_api)?
        .into_iter()
        .filter(|receipt| receipt.session_id.as_deref() == Some(&input.session_id))
        .collect::<Vec<_>>();
    if matching.is_empty() {
        return Ok(None);
    }
    if matching.iter().all(|receipt| {
        receipt.session_type.as_deref() == Some("conversation")
            && receipt.created_at.as_deref() == Some(input.started_at.as_str())
    }) {
        return Ok(Some(result.clone()));
    }
    Err(ApiError::conflict(
        "The launch session ID belongs to a different completed session.",
    ))
}

struct PreparedLaunch {
    metadata: chatend::SessionMetadata,
    command: SessionCommand,
    record: SessionRecord,
    result: SessionLaunch,
}

fn prepare_launch(input: &LaunchSession) -> Result<PreparedLaunch, ApiError> {
    validate_session_id(&input.session_id)?;
    validate_started_at(&input.started_at)?;
    if input.effective_context_tokens == 0 {
        return Err(ApiError::bad(
            "effective_context_tokens must be greater than zero",
        ));
    }
    if !input.channel.is_object() {
        return Err(ApiError::bad("channel must be a JSON object"));
    }
    if !input.state.is_object() {
        return Err(ApiError::bad("state must be a JSON object"));
    }
    if !input.initial_message.is_object() {
        return Err(ApiError::bad("initial_message must be a JSON object"));
    }
    if input.state.get("sessionId").and_then(Value::as_str) != Some(input.session_id.as_str()) {
        return Err(ApiError::bad(
            "state.sessionId must match the launch session ID",
        ));
    }
    if input.state.get("sessionType").and_then(Value::as_str) != Some("conversation") {
        return Err(ApiError::bad("state.sessionType must be conversation"));
    }
    if input.state.get("startedAt").and_then(Value::as_str) != Some(input.started_at.as_str()) {
        return Err(ApiError::bad(
            "state.startedAt must match the launch timestamp",
        ));
    }

    let metadata = chatend::SessionMetadata {
        session_id: input.session_id.clone(),
        kind: chatend::SessionKind::Conversation,
        created_at: input.started_at.clone(),
        effective_context_tokens: input.effective_context_tokens,
        channel: input.channel.clone(),
    };
    let encoded_metadata = serde_json::to_value(&metadata).map_err(ApiError::internal)?;
    if input.state.get("chatendMetadata") != Some(&encoded_metadata) {
        return Err(ApiError::bad(
            "state.chatendMetadata must match the launch session metadata",
        ));
    }

    let namespace =
        Uuid::parse_str(&input.session_id).map_err(|_| ApiError::bad("invalid session ID"))?;
    let command_id = Uuid::new_v5(&namespace, LAUNCH_COMMAND_NAME).to_string();
    let command = SessionCommand {
        id: command_id.clone(),
        conversation_id: input.session_id.clone(),
        sequence: 1,
        kind: "message".into(),
        payload: input.initial_message.clone(),
        status: "pending".into(),
        cancel_requested: false,
        outcome: None,
        created_at: input.started_at.clone(),
        processing_started_at: None,
        completed_at: None,
        idempotency_id: command_id.clone(),
    };
    let record = SessionRecord {
        id: input.session_id.clone(),
        phase: "active".into(),
        started_at: input.started_at.clone(),
        updated_at: input.started_at.clone(),
        state: input.state.clone(),
        provenance_id: Some(format!("launch:{}", input.session_id)),
        version: 1,
        last_user_message_at: None,
        ended_at: None,
        ingress_failure_count: 0,
        ingress_failures: json!([]),
        ingress_next_attempt_at: None,
        summary: false,
    };
    Ok(PreparedLaunch {
        metadata,
        command,
        record,
        result: SessionLaunch {
            session_id: input.session_id.clone(),
            command_id,
        },
    })
}

fn launch_command_matches(existing: &SessionCommand, expected: &SessionCommand) -> bool {
    existing.id == expected.id
        && existing.conversation_id == expected.conversation_id
        && existing.sequence == 1
        && existing.kind == "message"
        && existing.payload == expected.payload
        && existing.created_at == expected.created_at
        && existing.idempotency_id == expected.idempotency_id
}

fn launch_lifecycle_matches(existing: &SessionRecord, expected: &SessionRecord) -> bool {
    existing.id == expected.id
        && existing.started_at == expected.started_at
        && existing.provenance_id == expected.provenance_id
        && immutable_launch_state_matches(&existing.state, &expected.state)
}

fn immutable_launch_state_matches(current: &Value, expected: &Value) -> bool {
    const KEYS: &[&str] = &[
        "sessionId",
        "sessionType",
        "startedAt",
        "chatendMetadata",
        "launchContextNodeIds",
        "rootNodeIds",
        "launchProvenance",
    ];
    KEYS.iter()
        .all(|key| current.get(*key) == expected.get(*key))
}

async fn list_session_summaries(state: AppState) -> Result<Vec<SessionRecord>, ApiError> {
    let model = read_model(&state);
    let mut sessions = Vec::new();
    for projection in open_listed_control_projections(journal_paths(&state.config.directory)?)? {
        if let Some(record) = projection.lifecycle {
            sessions.push(model.active_summary(record));
        }
    }
    let receipts = state.catalog.receipts().map_err(catalog_api)?;
    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(model.completed(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) = state
        .catalog
        .receipts()
        .map_err(catalog_api)?
        .into_iter()
        .find(|receipt| receipt.session_object_id == id)
    {
        return Ok(read_model(&state).completed(receipt, false));
    }
    let journal = open_by_id(&state, &id)?;
    let record = latest_lifecycle(&journal).ok_or_else(ApiError::not_found)?;
    Ok(materialize(&state, record, &journal))
}

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 Some(control) = open_listed_control(&path).map_err(ApiError::internal)? else {
            continue;
        };
        let projection = control.projection();
        if projection.lifecycle.is_none() {
            continue;
        }
        if let Some(command) = projection.commands.get(command_id) {
            target = Some((path, command.conversation_id.clone()));
            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)?;
    if latest_lifecycle(&journal).is_none() {
        return Err(ApiError::not_found());
    }
    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(&state, &mut journal, &mut record)?;
    Ok(materialize(&state, record, &journal))
}

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(&state, &mut journal, &mut record)?;
    Ok(materialize(&state, record, &journal))
}

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(&state, &mut journal, &mut record)?;
    Ok(materialize(&state, record, &journal))
}

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 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 = "ingress_failed".into();
    record.ingress_next_attempt_at = None;
    record.version += 1;
    record.updated_at = now();
    append_lifecycle(&state, &mut journal, &mut record)?;
    tracing::error!(
        session_id = id,
        attempt,
        stage = %input.stage,
        code = input.code.as_deref().unwrap_or("ingress_error"),
        "Session History ingress stopped until manual retry"
    );
    Ok(materialize(&state, record, &journal))
}

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;
    if let Some(state) = record.state.as_object_mut() {
        state.remove("historyIngress");
    }
    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(&state, &mut journal, &mut record)?;
    Ok(materialize(&state, record, &journal))
}

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;
        }
        let attempt = record.ingress_failure_count.saturating_add(1);
        let failed_at = now();
        let mut failures = record
            .ingress_failures
            .as_array()
            .cloned()
            .unwrap_or_default();
        failures.push(json!({
            "attempt":attempt,
            "at":failed_at.clone(),
            "stage":"process_restart",
            "code":"ingress_interrupted",
            "message":"The ingress process stopped before the attempt completed.",
            "roundsUsed":Value::Null,
            "contextTokens":Value::Null,
            "contextWindowTokens":Value::Null,
        }));
        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 = "ingress_failed".into();
        record.ingress_next_attempt_at = None;
        record.version += 1;
        record.updated_at = failed_at;
        append_lifecycle(&state, &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(&state, 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 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();
    state
        .catalog
        .record(
            RecordCompletion {
                session_object_id: object_id.clone(),
                commit_receipt: Some(receipt),
                session_id: Some(record.id.clone()),
                session_type: record
                    .state
                    .get("sessionType")
                    .and_then(Value::as_str)
                    .map(str::to_owned),
                created_at: Some(record.started_at.clone()),
            },
            completed_at.clone(),
        )
        .map_err(catalog_api)?;
    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(&state, record, &journal);
    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(state: &AppState, mut record: SessionRecord) -> SessionRecord {
    read_model(state).prepare_control_state(&mut record.state);
    let ControlUpdate::Lifecycle(record) = ControlUpdate::Lifecycle(record).projected() else {
        unreachable!("lifecycle projection changed update kind");
    };
    record
}

fn append_lifecycle(
    state: &AppState,
    journal: &mut SessionJournal,
    record: &mut SessionRecord,
) -> Result<(), ApiError> {
    read_model(state).prepare_control_state(&mut record.state);
    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(state: &AppState, record: SessionRecord, journal: &SessionJournal) -> SessionRecord {
    read_model(state).active(record, &journal.list())
}

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

#[cfg(test)]
fn open_listed_journals(paths: Vec<PathBuf>) -> Result<Vec<SessionJournal>, ApiError> {
    let mut journals = Vec::new();
    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 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
    }

    fn launch_input(id: &str, text: &str) -> LaunchSession {
        let metadata = chatend::SessionMetadata {
            session_id: id.into(),
            kind: chatend::SessionKind::Conversation,
            created_at: "2026-08-14T00:00:00Z".into(),
            effective_context_tokens: 100_000,
            channel: json!({"type":"browser"}),
        };
        LaunchSession {
            session_id: id.into(),
            started_at: metadata.created_at.clone(),
            effective_context_tokens: metadata.effective_context_tokens,
            channel: metadata.channel.clone(),
            state: json!({
                "stateVersion":3,
                "sessionId":id,
                "sessionType":"conversation",
                "startedAt":metadata.created_at,
                "chatendMetadata":metadata,
                "launchContextNodeIds":[],
                "rootNodeIds":["US3fFBqB"],
                "launchProvenance":{"syntheticBootstrap":true},
                "orchestration":{"owner":"backend","status":"idle"},
            }),
            initial_message: json!({
                "text":text,
                "metadata":{"kennedyLaunchBootstrap":true}
            }),
        }
    }

    #[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(
                kcode_session_log::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."
        );
    }

    #[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(kcode_session_log::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"
        );
    }

    #[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!(service.command_heads().await.unwrap()[0].cancel_requested);
    }

    #[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 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");
    }

    #[tokio::test]
    async fn launch_is_lifecycle_last_deterministic_and_idempotent() {
        let service = service("launch-cold");
        let id = Uuid::new_v4().to_string();
        let input = launch_input(&id, "do the work");

        let first = service.launch_session(input.clone()).await.unwrap();
        let second = service.launch_session(input.clone()).await.unwrap();
        let expected =
            Uuid::new_v5(&Uuid::parse_str(&id).unwrap(), LAUNCH_COMMAND_NAME).to_string();

        assert_eq!(first, second);
        assert_eq!(first.session_id, id);
        assert_eq!(first.command_id, expected);
        let heads = service.command_heads().await.unwrap();
        assert_eq!(heads.len(), 1);
        assert_eq!(heads[0].id, expected);
        assert_eq!(heads[0].sequence, 1);
        assert_eq!(heads[0].payload, input.initial_message);
        assert_eq!(service.list().await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn launch_conflict_overwrites_nothing() {
        let service = service("launch-conflict");
        let id = Uuid::new_v4().to_string();
        let first = launch_input(&id, "first");
        service.launch_session(first.clone()).await.unwrap();

        let error = service
            .launch_session(launch_input(&id, "different"))
            .await
            .unwrap_err();
        assert_eq!(error.kind, ErrorKind::Conflict);
        assert_eq!(
            service.command_heads().await.unwrap()[0].payload,
            first.initial_message
        );
    }

    #[tokio::test]
    async fn invalid_launch_has_no_filesystem_effect() {
        let service = service("launch-invalid");
        let mut input = launch_input(&Uuid::new_v4().to_string(), "work");
        input.state["sessionType"] = json!("telegram");

        let error = service.launch_session(input).await.unwrap_err();
        assert_eq!(error.kind, ErrorKind::InvalidInput);
        assert_eq!(
            std::fs::read_dir(&service.state.config.directory)
                .unwrap()
                .count(),
            0
        );
    }

    #[tokio::test]
    async fn log_only_launch_is_reconciled() {
        let service = service("launch-log-only");
        let id = Uuid::new_v4().to_string();
        let input = launch_input(&id, "work");
        let prepared = prepare_launch(&input).unwrap();
        let path = service
            .state
            .config
            .directory
            .join(format!("{id}.session-log"));
        SessionHistoryIntegration::create_session(path, prepared.metadata).unwrap();

        let result = service.launch_session(input).await.unwrap();
        assert_eq!(result.session_id, id);
        assert_eq!(service.command_heads().await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn lifecycle_less_command_is_not_visible_or_mutable() {
        let service = service("launch-hidden-command");
        let id = Uuid::new_v4().to_string();
        let input = launch_input(&id, "work");
        let prepared = prepare_launch(&input).unwrap();
        let path = service
            .state
            .config
            .directory
            .join(format!("{id}.session-log"));
        SessionHistoryIntegration::create_session(&path, prepared.metadata).unwrap();
        let control =
            SessionControl::open(&service.state.config.directory, &id, OpenMode::CreateNew)
                .unwrap()
                .unwrap();
        let log = SessionStore::new(&service.state.config.directory)
            .open_session(&id)
            .unwrap();
        let mut journal = SessionJournal { log, control };
        append_command(&mut journal, &prepared.command).unwrap();

        assert!(service.command_heads().await.unwrap().is_empty());
        let error = service
            .claim_command(&prepared.command.id)
            .await
            .unwrap_err();
        assert_eq!(error.kind, ErrorKind::NotFound);

        let result = service.launch_session(input).await.unwrap();
        assert_eq!(result.command_id, prepared.command.id);
        assert_eq!(service.command_heads().await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn control_without_log_is_a_stable_conflict() {
        let service = service("launch-control-only");
        let id = Uuid::new_v4().to_string();
        SessionControl::open(&service.state.config.directory, &id, OpenMode::CreateNew)
            .unwrap()
            .unwrap();

        let error = service
            .launch_session(launch_input(&id, "work"))
            .await
            .unwrap_err();
        assert_eq!(error.kind, ErrorKind::Conflict);
        assert!(
            !service
                .state
                .config
                .directory
                .join(format!("{id}.session-log"))
                .exists()
        );
    }

    #[tokio::test]
    async fn concurrent_identical_launches_converge() {
        let service = service("launch-concurrent");
        let id = Uuid::new_v4().to_string();
        let input = launch_input(&id, "work");
        let left_service = service.clone();
        let left_input = input.clone();
        let right_service = service.clone();
        let right_input = input.clone();

        let (left, right) = tokio::join!(
            left_service.launch_session(left_input),
            right_service.launch_session(right_input)
        );
        assert_eq!(left.unwrap(), right.unwrap());
        assert_eq!(service.command_heads().await.unwrap().len(), 1);
    }

    #[tokio::test]
    async fn completed_launch_replays_without_live_files() {
        let service = service("launch-completed");
        let id = Uuid::new_v4().to_string();
        let input = launch_input(&id, "work");
        let result = service.launch_session(input.clone()).await.unwrap();

        service
            .record_completion(RecordCompletion {
                session_object_id: "A1234567".into(),
                commit_receipt: Some(CompletionReceipt {
                    transaction_id: Some("T1234567".into()),
                    session_object_id: "A1234567".into(),
                    session_id: Some(id.clone()),
                    session_type: Some("conversation".into()),
                    created_at: Some(input.started_at.clone()),
                    committed_at: Some("2026-08-14T00:01:00Z".into()),
                    ingress_source: None,
                    node_ids: BTreeMap::new(),
                    object_ids: BTreeMap::new(),
                }),
                session_id: Some(id.clone()),
                session_type: Some("conversation".into()),
                created_at: Some(input.started_at.clone()),
            })
            .await
            .unwrap();

        assert_eq!(service.launch_session(input).await.unwrap(), result);
        assert!(
            !service
                .state
                .config
                .directory
                .join(format!("{id}.session-log"))
                .exists()
        );
    }
}