mj_core/

state.rs

//! Durable controller-side state for Hel-managed sessions.

use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::path::{Component, Path, PathBuf};
use std::sync::Arc;

use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};

use crate::config::{
    Config, HarnessKind, ProjectRepository, TargetTemplate, atomic_write, data_dir, validate_id,
};
use crate::credentials::CredentialSyncSignal;
use crate::relay::{
    RELAY_EVENT_GENESIS_DIGEST, RelayOperationalState, SequencedEvent, WorkerEvent,
};
use crate::subagent::SubagentRecord;
use crate::targets::{AdditionalMount, validate_additional_mounts};

pub const STATE_VERSION: u32 = 1;

mod session_move;
pub use session_move::*;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum SessionState {
    Provisioning,
    Running,
    Disconnected,
    Checkpointing,
    Closing,
    Destroying,
    /// Checkpointed and torn down. Persisted as `"archived"` before the verb
    /// was renamed, so the alias keeps those records loading.
    #[serde(alias = "archived")]
    Stopped,
    Lost,
    Error,
    DestroyedWithDataLoss,
}

/// A lifecycle transition temporarily replaces the conversation in control surfaces.
/// Operation ownership takes precedence over intermediate durable session states.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum SessionTransitionKind {
    Starting,
    Resuming,
    Moving,
    Stopping,
    Destroying,
}

impl SessionTransitionKind {
    pub const fn label(self) -> &'static str {
        match self {
            Self::Starting => "Starting",
            Self::Resuming => "Resuming",
            Self::Moving => "Moving",
            Self::Stopping => "Stopping",
            Self::Destroying => "Destroying",
        }
    }

    pub fn for_session(state: SessionState, operation: Option<Self>) -> Option<Self> {
        operation.or_else(|| state.transition_kind())
    }
}

#[cfg(test)]
mod transition_tests {
    use super::{SessionState, SessionTransitionKind};

    #[test]
    fn operation_ownership_hides_intermediate_move_states_but_not_ordinary_live_work() {
        for state in [
            SessionState::Stopped,
            SessionState::Running,
            SessionState::Disconnected,
        ] {
            assert_eq!(
                SessionTransitionKind::for_session(state, Some(SessionTransitionKind::Moving)),
                Some(SessionTransitionKind::Moving)
            );
            assert_eq!(SessionTransitionKind::for_session(state, None), None);
        }
        assert_eq!(SessionState::Checkpointing.transition_kind(), None);
        assert_eq!(
            SessionState::Closing.transition_kind(),
            Some(SessionTransitionKind::Stopping)
        );
    }
}

/// Controller-owned execution state derived from the relay event stream.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum MaterializedExecutionState {
    #[default]
    Idle,
    Running {
        started_at_ms: i64,
    },
    Closing,
    Closed,
}

pub use crate::transcript::{TerminalOutputRecord, TranscriptBody, TranscriptItem};

/// What a durable queue entry does when its turn comes.
///
/// Serialized without a tag for prompts so entries written before configuration
/// changes could be queued keep loading unchanged.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum QueuedCommandKind {
    #[default]
    Prompt,
    SetConfig {
        key: String,
        value: String,
    },
}

impl QueuedCommandKind {
    pub fn is_prompt(&self) -> bool {
        matches!(self, Self::Prompt)
    }
}

/// The composer form of a configuration change, used both as the queue entry's
/// display text and as the text peeled back into the composer for editing.
pub fn config_command_text(key: &str, value: &str) -> String {
    if key == "fast-mode" {
        "/fast".to_owned()
    } else {
        format!("/{key} {value}")
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MaterializedQueuedPrompt {
    pub command_id: String,
    #[serde(default, skip_serializing_if = "QueuedCommandKind::is_prompt")]
    pub kind: QueuedCommandKind,
    pub content: Vec<serde_json::Value>,
    pub queued_at_ms: i64,
    /// Relay acceptance ordinal of the `CommandQueued` event that created this
    /// entry. It is the turn identity the API hands back to callers, so wait
    /// can tell one queued prompt's outcome from another's.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub accepted_ordinal: Option<u64>,
}

/// The prompt currently executing, recorded when its `CommandStarted` event is
/// projected and cleared when the command completes.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MaterializedTurn {
    pub command_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub accepted_ordinal: Option<u64>,
    /// Ordinal of the `CommandStarted` event, which is also the transcript
    /// position of the turn's first item.
    pub turn_start_position: u64,
    pub started_at_ms: i64,
}

/// How a prompt ended.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum TurnOutcomeKind {
    /// The harness finished the turn and reported this stop reason.
    Completed { stop_reason: String },
    /// The relay refused the command before it ran.
    Rejected { message: String },
    /// The command was interrupted after being accepted.
    Interrupted { message: String },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PromptCompletion {
    Finished,
    Cancelled,
    QuotaLimit,
    Error,
}

/// Shared interpretation for wait responses and durable completion events.
pub fn classify_prompt_completion(stop_reason: &str) -> PromptCompletion {
    let normalized = stop_reason
        .chars()
        .filter(|character| *character != '_' && *character != '-')
        .flat_map(char::to_lowercase)
        .collect::<String>();
    match normalized.as_str() {
        "endturn" => PromptCompletion::Finished,
        "cancelled" | "canceled" => PromptCompletion::Cancelled,
        "quotalimit" => PromptCompletion::QuotaLimit,
        _ if crate::relay::is_capacity_stop_reason(stop_reason) => PromptCompletion::QuotaLimit,
        _ => PromptCompletion::Error,
    }
}

/// The most recent finished prompt on a session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MaterializedTurnOutcome {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub diagnostic: Option<crate::diagnostic::TurnDiagnostic>,

    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub usage: Option<crate::usage::TokenUsage>,
    pub command_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub accepted_ordinal: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub turn_start_position: Option<u64>,
    pub completed_ordinal: u64,
    pub completed_at_ms: i64,
    pub outcome: TurnOutcomeKind,
}

/// Canonical controller projection for one logical ACP session.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MaterializedSession {
    pub session_id: String,
    pub applied_event_ordinal: u64,
    pub applied_event_digest: String,
    /// Monotonic controller projection watermark derived from relay event
    /// receipt times. It is deliberately independent of retained rows.
    pub last_activity_at_ms: Option<i64>,
    pub execution: MaterializedExecutionState,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_title: Option<String>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub configuration: BTreeMap<String, serde_json::Value>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    /// Transcript items are shared by pointer so cloning a snapshot copies
    /// handles rather than the whole conversation.
    pub transcript: Vec<Arc<TranscriptItem>>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub queued_prompts: Vec<MaterializedQueuedPrompt>,
    /// In-flight form requests are projected durably, but their answers are
    /// connection-only and never enter this state.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pending_elicitations: Vec<crate::elicitation::ElicitationRequest>,
    /// The prompt running right now, if any.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub active_turn: Option<MaterializedTurn>,
    /// The most recently finished prompt, kept after the session stops so a
    /// caller can still read how the last turn ended.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_turn_outcome: Option<MaterializedTurnOutcome>,
}

/// The small portion of a durable projection needed to populate dashboard
/// rows before the live session delivers its full transcript snapshot.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MaterializedSessionSummary {
    pub session_id: String,
    pub applied_event_ordinal: u64,
    pub last_activity_at_ms: Option<i64>,
    pub execution: MaterializedExecutionState,
    pub session_title: Option<String>,
    pub last_agent_message: Option<String>,
    pub last_user_message: Option<String>,
    /// Whether the last nonempty agent message appears after the last
    /// nonempty user message in transcript order.
    pub last_agent_message_follows_last_user: bool,
    pub agent_message_latest_content_ordinals: Vec<u64>,
    pub session_restart_event_ordinals: Vec<u64>,
}

impl MaterializedSession {
    pub fn empty(session_id: impl Into<String>) -> Self {
        Self {
            session_id: session_id.into(),
            applied_event_ordinal: 0,
            applied_event_digest: RELAY_EVENT_GENESIS_DIGEST.into(),
            last_activity_at_ms: None,
            execution: MaterializedExecutionState::Idle,
            session_title: None,
            configuration: BTreeMap::new(),
            transcript: Vec::new(),
            queued_prompts: Vec::new(),
            pending_elicitations: Vec::new(),
            active_turn: None,
            last_turn_outcome: None,
        }
    }

    pub fn last_activity_at_ms(&self) -> Option<i64> {
        self.last_activity_at_ms
    }

    /// Resolve the title exposed by a live materialized session.
    ///
    /// Sessions created before provisional titles were projected can still
    /// have an untitled transcript. Derive the same bounded fallback from
    /// their first visible user prompt when reading them.
    pub fn resolved_title(&self) -> Option<String> {
        self.session_title
            .as_deref()
            .and_then(normalize_session_title)
            .or_else(|| {
                self.transcript.iter().find_map(|item| {
                    let TranscriptBody::User { content } = &item.body else {
                        return None;
                    };
                    provisional_session_title(&crate::transcript::materialized_content_text(
                        content,
                    ))
                })
            })
            .or_else(|| {
                self.queued_prompts
                    .iter()
                    .filter(|prompt| prompt.kind.is_prompt())
                    .find_map(|prompt| {
                        provisional_session_title(&crate::transcript::materialized_content_text(
                            &prompt.content,
                        ))
                    })
            })
    }

    pub fn unread_agent_messages_after(&self, viewed_through_event_ordinal: u64) -> u64 {
        self.transcript
            .iter()
            .filter(|item| {
                item.latest_content_event_ordinal
                    .is_some_and(|ordinal| ordinal > viewed_through_event_ordinal)
                    && item.is_nonempty_agent_message()
            })
            .count() as u64
    }

    pub fn unread_session_restarts_after(&self, viewed_through_event_ordinal: u64) -> u64 {
        self.transcript
            .iter()
            .filter(|item| {
                item.position > viewed_through_event_ordinal && item.is_session_restart()
            })
            .count() as u64
    }

    pub fn validate(&self) -> Result<()> {
        validate_id("session", &self.session_id)?;
        validate_relay_event_frontier(
            self.applied_event_ordinal,
            &self.applied_event_digest,
            "materialized session event frontier",
        )?;
        if self
            .session_title
            .as_ref()
            .is_some_and(|title| title.trim().is_empty())
        {
            bail!("materialized session has an empty title");
        }
        let mut item_ids = BTreeSet::new();
        for item in &self.transcript {
            item.validate(self.applied_event_ordinal)?;
            if !item_ids.insert(item.stable_id.as_str()) {
                bail!(
                    "materialized transcript contains duplicate item {:?}",
                    item.stable_id
                );
            }
        }
        let mut command_ids = BTreeSet::new();
        for prompt in &self.queued_prompts {
            if prompt.command_id.trim().is_empty() {
                bail!("materialized prompt queue has an empty command id");
            }
            if !command_ids.insert(prompt.command_id.as_str()) {
                bail!(
                    "materialized prompt queue contains duplicate command {:?}",
                    prompt.command_id
                );
            }
            if let QueuedCommandKind::SetConfig { key, value } = &prompt.kind
                && (key.trim().is_empty() || value.trim().is_empty())
            {
                bail!(
                    "materialized queued configuration change {:?} is incomplete",
                    prompt.command_id
                );
            }
        }
        Ok(())
    }
}

