pub(crate) mod harness;
pub(crate) mod opencode_runtime;
pub mod types;
pub mod usage;

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::Arc;

use tokio::sync::{broadcast, mpsc, Mutex};

use crate::engine::agent::AgentConfig;
use crate::engine::config::load_config_or_default;
use crate::engine::launch::{prepare_launch_prompt, LaunchPromptInput};
use crate::engine::prompt::ContextBreakdown;
use crate::engine::prompt::{write_prompt_log, RelatedRepoContext, Surface};
use crate::engine::structured_reply::ClientContext;
use crate::lfd::id::LfdId;
use crate::lfd::providers::{lookup_cost_rates, CostRates};
use crate::lfd::scheduler::Scheduler;
use crate::lfd::sessions::harness::{is_terminal_harness_error, CreateHarnessFn, Harness};
use crate::lfd::sessions::types::{
    ContextSnapshot, CreateSessionParams, PersistedSessionEvent, Session, SessionConfig,
    SessionEvent, SessionItem, SessionStatus,
};
use crate::lfd::store::{SharedStore, Store, StoreError};
use crate::lfd::types::RepoId;

const LIVE_EVENT_BUFFER: usize = 256;

#[derive(Debug, thiserror::Error)]
pub enum SessionManagerError {
    #[error("store error: {0}")]
    Store(#[from] StoreError),
    #[error("session not found")]
    NotFound,
    #[error("invalid session state: expected {expected}, got {actual:?}")]
    InvalidState {
        expected: &'static str,
        actual: SessionStatus,
    },
    #[error("unsupported harness: {0}")]
    UnsupportedHarness(String),
    #[error("harness not implemented yet: {0}")]
    HarnessNotImplemented(String),
    #[error("wave run already has an active session: {0}")]
    WaveRunSessionConflict(String),
    #[error("invalid session config: {0}")]
    InvalidConfig(String),
    #[error("invalid repo_root: {0}")]
    InvalidRepoRoot(String),
    #[error("turn already in progress")]
    TurnAlreadyInProgress,
    #[error("input not supported for this harness: {0}")]
    InputNotSupported(String),
    #[error("harness error: {0}")]
    Harness(String),
}

/// Resolve related repos (parents + children) for a session repo.
///
/// Queries the store for edges, validates paths exist on disk,
/// and deduplicates by path.
async fn resolve_related_repos(
    store: &Store,
    repo_id: &RepoId,
) -> Result<Vec<RelatedRepoContext>, StoreError> {
    let mut seen_paths = std::collections::HashSet::new();
    let mut related = Vec::new();

    let parents = store.parents(repo_id).await?;
    let children = store.children(repo_id).await?;

    for repo in parents.into_iter().chain(children) {
        let path = PathBuf::from(&repo.path);
        if !path.is_dir() {
            tracing::warn!(
                repo_id = %repo.repo_id,
                path = %repo.path,
                "related repo missing from disk, skipping"
            );
            continue;
        }
        if seen_paths.insert(repo.path.clone()) {
            related.push(RelatedRepoContext {
                repo_id: repo.repo_id,
                path,
            });
        }
    }

    Ok(related)
}

struct SessionRuntime {
    harness_name: String,
    harness: Mutex<Box<dyn Harness>>,
    events_tx: broadcast::Sender<PersistedSessionEvent>,
    next_seq: AtomicI64,
    seeded_user_prompt: Option<String>,
}

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

struct SessionManagerInner {
    store: SharedStore,
    create_harness: CreateHarnessFn,
    scheduler: Option<Arc<Scheduler>>,
    runtimes: Mutex<HashMap<LfdId, Arc<SessionRuntime>>>,
}

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

#[derive(Clone, Debug)]
pub struct SessionManager {
    inner: Arc<SessionManagerInner>,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SessionStartupRecovery {
    pub sessions_failed: u32,
    pub opencode_servers_reaped: u32,
    pub reap_errors: u32,
}

impl SessionManager {
    pub fn new(store: SharedStore) -> Self {
        Self::with_create_harness_and_scheduler(store, harness::default_create_harness, None)
    }

    pub fn new_with_scheduler(store: SharedStore, scheduler: Arc<Scheduler>) -> Self {
        Self::with_create_harness_and_scheduler(
            store,
            harness::default_create_harness,
            Some(scheduler),
        )
    }

    #[cfg(test)]
    fn with_create_harness(store: SharedStore, create_harness: CreateHarnessFn) -> Self {
        Self::with_create_harness_and_scheduler(store, create_harness, None)
    }

    fn with_create_harness_and_scheduler(
        store: SharedStore,
        create_harness: CreateHarnessFn,
        scheduler: Option<Arc<Scheduler>>,
    ) -> Self {
        Self {
            inner: Arc::new(SessionManagerInner {
                store,
                create_harness,
                scheduler,
                runtimes: Mutex::new(HashMap::new()),
            }),
        }
    }

    pub async fn create_session(
        &self,
        params: CreateSessionParams,
    ) -> Result<Session, SessionManagerError> {
        let harness_name = resolve_harness(&params.harness)?;
        let (session_config, prepared_prompt, breakdown) =
            self.prepare_session_prompt(params.config).await?;

        if let Some(wave_run_id) = params.wave_run_id.as_deref() {
            if self
                .inner
                .store
                .get_active_session_for_wave_run(wave_run_id)
                .await?
                .is_some()
            {
                return Err(SessionManagerError::WaveRunSessionConflict(
                    wave_run_id.to_string(),
                ));
            }
        }

        let session = Session {
            id: LfdId::new(),
            harness: harness_name.clone(),
            status: SessionStatus::Starting,
            wave_run_id: params.wave_run_id,
            provider_session_id: None,
            config: session_config,
            created_at: time::OffsetDateTime::now_utc(),
            ended_at: None,
        };
        let (harness_events_tx, harness_events_rx) = mpsc::unbounded_channel();
        let harness = (self.inner.create_harness)(&harness_name, harness_events_tx)
            .map_err(|err| SessionManagerError::Harness(err.to_string()))?;
        self.inner.store.create_session(&session).await?;

        let (events_tx, _) = broadcast::channel(LIVE_EVENT_BUFFER);
        let seeded_user_prompt = normalized_seeded_user_prompt(&prepared_prompt.task_prompt);
        let runtime = Arc::new(SessionRuntime {
            harness_name,
            harness: Mutex::new(harness),
            events_tx,
            next_seq: AtomicI64::new(0),
            seeded_user_prompt,
        });

        {
            let mut runtimes = self.inner.runtimes.lock().await;
            runtimes.insert(session.id.clone(), runtime.clone());
        }

        self.append_runtime_event(
            &session.id,
            &runtime,
            SessionEvent::StatusChanged {
                status: SessionStatus::Starting,
            },
        )
        .await?;
        self.append_runtime_event(
            &session.id,
            &runtime,
            SessionEvent::ContextSnapshot {
                snapshot: ContextSnapshot::from(&breakdown),
            },
        )
        .await?;

        self.register_wave_session(session.wave_run_id.as_deref())
            .await;
        self.spawn_harness_event_bridge(session.id.clone(), runtime.clone(), harness_events_rx);
        let auto_start = session.wave_run_id.is_some();
        self.spawn_harness_startup(
            session.id.clone(),
            runtime.clone(),
            prepared_prompt,
            auto_start,
        );

        Ok(session)
    }

    async fn prepare_session_prompt(
        &self,
        mut config: SessionConfig,
    ) -> Result<(SessionConfig, AgentConfig, ContextBreakdown), SessionManagerError> {
        let repo_root = validate_repo_root(&config.repo_root)?;
        let step = config.step.trim().to_string();
        if step.is_empty() {
            return Err(SessionManagerError::InvalidConfig(
                "step is required".to_string(),
            ));
        }

        let cwd = resolve_cwd(&repo_root, config.cwd.as_deref())?;
        config.step = step.clone();
        config.repo_root = repo_root.to_string_lossy().to_string();
        config.cwd = cwd.as_ref().map(|path| path.to_string_lossy().to_string());
        // Sessions are primarily a Concerto feature; default to macOS surface.
        let surface = config.surface.unwrap_or(Surface::ConcertoMac);

        // Resolve related repos from the edge graph.
        let repo_root_str = repo_root.to_string_lossy().to_string();
        let related_repos = match self.inner.store.get_repo(&repo_root_str).await? {
            Some(repo) => resolve_related_repos(&self.inner.store, &repo.repo_id).await?,
            None => Vec::new(),
        };

        let file_config = load_config_or_default(Some(&repo_root));
        let prepared = prepare_launch_prompt(
            &file_config,
            LaunchPromptInput {
                repo_root: repo_root.to_path_buf(),
                step: Some(step.clone()),
                resolved_step: None,
                surface,
                directions: config.directions.clone(),
                area: config.area.clone(),
                wave: config.wave.clone(),
                message: config.message.clone(),
                agent: config.agent.clone(),
                cwd,
                max_turns: config.max_turns,
                yolo_mode: config.yolo_mode,
                include_config_directions: false,
                include_config_area: true,
                source_overrides: Default::default(),
                summary: None,
                client_context: ClientContext {
                    has_ui: config.client_has_ui.unwrap_or(false),
                    compact: config.client_compact.unwrap_or(false),
                },
                related_repos,
            },
        )
        .map_err(|err| SessionManagerError::InvalidConfig(err.to_string()))?;

        let _ = write_prompt_log(&repo_root, &prepared.prompt, &step, None);
        Ok((config, prepared.config, prepared.breakdown))
    }

    pub async fn get_session(&self, session_id: &LfdId) -> Result<Session, SessionManagerError> {
        self.inner
            .store
            .get_session(session_id)
            .await?
            .ok_or(SessionManagerError::NotFound)
    }

    pub async fn send_input(
        &self,
        session_id: &LfdId,
        content: &str,
    ) -> Result<(), SessionManagerError> {
        let session = self.get_session(session_id).await?;
        if session.status != SessionStatus::Active {
            return Err(SessionManagerError::InvalidState {
                expected: "active",
                actual: session.status,
            });
        }
        if !harness::HarnessKind::parse(&session.harness)
            .map(|kind| kind.input_supported())
            .unwrap_or(false)
        {
            return Err(SessionManagerError::InputNotSupported(session.harness));
        }

        let runtime = self
            .runtime(session_id)
            .await
            .ok_or(SessionManagerError::InvalidState {
                expected: "runtime available",
                actual: session.status,
            })?;

        let send_result = {
            let mut harness = runtime.harness.lock().await;
            harness.send_input(content).await
        };

        if let Err(err) = send_result {
            if harness::is_turn_in_progress(&err) {
                return Err(SessionManagerError::TurnAlreadyInProgress);
            }

            self.mark_session_failed(session_id, &runtime, "send_input_failed", err.to_string())
                .await;
            return Err(SessionManagerError::Harness(err.to_string()));
        }

        Ok(())
    }

    pub async fn stop_session(&self, session_id: &LfdId) -> Result<Session, SessionManagerError> {
        let session = self.get_session(session_id).await?;
        if session.status.is_terminal() {
            return Ok(session);
        }

        if session.status == SessionStatus::Starting {
            let runtime = self.runtime(session_id).await;
            self.set_status(
                session_id,
                SessionStatus::Failed,
                Some(time::OffsetDateTime::now_utc().unix_timestamp()),
                runtime.clone(),
            )
            .await?;
            return self.get_session(session_id).await;
        }

        let runtime = self.runtime(session_id).await;
        if session.status != SessionStatus::Ending {
            self.set_status(session_id, SessionStatus::Ending, None, runtime.clone())
                .await?;
        }

        let final_status = if let Some(ref runtime) = runtime {
            let stop_result = {
                let mut harness = runtime.harness.lock().await;
                harness.stop().await
            };
            if let Err(err) = stop_result {
                self.append_runtime_event(
                    session_id,
                    runtime,
                    SessionEvent::Error {
                        code: "session_stop_failed".to_string(),
                        message: err.to_string(),
                    },
                )
                .await?;
                SessionStatus::Failed
            } else {
                SessionStatus::Ended
            }
        } else {
            SessionStatus::Ended
        };

        let has_runtime = runtime.is_some();
        self.set_status(
            session_id,
            final_status,
            Some(time::OffsetDateTime::now_utc().unix_timestamp()),
            runtime,
        )
        .await?;
        if has_runtime {
            self.remove_runtime(session_id).await;
        }
        self.get_session(session_id).await
    }

    pub async fn list_events(
        &self,
        session_id: &LfdId,
        after_seq: Option<i64>,
    ) -> Result<Vec<PersistedSessionEvent>, SessionManagerError> {
        // Ensure we return a clean 404 for unknown sessions.
        let _ = self.get_session(session_id).await?;
        Ok(self
            .inner
            .store
            .list_session_events(session_id, after_seq)
            .await?)
    }

    pub async fn subscribe(
        &self,
        session_id: &LfdId,
    ) -> Result<Option<broadcast::Receiver<PersistedSessionEvent>>, SessionManagerError> {
        let _ = self.get_session(session_id).await?;
        Ok(self
            .runtime(session_id)
            .await
            .map(|runtime| runtime.events_tx.subscribe()))
    }

    pub async fn seeded_user_prompt(
        &self,
        session_id: &LfdId,
    ) -> Result<Option<String>, SessionManagerError> {
        let _ = self.get_session(session_id).await?;
        Ok(self
            .runtime(session_id)
            .await
            .and_then(|runtime| runtime.seeded_user_prompt.clone()))
    }