/// A materialized session paired with the live worker's relay state. The
/// session manager hands this to every reader that needs both the durable
/// projection and the connection's operational status.
#[derive(Debug, Clone, PartialEq)]
pub struct ManagedSessionSnapshot {
    pub materialized: MaterializedSession,
    /// What `materialized.transcript` leaves out, and the facts that live
    /// there. See [`ProjectionWindow`].
    pub window: ProjectionWindow,
    pub operational: RelayOperationalState,
    /// Newest relay event observed by this live actor that asks for immediate
    /// credential reconciliation. This is intentionally ephemeral: it avoids
    /// retaining raw replay pages or rescanning projected history.
    pub latest_credential_sync_signal: Option<CredentialSyncSignal>,
    /// Content address of the executable the connected worker is running, as
    /// it reported in hello. `None` when the connection did not come from a
    /// live worker or the worker predates the field; either way the worker is
    /// not known to be the build this controller would install.
    pub worker_build: Option<String>,
    /// Pending parent-tool work fetched from the target worker.
    pub subagent_requests: Vec<crate::subagent::SubagentToolRequest>,
    /// Recently completed tool work cached by the worker for idempotent calls.
    pub subagent_results: Vec<crate::subagent::SubagentToolResult>,
}

/// What a projection's transcript window leaves out.
///
/// A polled projection carries only the end of the transcript, because that is
/// all any viewer shows and loading the rest is work proportional to history.
/// Two facts a reader needs live outside that window: the provisional title
/// comes from the *first* user message, and the newest turn start is outside
/// it whenever a single turn is longer than the window. Both are read
/// separately, with one indexed query each, rather than found by scanning.
///
/// A complete projection answers both by scanning what it already holds, which
/// is what [`ProjectionWindow::of`] does.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProjectionWindow {
    /// Transcript items before the window. Zero when the projection is whole.
    pub omitted_items: usize,
    /// The title derived from the first user message.
    pub provisional_title: Option<String>,
    /// Position of the newest turn start — a user message or the marker for a
    /// turn the harness began on its own — whether or not it is in the
    /// window. `None` when the session has none.
    pub latest_turn_start_position: Option<u64>,
}

impl ProjectionWindow {
    /// The window of a projection that omits nothing.
    #[must_use]
    pub fn of(session: &MaterializedSession) -> Self {
        Self {
            omitted_items: 0,
            provisional_title: session.transcript.iter().find_map(|item| {
                let TranscriptBody::User { content } = &item.body else {
                    return None;
                };
                provisional_session_title(&crate::transcript::materialized_content_text(content))
            }),
            latest_turn_start_position: session
                .transcript
                .iter()
                .rev()
                .find(|item| item.is_turn_start())
                .map(|item| item.position),
        }
    }
}

impl ManagedSessionSnapshot {
    /// The session's title, using the same precedence as
    /// [`MaterializedSession::resolved_title`] but taking the provisional
    /// title from the window rather than from a transcript head that a polled
    /// projection does not carry.
    #[must_use]
    pub fn resolved_title(&self) -> Option<String> {
        self.materialized
            .session_title
            .as_deref()
            .and_then(normalize_session_title)
            .or_else(|| self.window.provisional_title.clone())
            .or_else(|| {
                self.materialized
                    .queued_prompts
                    .iter()
                    .filter(|prompt| prompt.kind.is_prompt())
                    .find_map(|prompt| {
                        provisional_session_title(&crate::transcript::materialized_content_text(
                            &prompt.content,
                        ))
                    })
            })
    }

    /// The position of the turn this session most recently finished, or `None`
    /// while it is still working. Same answer as
    /// [`latest_completed_turn_ordinal`], from a position the window carries
    /// rather than a scan back through the transcript.
    #[must_use]
    pub fn latest_completed_turn_ordinal(&self) -> Option<u64> {
        if self.materialized.execution != MaterializedExecutionState::Idle {
            return None;
        }
        self.window.latest_turn_start_position
    }
}

/// One session's activity, reported to the recovery coordinator.
#[derive(Debug, Clone)]
pub struct RecoveryObservation {
    pub session: SessionRecord,
    pub config: Config,
    pub latest_completed_turn_ordinal: Option<u64>,
    pub execution: MaterializedExecutionState,
    /// Whether live provider-owned work permits an automatic checkpoint now.
    /// This is separate from materialized execution because Kimi detached
    /// agents outlive the parent turn that returned the session to `Idle`.
    pub checkpoint_safe: bool,
}

/// The position where the session's most recent finished turn began, or
/// `None` while it is still working. A turn starts at a user message or at the
/// marker for a turn the harness began on its own, so autonomous work is
/// covered once it settles.
pub fn latest_completed_turn_ordinal(session: &MaterializedSession) -> Option<u64> {
    if session.execution != MaterializedExecutionState::Idle {
        return None;
    }
    session
        .transcript
        .iter()
        .rev()
        .find(|item| item.is_turn_start())
        .map(|item| item.position)
}

pub fn validate_relay_event_digest(digest: &str, name: &str) -> Result<()> {
    if digest.len() != 64
        || !digest
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        bail!("{name} must be a lowercase SHA-256 digest");
    }
    Ok(())
}

pub fn validate_relay_event_frontier(ordinal: u64, digest: &str, name: &str) -> Result<()> {
    validate_relay_event_digest(digest, name)?;
    if (ordinal == 0) != (digest == RELAY_EVENT_GENESIS_DIGEST) {
        bail!("{name} has inconsistent ordinal {ordinal} and digest {digest}");
    }
    Ok(())
}

fn is_false(value: &bool) -> bool {
    !*value
}

impl SessionState {
    /// Recovery without a live operation still hides an unfinished target transition.
    /// Ordinary checkpoints and reconnects deliberately keep their conversation visible.
    pub const fn transition_kind(self) -> Option<SessionTransitionKind> {
        match self {
            Self::Provisioning => Some(SessionTransitionKind::Starting),
            Self::Closing => Some(SessionTransitionKind::Stopping),
            Self::Destroying => Some(SessionTransitionKind::Destroying),
            _ => None,
        }
    }