    pub async fn recover_orphaned_sessions(
        &self,
    ) -> Result<SessionStartupRecovery, SessionManagerError> {
        let mut recovery = SessionStartupRecovery::default();
        let now = time::OffsetDateTime::now_utc();
        let now_ts = now.unix_timestamp();
        let sessions = self
            .inner
            .store
            .list_sessions_by_statuses(&[SessionStatus::Starting, SessionStatus::Active])
            .await?;

        for session in &sessions {
            let existing_events = self
                .inner
                .store
                .list_session_events(&session.id, None)
                .await?;
            let mut next_seq = existing_events
                .last()
                .map(|event| event.seq + 1)
                .unwrap_or(0);

            let error_event = SessionEvent::Error {
                code: "lfd_restarted_orphaned_session".to_string(),
                message: "session was orphaned when lfd restarted".to_string(),
            };
            self.inner
                .store
                .append_session_event(&session.id, next_seq, &error_event, now_ts)
                .await?;
            next_seq += 1;

            let status_event = SessionEvent::StatusChanged {
                status: SessionStatus::Failed,
            };
            self.inner
                .store
                .append_session_event(&session.id, next_seq, &status_event, now_ts)
                .await?;

            self.inner
                .store
                .update_session_status(&session.id, SessionStatus::Failed, Some(now_ts))
                .await?;
            recovery.sessions_failed += 1;
        }

        let opencode_reap = opencode_runtime::reap_orphaned_opencode_servers();
        recovery.opencode_servers_reaped = opencode_reap.reaped;
        recovery.reap_errors = opencode_reap.errors;
        Ok(recovery)
    }

    fn spawn_harness_event_bridge(
        &self,
        session_id: LfdId,
        runtime: Arc<SessionRuntime>,
        mut harness_events_rx: mpsc::UnboundedReceiver<SessionEvent>,
    ) {
        let manager = self.clone();
        tokio::spawn(async move {
            while let Some(event) = harness_events_rx.recv().await {
                match event {
                    SessionEvent::ProviderSessionId {
                        provider_session_id,
                    } => {
                        manager
                            .handle_provider_session_id(&session_id, &runtime, provider_session_id)
                            .await;
                    }
                    SessionEvent::Error {
                        ref code,
                        ref message,
                    } => {
                        let fatal = is_terminal_harness_error(code);
                        manager
                            .append_runtime_event_or_warn(
                                &session_id,
                                &runtime,
                                SessionEvent::Error {
                                    code: code.clone(),
                                    message: message.clone(),
                                },
                                "error",
                            )
                            .await;
                        if fatal {
                            manager
                                .mark_session_failed(&session_id, &runtime, code, message.clone())
                                .await;
                        }
                    }
                    event => {
                        manager
                            .append_runtime_event_or_warn(&session_id, &runtime, event, "event")
                            .await;
                    }
                }
            }
        });
    }

    fn spawn_harness_startup(
        &self,
        session_id: LfdId,
        runtime: Arc<SessionRuntime>,
        launch: AgentConfig,
        auto_start: bool,
    ) {
        let manager = self.clone();
        tokio::spawn(async move {
            let result = {
                let mut harness = runtime.harness.lock().await;
                harness.start(&launch).await
            };

            let current_status = match manager.inner.store.get_session(&session_id).await {
                Ok(Some(session)) => Some(session.status),
                Ok(None) => None,
                Err(err) => {
                    tracing::warn!(
                        session_id = %session_id,
                        error = %err,
                        "failed to read session status after startup"
                    );
                    None
                }
            };

            match result {
                Ok(()) => {
                    if current_status != Some(SessionStatus::Starting) {
                        if current_status.is_some_and(SessionStatus::is_terminal) {
                            manager
                                .stop_harness_runtime(&session_id, &runtime, "startup_aborted")
                                .await;
                        }
                        return;
                    }
                    if let Err(err) = manager
                        .set_status(
                            &session_id,
                            SessionStatus::Active,
                            None,
                            Some(runtime.clone()),
                        )
                        .await
                    {
                        tracing::warn!(
                            session_id = %session_id,
                            error = %err,
                            "failed to set active session status"
                        );
                        return;
                    }

                    // Auto-start: send empty input so the harness delivers
                    // the task_prompt (step content) as the agent's first turn.
                    if auto_start {
                        if let Err(err) = manager
                            .append_seeded_user_prompt_event(
                                &session_id,
                                &runtime,
                                launch.task_prompt.trim(),
                            )
                            .await
                        {
                            tracing::warn!(
                                session_id = %session_id,
                                error = %err,
                                "failed to append seeded user prompt event"
                            );
                        }
                        let send_result = {
                            let mut harness = runtime.harness.lock().await;
                            harness.send_input("").await
                        };
                        if let Err(err) = send_result {
                            tracing::warn!(
                                session_id = %session_id,
                                error = %err,
                                "session auto-start failed"
                            );
                        }
                    }
                }
                Err(err) => {
                    if current_status.is_some_and(SessionStatus::is_terminal) {
                        manager
                            .stop_harness_runtime(&session_id, &runtime, "startup_aborted")
                            .await;
                        return;
                    }
                    manager
                        .mark_session_failed(
                            &session_id,
                            &runtime,
                            "session_start_failed",
                            err.to_string(),
                        )
                        .await;
                }
            }
        });
    }

    async fn append_seeded_user_prompt_event(
        &self,
        session_id: &LfdId,
        runtime: &Arc<SessionRuntime>,
        prompt: &str,
    ) -> Result<(), SessionManagerError> {
        if prompt.is_empty() {
            return Ok(());
        }

        self.append_runtime_event(
            session_id,
            runtime,
            SessionEvent::ItemCompleted {
                turn_id: "turn_seed_user_prompt".to_string(),
                item: SessionItem::Message {
                    id: "msg_seed_user_prompt".to_string(),
                    text: prompt.to_string(),
                    phase: Some("user".to_string()),
                },
            },
        )
        .await
        .map(|_| ())
    }

    async fn handle_provider_session_id(
        &self,
        session_id: &LfdId,
        runtime: &Arc<SessionRuntime>,
        provider_session_id: String,
    ) {
        {
            let mut harness = runtime.harness.lock().await;
            harness.set_provider_session_id(Some(provider_session_id.clone()));
        }
        if let Err(err) = self
            .inner
            .store
            .update_provider_session_id(session_id, &provider_session_id)
            .await
        {
            tracing::warn!(
                session_id = %session_id,
                error = %err,
                "failed to persist provider session id"
            );
        }
    }

    async fn mark_session_failed(
        &self,
        session_id: &LfdId,
        runtime: &Arc<SessionRuntime>,
        code: &str,
        message: String,
    ) {
        let _ = self
            .append_runtime_event(
                session_id,
                runtime,
                SessionEvent::Error {
                    code: code.to_string(),
                    message,
                },
            )
            .await;
        let _ = self
            .set_status(
                session_id,
                SessionStatus::Failed,
                Some(time::OffsetDateTime::now_utc().unix_timestamp()),
                Some(runtime.clone()),
            )
            .await;
        self.remove_runtime(session_id).await;
    }

    async fn set_status(
        &self,
        session_id: &LfdId,
        status: SessionStatus,
        ended_at: Option<i64>,
        runtime: Option<Arc<SessionRuntime>>,
    ) -> Result<(), SessionManagerError> {
        self.inner
            .store
            .update_session_status(session_id, status, ended_at)
            .await?;

        if let Some(runtime) = runtime {
            self.append_runtime_event(session_id, &runtime, SessionEvent::StatusChanged { status })
                .await?;
        }

        if status.is_terminal() {
            self.on_session_terminal(session_id).await;
        }

        Ok(())
    }

    async fn append_runtime_event_or_warn(
        &self,
        session_id: &LfdId,
        runtime: &Arc<SessionRuntime>,
        event: SessionEvent,
        context: &'static str,
    ) {
        if let Err(err) = self.append_runtime_event(session_id, runtime, event).await {
            tracing::warn!(
                session_id = %session_id,
                error = %err,
                context,
                "failed to persist harness event"
            );
        }
    }

    async fn append_runtime_event(
        &self,
        session_id: &LfdId,
        runtime: &Arc<SessionRuntime>,
        mut event: SessionEvent,
    ) -> Result<(), SessionManagerError> {
        populate_turn_usage_cost(&runtime.harness_name, &mut event);

        let now = time::OffsetDateTime::now_utc();
        let seq = runtime.next_seq.fetch_add(1, Ordering::Relaxed);

        self.inner
            .store
            .append_session_event(session_id, seq, &event, now.unix_timestamp())
            .await?;

        let persisted = PersistedSessionEvent {
            session_id: session_id.clone(),
            seq,
            event,
            created_at: now,
        };
        let _ = runtime.events_tx.send(persisted);
        Ok(())
    }

    async fn runtime(&self, session_id: &LfdId) -> Option<Arc<SessionRuntime>> {
        let runtimes = self.inner.runtimes.lock().await;
        runtimes.get(session_id).cloned()
    }

    async fn remove_runtime(&self, session_id: &LfdId) {
        let mut runtimes = self.inner.runtimes.lock().await;
        runtimes.remove(session_id);
    }

    async fn stop_harness_runtime(
        &self,
        session_id: &LfdId,
        runtime: &Arc<SessionRuntime>,
        context: &'static str,
    ) {
        let stop_result = {
            let mut harness = runtime.harness.lock().await;
            harness.stop().await
        };
        if let Err(err) = stop_result {
            tracing::warn!(
                session_id = %session_id,
                error = %err,
                context,
                "failed to stop session harness"
            );
        }
        self.remove_runtime(session_id).await;
    }

    async fn register_wave_session(&self, wave_run_id: Option<&str>) {
        let Some((scheduler, wave_id)) = self.scheduler_wave_id_for_run(wave_run_id).await else {
            return;
        };
        if !scheduler.register_session(&wave_id) {
            tracing::debug!(wave_id, "session already registered for wave");
        }
    }

    async fn on_session_terminal(&self, session_id: &LfdId) {
        let session = match self.inner.store.get_session(session_id).await {
            Ok(Some(session)) => session,
            Ok(None) => return,
            Err(err) => {
                tracing::warn!(session_id = %session_id, error = %err, "failed to load terminal session");
                return;
            }
        };

        let Some((scheduler, wave_id)) = self
            .scheduler_wave_id_for_run(session.wave_run_id.as_deref())
            .await
        else {
            return;
        };
        scheduler.unregister_session(&wave_id);
    }

    async fn scheduler_wave_id_for_run(
        &self,
        wave_run_id: Option<&str>,
    ) -> Option<(Arc<Scheduler>, String)> {
        let scheduler = self.inner.scheduler.clone()?;
        let wave_run_id = wave_run_id?;
        let wave_id = self.wave_id_for_wave_run(wave_run_id).await?;
        Some((scheduler, wave_id))
    }