    /// True while the session still belongs on the dashboard. `Closing` and
    /// `Checkpointing` stay active on purpose: a stop that has not produced a
    /// verified checkpoint must not make its row disappear.
    pub const fn is_active(self) -> bool {
        matches!(
            self,
            Self::Provisioning
                | Self::Running
                | Self::Disconnected
                | Self::Checkpointing
                | Self::Closing
                | Self::Destroying
                | Self::Error
        )
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum PodmanWorkspaceLocator {
    #[default]
    ContainerLayer,
    Volume {
        name: String,
    },
    HostPath {
        path: PathBuf,
        helper: Vec<String>,
        resource: String,
    },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum TargetLocator {
    LocalBare {
        worker_root: PathBuf,
    },
    LocalPodman {
        container_id: String,
        #[serde(default)]
        workspace_storage: PodmanWorkspaceLocator,
    },
    LocalDocker {
        container_id: String,
    },
    AppleContainer {
        container_id: String,
    },
    AwsEc2 {
        instance_id: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        address: Option<String>,
    },
    SshBare {
        host: String,
        workspace: PathBuf,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        worker_id: Option<String>,
    },
    SshPodman {
        host: String,
        container_id: String,
        #[serde(default)]
        workspace_storage: PodmanWorkspaceLocator,
    },
    SshDocker {
        host: String,
        container_id: String,
    },
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum ManagedWorktreeTarget {
    Local,
    Ssh {
        destination: String,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        ssh_args: Vec<String>,
    },
}

/// Whether a selected project can create a session-owned Git checkout.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ManagedWorktreeOptions {
    pub available: bool,
    pub default_create: bool,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ManagedWorktree {
    pub source_project_directory: PathBuf,
    pub source_repository: PathBuf,
    pub worktree_root: PathBuf,
    pub branch: String,
    pub target: ManagedWorktreeTarget,
    /// The commit the session branch was created at. Recorded so an export can
    /// diff against it in one read; sessions created before this field existed
    /// fall back to the branch reflog, which expires.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub base_commit: Option<String>,
}

impl ManagedWorktree {
    fn validate(&self, session_id: &str, project_directory: Option<&Path>) -> Result<()> {
        for (label, path) in [
            ("source project directory", &self.source_project_directory),
            ("source repository", &self.source_repository),
            ("worktree root", &self.worktree_root),
        ] {
            if !path.is_absolute() || path.components().any(|part| part == Component::ParentDir) {
                bail!("managed worktree {label} must be an absolute safe path");
            }
        }
        if !self
            .source_project_directory
            .starts_with(&self.source_repository)
        {
            bail!("managed worktree source directory is outside its repository");
        }
        let expected_root = self
            .source_repository
            .join(".mj")
            .join("worktrees")
            .join(session_id);
        if self.worktree_root != expected_root {
            bail!("managed worktree root does not match the session-owned path");
        }
        if self.branch != format!("mj/{session_id}") {
            bail!("managed worktree branch does not match the session id");
        }
        let relative = self
            .source_project_directory
            .strip_prefix(&self.source_repository)
            .expect("source relationship checked above");
        if project_directory != Some(self.worktree_root.join(relative).as_path()) {
            bail!("session project directory does not match its managed worktree");
        }
        match &self.target {
            ManagedWorktreeTarget::Local => {}
            ManagedWorktreeTarget::Ssh { destination, .. } if destination.trim().is_empty() => {
                bail!("managed SSH worktree has an empty destination")
            }
            ManagedWorktreeTarget::Ssh { .. } => {}
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum SessionResourceAllocation {
    Container {
        cpus: u64,
        memory_bytes: u64,
    },
    AwsEc2 {
        instance_type: String,
        vcpus: u64,
        memory_bytes: u64,
    },
}

impl SessionResourceAllocation {
    pub fn validate(&self) -> Result<()> {
        match self {
            Self::Container { cpus, memory_bytes } if *cpus == 0 || *memory_bytes == 0 => {
                bail!("container resource allocation must have non-zero CPU and memory")
            }
            Self::AwsEc2 {
                instance_type,
                vcpus,
                memory_bytes,
            } if instance_type.trim().is_empty() || *vcpus == 0 || *memory_bytes == 0 => {
                bail!("EC2 resource allocation must have an instance type, CPU, and memory")
            }
            _ => Ok(()),
        }
    }
}

/// The CPU count an allocation grants, regardless of target kind.
pub fn allocation_cpus(allocation: &SessionResourceAllocation) -> u64 {
    match allocation {
        SessionResourceAllocation::Container { cpus, .. } => *cpus,
        SessionResourceAllocation::AwsEc2 { vcpus, .. } => *vcpus,
    }
}

/// The memory, in bytes, an allocation grants, regardless of target kind.
pub fn allocation_memory(allocation: &SessionResourceAllocation) -> u64 {
    match allocation {
        SessionResourceAllocation::Container { memory_bytes, .. }
        | SessionResourceAllocation::AwsEc2 { memory_bytes, .. } => *memory_bytes,
    }
}

impl TargetLocator {
    fn validate(&self, session_id: &str) -> Result<()> {
        match self {
            Self::LocalBare { worker_root } => {
                if !worker_root.is_absolute()
                    || worker_root
                        .components()
                        .any(|part| part == Component::ParentDir)
                    || !worker_root.ends_with(session_id)
                {
                    bail!(
                        "local bare worker root must be an absolute safe path ending in the session id"
                    );
                }
            }
            Self::LocalPodman { container_id, .. }
            | Self::LocalDocker { container_id }
            | Self::AppleContainer { container_id }
            | Self::SshPodman { container_id, .. }
            | Self::SshDocker { container_id, .. }
                if container_id.trim().is_empty() =>
            {
                bail!("target locator has an empty container id")
            }
            Self::AwsEc2 { instance_id, .. } if instance_id.trim().is_empty() => {
                bail!("target locator has an empty AWS instance id")
            }
            Self::SshBare {
                host, workspace, ..
            } => {
                if host.trim().is_empty() {
                    bail!("bare SSH target locator has an empty host");
                }
                if workspace.as_os_str().is_empty()
                    || workspace
                        .components()
                        .any(|part| part == Component::ParentDir)
                    || !workspace.ends_with(session_id)
                {
                    bail!("bare SSH target locator must be a safe path ending in the session id");
                }
            }
            Self::SshPodman { host, .. } if host.trim().is_empty() => {
                bail!("SSH Podman target locator has an empty host")
            }
            Self::SshDocker { host, .. } if host.trim().is_empty() => {
                bail!("SSH Docker target locator has an empty host")
            }
            _ => {}
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CheckpointMetadata {
    pub archive_path: PathBuf,
    /// Lowercase SHA-256 digest of the verified archive.
    pub sha256: String,
    pub created_at: String,
    pub event_frontier: u64,
}

impl CheckpointMetadata {
    fn validate(&self) -> Result<()> {
        if self.archive_path.as_os_str().is_empty() {
            bail!("checkpoint archive path is empty");
        }
        if self.sha256.len() != 64
            || !self
                .sha256
                .bytes()
                .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
        {
            bail!("checkpoint SHA-256 must be 64 lowercase hexadecimal characters");
        }
        if self.created_at.trim().is_empty() {
            bail!("checkpoint timestamp is empty");
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SessionRecord {
    pub id: String,
    /// Owning workspace while active, or the most recent workspace while inactive.
    ///
    /// Inactive histories are globally resumable, so this id may refer to a
    /// workspace that has since been deleted.
    #[serde(default = "default_session_workspace_id")]
    pub workspace_id: String,
    pub title: String,
    pub harness_kind: HarnessKind,
    pub last_profile: String,
    pub bundle_id: String,
    /// Existing project directory used directly by a local or SSH bare target.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub project_directory: Option<PathBuf>,
    /// Git worktree created and owned by Hel for this raw-project session.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub managed_worktree: Option<ManagedWorktree>,
    /// None preserves automatic selection; false uses the selected directory.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub create_managed_worktree: Option<bool>,
    pub target_template_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub resource_allocation: Option<SessionResourceAllocation>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub additional_mounts: Vec<AdditionalMount>,
    /// Per-session container CPU limit that overrides the target template's
    /// value. It is applied the next time the container is created.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub container_cpus: Option<String>,
    /// Per-session container memory limit that overrides the target
    /// template's value. It is applied the next time the container is created.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub container_memory: Option<String>,
    pub state: SessionState,
    /// Legacy visibility preference, retained for record compatibility.
    /// Current surfaces do not hide sessions based on this flag.
    #[serde(default, skip_serializing_if = "is_false")]
    pub archived: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target: Option<TargetLocator>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub native_session_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub acp_session_title: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_title_override: Option<String>,
    pub created_at: String,
    pub updated_at: String,
    #[serde(default, alias = "detached_after_event_ordinal")]
    pub viewed_through_event_ordinal: u64,
    /// Unsent chat input carried across a detach, so returning to a session
    /// restores what the user was typing. Empty means no draft.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub draft_input: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_error: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_checkpoint_error: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub checkpoint: Option<CheckpointMetadata>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HostContainerSize {
    pub cpus: u64,
    pub memory_bytes: u64,
}

fn default_session_workspace_id() -> String {
    crate::workspace::DEFAULT_WORKSPACE_ID.to_owned()
}

impl SessionRecord {
    /// Configuration drift belongs to this session, not the entire controller.
    /// The diagnostic contains only public identifiers, so both UIs can show it.
    pub fn configuration_issue(&self, config: &Config) -> Option<String> {
        if !self.state.is_active() {
            return None;
        }
        let mut issues = Vec::new();
        match config.profiles.get(&self.last_profile) {
            None => issues.push(format!("missing profile {:?}", self.last_profile)),
            Some(profile) if profile.kind != self.harness_kind => issues.push(format!(
                "expects {:?}, but profile {:?} is {:?}",
                self.harness_kind, self.last_profile, profile.kind
            )),
            Some(_) => {}
        }
        if self.project_directory.is_none() && !config.bundles.contains_key(&self.bundle_id) {
            issues.push(format!("missing bundle {:?}", self.bundle_id));
        }
        if !config.targets.contains_key(&self.target_template_id) {
            issues.push(format!(
                "missing target template {:?}",
                self.target_template_id
            ));
        }
        (!issues.is_empty()).then(|| format!(
            "Session {:?} needs configuration repair: {}. Restore these entries in config.toml, then retry. Run mj setup to rediscover installed profiles and targets; existing sessions are preserved.",
            self.id, issues.join("; ")
        ))
    }

    pub fn validate_configuration(&self, config: &Config) -> Result<()> {
        if let Some(issue) = self.configuration_issue(config) {
            bail!("{issue}");
        }
        Ok(())
    }

    /// User-visible session name, independent of the initial prompt stored in `title`.
    pub fn display_title(&self) -> &str {
        self.session_title_override
            .as_deref()
            .or(self.acp_session_title.as_deref())
            .unwrap_or(&self.id)
    }

    /// Project this session works in, as the session list and the chat header
    /// both name it: the source repository of a managed worktree, else the
    /// project directory, else the bundle's primary repository, else the
    /// bundle id.
    pub fn project_name(&self, config: &Config) -> String {
        if let Some(worktree) = &self.managed_worktree {
            return path_leaf(&worktree.source_repository);
        }
        if let Some(project_directory) = &self.project_directory {
            return path_leaf(project_directory);
        }
        self.bundle_source_name(config)
    }

    /// Target label used by the live session summary. Bare targets identify
    /// the project directory they open directly; workspace targets already
    /// identify the provisioned environment on their own.
    pub fn project_target(&self, config: &Config, target_id: &str) -> String {
        if !matches!(
            config.targets.get(target_id),
            Some(TargetTemplate::LocalBare | TargetTemplate::SshBare { .. })
        ) {
            return target_id.to_owned();
        }
        self.managed_worktree
            .as_ref()
            .map(|worktree| &worktree.source_project_directory)
            .or(self.project_directory.as_ref())
            .and_then(|path| path.file_name())
            .map_or_else(
                || target_id.to_owned(),
                |directory| format!("{target_id}/{}", directory.to_string_lossy()),
            )
    }

    /// Stable source identity used to group sessions. Managed worktrees point
    /// back at their source repository, raw sessions use their project
    /// directory until their Git origin is resolved, and bundle sessions use
    /// their complete canonical repository set when configured.
    pub fn project_source(&self, config: &Config) -> ProjectSourceIdentity {
        if let Some(worktree) = &self.managed_worktree {
            return ProjectSourceIdentity::path(&worktree.source_repository, None);
        }
        if let Some(project_directory) = &self.project_directory {
            let remote = match &self.target {
                Some(TargetLocator::SshBare { host, .. }) => Some(host.as_str()),
                _ => None,
            };
            return ProjectSourceIdentity::path(project_directory, remote);
        }
        self.bundle_source_identity(config)
            .unwrap_or_else(|| ProjectSourceIdentity {
                key: format!("bundle:{}", self.bundle_id),
                short: path_leaf(Path::new(&self.bundle_id)),
                full: self.bundle_id.clone(),
            })
    }

    /// Resolve the display name shared by session headings, chat headers, and
    /// resume details for a bundle-backed session.
    fn bundle_source_name(&self, config: &Config) -> String {
        self.bundle_source_identity(config)
            .map(|source| source.short)
            .unwrap_or_else(|| path_leaf(Path::new(&self.bundle_id)))
    }

    /// Resolve the canonical identity of every repository in a bundle for
    /// grouping and display naming.
    fn bundle_source_identity(&self, config: &Config) -> Option<ProjectSourceIdentity> {
        let bundle = config.bundles.get(&self.bundle_id)?;
        let sources = bundle
            .repositories
            .iter()
            .map(repository_source_identity)
            .collect::<Option<Vec<_>>>()?;
        ProjectSourceIdentity::bundle(sources)
    }

    /// Orders two sessions the way the session list's sequence view does:
    /// oldest first by creation time, with the id as a stable tiebreak. A
    /// session whose timestamp does not parse sorts last.
    pub fn compare_by_creation(&self, other: &Self) -> std::cmp::Ordering {
        self.creation_order_key().cmp(&other.creation_order_key())
    }

    /// Parse once per session when used with `sort_by_cached_key`.
    pub fn creation_order_key(&self) -> (bool, Option<i64>, &str) {
        let timestamp = created_at_seconds(&self.created_at);
        (timestamp.is_none(), timestamp, &self.id)
    }

    fn validate(&self, map_id: &str) -> Result<()> {
        validate_id("session", &self.id)?;
        if self.id != map_id {
            bail!(
                "session map key {map_id:?} does not match record id {:?}",
                self.id
            );
        }
        validate_id("workspace", &self.workspace_id)?;
        validate_id("profile", &self.last_profile)?;
        validate_id("bundle", &self.bundle_id)?;
        if let Some(project_directory) = &self.project_directory
            && (!project_directory.is_absolute()
                || project_directory
                    .components()
                    .any(|part| part == Component::ParentDir))
        {
            bail!("session {:?} has an unsafe project directory", self.id);
        }
        if let Some(managed_worktree) = &self.managed_worktree {
            managed_worktree.validate(&self.id, self.project_directory.as_deref())?;
        }
        validate_id("target template", &self.target_template_id)?;
        if let Some(allocation) = &self.resource_allocation {
            allocation.validate()?;
        }
        validate_additional_mounts(&self.additional_mounts)?;
        if self.title.trim().is_empty() {
            bail!("session {:?} has an empty title", self.id);
        }
        if self
            .acp_session_title
            .as_ref()
            .is_some_and(|title| title.trim().is_empty())
            || self
                .session_title_override
                .as_ref()
                .is_some_and(|title| title.trim().is_empty())
        {
            bail!("session {:?} has an empty display title", self.id);
        }
        if self.created_at.trim().is_empty() || self.updated_at.trim().is_empty() {
            bail!("session {:?} has an empty timestamp", self.id);
        }
        if let Some(target) = &self.target {
            target.validate(&self.id)?;
        }
        if let Some(checkpoint) = &self.checkpoint {
            checkpoint.validate()?;
        }
        Ok(())
    }
}

fn repository_source_identity(repository: &ProjectRepository) -> Option<ProjectSourceIdentity> {
    repository
        .github
        .as_deref()
        .and_then(ProjectSourceIdentity::git_remote)
        .or_else(|| {
            repository
                .local
                .as_deref()
                .map(|path| ProjectSourceIdentity::path(path, None))
        })
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct ProjectSourceIdentity {
    pub key: String,
    pub short: String,
    pub full: String,
}

impl ProjectSourceIdentity {
    /// Combine repository identities into one stable bundle identity.
    pub fn bundle(mut sources: Vec<Self>) -> Option<Self> {
        if sources.is_empty() {
            return None;
        }
        sources.sort_by(|left, right| {
            left.key
                .cmp(&right.key)
                .then_with(|| left.full.cmp(&right.full))
                .then_with(|| left.short.cmp(&right.short))
        });
        sources.dedup_by(|left, right| left.key == right.key);
        if sources.len() == 1 {
            return sources.pop();
        }
        let keys = sources
            .iter()
            .map(|source| source.key.clone())
            .collect::<Vec<_>>();
        let key = serde_json::to_string(&keys).ok()?;
        Some(Self {
            key: format!("bundle:{key}"),
            short: sources
                .iter()
                .map(|source| source.short.as_str())
                .collect::<Vec<_>>()
                .join(" + "),
            full: sources
                .iter()
                .map(|source| source.full.as_str())
                .collect::<Vec<_>>()
                .join(" + "),
        })
    }

    /// Canonicalizes a Git remote so raw checkouts group as the same project
    /// even when their worktree paths differ.
    pub fn git_remote(source: &str) -> Option<Self> {
        if let Some(normalized) = normalize_github_source(source) {
            let short = normalized
                .rsplit_once('/')
                .map_or(normalized.as_str(), |(_, repository)| repository)
                .to_owned();
            return Some(Self {
                key: format!("github:{}", normalized.to_lowercase()),
                short,
                full: normalized,
            });
        }
        let normalized = source.trim().trim_end_matches('/').trim_end_matches(".git");
        if normalized.is_empty() {
            return None;
        }
        let short = normalized
            .rsplit(['/', ':'])
            .find(|part| !part.is_empty())
            .unwrap_or(normalized)
            .to_owned();
        Some(Self {
            key: format!("git:{}", normalized.to_lowercase()),
            short,
            full: normalized.to_owned(),
        })
    }

    /// Build a local-root identity, qualified by host for remote directories.
    pub fn path(path: &Path, remote: Option<&str>) -> Self {
        let normalized = path.components().collect::<PathBuf>();
        let path_text = normalized.to_string_lossy().into_owned();
        let full = remote.map_or_else(|| path_text.clone(), |host| format!("{host}:{path_text}"));
        let key = remote.map_or_else(
            || format!("path:{path_text}"),
            |host| format!("path:{}:{path_text}", host.to_lowercase()),
        );
        Self {
            key,
            short: path_leaf(path),
            full,
        }
    }
}

fn normalize_github_source(source: &str) -> Option<String> {
    let source = source.trim();
    let path = source
        .strip_prefix("https://github.com/")
        .or_else(|| source.strip_prefix("http://github.com/"))
        .or_else(|| source.strip_prefix("git@github.com:"))
        .or_else(|| source.strip_prefix("ssh://git@github.com/"))
        .or_else(|| {
            (!source.contains("://") && !source.contains('@') && !source.contains(':'))
                .then_some(source)
        })?
        .trim_end_matches(".git");
    let mut parts = path.split('/');
    let owner = parts.next()?;
    let repository = parts.next()?;
    (!owner.is_empty() && !repository.is_empty() && parts.next().is_none())
        .then(|| format!("{owner}/{repository}"))
}

/// Last component of a path, falling back to the whole path when it has none.
fn path_leaf(path: &Path) -> String {
    path.file_name()
        .unwrap_or(path.as_os_str())
        .to_string_lossy()
        .into_owned()
}

fn created_at_seconds(timestamp: &str) -> Option<i64> {
    chrono::DateTime::parse_from_rfc3339(timestamp)
        .ok()
        .map(|timestamp| timestamp.timestamp())
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct State {
    pub version: u32,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub sessions: BTreeMap<String, SessionRecord>,
    /// Child sessions keyed by their session id. The relationship lives in
    /// controller state so every control surface sees the same session family.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub subagents: BTreeMap<String, SubagentRecord>,
    /// Recently used source directories, keyed by `local` or SSH host name.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub mount_history: BTreeMap<String, Vec<PathBuf>>,
    /// Most recently launched container size on each physical target host.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub container_sizes: BTreeMap<String, HostContainerSize>,
}

impl Default for State {
    fn default() -> Self {
        Self {
            version: STATE_VERSION,
            sessions: BTreeMap::new(),
            subagents: BTreeMap::new(),
            mount_history: BTreeMap::new(),
            container_sizes: BTreeMap::new(),
        }
    }
}

impl State {
    pub fn validate(&self) -> Result<()> {
        if self.version != STATE_VERSION {
            bail!(
                "unsupported Mjolnir state version {}; expected {STATE_VERSION}",
                self.version
            );
        }
        for (id, session) in &self.sessions {
            session.validate(id)?;
        }
        for (child_id, subagent) in &self.subagents {
            if child_id != &subagent.child_session_id {
                bail!("sub-agent key {child_id:?} does not match its child session id");
            }
            if child_id == &subagent.parent_session_id {
                bail!("sub-agent {child_id:?} cannot be its own parent");
            }
            if !self.sessions.contains_key(child_id) {
                bail!("sub-agent {child_id:?} has no child session");
            }
            if !self.sessions.contains_key(&subagent.parent_session_id) {
                bail!(
                    "sub-agent {child_id:?} has unknown parent {:?}",
                    subagent.parent_session_id
                );
            }
            if self.subagents.contains_key(&subagent.parent_session_id) {
                bail!("sub-agent {child_id:?} cannot belong to another sub-agent");
            }
            if subagent.task_name.trim().is_empty()
                || subagent.profile_id.trim().is_empty()
                || subagent.request_key.trim().is_empty()
            {
                bail!("sub-agent {child_id:?} has incomplete relationship metadata");
            }
            if subagent.working_directory.is_absolute()
                || subagent
                    .working_directory
                    .components()
                    .any(|component| component == Component::ParentDir)
            {
                bail!("sub-agent {child_id:?} has an unsafe working directory");
            }
        }
        for (host, sources) in &self.mount_history {
            if host.trim().is_empty() {
                bail!("mount history contains an empty host key");
            }
            if sources.iter().any(|source| !source.is_absolute()) {
                bail!("mount history for {host:?} contains a non-absolute source path");
            }
        }
        for (host, size) in &self.container_sizes {
            if host.trim().is_empty() {
                bail!("container size history contains an empty host key");
            }
            if size.cpus == 0 || size.memory_bytes == 0 {
                bail!("container size history for {host:?} contains a zero value");
            }
            if size.cpus > i64::MAX as u64 || size.memory_bytes > i64::MAX as u64 {
                bail!("container size history for {host:?} exceeds SQLite integer range");
            }
        }
        Ok(())
    }

    pub fn remember_mount_sources(&mut self, host: &str, mounts: &[AdditionalMount]) {
        if mounts.is_empty() {
            return;
        }
        let sources = self.mount_history.entry(host.to_owned()).or_default();
        for mount in mounts.iter().rev() {
            sources.retain(|source| source != &mount.source);
            sources.insert(0, mount.source.clone());
        }
        sources.truncate(20);
    }

    pub fn remember_container_size(&mut self, host: &str, size: HostContainerSize) {
        self.container_sizes.insert(host.to_owned(), size);
    }

    pub fn project_directories(&self, host: &str) -> &[PathBuf] {
        self.mount_history
            .get(&project_history_key(host))
            .map(Vec::as_slice)
            .unwrap_or_default()
    }

    pub fn remember_project_directory(&mut self, host: &str, directory: &Path) {
        let key = project_history_key(host);
        let directories = self.mount_history.entry(key).or_default();
        directories.retain(|existing| existing != directory);
        directories.insert(0, directory.to_path_buf());
        directories.truncate(20);
    }

    pub fn destroy_stopped_session(&mut self, session_id: &str) -> Result<SessionRecord> {
        let session = self
            .sessions
            .get(session_id)
            .with_context(|| format!("unknown session {session_id}"))?;
        if session.state.is_active() {
            bail!("refusing to destroy active session {session_id}");
        }
        Ok(self
            .sessions
            .remove(session_id)
            .expect("session checked above"))
    }

    /// Remove a session record from state regardless of its lifecycle state.
    ///
    /// Force destruction is the one caller: by the time it runs, every
    /// external artifact has been torn down or its loss accepted, so no state
    /// is refused here.
    pub fn destroy_session_force(&mut self, session_id: &str) -> Result<SessionRecord> {
        self.sessions
            .get(session_id)
            .with_context(|| format!("unknown session {session_id}"))?;
        Ok(self
            .sessions
            .remove(session_id)
            .expect("session checked above"))
    }

    /// Setup may add replacements under new names, but must not rewrite
    /// dependencies still owned by active sessions.
    pub fn validate_setup_update(&self, before: &Config, after: &Config) -> Result<()> {
        for session in self
            .sessions
            .values()
            .filter(|session| session.state.is_active())
        {
            let protected = if let Some(profile) = before.profiles.get(&session.last_profile) {
                let mut comparable = profile.clone();
                if let Some(updated) = after.profiles.get(&session.last_profile) {
                    comparable.enabled = updated.enabled;
                }
                // A mismatched harness is already broken; allow repairing it.
                profile.kind == session.harness_kind
                    && after.profiles.get(&session.last_profile) != Some(&comparable)
            } else {
                false
            };
            let bundle_changed = session.project_directory.is_none()
                && before
                    .bundles
                    .get(&session.bundle_id)
                    .is_some_and(|bundle| after.bundles.get(&session.bundle_id) != Some(bundle));
            let target_changed =
                before
                    .targets
                    .get(&session.target_template_id)
                    .is_some_and(|target| {
                        after.targets.get(&session.target_template_id) != Some(target)
                    });
            if protected || bundle_changed || target_changed {
                bail!(
                    "Setup would change configuration used by active session {:?}. Keep its profile {:?}, bundle {:?}, and target {:?}; add a separate entry for new settings, or stop the session before editing its configuration.",
                    session.id,
                    session.last_profile,
                    session.bundle_id,
                    session.target_template_id
                );
            }
        }
        Ok(())
    }

    /// Strict validation for callers that need all active references intact.
    pub fn validate_against_config(&self, config: &Config) -> Result<()> {
        self.validate()?;
        config.validate()?;
        for session in self.sessions.values() {
            session.validate_configuration(config)?;
        }
        Ok(())
    }

    pub fn load_from(path: &Path) -> Result<Self> {
        Self::load_json_from(path)
    }

    pub fn load_json_from(path: &Path) -> Result<Self> {
        if !path.exists() {
            return Ok(Self::default());
        }
        let body =
            fs::read(path).with_context(|| format!("read Mjolnir state {}", path.display()))?;
        let state: Self = serde_json::from_slice(&body)
            .with_context(|| format!("parse Mjolnir state {}", path.display()))?;
        state.validate()?;
        Ok(state)
    }

    pub fn save_to(&self, path: &Path) -> Result<()> {
        self.validate()?;
        let body = serde_json::to_vec_pretty(self).context("serialize Mjolnir state")?;
        atomic_write(path, &body)
    }
}

fn project_history_key(host: &str) -> String {
    format!("project:{host}")
}

pub fn state_path() -> PathBuf {
    data_dir().join("state.json")
}

/// Generate an opaque, filesystem-safe stable id for a new logical session.
pub fn new_session_id() -> Result<String> {
    let mut random = [0u8; 16];
    getrandom::fill(&mut random)
        .map_err(|error| anyhow::anyhow!("generate Mjolnir session id: {error}"))?;
    let mut encoded = String::with_capacity(32);
    for byte in random {
        use std::fmt::Write as _;
        write!(encoded, "{byte:02x}").expect("writing to a String cannot fail");
    }
    Ok(encoded)
}

/// Return the newest clean ACP session title from canonical worker events.
pub fn harness_session_title(events: &[SequencedEvent]) -> Option<String> {
    events.iter().rev().find_map(|event| {
        let WorkerEvent::Adapter { payload, .. } = &event.event else {
            return None;
        };
        let crate::acp::RuntimeEvent::SessionUpdate { update } =
            serde_json::from_value(payload.clone()).ok()?
        else {
            return None;
        };
        let kind = update
            .get("sessionUpdate")
            .and_then(serde_json::Value::as_str)?;
        let title = match kind {
            "session_info_update" | "session_title" => {
                update.get("title").and_then(serde_json::Value::as_str)
            }
            _ => None,
        }?;
        normalize_session_title(title)
    })
}

pub fn normalize_session_title(title: &str) -> Option<String> {
    let normalized = crate::relay::strip_hidden_prompt_context(title)
        .split_whitespace()
        .collect::<Vec<_>>()
        .join(" ");
    (!normalized.is_empty()).then_some(normalized)
}

/// Build the short-lived title shown before the harness supplies its own.
///
/// The first visible user prompt is immediately useful for identifying a
/// session, but it can be arbitrarily large. Keep this fallback bounded; a
/// later ACP session-info update remains authoritative and replaces it.
pub fn provisional_session_title(prompt: &str) -> Option<String> {
    const MAX_TITLE_CHARS: usize = 64;

    let normalized = normalize_session_title(prompt)?;
    if normalized.chars().count() <= MAX_TITLE_CHARS {
        return Some(normalized);
    }

    let mut truncated = normalized
        .chars()
        .take(MAX_TITLE_CHARS - 1)
        .collect::<String>();
    if let Some(boundary) = truncated.rfind(char::is_whitespace) {
        truncated.truncate(boundary);
    }
    truncated.push('…');
    Some(truncated)
}

pub fn short_id(id: &str) -> &str {
    id.get(..8).unwrap_or(id)
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct RecoveryCandidate {
    pub session_id: String,
    pub target_template_id: String,
    pub locator: TargetLocator,
    pub ownership: Option<crate::worker_launch::WorkerOwnership>,
}

#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct RecoveryScan {
    pub candidates: Vec<RecoveryCandidate>,
    pub warnings: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ResumeRepositorySourceReceipt {
    pub session_id: String,
    pub bundle_id: String,
    pub checkpoint_sha256: String,
    pub repositories: Vec<crate::config::ProjectRepository>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::{
        CONFIG_VERSION, ContainerTemplate, HarnessProfile, ProjectBundle, ProjectRepository,
        TargetTemplate,
    };

    fn user_item(position: u64, text: &str) -> Arc<TranscriptItem> {
        Arc::new(TranscriptItem {
            stable_id: format!("user:{position}"),
            position,
            latest_content_event_ordinal: None,
            created_at_ms: 1_000,
            last_changed_at_ms: 1_000,
            body: TranscriptBody::User {
                content: vec![serde_json::json!({"type": "text", "text": text})],
            },
        })
    }

    fn agent_item(position: u64) -> Arc<TranscriptItem> {
        Arc::new(TranscriptItem {
            stable_id: format!("agent:{position}"),
            position,
            latest_content_event_ordinal: Some(position),
            created_at_ms: 1_000,
            last_changed_at_ms: 1_000,
            body: TranscriptBody::Agent {
                chunks: vec![serde_json::json!({
                    "content": {"type": "text", "text": "working"},
                    "messageId": "answer"
                })],
                streaming: false,
            },
        })
    }

    fn snapshot(session: MaterializedSession, window: ProjectionWindow) -> ManagedSessionSnapshot {
        ManagedSessionSnapshot {
            materialized: session,
            window,
            worker_build: None,
            subagent_requests: Vec::new(),
            subagent_results: Vec::new(),
            operational: serde_json::from_value(serde_json::json!({
                "session_id": "session-1",
                "execution": "idle",
                "latest_ordinal": 0,
                "latest_digest": crate::relay::RELAY_EVENT_GENESIS_DIGEST,
                "acknowledged_through": 0,
                "acknowledged_digest": crate::relay::RELAY_EVENT_GENESIS_DIGEST,
                "recovery_floor_ordinal": 0,
                "recovery_floor_digest": crate::relay::RELAY_EVENT_GENESIS_DIGEST,
                "native_session_id": null,
                "agent_capabilities": null,
                "agent_info": null,
                "config_options": [],
                "available_commands": [],
                "config": {},
                "active_prompt": null,
                "queued_prompts": [],
                "checkpoint_barrier": null,
                "checkpoint_ready": null,
            }))
            .expect("an idle operational state"),
            latest_credential_sync_signal: None,
        }
    }

    /// A polled projection carries only the end of the transcript. The title
    /// comes from the first user message and the completed turn from the last
    /// one, so both have to survive the head being outside the window.
    #[test]
    fn a_windowed_projection_answers_the_same_title_and_turn_as_a_whole_one() {
        let mut whole = MaterializedSession::empty("session-1");
        whole.transcript = vec![
            user_item(1, "build the relay"),
            agent_item(2),
            agent_item(3),
            user_item(4, "now test it"),
            agent_item(5),
        ];
        let complete = snapshot(whole.clone(), ProjectionWindow::of(&whole));

        // The same session, loaded as a two-item window: the head is gone and
        // so is the last user message.
        let mut windowed_session = whole.clone();
        windowed_session.transcript = whole.transcript[3..].to_vec();
        let mut windowed = snapshot(windowed_session, ProjectionWindow::of(&whole));
        windowed.window.omitted_items = 3;

        assert_eq!(
            complete.resolved_title().as_deref(),
            Some("build the relay")
        );
        assert_eq!(windowed.resolved_title(), complete.resolved_title());
        assert_eq!(complete.latest_completed_turn_ordinal(), Some(4));
        assert_eq!(
            windowed.latest_completed_turn_ordinal(),
            complete.latest_completed_turn_ordinal()
        );
    }

    /// A session still working has not completed a turn, whatever its
    /// transcript says.
    #[test]
    fn a_running_session_reports_no_completed_turn() {
        let mut session = MaterializedSession::empty("session-1");
        session.transcript = vec![user_item(1, "build it")];
        session.execution = MaterializedExecutionState::Running { started_at_ms: 1 };
        let window = ProjectionWindow::of(&session);

        assert_eq!(
            snapshot(session, window).latest_completed_turn_ordinal(),
            None
        );
    }

    #[test]
    fn fast_mode_configuration_uses_its_user_facing_toggle_command() {
        assert_eq!(config_command_text("fast-mode", "on"), "/fast");
        assert_eq!(config_command_text("fast-mode", "off"), "/fast");
        assert_eq!(config_command_text("model", "sol"), "/model sol");
    }

    fn sample_state() -> State {
        let session = SessionRecord {
            create_managed_worktree: None,
            workspace_id: crate::workspace::DEFAULT_WORKSPACE_ID.to_owned(),
            archived: false,
            container_cpus: None,
            container_memory: None,
            id: "0123456789abcdef".into(),
            title: "Build Hel".into(),
            harness_kind: HarnessKind::Codex,
            last_profile: "codex-1".into(),
            bundle_id: "hel".into(),
            project_directory: None,
            managed_worktree: None,
            target_template_id: "podman".into(),
            resource_allocation: None,
            additional_mounts: vec![AdditionalMount {
                source: PathBuf::from("/home/test/cache"),
                destination: PathBuf::from("/mnt/cache"),
                read_only: false,
            }],
            state: SessionState::Running,
            target: Some(TargetLocator::LocalPodman {
                container_id: "afb67d".into(),
                workspace_storage: Default::default(),
            }),
            native_session_id: Some("native-1".into()),
            acp_session_title: Some("Build Hel".into()),
            session_title_override: None,
            created_at: "2026-08-09T12:00:00Z".into(),
            updated_at: "2026-08-09T12:01:00Z".into(),
            viewed_through_event_ordinal: 0,
            draft_input: String::new(),
            last_error: None,
            last_checkpoint_error: None,
            checkpoint: Some(CheckpointMetadata {
                archive_path: PathBuf::from("sessions/0123456789abcdef.hel.zip"),
                sha256: "a".repeat(64),
                created_at: "2026-08-09T12:01:00Z".into(),
                event_frontier: 42,
            }),
        };
        State {
            version: STATE_VERSION,
            sessions: BTreeMap::from([(session.id.clone(), session)]),
            subagents: BTreeMap::new(),
            mount_history: BTreeMap::from([(
                "local".into(),
                vec![PathBuf::from("/home/test/cache")],
            )]),
            container_sizes: BTreeMap::new(),
        }
    }

    fn sample_config() -> Config {
        Config {
            advanced: Default::default(),
            version: CONFIG_VERSION,
            sessions_side: Default::default(),
            show_stopped_sessions: false,
            newer_config_version: None,
            spinner: Default::default(),
            theme: Default::default(),
            phone: Default::default(),
            review: Default::default(),
            subagents: Default::default(),
            legacy_startup: (),
            profiles: BTreeMap::from([(
                "codex-1".into(),
                HarnessProfile {
                    enabled: true,
                    context_window_bytes: None,
                    kind: HarnessKind::Codex,
                    home: PathBuf::from("/home/test/.codex"),
                    environment: BTreeMap::new(),
                },
            )]),
            bundles: BTreeMap::from([(
                "hel".into(),
                ProjectBundle {
                    primary_repo: "hel".into(),
                    repositories: vec![ProjectRepository {
                        id: "hel".into(),
                        github: Some("BrokkAi/hel".into()),
                        local: None,
                        destination: PathBuf::from("hel"),
                        git_ref: None,
                    }],
                },
            )]),
            targets: BTreeMap::from([(
                "podman".into(),
                TargetTemplate::LocalPodman {
                    container: ContainerTemplate {
                        image: "ubuntu:24.04".into(),
                        pull_policy: Default::default(),
                        platform: None,
                        cpus: None,
                        memory: None,
                        environment: BTreeMap::new(),
                        workspace_storage: Default::default(),
                    },
                },
            )]),
        }
    }

    fn sample_session() -> SessionRecord {
        sample_state()
            .sessions
            .remove("0123456789abcdef")
            .expect("sample session")
    }

    #[test]
    fn session_records_written_before_container_overrides_still_load() {
        let session = sample_session();
        let mut json = serde_json::to_value(&session).expect("serialize session");
        let object = json.as_object_mut().expect("session object");
        assert!(object.remove("container_cpus").is_none());
        assert!(object.remove("container_memory").is_none());

        let loaded: SessionRecord = serde_json::from_value(json).expect("load older session");
        assert_eq!(loaded.container_cpus, None);
        assert_eq!(loaded.container_memory, None);
        assert_eq!(loaded, session);

        let mut edited = session.clone();
        edited.container_cpus = Some("4".into());
        edited.container_memory = Some("8g".into());
        let round_tripped: SessionRecord =
            serde_json::from_str(&serde_json::to_string(&edited).expect("serialize"))
                .expect("reload edited session");
        assert_eq!(round_tripped, edited);
    }

    #[test]
    fn container_size_history_rejects_invalid_keys_and_values() {
        let mut state = State::default();
        state.container_sizes.insert(
            String::new(),
            HostContainerSize {
                cpus: 8,
                memory_bytes: 32,
            },
        );
        assert!(
            state
                .validate()
                .unwrap_err()
                .to_string()
                .contains("empty host")
        );

        state.container_sizes = BTreeMap::from([(
            "local".into(),
            HostContainerSize {
                cpus: 0,
                memory_bytes: 32,
            },
        )]);
        assert!(state.validate().unwrap_err().to_string().contains("zero"));
    }

    #[test]
    fn project_name_prefers_a_worktree_source_then_a_project_directory_then_the_bundle() {
        let mut config = sample_config();
        config
            .bundles
            .get_mut("hel")
            .expect("bundle")
            .repositories
            .push(ProjectRepository {
                id: "docs".into(),
                github: Some("BrokkAi/docs".into()),
                local: None,
                destination: PathBuf::from("documentation"),
                git_ref: None,
            });
        let mut session = sample_session();

        assert_eq!(session.project_name(&config), "docs + hel");

        session.project_directory = Some(PathBuf::from("/home/test/Projects/raw-project"));
        assert_eq!(session.project_name(&config), "raw-project");

        session.project_directory = Some(PathBuf::from(
            "/home/test/Projects/source/.mj/worktrees/0123456789abcdef",
        ));
        session.managed_worktree = Some(ManagedWorktree {
            source_project_directory: PathBuf::from("/home/test/Projects/source"),
            source_repository: PathBuf::from("/home/test/Projects/source"),
            worktree_root: PathBuf::from(
                "/home/test/Projects/source/.mj/worktrees/0123456789abcdef",
            ),
            branch: "mj/0123456789abcdef".into(),
            target: ManagedWorktreeTarget::Local,
            base_commit: None,
        });
        assert_eq!(session.project_name(&config), "source");
    }

    #[test]
    fn bundle_project_name_uses_the_primary_github_repository_name() {
        let mut config = sample_config();
        config.bundles.insert(
            "bifrost".into(),
            ProjectBundle {
                primary_repo: "bifrost".into(),
                repositories: vec![ProjectRepository {
                    id: "bifrost".into(),
                    github: Some("BrokkAi/bifrost-dev".into()),
                    local: None,
                    destination: PathBuf::from("bifrost"),
                    git_ref: None,
                }],
            },
        );
        let mut session = sample_session();
        session.bundle_id = "bifrost".into();

        assert_eq!(session.project_name(&config), "bifrost-dev");
        assert_eq!(
            session.project_source(&config),
            ProjectSourceIdentity {
                key: "github:brokkai/bifrost-dev".into(),
                short: "bifrost-dev".into(),
                full: "BrokkAi/bifrost-dev".into(),
            }
        );
    }

    #[test]
    fn bundle_project_name_uses_a_local_source_or_bundle_id_fallback() {
        let mut config = sample_config();
        config.bundles.insert(
            "local-bundle".into(),
            ProjectBundle {
                primary_repo: "local".into(),
                repositories: vec![ProjectRepository {
                    id: "local".into(),
                    github: None,
                    local: Some(PathBuf::from("/home/test/Projects/bifrost-dev")),
                    destination: PathBuf::from("bifrost"),
                    git_ref: None,
                }],
            },
        );
        let mut session = sample_session();
        session.bundle_id = "local-bundle".into();

        assert_eq!(session.project_name(&config), "bifrost-dev");
        assert_eq!(
            session.project_source(&config),
            ProjectSourceIdentity {
                key: "path:/home/test/Projects/bifrost-dev".into(),
                short: "bifrost-dev".into(),
                full: "/home/test/Projects/bifrost-dev".into(),
            }
        );

        session.bundle_id = "missing-bundle".into();
        assert_eq!(session.project_name(&config), "missing-bundle");
        assert_eq!(
            session.project_source(&config),
            ProjectSourceIdentity {
                key: "bundle:missing-bundle".into(),
                short: "missing-bundle".into(),
                full: "missing-bundle".into(),
            }
        );

        let mut other_missing = session.clone();
        other_missing.bundle_id = "another-missing-bundle".into();
        assert_ne!(
            session.project_source(&config).key,
            other_missing.project_source(&config).key
        );
    }

    #[test]
    fn project_target_adds_the_raw_project_name_only_for_bare_targets() {
        let mut config = sample_config();
        config
            .targets
            .insert("localhost".into(), TargetTemplate::LocalBare);
        let mut session = sample_session();
        session.project_directory = Some(PathBuf::from("/mnt/optane/bifrost-fird"));

        assert_eq!(session.project_target(&config, "podman"), "podman");
        assert_eq!(
            session.project_target(&config, "localhost"),
            "localhost/bifrost-fird"
        );
        assert_eq!(
            session.project_target(&config, "retired-target"),
            "retired-target"
        );
    }

    #[test]
    fn project_source_uses_bundle_repository_and_ignores_managed_worktree_destinations() {
        let config = sample_config();
        let mut session = sample_session();
        let source = session.project_source(&config);
        assert_eq!(source.key, "github:brokkai/hel");
        assert_eq!(source.short, "hel");
        assert_eq!(source.full, "BrokkAi/hel");
        assert_eq!(
            ProjectSourceIdentity::git_remote("git@github.com:BrokkAi/bifrost-dev.git"),
            ProjectSourceIdentity::git_remote("https://github.com/BrokkAi/bifrost-dev.git")
        );
        assert_ne!(
            ProjectSourceIdentity::git_remote("BrokkAi/bifrost-dev"),
            ProjectSourceIdentity::git_remote("OtherOrg/bifrost-dev")
        );

        session.project_directory = Some(PathBuf::from(
            "/home/test/Projects/source/.mj/worktrees/0123456789abcdef",
        ));
        session.managed_worktree = Some(ManagedWorktree {
            source_project_directory: PathBuf::from("/home/test/Projects/source/crate"),
            source_repository: PathBuf::from("/home/test/Projects/source"),
            worktree_root: PathBuf::from(
                "/home/test/Projects/source/.mj/worktrees/0123456789abcdef",
            ),
            branch: "mj/0123456789abcdef".into(),
            target: ManagedWorktreeTarget::Local,
            base_commit: None,
        });
        let source = session.project_source(&config);
        assert_eq!(source.short, "source");
        assert_eq!(source.full, "/home/test/Projects/source");
        assert!(!source.full.contains(".mj/worktrees"));
    }

    #[test]
    fn single_repository_bundle_uses_the_standalone_repository_identity() {
        let mut config = sample_config();
        let shared_bundle = config.bundles["hel"].clone();
        config.bundles.insert("other".into(), shared_bundle);

        let first = sample_session();
        let mut second = first.clone();
        second.bundle_id = "other".into();

        assert_eq!(
            config.bundles["hel"].primary_repo,
            config.bundles["other"].primary_repo
        );
        let first_source = first.project_source(&config);
        let second_source = second.project_source(&config);
        let standalone = ProjectSourceIdentity::git_remote("BrokkAi/hel").unwrap();
        assert_eq!(first_source, standalone);
        assert_eq!(second_source, standalone);
    }

    #[test]
    fn multi_repository_bundles_include_all_repositories_in_sorted_identity_order() {
        let mut config = sample_config();
        let primary = config.bundles["hel"].repositories[0].clone();
        let secondary = ProjectRepository {
            id: "docs".into(),
            github: Some("BrokkAi/docs".into()),
            local: None,
            destination: PathBuf::from("docs"),
            git_ref: None,
        };
        config.bundles.insert(
            "with-docs".into(),
            ProjectBundle {
                primary_repo: primary.id.clone(),
                repositories: vec![primary.clone(), secondary.clone()],
            },
        );
        let mut session = sample_session();
        session.bundle_id = "with-docs".into();

        assert_eq!(session.project_name(&config), "docs + hel");
        assert_eq!(
            session.project_source(&config),
            ProjectSourceIdentity {
                key: "bundle:[\"github:brokkai/docs\",\"github:brokkai/hel\"]".into(),
                short: "docs + hel".into(),
                full: "BrokkAi/docs + BrokkAi/hel".into(),
            }
        );

        let mut other_secondary = secondary;
        other_secondary.github = Some("OtherOrg/docs".into());
        config.bundles.insert(
            "with-other-docs".into(),
            ProjectBundle {
                primary_repo: primary.id.clone(),
                repositories: vec![primary, other_secondary],
            },
        );
        let mut other_session = session.clone();
        other_session.bundle_id = "with-other-docs".into();
        assert_ne!(
            session.project_source(&config).key,
            other_session.project_source(&config).key
        );
    }

    #[test]
    fn multi_repository_bundle_identity_ignores_repository_order_and_primary_selection() {
        let mut config = sample_config();
        let primary = config.bundles["hel"].repositories[0].clone();
        let secondary = ProjectRepository {
            id: "docs".into(),
            github: Some("BrokkAi/docs".into()),
            local: None,
            destination: PathBuf::from("docs"),
            git_ref: None,
        };
        config.bundles.insert(
            "first-order".into(),
            ProjectBundle {
                primary_repo: primary.id.clone(),
                repositories: vec![primary.clone(), secondary.clone()],
            },
        );
        config.bundles.insert(
            "second-order".into(),
            ProjectBundle {
                primary_repo: secondary.id.clone(),
                repositories: vec![secondary, primary],
            },
        );

        let mut first = sample_session();
        first.bundle_id = "first-order".into();
        let mut second = first.clone();
        second.bundle_id = "second-order".into();
        assert_eq!(
            first.project_source(&config),
            second.project_source(&config)
        );
    }

    #[test]
    fn duplicate_repository_sources_collapse_to_the_single_repository_identity() {
        let mut config = sample_config();
        let primary = config.bundles["hel"].repositories[0].clone();
        let duplicate = ProjectRepository {
            id: "hel-copy".into(),
            github: primary.github.clone(),
            local: None,
            destination: PathBuf::from("hel-copy"),
            git_ref: None,
        };
        config.bundles.insert(
            "duplicate".into(),
            ProjectBundle {
                primary_repo: primary.id.clone(),
                repositories: vec![primary, duplicate],
            },
        );
        let mut session = sample_session();
        session.bundle_id = "duplicate".into();

        let source = session.project_source(&config);
        assert_eq!(
            source,
            ProjectSourceIdentity::git_remote("BrokkAi/hel").unwrap()
        );
    }

    #[test]
    fn unresolved_bundle_repository_uses_the_bundle_fallback() {
        let mut config = sample_config();
        config.bundles.insert(
            "incomplete".into(),
            ProjectBundle {
                primary_repo: "broken".into(),
                repositories: vec![ProjectRepository {
                    id: "broken".into(),
                    github: None,
                    local: None,
                    destination: PathBuf::from("broken"),
                    git_ref: None,
                }],
            },
        );
        let mut session = sample_session();
        session.bundle_id = "incomplete".into();

        assert_eq!(session.project_name(&config), "incomplete");
        assert_eq!(
            session.project_source(&config),
            ProjectSourceIdentity {
                key: "bundle:incomplete".into(),
                short: "incomplete".into(),
                full: "incomplete".into(),
            }
        );
    }

    #[test]
    fn sessions_order_by_creation_time_and_fall_back_to_the_id() {
        let older = sample_session();
        let mut newer = sample_session();
        newer.id = "0000000000000001".into();
        newer.created_at = "2026-08-09T13:00:00Z".into();
        let mut unparsable = sample_session();
        unparsable.id = "0000000000000002".into();
        unparsable.created_at = "not a timestamp".into();
        let mut same_time = sample_session();
        same_time.id = "zzzzzzzzzzzzzzzz".into();

        let mut sessions = [&unparsable, &newer, &same_time, &older];
        sessions.sort_by(|left, right| left.compare_by_creation(right));

        assert_eq!(
            sessions
                .iter()
                .map(|session| &session.id)
                .collect::<Vec<_>>(),
            [&older.id, &same_time.id, &newer.id, &unparsable.id]
        );
    }

    #[test]
    fn retired_checkpoint_and_detach_cursor_names_are_rejected() {
        let session_id = "0123456789abcdef";

        let mut old_checkpoint = serde_json::to_value(sample_state()).unwrap();
        let checkpoint = old_checkpoint["sessions"][session_id]["checkpoint"]
            .as_object_mut()
            .unwrap();
        let frontier = checkpoint.remove("event_frontier").unwrap();
        checkpoint.insert("event_sequence".into(), frontier);
        assert!(serde_json::from_value::<State>(old_checkpoint).is_err());

        let mut old_detach_cursor = serde_json::to_value(sample_state()).unwrap();
        let session = old_detach_cursor["sessions"][session_id]
            .as_object_mut()
            .unwrap();
        let ordinal = session.remove("viewed_through_event_ordinal").unwrap();
        session.insert("last_viewed_event_sequence".into(), ordinal);
        assert!(serde_json::from_value::<State>(old_detach_cursor).is_err());
    }

    #[test]
    fn detached_cursor_field_loads_as_the_viewed_cursor() {
        let session_id = "0123456789abcdef";
        let mut legacy = serde_json::to_value(sample_state()).unwrap();
        let session = legacy["sessions"][session_id].as_object_mut().unwrap();
        let ordinal = session.remove("viewed_through_event_ordinal").unwrap();
        session.insert("detached_after_event_ordinal".into(), ordinal);

        let loaded: State = serde_json::from_value(legacy).unwrap();
        assert_eq!(
            loaded.sessions[session_id].viewed_through_event_ordinal,
            sample_state().sessions[session_id].viewed_through_event_ordinal
        );
    }

    #[test]
    fn state_written_before_drafts_loads_with_an_empty_draft() {
        let session_id = "0123456789abcdef";
        let mut without_draft = serde_json::to_value(sample_state()).unwrap();
        let session = without_draft["sessions"][session_id]
            .as_object_mut()
            .unwrap();
        session.remove("draft_input");

        let state = serde_json::from_value::<State>(without_draft).unwrap();
        assert_eq!(state.sessions[session_id].draft_input, "");
    }

    #[test]
    fn json_state_round_trip_is_atomic() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("nested/state.json");
        let state = sample_state();
        state.save_to(&path).unwrap();
        assert_eq!(State::load_from(&path).unwrap(), state);
        assert!(
            fs::read_dir(directory.path().join("nested"))
                .unwrap()
                .all(|entry| {
                    !entry
                        .unwrap()
                        .file_name()
                        .to_string_lossy()
                        .ends_with(".tmp")
                })
        );
    }

    #[test]
    fn mount_history_keeps_unique_recent_sources_per_host() {
        let mut state = State::default();
        state.remember_mount_sources(
            "builder.example.test",
            &[
                AdditionalMount {
                    source: "/srv/first".into(),
                    destination: "/mnt/first".into(),
                    read_only: false,
                },
                AdditionalMount {
                    source: "/srv/second".into(),
                    destination: "/mnt/second".into(),
                    read_only: false,
                },
            ],
        );
        state.remember_mount_sources(
            "builder.example.test",
            &[AdditionalMount {
                source: "/srv/first".into(),
                destination: "/mnt/again".into(),
                read_only: false,
            }],
        );

        assert_eq!(
            state.mount_history["builder.example.test"],
            vec![PathBuf::from("/srv/first"), PathBuf::from("/srv/second")]
        );
    }

    #[test]
    fn materialized_activity_watermark_does_not_regress_when_detail_is_removed() {
        let mut materialized = MaterializedSession::empty("session-1");
        assert_eq!(materialized.last_activity_at_ms(), None);

        materialized.execution = MaterializedExecutionState::Running { started_at_ms: 300 };
        materialized.transcript.push(Arc::new(TranscriptItem {
            stable_id: "system:1".into(),
            position: 1,
            latest_content_event_ordinal: None,
            created_at_ms: 350,
            last_changed_at_ms: 400,
            body: TranscriptBody::System {
                text: "working".into(),
            },
        }));
        materialized.queued_prompts.push(MaterializedQueuedPrompt {
            accepted_ordinal: None,
            command_id: "prompt-2".into(),
            kind: QueuedCommandKind::Prompt,
            content: Vec::new(),
            queued_at_ms: 500,
        });
        materialized.last_activity_at_ms = Some(500);
        assert_eq!(materialized.last_activity_at_ms(), Some(500));

        materialized.queued_prompts.clear();
        assert_eq!(materialized.last_activity_at_ms(), Some(500));
        materialized.transcript.clear();
        assert_eq!(materialized.last_activity_at_ms(), Some(500));
        materialized.execution = MaterializedExecutionState::Idle;
        assert_eq!(materialized.last_activity_at_ms(), Some(500));
    }

    /// Shared transcript items must stay plain JSON on the wire: sharing is a
    /// controller memory concern, not part of the serialized shape.
    #[test]
    fn shared_transcript_items_serialize_as_plain_items() {
        let mut materialized = MaterializedSession::empty("session-1");
        materialized.applied_event_ordinal = 1;
        materialized.applied_event_digest = "a".repeat(64);
        let item = Arc::new(TranscriptItem {
            stable_id: "system:1".into(),
            position: 1,
            latest_content_event_ordinal: None,
            created_at_ms: 10,
            last_changed_at_ms: 10,
            body: TranscriptBody::System {
                text: "started".into(),
            },
        });
        // The same item twice would be deduplicated by serde's pointer-aware
        // encodings; stable ids keep it a legal transcript.
        materialized.transcript.push(Arc::clone(&item));
        let mut second = TranscriptItem::clone(&item);
        second.stable_id = "system:2".into();
        materialized.transcript.push(Arc::new(second));
        materialized.validate().unwrap();

        let encoded = serde_json::to_value(&materialized).unwrap();
        assert_eq!(encoded["transcript"][0]["stable_id"], "system:1");
        assert_eq!(encoded["transcript"][0]["body"]["kind"], "system");
        assert_eq!(encoded["transcript"][0]["body"]["text"], "started");
        assert_eq!(encoded["transcript"][1]["stable_id"], "system:2");

        let restored: MaterializedSession = serde_json::from_value(encoded).unwrap();
        assert_eq!(restored, materialized);
    }

    #[test]
    fn materialized_event_frontier_requires_the_matching_digest_kind() {
        let mut materialized = MaterializedSession::empty("session-1");
        materialized.validate().unwrap();

        materialized.applied_event_ordinal = 1;
        assert!(
            materialized
                .validate()
                .unwrap_err()
                .to_string()
                .contains("inconsistent ordinal")
        );

        materialized.applied_event_digest = "A".repeat(64);
        assert!(
            materialized
                .validate()
                .unwrap_err()
                .to_string()
                .contains("lowercase SHA-256")
        );
    }

    #[test]
    fn project_directory_history_is_recent_and_isolated_per_remote_host() {
        let mut state = State::default();
        state.remember_project_directory("builder-a", Path::new("/srv/one"));
        state.remember_project_directory("builder-a", Path::new("/srv/two"));
        state.remember_project_directory("builder-a", Path::new("/srv/one"));
        state.remember_project_directory("builder-b", Path::new("/work/other"));

        assert_eq!(
            state.project_directories("builder-a"),
            [PathBuf::from("/srv/one"), PathBuf::from("/srv/two")]
        );
        assert_eq!(
            state.project_directories("builder-b"),
            [PathBuf::from("/work/other")]
        );
    }

    #[test]
    fn setup_protects_active_dependencies_but_allows_additions_repairs_and_defaults() {
        let state = sample_state();
        let before = sample_config();
        let session = state.sessions.values().next().unwrap();
        for section in ["profile", "bundle", "target"] {
            let mut after = before.clone();
            match section {
                "profile" => {
                    after.profiles.remove(&session.last_profile);
                }
                "bundle" => {
                    after.bundles.remove(&session.bundle_id);
                }
                _ => {
                    after.targets.remove(&session.target_template_id);
                }
            }
            assert!(
                state
                    .validate_setup_update(&before, &after)
                    .unwrap_err()
                    .to_string()
                    .contains("active session")
            );
            // Restoring a removed entry is always permitted.
            state.validate_setup_update(&after, &before).unwrap();
        }
        let mut after = before.clone();
        after.profiles.get_mut(&session.last_profile).unwrap().home = PathBuf::from("/new/home");
        assert!(state.validate_setup_update(&before, &after).is_err());
        let mut after = before.clone();
        after
            .profiles
            .get_mut(&session.last_profile)
            .unwrap()
            .enabled = false;
        after.advanced.show_stopped_sessions = !before.advanced.show_stopped_sessions;
        after.targets.insert(
            "alternative".into(),
            crate::config::TargetTemplate::LocalBare,
        );
        state.validate_setup_update(&before, &after).unwrap();
        let mut stopped = state.clone();
        stopped.sessions.values_mut().next().unwrap().state = SessionState::Stopped;
        stopped
            .validate_setup_update(&before, &Config::default())
            .unwrap();
    }

    #[test]
    fn configuration_repair_reports_all_missing_entries_and_clears_after_restoration() {
        let state = sample_state();
        let session = state.sessions.values().next().unwrap();
        let mut config = sample_config();
        config.profiles.clear();
        config.bundles.clear();
        config.targets.clear();
        let issue = session.configuration_issue(&config).unwrap();
        assert!(issue.contains("missing profile"));
        assert!(issue.contains("missing bundle"));
        assert!(issue.contains("missing target template"));
        assert!(issue.contains("config.toml"));
        assert!(session.configuration_issue(&sample_config()).is_none());
        let mut raw = session.clone();
        raw.project_directory = Some(PathBuf::from("/project"));
        let mut config = sample_config();
        config.bundles.clear();
        assert!(raw.configuration_issue(&config).is_none());
        let mut stopped = session.clone();
        stopped.state = SessionState::Stopped;
        assert!(stopped.configuration_issue(&Config::default()).is_none());
    }

    #[test]
    fn active_state_validates_references_and_harness_kind() {
        let state = sample_state();
        state.validate_against_config(&sample_config()).unwrap();

        let mut config = sample_config();
        config.profiles.get_mut("codex-1").unwrap().kind = HarnessKind::Claude;
        assert!(
            state
                .validate_against_config(&config)
                .unwrap_err()
                .to_string()
                .contains("expects Codex")
        );
    }

    /// Records written before the verb was renamed say "archived". They must
    /// still load, and they must be written back with the new name.
    #[test]
    fn the_stopped_state_reads_the_retired_archived_name_and_writes_the_new_one() {
        assert_eq!(
            serde_json::from_str::<SessionState>("\"archived\"").unwrap(),
            SessionState::Stopped
        );
        assert_eq!(
            serde_json::from_str::<SessionState>("\"stopped\"").unwrap(),
            SessionState::Stopped
        );
        assert_eq!(
            serde_json::to_string(&SessionState::Stopped).unwrap(),
            "\"stopped\""
        );
        assert!(!SessionState::Stopped.is_active());
    }

    /// The archived flag is a later addition, so records written without it
    /// load as visible and stay out of the serialized form until it is set.
    #[test]
    fn the_archived_flag_defaults_off_and_is_omitted_when_it_is_off() {
        let mut state = sample_state();
        let session = state.sessions.values_mut().next().unwrap();
        assert!(!session.archived);
        let json = serde_json::to_string(&*session).unwrap();
        assert!(!json.contains("archived"), "{json}");

        session.archived = true;
        let json = serde_json::to_string(&*session).unwrap();
        assert!(json.contains("\"archived\":true"), "{json}");
        assert!(
            serde_json::from_str::<SessionRecord>(&json)
                .unwrap()
                .archived
        );
    }

    #[test]
    fn stopped_session_does_not_pin_renamed_config_entries() {
        let mut state = sample_state();
        state.sessions.values_mut().next().unwrap().state = SessionState::Stopped;
        state.validate_against_config(&Config::default()).unwrap();
    }

    #[test]
    fn only_inactive_sessions_can_be_removed_from_the_archive() {
        let mut state = sample_state();
        assert!(
            state
                .destroy_stopped_session("0123456789abcdef")
                .unwrap_err()
                .to_string()
                .contains("active session")
        );
        assert!(state.sessions.contains_key("0123456789abcdef"));

        state.sessions.values_mut().next().unwrap().state = SessionState::Stopped;
        let removed = state.destroy_stopped_session("0123456789abcdef").unwrap();
        assert_eq!(removed.id, "0123456789abcdef");
        assert!(state.sessions.is_empty());
    }

    #[test]
    fn force_removal_permits_an_active_session() {
        let mut state = sample_state();
        let removed = state.destroy_session_force("0123456789abcdef").unwrap();
        assert_eq!(removed.id, "0123456789abcdef");
        assert!(state.sessions.is_empty());
        assert!(
            state
                .destroy_session_force("0123456789abcdef")
                .unwrap_err()
                .to_string()
                .contains("unknown session")
        );
    }

    #[test]
    fn harness_title_prefers_the_newest_session_info_update() {
        let events = vec![
            SequencedEvent {
                seq: 1,
                recorded_at_ms: None,
                request_id: None,
                event: WorkerEvent::Adapter {
                    kind: "session_update".into(),
                    payload: serde_json::json!({
                        "type": "session_update",
                        "update": {
                            "sessionUpdate": "session_info_update",
                            "title": "First title"
                        }
                    }),
                },
            },
            SequencedEvent {
                seq: 2,
                recorded_at_ms: None,
                request_id: None,
                event: WorkerEvent::Adapter {
                    kind: "session_update".into(),
                    payload: serde_json::json!({
                        "type": "session_update",
                        "update": {
                            "sessionUpdate": "session_summary",
                            "summary": "  Build   the dashboard  "
                        }
                    }),
                },
            },
        ];

        assert_eq!(
            harness_session_title(&events).as_deref(),
            Some("First title")
        );
    }

    #[test]
    fn extension_session_title_is_cleaned_without_losing_available_text() {
        let first_prompt = format!("{}overflow", "word ".repeat(20));
        let expected = first_prompt.trim().to_string();
        let events = vec![
            SequencedEvent {
                seq: 1,
                recorded_at_ms: None,
                request_id: Some("prompt-1".into()),
                event: WorkerEvent::PromptAccepted {
                    request_id: "prompt-1".into(),
                    text: format!("  {first_prompt}\n"),
                    attachments: vec![],
                },
            },
            SequencedEvent {
                seq: 2,
                recorded_at_ms: None,
                request_id: None,
                event: WorkerEvent::Adapter {
                    kind: "session_update".into(),
                    payload: serde_json::json!({
                        "type": "session_update",
                        "update": {
                            "sessionUpdate": "session_title",
                            "title": first_prompt
                        }
                    }),
                },
            },
        ];

        assert_eq!(
            harness_session_title(&events).as_deref(),
            Some(expected.as_str())
        );
    }

    #[test]
    fn first_prompt_is_not_used_as_an_acp_session_title() {
        let events = vec![SequencedEvent {
            seq: 1,
            recorded_at_ms: None,
            request_id: Some("prompt-1".into()),
            event: WorkerEvent::PromptAccepted {
                request_id: "prompt-1".into(),
                text: "Do not use me as a title".into(),
                attachments: vec![],
            },
        }];

        assert_eq!(harness_session_title(&events), None);
    }

    #[test]
    fn provisional_title_is_cleaned_and_bounded() {
        assert_eq!(
            provisional_session_title(concat!(
                "<mj-project-memory>private</mj-project-memory> ",
                "  fix the flaky\nresume test  "
            ))
            .as_deref(),
            Some("fix the flaky resume test")
        );

        let prompt = format!("{}overflow", "word ".repeat(20));
        assert_eq!(
            provisional_session_title(&prompt).as_deref(),
            Some(format!("{}word…", "word ".repeat(11)).as_str())
        );
    }

    #[test]
    fn harness_title_elides_hidden_context_instead_of_naming_the_session_from_it() {
        let titled = |title: &str| SequencedEvent {
            seq: 1,
            recorded_at_ms: None,
            request_id: None,
            event: WorkerEvent::Adapter {
                kind: "session_update".into(),
                payload: serde_json::json!({
                    "type": "session_update",
                    "update": {
                        "sessionUpdate": "session_title",
                        "title": title
                    }
                }),
            },
        };

        assert_eq!(
            harness_session_title(&[titled(concat!(
                "<mj-project-memory>private</mj-project-memory> ",
                "Visible session name"
            ))])
            .as_deref(),
            Some("Visible session name")
        );
        assert_eq!(
            harness_session_title(&[titled("<mj-project-memory>truncated")]),
            None
        );
    }

    #[test]
    fn harness_titles_are_normalized_to_one_complete_line() {
        let events = vec![SequencedEvent {
            seq: 1,
            recorded_at_ms: None,
            request_id: None,
            event: WorkerEvent::Adapter {
                kind: "session_update".into(),
                payload: serde_json::json!({
                    "type": "session_update",
                    "update": {
                        "sessionUpdate": "session_title",
                        "title": "first\nsecond\tthird fourth fifth sixth seventh eighth ninth tenth eleventh twelfth thirteenth"
                    }
                }),
            },
        }];

        assert_eq!(
            harness_session_title(&events).as_deref(),
            Some(
                "first second third fourth fifth sixth seventh eighth ninth tenth eleventh twelfth thirteenth"
            )
        );
    }

    #[test]
    fn locator_rejects_parent_traversal() {
        let mut state = sample_state();
        state.sessions.values_mut().next().unwrap().target = Some(TargetLocator::SshBare {
            host: "builder".into(),
            workspace: PathBuf::from("~/hel/../other"),
            worker_id: None,
        });
        assert!(
            state
                .validate()
                .unwrap_err()
                .to_string()
                .contains("safe path ending")
        );
    }

    #[test]
    fn generated_session_ids_are_valid_and_distinct() {
        let first = new_session_id().unwrap();
        let second = new_session_id().unwrap();
        validate_id("session", &first).unwrap();
        assert_eq!(first.len(), 32);
        assert_ne!(first, second);
    }
}