    async fn wave_id_for_wave_run(&self, wave_run_id: &str) -> Option<String> {
        let run_id = match wave_run_id.parse::<LfdId>() {
            Ok(run_id) => run_id,
            Err(err) => {
                tracing::warn!(wave_run_id, error = %err, "invalid wave_run_id on session");
                return None;
            }
        };

        match self.inner.store.get_wave_run(&run_id).await {
            Ok(Some(run)) => Some(run.wave_id.to_string()),
            Ok(None) => {
                tracing::warn!(wave_run_id, "wave run referenced by session not found");
                None
            }
            Err(err) => {
                tracing::warn!(wave_run_id, error = %err, "failed to resolve wave for session");
                None
            }
        }
    }
}

fn resolve_harness(name: &str) -> Result<String, SessionManagerError> {
    let requested = name.trim().to_ascii_lowercase();
    if matches!(requested.as_str(), "gemini") {
        return Err(SessionManagerError::HarnessNotImplemented(requested));
    }

    harness::canonical_harness(&requested)
        .map(ToString::to_string)
        .ok_or(SessionManagerError::UnsupportedHarness(requested))
}

fn validate_repo_root(repo_root: &str) -> Result<PathBuf, SessionManagerError> {
    let raw = repo_root.trim();
    if raw.is_empty() {
        return Err(SessionManagerError::InvalidRepoRoot(
            "path is empty".to_string(),
        ));
    }

    let path = PathBuf::from(raw);
    if !path.exists() {
        return Err(SessionManagerError::InvalidRepoRoot(format!(
            "path does not exist: {raw}"
        )));
    }
    if !path.is_dir() {
        return Err(SessionManagerError::InvalidRepoRoot(format!(
            "path is not a directory: {raw}"
        )));
    }
    let canonical = path.canonicalize().map_err(|err| {
        SessionManagerError::InvalidRepoRoot(format!("failed to resolve repo_root '{raw}': {err}"))
    })?;
    if !canonical.join(".lf").is_dir() {
        return Err(SessionManagerError::InvalidRepoRoot(format!(
            "missing .lf/ in repo root: {raw}"
        )));
    }
    Ok(canonical)
}

fn resolve_cwd(
    repo_root: &Path,
    cwd: Option<&str>,
) -> Result<Option<PathBuf>, SessionManagerError> {
    let Some(trimmed) = cwd.map(str::trim).filter(|value| !value.is_empty()) else {
        return Ok(None);
    };

    let path = PathBuf::from(trimmed);
    let resolved = if path.is_absolute() {
        path
    } else {
        repo_root.join(path)
    };

    if !resolved.exists() {
        return Err(SessionManagerError::InvalidConfig(format!(
            "cwd does not exist: {}",
            resolved.display()
        )));
    }
    if !resolved.is_dir() {
        return Err(SessionManagerError::InvalidConfig(format!(
            "cwd is not a directory: {}",
            resolved.display()
        )));
    }

    let canonical_cwd = resolved.canonicalize().map_err(|err| {
        SessionManagerError::InvalidConfig(format!(
            "failed to resolve cwd '{}': {err}",
            resolved.display()
        ))
    })?;
    if !canonical_cwd.starts_with(repo_root) {
        return Err(SessionManagerError::InvalidConfig(format!(
            "cwd must be inside repo_root: {}",
            canonical_cwd.display()
        )));
    }

    Ok(Some(canonical_cwd))
}

fn normalized_seeded_user_prompt(prompt: &str) -> Option<String> {
    let trimmed = prompt.trim();
    if trimmed.is_empty() {
        return None;
    }
    Some(trimmed.to_string())
}

fn populate_turn_usage_cost(harness: &str, event: &mut SessionEvent) {
    let SessionEvent::TurnUsage { usage, .. } = event else {
        return;
    };
    if usage.cost_usd.is_some() {
        return;
    }
    let Some(model) = usage.model.as_deref() else {
        return;
    };
    let Some(rates) = lookup_cost_rates(harness, model) else {
        return;
    };
    usage.cost_usd = Some(compute_usage_cost(usage, rates));
}

fn compute_usage_cost(usage: &crate::lfd::sessions::types::TurnUsage, rates: CostRates) -> f64 {
    let mtok = |tokens: u64| tokens as f64 / 1_000_000.0;
    mtok(usage.input_tokens) * rates.input_per_mtok
        + mtok(usage.output_tokens) * rates.output_per_mtok
        + mtok(usage.cache_read_tokens.unwrap_or(0)) * rates.cache_read_per_mtok
        + mtok(usage.cache_write_tokens.unwrap_or(0)) * rates.cache_write_per_mtok
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::lfd::scheduler::Scheduler;
    use crate::lfd::sessions::harness::{Harness, HarnessError};
    use crate::lfd::sessions::types::{
        SessionConfig, SessionEvent, SessionItem, TurnStatus, TurnUsage,
    };
    use crate::lfd::store::{open_store, StorageConfig};
    use crate::lfd::types::{Wave, WaveRun};
    use anyhow::Result;
    use async_trait::async_trait;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use std::time::Duration;
    use tempfile::tempdir;
    use tokio::sync::mpsc;

    #[derive(Debug)]
    struct FakeHarness {
        tx: mpsc::UnboundedSender<SessionEvent>,
    }

    #[async_trait]
    impl Harness for FakeHarness {
        async fn start(&mut self, _config: &AgentConfig) -> Result<()> {
            Ok(())
        }

        async fn send_input(&mut self, content: &str) -> Result<()> {
            let turn_id = format!("turn_{}", uuid::Uuid::new_v4());
            let _ = self.tx.send(SessionEvent::TurnStarted {
                turn_id: turn_id.clone(),
            });
            let _ = self.tx.send(SessionEvent::TextDelta {
                turn_id: turn_id.clone(),
                content: content.to_string(),
            });
            let _ = self.tx.send(SessionEvent::TurnCompleted {
                turn_id,
                status: TurnStatus::Completed,
            });
            Ok(())
        }

        async fn stop(&mut self) -> Result<()> {
            Ok(())
        }
    }

    fn fake_create_harness(
        _harness: &str,
        event_tx: mpsc::UnboundedSender<SessionEvent>,
    ) -> Result<Box<dyn Harness>> {
        Ok(Box::new(FakeHarness { tx: event_tx }))
    }

    #[derive(Debug)]
    struct UsageHarness {
        tx: mpsc::UnboundedSender<SessionEvent>,
    }

    #[async_trait]
    impl Harness for UsageHarness {
        async fn start(&mut self, _config: &AgentConfig) -> Result<()> {
            Ok(())
        }

        async fn send_input(&mut self, _content: &str) -> Result<()> {
            let turn_id = format!("turn_{}", uuid::Uuid::new_v4());
            let _ = self.tx.send(SessionEvent::TurnStarted {
                turn_id: turn_id.clone(),
            });
            let _ = self.tx.send(SessionEvent::TurnCompleted {
                turn_id: turn_id.clone(),
                status: TurnStatus::Completed,
            });
            let _ = self.tx.send(SessionEvent::TurnUsage {
                turn_id,
                usage: TurnUsage {
                    input_tokens: 1_000_000,
                    output_tokens: 2_000_000,
                    reasoning_tokens: None,
                    cache_read_tokens: Some(500_000),
                    cache_write_tokens: Some(250_000),
                    model: Some("kimi-k2".to_string()),
                    cost_usd: None,
                },
            });
            Ok(())
        }

        async fn stop(&mut self) -> Result<()> {
            Ok(())
        }
    }

    fn usage_create_harness(
        _harness: &str,
        event_tx: mpsc::UnboundedSender<SessionEvent>,
    ) -> Result<Box<dyn Harness>> {
        Ok(Box::new(UsageHarness { tx: event_tx }))
    }

    #[derive(Debug)]
    struct BusyHarness;

    #[async_trait]
    impl Harness for BusyHarness {
        async fn start(&mut self, _config: &AgentConfig) -> Result<()> {
            Ok(())
        }

        async fn send_input(&mut self, _content: &str) -> Result<()> {
            Err(HarnessError::TurnAlreadyInProgress.into())
        }

        async fn stop(&mut self) -> Result<()> {
            Ok(())
        }
    }

    fn busy_create_harness(
        _harness: &str,
        _event_tx: mpsc::UnboundedSender<SessionEvent>,
    ) -> Result<Box<dyn Harness>> {
        Ok(Box::new(BusyHarness))
    }

    #[derive(Debug)]
    struct ResumeAwareHarness {
        tx: mpsc::UnboundedSender<SessionEvent>,
        send_count: usize,
        provider_session_id: Option<String>,
    }

    #[async_trait]
    impl Harness for ResumeAwareHarness {
        async fn start(&mut self, _config: &AgentConfig) -> Result<()> {
            Ok(())
        }

        async fn send_input(&mut self, _content: &str) -> Result<()> {
            self.send_count += 1;
            let turn_id = format!("turn_{}", uuid::Uuid::new_v4());
            let _ = self.tx.send(SessionEvent::TurnStarted {
                turn_id: turn_id.clone(),
            });
            if self.send_count == 1 {
                let _ = self.tx.send(SessionEvent::ProviderSessionId {
                    provider_session_id: "sess_resume_1".to_string(),
                });
            }
            let resume = self
                .provider_session_id
                .clone()
                .unwrap_or_else(|| "none".to_string());
            let _ = self.tx.send(SessionEvent::TextDelta {
                turn_id: turn_id.clone(),
                content: format!("resume:{resume}"),
            });
            let _ = self.tx.send(SessionEvent::TurnCompleted {
                turn_id,
                status: TurnStatus::Completed,
            });
            Ok(())
        }

        async fn stop(&mut self) -> Result<()> {
            Ok(())
        }

        fn set_provider_session_id(&mut self, provider_session_id: Option<String>) {
            self.provider_session_id = provider_session_id;
        }
    }

    fn resume_aware_create_harness(
        _harness: &str,
        event_tx: mpsc::UnboundedSender<SessionEvent>,
    ) -> Result<Box<dyn Harness>> {
        Ok(Box::new(ResumeAwareHarness {
            tx: event_tx,
            send_count: 0,
            provider_session_id: None,
        }))
    }

    static STARTING_STOP_CALLED: AtomicBool = AtomicBool::new(false);

    #[derive(Debug)]
    struct SlowStartHarness;

    #[async_trait]
    impl Harness for SlowStartHarness {
        async fn start(&mut self, _config: &AgentConfig) -> Result<()> {
            tokio::time::sleep(Duration::from_millis(200)).await;
            Ok(())
        }

        async fn send_input(&mut self, _content: &str) -> Result<()> {
            Ok(())
        }

        async fn stop(&mut self) -> Result<()> {
            STARTING_STOP_CALLED.store(true, Ordering::SeqCst);
            Ok(())
        }
    }

    fn slow_start_create_harness(
        _harness: &str,
        _event_tx: mpsc::UnboundedSender<SessionEvent>,
    ) -> Result<Box<dyn Harness>> {
        Ok(Box::new(SlowStartHarness))
    }

    async fn wait_for_status(
        manager: &SessionManager,
        session_id: &LfdId,
        expected: SessionStatus,
    ) -> Session {
        for _ in 0..50 {
            let session = manager
                .get_session(session_id)
                .await
                .expect("session should exist");
            if session.status == expected {
                return session;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        panic!("session never reached expected status");
    }

    async fn wait_for_provider_session_id(
        manager: &SessionManager,
        session_id: &LfdId,
        expected: &str,
    ) {
        for _ in 0..50 {
            let session = manager
                .get_session(session_id)
                .await
                .expect("session should exist");
            if session.provider_session_id.as_deref() == Some(expected) {
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        panic!("session never captured expected provider session id");
    }

    fn test_session_config(repo_root: &std::path::Path) -> SessionConfig {
        std::fs::create_dir_all(repo_root.join(".lf")).expect("create .lf for tests");
        SessionConfig {
            step: "design".to_string(),
            repo_root: repo_root.to_string_lossy().to_string(),
            ..Default::default()
        }
    }

    #[test]
    fn populate_turn_usage_cost_applies_opencode_model_rates() {
        let mut event = SessionEvent::TurnUsage {
            turn_id: "turn_1".to_string(),
            usage: TurnUsage {
                input_tokens: 1_000_000,
                output_tokens: 2_000_000,
                reasoning_tokens: None,
                cache_read_tokens: Some(500_000),
                cache_write_tokens: Some(250_000),
                model: Some("kimi-k2".to_string()),
                cost_usd: None,
            },
        };

        populate_turn_usage_cost("opencode", &mut event);

        let SessionEvent::TurnUsage { usage, .. } = event else {
            panic!("expected turn usage event");
        };
        let cost = usage.cost_usd.expect("cost should be populated");
        assert!((cost - 5.825).abs() < 1e-9);
    }

    #[test]
    fn populate_turn_usage_cost_does_not_override_existing_value() {
        let mut event = SessionEvent::TurnUsage {
            turn_id: "turn_1".to_string(),
            usage: TurnUsage {
                input_tokens: 1_000_000,
                output_tokens: 1_000_000,
                reasoning_tokens: None,
                cache_read_tokens: None,
                cache_write_tokens: None,
                model: Some("kimi-k2".to_string()),
                cost_usd: Some(0.42),
            },
        };

        populate_turn_usage_cost("opencode", &mut event);

        let SessionEvent::TurnUsage { usage, .. } = event else {
            panic!("expected turn usage event");
        };
        assert_eq!(usage.cost_usd, Some(0.42));
    }

    #[tokio::test]
    async fn prepare_session_prompt_uses_surface_from_config() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        std::fs::create_dir_all(tmp.path().join(".lf/steps")).expect("create .lf/steps");
        std::fs::write(tmp.path().join(".lf/steps/design.md"), "Design it.").expect("write step");

        let manager = SessionManager::with_create_harness(store, fake_create_harness);
        let (_, prompt, _) = manager
            .prepare_session_prompt(SessionConfig {
                step: "design".to_string(),
                repo_root: tmp.path().to_string_lossy().to_string(),
                surface: Some(Surface::ConcertoIphone),
                ..Default::default()
            })
            .await
            .expect("prepare session prompt");

        assert!(prompt.system_prompt.contains("Surface: Concerto (iPhone)"));
    }

    #[tokio::test]
    async fn create_session_emits_context_snapshot_event() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );

        std::fs::create_dir_all(tmp.path().join(".lf/steps")).expect("create .lf/steps");
        std::fs::write(tmp.path().join(".lf/steps/design.md"), "Design the system.")
            .expect("write step");

        let manager = SessionManager::with_create_harness(store, fake_create_harness);
        let created = manager
            .create_session(CreateSessionParams {
                harness: "codex".to_string(),
                wave_run_id: None,
                config: SessionConfig {
                    step: "design".to_string(),
                    repo_root: tmp.path().to_string_lossy().to_string(),
                    ..Default::default()
                },
            })
            .await
            .expect("create session");

        let events = manager
            .list_events(&created.id, None)
            .await
            .expect("list events");

        let snapshot = events.iter().find_map(|event| match &event.event {
            SessionEvent::ContextSnapshot { snapshot } => Some(snapshot.clone()),
            _ => None,
        });
        let snapshot = snapshot.expect("context snapshot event");
        assert!(snapshot.total > 0);
        assert!(snapshot.sources.contains_key("step"));
    }

    #[tokio::test]
    async fn opencode_turn_usage_events_include_computed_cost() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );

        std::fs::create_dir_all(tmp.path().join(".lf/steps")).expect("create .lf/steps");
        std::fs::write(tmp.path().join(".lf/steps/design.md"), "Design the system.")
            .expect("write step");

        let manager = SessionManager::with_create_harness(store, usage_create_harness);
        let created = manager
            .create_session(CreateSessionParams {
                harness: "opencode".to_string(),
                wave_run_id: Some("run_cost".to_string()),
                config: SessionConfig {
                    step: "design".to_string(),
                    repo_root: tmp.path().to_string_lossy().to_string(),
                    ..Default::default()
                },
            })
            .await
            .expect("create session");
        let _ = wait_for_status(&manager, &created.id, SessionStatus::Active).await;

        let mut cost = None;
        for _ in 0..50 {
            let events = manager
                .list_events(&created.id, None)
                .await
                .expect("list events");
            cost = events.iter().find_map(|event| match &event.event {
                SessionEvent::TurnUsage { usage, .. } => usage.cost_usd,
                _ => None,
            });
            if cost.is_some() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        let cost = cost.expect("turn usage event with computed cost");
        assert!((cost - 5.825).abs() < 1e-9);
    }

    #[tokio::test]
    async fn session_lifecycle_create_input_events_end() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );

        let manager = SessionManager::with_create_harness(store, fake_create_harness);
        std::fs::create_dir_all(tmp.path().join(".lf")).expect("create .lf for tests");
        let created = manager
            .create_session(CreateSessionParams {
                harness: "codex".to_string(),
                wave_run_id: Some("run_1".to_string()),
                config: SessionConfig {
                    step: "design".to_string(),
                    repo_root: tmp.path().to_string_lossy().to_string(),
                    agent: Some("gpt-5.1-codex".to_string()),
                    cwd: Some(tmp.path().to_string_lossy().to_string()),
                    ..Default::default()
                },
            })
            .await
            .expect("create session");

        assert_eq!(created.status, SessionStatus::Starting);
        let _ = wait_for_status(&manager, &created.id, SessionStatus::Active).await;

        manager
            .send_input(&created.id, "fix the failing tests")
            .await
            .expect("send input");

        let mut saw_text_delta = false;
        for _ in 0..50 {
            let events = manager
                .list_events(&created.id, None)
                .await
                .expect("list events");
            saw_text_delta = events.iter().any(|event| {
                matches!(
                    &event.event,
                    SessionEvent::TextDelta { content, .. } if content == "fix the failing tests"
                )
            });
            if saw_text_delta {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        assert!(saw_text_delta);

        let ended = manager
            .stop_session(&created.id)
            .await
            .expect("stop session");
        assert_eq!(ended.status, SessionStatus::Ended);

        let replay = manager
            .list_events(&created.id, Some(0))
            .await
            .expect("replay events");
        assert!(!replay.is_empty());
    }

    #[tokio::test]
    async fn auto_start_seeds_user_prompt_event() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );

        let steps_dir = tmp.path().join(".lf").join("steps");
        std::fs::create_dir_all(&steps_dir).expect("create steps dir");
        std::fs::write(
            steps_dir.join("seed-step.md"),
            "Seeded prompt text for interactive wave start.",
        )
        .expect("write test step");

        let manager = SessionManager::with_create_harness(store, fake_create_harness);
        let created = manager
            .create_session(CreateSessionParams {
                harness: "codex".to_string(),
                wave_run_id: Some("run_seed".to_string()),
                config: SessionConfig {
                    step: "seed-step".to_string(),
                    repo_root: tmp.path().to_string_lossy().to_string(),
                    agent: Some("gpt-5.1-codex".to_string()),
                    cwd: Some(tmp.path().to_string_lossy().to_string()),
                    ..Default::default()
                },
            })
            .await
            .expect("create session");

        let _ = wait_for_status(&manager, &created.id, SessionStatus::Active).await;

        let mut saw_seeded_prompt = false;
        for _ in 0..50 {
            let events = manager
                .list_events(&created.id, None)
                .await
                .expect("list events");
            saw_seeded_prompt = events.iter().any(|event| {
                matches!(
                    &event.event,
                    SessionEvent::ItemCompleted {
                        item:
                            SessionItem::Message {
                                text,
                                phase: Some(phase),
                                ..
                            },
                        ..
                    } if phase == "user" && text.contains("Seeded prompt text for interactive wave start.")
                )
            });
            if saw_seeded_prompt {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }

        assert!(saw_seeded_prompt);
    }

    #[tokio::test]
    async fn create_session_rejects_unsupported_harness() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        // Use default_create_harness (not fake) so unsupported harnesses are rejected.
        let manager = SessionManager::new(store);

        let err = manager
            .create_session(CreateSessionParams {
                harness: "nonexistent".to_string(),
                wave_run_id: None,
                config: SessionConfig::default(),
            })
            .await
            .expect_err("unsupported harness should fail");

        assert!(matches!(
            err,
            SessionManagerError::UnsupportedHarness(ref name) if name == "nonexistent"
        ));
    }

    #[tokio::test]
    async fn create_session_marks_known_unimplemented_harness() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let manager = SessionManager::new(store);

        let err = manager
            .create_session(CreateSessionParams {
                harness: "gemini".to_string(),
                wave_run_id: None,
                config: SessionConfig::default(),
            })
            .await
            .expect_err("gemini should be explicitly not implemented");

        assert!(matches!(
            err,
            SessionManagerError::HarnessNotImplemented(ref name) if name == "gemini"
        ));
    }

    #[tokio::test]
    async fn create_session_accepts_opencode_harness_name() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let repo_root = tmp.path().join("repo");
        std::fs::create_dir_all(repo_root.join(".lf")).expect("create .lf for tests");
        std::fs::create_dir_all(repo_root.join(".lf/steps")).expect("create .lf/steps");
        std::fs::write(repo_root.join(".lf/steps/design.md"), "stub design")
            .expect("write stub step");

        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let manager = SessionManager::with_create_harness(store, fake_create_harness);

        let created = manager
            .create_session(CreateSessionParams {
                harness: "opencode".to_string(),
                wave_run_id: None,
                config: SessionConfig {
                    step: "design".to_string(),
                    repo_root: repo_root.to_string_lossy().to_string(),
                    ..Default::default()
                },
            })
            .await
            .expect("opencode should resolve as a supported harness");

        assert_eq!(created.harness, "opencode");
    }

    #[tokio::test]
    async fn create_session_rejects_invalid_repo_root() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let manager = SessionManager::with_create_harness(store, fake_create_harness);

        let err = manager
            .create_session(CreateSessionParams {
                harness: "claude".to_string(),
                wave_run_id: None,
                config: SessionConfig {
                    step: "design".to_string(),
                    repo_root: tmp.path().join("missing").to_string_lossy().to_string(),
                    ..Default::default()
                },
            })
            .await
            .expect_err("invalid repo root should fail");

        assert!(matches!(err, SessionManagerError::InvalidRepoRoot(_)));
    }

    #[tokio::test]
    async fn create_session_rejects_cwd_outside_repo_root() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let repo_root = tmp.path().join("repo");
        std::fs::create_dir_all(repo_root.join(".lf")).expect("create .lf for tests");
        let outside = tmp.path().join("outside");
        std::fs::create_dir_all(&outside).expect("create outside dir");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let manager = SessionManager::with_create_harness(store, fake_create_harness);

        let err = manager
            .create_session(CreateSessionParams {
                harness: "claude".to_string(),
                wave_run_id: None,
                config: SessionConfig {
                    step: "design".to_string(),
                    repo_root: repo_root.to_string_lossy().to_string(),
                    cwd: Some(outside.to_string_lossy().to_string()),
                    ..Default::default()
                },
            })
            .await
            .expect_err("cwd outside repo root should fail");

        assert!(matches!(err, SessionManagerError::InvalidConfig(_)));
    }

    #[tokio::test]
    async fn create_session_enforces_single_active_session_per_wave_run() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let manager = SessionManager::with_create_harness(store, fake_create_harness);

        let created = manager
            .create_session(CreateSessionParams {
                harness: "codex".to_string(),
                wave_run_id: Some("run_1".to_string()),
                config: test_session_config(tmp.path()),
            })
            .await
            .expect("first session should create");
        let _ = wait_for_status(&manager, &created.id, SessionStatus::Active).await;

        let err = manager
            .create_session(CreateSessionParams {
                harness: "codex".to_string(),
                wave_run_id: Some("run_1".to_string()),
                config: test_session_config(tmp.path()),
            })
            .await
            .expect_err("second active session should be rejected");

        assert!(matches!(
            err,
            SessionManagerError::WaveRunSessionConflict(ref wave_run_id) if wave_run_id == "run_1"
        ));
    }

    #[tokio::test]
    async fn send_input_busy_error_does_not_fail_session() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let manager = SessionManager::with_create_harness(store, busy_create_harness);

        let created = manager
            .create_session(CreateSessionParams {
                harness: "codex".to_string(),
                wave_run_id: None,
                config: test_session_config(tmp.path()),
            })
            .await
            .expect("create session");
        let _ = wait_for_status(&manager, &created.id, SessionStatus::Active).await;

        let err = manager
            .send_input(&created.id, "hello")
            .await
            .expect_err("busy harness should reject concurrent turn");
        assert!(matches!(err, SessionManagerError::TurnAlreadyInProgress));

        let session = manager
            .get_session(&created.id)
            .await
            .expect("session should still exist");
        assert_eq!(session.status, SessionStatus::Active);
    }

    #[tokio::test]
    async fn provider_session_id_is_applied_to_harness_for_resume() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let manager = SessionManager::with_create_harness(store, resume_aware_create_harness);

        let created = manager
            .create_session(CreateSessionParams {
                harness: "codex".to_string(),
                wave_run_id: None,
                config: test_session_config(tmp.path()),
            })
            .await
            .expect("create session");
        let _ = wait_for_status(&manager, &created.id, SessionStatus::Active).await;

        manager
            .send_input(&created.id, "first turn")
            .await
            .expect("first input should succeed");
        wait_for_provider_session_id(&manager, &created.id, "sess_resume_1").await;

        manager
            .send_input(&created.id, "second turn")
            .await
            .expect("second input should succeed");

        let mut saw_resume = false;
        for _ in 0..50 {
            let events = manager
                .list_events(&created.id, None)
                .await
                .expect("list events");
            saw_resume = events.iter().any(|event| {
                matches!(
                    &event.event,
                    SessionEvent::TextDelta { content, .. } if content == "resume:sess_resume_1"
                )
            });
            if saw_resume {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        assert!(saw_resume);
    }

    #[tokio::test]
    async fn recover_orphaned_sessions_marks_active_as_failed() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );

        let manager = SessionManager::with_create_harness(store.clone(), fake_create_harness);
        let created = manager
            .create_session(CreateSessionParams {
                harness: "claude".to_string(),
                wave_run_id: None,
                config: test_session_config(tmp.path()),
            })
            .await
            .expect("create session");
        let _ = wait_for_status(&manager, &created.id, SessionStatus::Active).await;

        // Wait for the Active StatusChanged event to be persisted (set_status
        // updates the DB status before appending the event, so wait_for_status
        // can return before the event is written).
        for _ in 0..50 {
            let events = manager
                .list_events(&created.id, None)
                .await
                .expect("list events");
            let has_active_event = events.iter().any(|e| {
                matches!(
                    &e.event,
                    SessionEvent::StatusChanged { status } if *status == SessionStatus::Active
                )
            });
            if has_active_event {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }

        // Simulate a new SessionManager on restart (no runtimes, orphaned session in DB).
        let fresh_manager = SessionManager::with_create_harness(store, fake_create_harness);
        let recovered = fresh_manager
            .recover_orphaned_sessions()
            .await
            .expect("orphan recovery");
        assert_eq!(recovered.sessions_failed, 1);

        let session = fresh_manager
            .get_session(&created.id)
            .await
            .expect("session should exist");
        assert_eq!(session.status, SessionStatus::Failed);

        let events = fresh_manager
            .list_events(&created.id, None)
            .await
            .expect("list events");
        let has_orphan_error = events.iter().any(|event| {
            matches!(
                &event.event,
                SessionEvent::Error { code, .. } if code == "lfd_restarted_orphaned_session"
            )
        });
        assert!(has_orphan_error);
    }

    #[tokio::test(start_paused = true)]
    async fn stop_session_while_starting_marks_failed_and_cleans_up_harness() {
        STARTING_STOP_CALLED.store(false, Ordering::SeqCst);
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let manager = SessionManager::with_create_harness(store, slow_start_create_harness);

        let created = manager
            .create_session(CreateSessionParams {
                harness: "claude".to_string(),
                wave_run_id: None,
                config: test_session_config(tmp.path()),
            })
            .await
            .expect("create session");

        let stopped = manager
            .stop_session(&created.id)
            .await
            .expect("stop session while starting");
        assert_eq!(stopped.status, SessionStatus::Failed);
        assert!(
            !STARTING_STOP_CALLED.load(Ordering::SeqCst),
            "stop_session should not block on harness.stop while startup is running"
        );

        tokio::time::sleep(Duration::from_millis(250)).await;
        let after_start_finishes = manager
            .get_session(&created.id)
            .await
            .expect("session should exist");
        assert_eq!(after_start_finishes.status, SessionStatus::Failed);
        assert!(
            STARTING_STOP_CALLED.load(Ordering::SeqCst),
            "startup cleanup should stop the harness once startup returns"
        );
    }

    #[tokio::test]
    async fn wave_session_registers_and_unregisters_scheduler_session() {
        let tmp = tempdir().expect("tempdir");
        let db_path = tmp.path().join("lfd.db");
        let store = Arc::new(
            open_store(&StorageConfig::sqlite(db_path))
                .await
                .expect("open sqlite store"),
        );
        let scheduler = Arc::new(Scheduler::new(1));
        let manager = SessionManager::with_create_harness_and_scheduler(
            store.clone(),
            fake_create_harness,
            Some(scheduler.clone()),
        );

        std::fs::create_dir_all(tmp.path().join(".lf")).expect("create .lf for tests");
        let wave_id = LfdId::new();
        let wave = Wave::new(
            wave_id.clone(),
            "wave-a".to_string(),
            tmp.path().to_string_lossy().to_string(),
        );
        store.create_wave(&wave).await.expect("create wave");

        let run_id = LfdId::new();
        let mut run = WaveRun::new(run_id.clone(), wave_id.clone());
        run.worktree = tmp.path().to_string_lossy().to_string();
        run.branch = "wave-a".to_string();
        store.create_wave_run(&run).await.expect("create wave run");

        let created = manager
            .create_session(CreateSessionParams {
                harness: "codex".to_string(),
                wave_run_id: Some(run_id.to_string()),
                config: test_session_config(tmp.path()),
            })
            .await
            .expect("create session");
        let _ = wait_for_status(&manager, &created.id, SessionStatus::Active).await;

        assert!(scheduler.has_active_session(wave_id.as_str()));

        manager
            .stop_session(&created.id)
            .await
            .expect("stop session");
        assert!(!scheduler.has_active_session(wave_id.as_str()));
    }
}
