ai-harness/crates/harness-app/src/lib.rs

//! Composition root: builds the pieces `harness-tui` (and later `harness-server`) need —
//! config, store, event bus, permission service, tool registry, provider registry — and
//! exposes a small headless API (`run_prompt`) used by the `harness run -p` debug command.

use std::collections::hash_map::DefaultHasher;
use std::collections::HashMap;
use std::hash::Hasher;
use std::path::Path;
use std::path::PathBuf;
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::time::{SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use harness_core::agent::{AgentDef, AgentRegistry};
use harness_core::config::{self, Config, ConfigError};
use harness_core::engine::jobs::{JobBoard, JobState, LaunchSpec};
use harness_core::engine::{run_session, RunConfig, StepContext};
use harness_core::event::{AppEvent, EventBus, RunOutcome};
use harness_core::permission::{PermissionReply, PermissionService, Rule, Ruleset};
use harness_core::store::{Store, StoreError};
use harness_core::tool::{
    ContextReporter, SpawnError, SpawnOutcome, SpawnRequest, SubagentSpawner, ToolRegistry,
};
use harness_core::types::{
    Message, MessageId, ModelRef, Part, PartBody, PartId, Session, SessionId,
};
use harness_providers::{AnthropicProvider, ModelCatalog, OpenAiProvider, ProviderRegistry};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;

/// Fallback prompt for an agent whose markdown body is empty or that isn't in the registry.
pub const DEFAULT_AGENT_PROMPT: &str =
    "You are a helpful coding assistant with access to tools for reading, editing, running \
     commands, and searching code. Use them as needed to satisfy the user's request, then \
     reply with a concise final answer.";

const DEFAULT_MAX_STEPS: u32 = 50;

#[derive(Debug, thiserror::Error)]
pub enum AppError {
    #[error(transparent)]
    Config(#[from] ConfigError),
    #[error(transparent)]
    Store(#[from] StoreError),
    #[error("model ref must be \"provider/model\", got {0:?}")]
    InvalidModelRef(String),
    #[error("no provider registered for {0:?} (missing API key?)")]
    UnknownProvider(String),
    #[error("session {0} already has an active run")]
    SessionRunning(SessionId),
}

fn now_ms() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as i64
}

fn db_path(cwd: &Path) -> PathBuf {
    let base = dirs::data_dir()
        .unwrap_or_else(std::env::temp_dir)
        .join("ai-harness")
        .join("db");
    let cwd_str = cwd.to_string_lossy();
    let sanitized: String = cwd_str
        .chars()
        .map(|c| if c.is_alphanumeric() { c } else { '_' })
        .collect();
    let mut hasher = DefaultHasher::new();
    hasher.write(cwd_str.as_bytes());
    let hash = hasher.finish();
    let hash_hex = format!("{:016x}", hash);
    let slug = format!("{}_{}", sanitized, &hash_hex[..8]);
    base.join(format!("{}.sqlite", slug))
}

/// Non-blocking, multi-turn engine API used by the TUI.
#[derive(Clone)]
pub struct EngineHandle {
    inner: Arc<EngineInner>,
}

struct EngineInner {
    config: Config,
    store: Store,
    bus: EventBus,
    permissions: Arc<PermissionService>,
    tools: ToolRegistry,
    providers: ProviderRegistry,
    catalog: ModelCatalog,
    agents: AgentRegistry,
    cwd: PathBuf,
    data_dir: PathBuf,
    runs: Mutex<HashMap<SessionId, RunHandle>>,
    /// One job board per session (scoped to that session as a parent). Lazily created and
    /// shared between a session's run loop (board injection) and the spawner (registration).
    boards: Mutex<HashMap<SessionId, Arc<JobBoard>>>,
    /// Self-reference so the spawner can hand an `Arc<EngineInner>` to child contexts (for
    /// nested delegation) and to detached background tasks. Set once in `new`.
    me: OnceLock<Weak<EngineInner>>,
}

struct RunHandle {
    cancel: CancellationToken,
}

impl EngineHandle {
    /// Persistent SQLite-backed store. Used by the TUI and the default headless `App`.
    pub fn init(cwd: PathBuf) -> Result<Self, AppError> {
        let path = db_path(&cwd);
        let store = Store::open(&path)?;
        let handle = Self::new(cwd, store)?;
        handle.spawn_catalog_refresh();
        Ok(handle)
    }

    /// In-memory store — useful for tests and ephemeral sessions.
    pub fn init_in_memory(cwd: PathBuf) -> Result<Self, AppError> {
        let store = Store::open_in_memory()?;
        Self::new(cwd, store)
    }

    fn new(cwd: PathBuf, store: Store) -> Result<Self, AppError> {
        let config = config::load(&cwd)?;

        let mut providers = ProviderRegistry::new();
        if let Some(key) = config
            .providers
            .get("anthropic")
            .and_then(|p| p.api_key.clone())
        {
            providers.register(Arc::new(AnthropicProvider::new(key)));
        }
        if let Some(key) = config
            .providers
            .get("openai")
            .and_then(|p| p.api_key.clone())
        {
            let provider = match config
                .providers
                .get("openai")
                .and_then(|p| p.base_url.clone())
            {
                Some(base_url) => OpenAiProvider::with_base_url(key, base_url),
                None => OpenAiProvider::new(key),
            };
            providers.register(Arc::new(provider));
        }
        // OpenCode Zen: OpenAI-compatible chat gateway under its own `opencode` provider id,
        // so it coexists with a real `openai` provider. `base_url` defaults to Zen's endpoint.
        if let Some(key) = config
            .providers
            .get("opencode")
            .and_then(|p| p.api_key.clone())
        {
            let base_url = config
                .providers
                .get("opencode")
                .and_then(|p| p.base_url.clone());
            providers.register(Arc::new(OpenAiProvider::opencode(key, base_url)));
        }

        Self::build(cwd, store, config, providers)
    }

    /// Shared construction over an explicit provider set — the seam tests use to inject a mock.
    fn build(
        cwd: PathBuf,
        store: Store,
        config: Config,
        providers: ProviderRegistry,
    ) -> Result<Self, AppError> {
        let bus = EventBus::new();
        let permissions = Arc::new(PermissionService::new(bus.clone()));

        let mut tools = ToolRegistry::new();
        harness_tools::register_builtins(&mut tools);
        harness_tools::register_task_tool(&mut tools);

        // Layered agent registry: bundled markdown → global dir → project dir → config patches.
        let global_agent_dir = dirs::config_dir().map(|d| d.join("ai-harness").join("agent"));
        let project_agent_dir = cwd.join(".harness").join("agent");
        let agents = AgentRegistry::load(
            &config.agents,
            global_agent_dir.as_deref(),
            Some(project_agent_dir.as_path()),
        );

        let data_dir = dirs::data_dir()
            .unwrap_or_else(std::env::temp_dir)
            .join("ai-harness")
            .join("tool-output");

        // No network at construction: use the cached copy if fresh, else the baked snapshot.
        // `init` warms the cache in the background for the next launch.
        let catalog = ModelCatalog::load_cached_or_baked(&ModelCatalog::default_cache_path());

        let inner = Arc::new(EngineInner {
            config,
            store,
            bus,
            permissions,
            tools,
            providers,
            catalog,
            agents,
            cwd,
            data_dir,
            runs: Mutex::new(HashMap::new()),
            boards: Mutex::new(HashMap::new()),
            me: OnceLock::new(),
        });
        let _ = inner.me.set(Arc::downgrade(&inner));
        Ok(Self { inner })
    }

    /// Fire-and-forget refresh of the models.dev cache so the next launch has current pricing.
    fn spawn_catalog_refresh(&self) {
        tokio::spawn(async {
            if let Err(e) = ModelCatalog::refresh_default_cache().await {
                tracing::debug!(error = %e, "models.dev refresh failed; using cached/baked metadata");
            }
        });
    }

    pub fn bus(&self) -> EventBus {
        self.inner.bus.clone()
    }

    pub fn permissions(&self) -> Arc<PermissionService> {
        self.inner.permissions.clone()
    }

    pub fn config(&self) -> Config {
        self.inner.config.clone()
    }

    pub async fn new_session(&self, agent: &str, model_ref: &str) -> Result<SessionId, AppError> {
        let (provider_id, model_id) = model_ref
            .split_once('/')
            .ok_or_else(|| AppError::InvalidModelRef(model_ref.to_string()))?;
        let model = ModelRef::new(provider_id, model_id);
        let now = now_ms();
        let session = Session::new_root(agent, model, now);
        let session_id = session.id.clone();
        self.inner.store.upsert_session(session.clone()).await?;
        self.inner.bus.publish(AppEvent::SessionCreated { session });
        Ok(session_id)
    }

    pub async fn prompt(
        &self,
        session_id: SessionId,
        text: String,
        model_ref: &str,
    ) -> Result<(), AppError> {
        let (provider_id, model_id) = model_ref
            .split_once('/')
            .ok_or_else(|| AppError::InvalidModelRef(model_ref.to_string()))?;
        let provider = self
            .inner
            .providers
            .get(provider_id)
            .ok_or_else(|| AppError::UnknownProvider(provider_id.to_string()))?;
        let model = ModelRef::new(provider_id, model_id);

        {
            let runs = self.inner.runs.lock().unwrap();
            if runs.contains_key(&session_id) {
                return Err(AppError::SessionRunning(session_id));
            }
        }

        let now = now_ms();
        let user_message = Message::new_user(session_id.clone(), now);
        self.inner
            .store
            .upsert_message(user_message.clone())
            .await?;
        // The store actor does not emit events, so publish the user message and its part
        // ourselves — otherwise the TUI (which builds its live transcript purely from bus
        // events) never shows the prompt the user just typed until the session is reloaded.
        self.inner.bus.publish(AppEvent::MessageCreated {
            message: user_message.clone(),
        });

        let user_part = Part {
            id: PartId::new(),
            message_id: user_message.id.clone(),
            session_id: session_id.clone(),
            idx: 0,
            body: PartBody::Text {
                text,
                synthetic: false,
            },
        };
        self.inner.store.upsert_part(user_part.clone()).await?;
        self.inner
            .bus
            .publish(AppEvent::PartUpdated { part: user_part });

        // Resolve the session's agent from the registry (falls back to a generic assistant).
        let session_agent = self
            .inner
            .store
            .session(session_id.clone())
            .await?
            .map(|s| s.agent)
            .unwrap_or_else(|| "orchestrator".to_string());
        let agent = self.inner.agent_def(&session_agent);
        let inject_job_board = self.inner.config.orchestration.job_board && agent.mode.is_primary();
        let board = self.inner.board_for(&session_id).await?;

        let ctx = StepContext {
            store: self.inner.store.clone(),
            bus: self.inner.bus.clone(),
            tools: self.inner.filtered_tools(&agent),
            permissions: self.inner.permissions.clone(),
            static_rules: self.inner.effective_static_rules(&agent, 0),
            extra_rules: Arc::new(Mutex::new(Vec::new())),
            parent_rules: Vec::new(), // root session: no intersection
            session_id: session_id.clone(),
            cwd: self.inner.cwd.clone(),
            data_dir: self.inner.data_dir.join(session_id.to_string()),
            cancel: CancellationToken::new(),
            now,
            spawner: Some(self.inner.clone()),
            job_board: Some(board),
            context_reporter: None, // root session has no parent board to report to
        };
        let run_config = RunConfig {
            agent_name: agent.name.clone(),
            agent_prompt: self.inner.agent_prompt(&agent),
            model,
            temperature: agent.temperature,
            max_steps: agent.max_steps.unwrap_or(DEFAULT_MAX_STEPS),
            instructions: self.inner.config.instructions.clone(),
            cost: Some(self.inner.catalog.cost(provider_id, model_id)),
            inject_job_board,
            reminder_turn_start: self.inner.reminder("turn_start"),
            reminder_after_file_tool: self.inner.reminder("after_file_tool"),
        };

        // Reserve the run slot *before* spawning. If we inserted after spawning, a run that
        // finished quickly could remove its (not-yet-inserted) entry first, and our later
        // insert would then strand the session as permanently "running".
        let cancel = ctx.cancel.clone();
        {
            let mut runs = self.inner.runs.lock().unwrap();
            if runs.contains_key(&session_id) {
                return Err(AppError::SessionRunning(session_id));
            }
            runs.insert(
                session_id.clone(),
                RunHandle {
                    cancel: cancel.clone(),
                },
            );
        }

        self.inner.bus.publish(AppEvent::RunStarted {
            session_id: session_id.clone(),
        });

        let inner = self.inner.clone();
        let spawn_session_id = session_id.clone();
        tokio::spawn(async move {
            let outcome = run_session(provider, ctx, &run_config, now_ms).await;
            inner.bus.publish(AppEvent::RunFinished {
                session_id: spawn_session_id.clone(),
                outcome: outcome.clone(),
            });
            let mut runs = inner.runs.lock().unwrap();
            runs.remove(&spawn_session_id);
        });

        Ok(())
    }

    pub fn abort(&self, session_id: &SessionId) {
        let runs = self.inner.runs.lock().unwrap();
        if let Some(run) = runs.get(session_id) {
            run.cancel.cancel();
        }
    }

    pub fn permission_reply(&self, id: &str, reply: PermissionReply) -> bool {
        self.inner.permissions.reply(id, reply)
    }

    pub async fn list_sessions(&self) -> Result<Vec<Session>, AppError> {
        Ok(self.inner.store.sessions().await?)
    }

    pub async fn session_messages(&self, session_id: SessionId) -> Result<Vec<Message>, AppError> {
        Ok(self.inner.store.messages(session_id).await?)
    }

    pub async fn message_parts(&self, message_id: MessageId) -> Result<Vec<Part>, AppError> {
        Ok(self.inner.store.parts(message_id).await?)
    }

    pub async fn get_session(&self, session_id: SessionId) -> Result<Option<Session>, AppError> {
        Ok(self.inner.store.session(session_id).await?)
    }

    /// Background jobs spawned by `session_id` (the parent), for the TUI jobs pane / drill-in.
    pub async fn jobs(
        &self,
        session_id: SessionId,
    ) -> Result<Vec<harness_core::engine::JobRecord>, AppError> {
        Ok(self.inner.store.jobs_for_parent(session_id).await?)
    }

    pub fn is_running(&self, session_id: &SessionId) -> bool {
        let runs = self.inner.runs.lock().unwrap();
        runs.contains_key(session_id)
    }

    pub async fn final_text(&self, session_id: &SessionId) -> Result<String, AppError> {
        let messages = self.inner.store.messages(session_id.clone()).await?;
        let Some(last) = messages.last() else {
            return Ok(String::new());
        };
        let parts = self.inner.store.parts(last.id.clone()).await?;
        let text = parts
            .iter()
            .filter_map(|p| match &p.body {
                PartBody::Text { text, .. } => Some(text.as_str()),
                _ => None,
            })
            .collect::<Vec<_>>()
            .join("");
        Ok(text)
    }
}

/// Deny rules injected as low-priority defaults for any session at depth > 0: a subagent may
/// not itself delegate (`task`) or manage todos unless config/agent rules re-enable it.
fn depth_guard_rules() -> Ruleset {
    ["task", "todo", "todo*"]
        .into_iter()
        .map(|perm| Rule {
            permission: perm.to_string(),
            pattern: "*".to_string(),
            action: harness_core::permission::Action::Deny,
        })
        .collect()
}

impl EngineInner {
    fn arc(&self) -> Arc<EngineInner> {
        self.me
            .get()
            .and_then(Weak::upgrade)
            .expect("EngineInner self-reference set in new()")
    }

    /// Resolves an agent definition by name, falling back to a generic assistant so an
    /// unknown/misconfigured agent still runs rather than failing the session.
    fn agent_def(&self, name: &str) -> AgentDef {
        self.agents.get(name).cloned().unwrap_or_else(|| AgentDef {
            name: name.to_string(),
            description: String::new(),
            mode: harness_core::agent::AgentMode::Primary,
            model: None,
            temperature: None,
            prompt: String::new(),
            permissions: Vec::new(),
            tools: std::collections::HashMap::new(),
            max_steps: None,
            source: harness_core::agent::AgentSource::Config,
        })
    }

    /// A tool registry filtered to what `agent` enables. Exact tool names override a `*`
    /// wildcard; absent = enabled. Honors the `tools:` frontmatter map (docs/04-multiagent.md).
    fn filtered_tools(&self, agent: &AgentDef) -> ToolRegistry {
        if agent.tools.is_empty() {
            return self.tools.clone();
        }
        let mut reg = ToolRegistry::new();
        for tool in self.tools.all() {
            let name = tool.name();
            let enabled = agent
                .tools
                .get(name)
                .or_else(|| agent.tools.get("*"))
                .copied()
                .unwrap_or(true);
            if enabled {
                reg.register(tool);
            }
        }
        reg
    }

    fn agent_prompt(&self, agent: &AgentDef) -> String {
        if agent.prompt.trim().is_empty() {
            DEFAULT_AGENT_PROMPT.to_string()
        } else {
            agent.prompt.clone()
        }
    }

    /// An optional orchestration reminder string by hook key (`turn_start`/`after_file_tool`).
    fn reminder(&self, key: &str) -> Option<String> {
        self.config
            .orchestration
            .reminders
            .as_ref()?
            .get(key)
            .cloned()
    }

    /// The ruleset a session evaluates its own tool calls against: low-priority depth guards
    /// (depth > 0), then global config permissions, then the agent's own rules (highest).
    fn effective_static_rules(&self, agent: &AgentDef, depth: u8) -> Ruleset {
        let mut rules = Ruleset::new();
        if depth > 0 {
            rules.extend(depth_guard_rules());
        }
        rules.extend(self.config.permissions.clone());
        rules.extend(agent.permissions.clone());
        rules
    }

    /// The parent-effective ruleset a *child* intersects against: global config, the parent
    /// agent's rules, then the parent session's `Always` grants. Frozen at spawn time.
    fn parent_effective_rules(&self, session: &Session, agent: &AgentDef) -> Ruleset {
        let mut rules = self.config.permissions.clone();
        rules.extend(agent.permissions.clone());
        rules.extend(session.extra_rules.clone());
        rules
    }

    /// Gets (or lazily loads) the job board scoped to `session_id` as a parent.
    async fn board_for(&self, session_id: &SessionId) -> Result<Arc<JobBoard>, StoreError> {
        if let Some(board) = self.boards.lock().unwrap().get(session_id).cloned() {
            return Ok(board);
        }
        let board = Arc::new(
            JobBoard::load(
                self.store.clone(),
                self.bus.clone(),
                session_id,
                self.config.orchestration.max_reusable_per_agent,
            )
            .await?,
        );
        // Double-check under lock in case a concurrent caller inserted first.
        let mut boards = self.boards.lock().unwrap();
        Ok(boards.entry(session_id.clone()).or_insert(board).clone())
    }

    /// Appends `text` as a user message + text part to `session_id` (mirrors the root prompt
    /// path so a child session's first turn has an instruction to act on).
    async fn append_user_message(
        &self,
        session_id: &SessionId,
        text: String,
        now: i64,
    ) -> Result<(), StoreError> {
        let message = Message::new_user(session_id.clone(), now);
        self.store.upsert_message(message.clone()).await?;
        self.bus.publish(AppEvent::MessageCreated {
            message: message.clone(),
        });
        let part = Part {
            id: PartId::new(),
            message_id: message.id.clone(),
            session_id: session_id.clone(),
            idx: 0,
            body: PartBody::Text {
                text,
                synthetic: false,
            },
        };
        self.store.upsert_part(part.clone()).await?;
        self.bus.publish(AppEvent::PartUpdated { part });
        Ok(())
    }

    /// Concatenates the `Text` parts of a session's last message — a child run's final answer.
    async fn session_final_text(&self, session_id: &SessionId) -> String {
        let Ok(messages) = self.store.messages(session_id.clone()).await else {
            return String::new();
        };
        let Some(last) = messages.last() else {
            return String::new();
        };
        let Ok(parts) = self.store.parts(last.id.clone()).await else {
            return String::new();
        };
        parts
            .iter()
            .filter_map(|p| match &p.body {
                PartBody::Text { text, .. } => Some(text.as_str()),
                _ => None,
            })
            .collect::<Vec<_>>()
            .join("")
    }

    /// Builds a child session's step context (permission intersection + its own board).
    #[allow(clippy::too_many_arguments)]
    fn child_ctx(
        &self,
        session_id: SessionId,
        tools: ToolRegistry,
        static_rules: Ruleset,
        parent_rules: Ruleset,
        extra_rules: Ruleset,
        board: Arc<JobBoard>,
        reporter: Option<Arc<dyn ContextReporter>>,
        cancel: CancellationToken,
        now: i64,
    ) -> StepContext {
        StepContext {
            store: self.store.clone(),
            bus: self.bus.clone(),
            tools,
            permissions: self.permissions.clone(),
            static_rules,
            extra_rules: Arc::new(Mutex::new(extra_rules)),
            parent_rules,
            session_id: session_id.clone(),
            cwd: self.cwd.clone(),
            data_dir: self.data_dir.join(session_id.to_string()),
            cancel,
            now,
            spawner: Some(self.arc()),
            job_board: Some(board),
            context_reporter: reporter,
        }
    }
}

/// Reports a child session's file reads to its job on the parent board.
struct BoardReporter {
    board: Arc<JobBoard>,
    task_id: String,
}

#[async_trait]
impl ContextReporter for BoardReporter {
    async fn report_file(&self, path: String, lines: u32) {
        let _ = self
            .board
            .report_context_file(&self.task_id, path, lines, now_ms())
            .await;
    }
}

#[async_trait]
impl SubagentSpawner for EngineInner {
    async fn spawn(&self, req: SpawnRequest) -> Result<SpawnOutcome, SpawnError> {
        let other = |e: StoreError| SpawnError::Other(e.to_string());

        // 1. Load the parent and enforce the depth limit.
        let parent = self
            .store
            .session(req.parent_session_id.clone())
            .await
            .map_err(other)?
            .ok_or_else(|| SpawnError::Other("parent session not found".into()))?;
        let child_depth = parent.depth + 1;
        if child_depth as u32 > self.config.orchestration.depth_limit {
            return Err(SpawnError::DepthExceeded);
        }

        // 2. Resolve the agent — must be usable as a subagent.
        let agent = self
            .agents
            .get(&req.agent)
            .filter(|a| a.mode.is_subagent())
            .cloned()
            .ok_or_else(|| SpawnError::InvalidAgent(req.agent.clone()))?;

        // Parent-effective rules for permission intersection (frozen snapshot).
        let parent_agent = self.agent_def(&parent.agent);
        let parent_rules = self.parent_effective_rules(&parent, &parent_agent);

        let now = now_ms();
        let board = self
            .board_for(&req.parent_session_id)
            .await
            .map_err(other)?;

        // 3. Reuse a completed child session by alias/id, or create a fresh one.
        let (child_session, task_id, reused) = if let Some(alias) = &req.reuse_task_id {
            let job = board
                .resolve_reusable(&req.parent_session_id, alias)
                .ok_or_else(|| SpawnError::ReuseNotFound(alias.clone()))?;
            let child = self
                .store
                .session(job.child_session.clone())
                .await
                .map_err(other)?
                .ok_or_else(|| SpawnError::ReuseNotFound(alias.clone()))?;
            let _ = board.touch(&job.task_id, now).await;
            (child, job.task_id, true)
        } else {
            let mut child = Session::new_child(&parent, req.agent.clone(), now);
            if let Some(model) = &agent.model {
                child.model = model.clone();
            }
            self.store
                .upsert_session(child.clone())
                .await
                .map_err(other)?;
            self.bus.publish(AppEvent::SessionCreated {
                session: child.clone(),
            });
            let task_id = child.id.to_string();
            (child, task_id, false)
        };
        let child_id = child_session.id.clone();

        // 4. Append the instruction as the child's next user turn.
        self.append_user_message(&child_id, req.prompt.clone(), now)
            .await
            .map_err(other)?;

        // 5. Resolve the provider for the child model.
        let provider = self
            .providers
            .get(&child_session.model.provider_id)
            .ok_or_else(|| {
                SpawnError::Other(format!(
                    "no provider registered for {:?}",
                    child_session.model.provider_id
                ))
            })?;

        let child_static = self.effective_static_rules(&agent, child_depth);
        let child_tools = self.filtered_tools(&agent);
        let child_board = self.board_for(&child_id).await.map_err(other)?;
        let run_config = RunConfig {
            agent_name: agent.name.clone(),
            agent_prompt: self.agent_prompt(&agent),
            model: child_session.model.clone(),
            temperature: agent.temperature,
            max_steps: agent.max_steps.unwrap_or(DEFAULT_MAX_STEPS),
            instructions: self.config.instructions.clone(),
            cost: Some(self.catalog.cost(
                &child_session.model.provider_id,
                &child_session.model.model_id,
            )),
            inject_job_board: agent.mode.is_primary(),
            reminder_turn_start: self.reminder("turn_start"),
            reminder_after_file_tool: self.reminder("after_file_tool"),
        };

        // Register the launch on the parent board (also makes foreground results reusable).
        if !reused {
            let _ = board
                .register_launch(
                    LaunchSpec {
                        task_id: task_id.clone(),
                        parent_session: req.parent_session_id.clone(),
                        child_session: child_id.clone(),
                        agent: req.agent.clone(),
                        description: req.description.clone(),
                        objective: Some(req.description.clone()),
                    },
                    now,
                )
                .await;
        }

        if req.background {
            // Child of the parent session's *run* token so the job dies with the session, not
            // when the (already-returned) tool call completes.
            let cancel = {
                let runs = self.runs.lock().unwrap();
                runs.get(&req.parent_session_id)
                    .map(|r| r.cancel.child_token())
            }
            .unwrap_or_default();

            let reporter: Arc<dyn ContextReporter> = Arc::new(BoardReporter {
                board: board.clone(),
                task_id: task_id.clone(),
            });
            let ctx = self.child_ctx(
                child_id.clone(),
                child_tools,
                child_static,
                parent_rules,
                child_session.extra_rules.clone(),
                child_board,
                Some(reporter),
                cancel,
                now,
            );
            let this = self.arc();
            let board_bg = board.clone();
            let task_id_bg = task_id.clone();
            let child_id_bg = child_id.clone();
            tokio::spawn(async move {
                let outcome = run_session(provider, ctx, &run_config, now_ms).await;
                let (state, summary) = match &outcome {
                    RunOutcome::Stopped => (
                        JobState::Completed,
                        Some(this.session_final_text(&child_id_bg).await),
                    ),
                    RunOutcome::Aborted => (JobState::Cancelled, None),
                    RunOutcome::Errored { message } => (JobState::Error, Some(message.clone())),
                };
                let _ = board_bg.finish(&task_id_bg, state, summary, now_ms()).await;
            });

            let alias = board
                .snapshot()
                .into_iter()
                .find(|j| j.task_id == task_id)
                .map(|j| j.alias);
            Ok(SpawnOutcome {
                child_session_id: child_id,
                background: true,
                alias,
                final_text: None,
            })
        } else {
            // Foreground: run to completion under the tool call's token and return the text.
            let reporter: Arc<dyn ContextReporter> = Arc::new(BoardReporter {
                board: board.clone(),
                task_id: task_id.clone(),
            });
            let ctx = self.child_ctx(
                child_id.clone(),
                child_tools,
                child_static,
                parent_rules,
                child_session.extra_rules.clone(),
                child_board,
                Some(reporter),
                req.cancel.clone(),
                now,
            );
            let outcome = run_session(provider, ctx, &run_config, now_ms).await;
            let final_text = self.session_final_text(&child_id).await;
            let (state, summary) = match &outcome {
                RunOutcome::Stopped => (JobState::Completed, Some(final_text.clone())),
                RunOutcome::Aborted => (JobState::Cancelled, None),
                RunOutcome::Errored { message } => (JobState::Error, Some(message.clone())),
            };
            let _ = board.finish(&task_id, state, summary, now_ms()).await;
            Ok(SpawnOutcome {
                child_session_id: child_id,
                background: false,
                alias: None,
                final_text: Some(final_text),
            })
        }
    }
}

pub struct App {
    engine: EngineHandle,
    // Keeps the auto-approve task alive for the lifetime of the App.
    _auto_approve_handle: Option<JoinHandle<()>>,
}

impl App {
    /// Persistent SQLite-backed store with an auto-approve permission frontend — the default
    /// headless configuration used by `harness run -p`.
    pub fn init(cwd: PathBuf) -> Result<Self, AppError> {
        let engine = EngineHandle::init(cwd)?;
        let _auto_approve_handle = Some(spawn_auto_approve_task(&engine));
        Ok(Self {
            engine,
            _auto_approve_handle,
        })
    }

    /// In-memory store with an auto-approve permission frontend — useful for tests.
    pub fn init_in_memory(cwd: PathBuf) -> Result<Self, AppError> {
        let engine = EngineHandle::init_in_memory(cwd)?;
        let _auto_approve_handle = Some(spawn_auto_approve_task(&engine));
        Ok(Self {
            engine,
            _auto_approve_handle,
        })
    }

    pub fn engine(&self) -> &EngineHandle {
        &self.engine
    }

    pub fn config(&self) -> Config {
        self.engine.config()
    }

    /// Runs a single headless turn: creates a root session, appends `prompt` as the user
    /// message, and drives the engine loop to completion. Returns the outcome plus the
    /// session id so the caller can fetch the transcript via `final_text`.
    pub async fn run_prompt(
        &self,
        prompt: String,
        model_ref: &str,
    ) -> Result<(RunOutcome, SessionId), AppError> {
        let session_id = self.engine.new_session("orchestrator", model_ref).await?;
        let mut rx = self.engine.bus().subscribe();
        // Subscribe before starting the run so we cannot miss the RunFinished event
        // even for an instant (mock) provider.
        self.engine
            .prompt(session_id.clone(), prompt, model_ref)
            .await?;

        while let Ok(event) = rx.recv().await {
            if let AppEvent::RunFinished {
                session_id: sid,
                outcome,
            } = event
            {
                if sid == session_id {
                    return Ok((outcome, session_id));
                }
            }
        }

        // Bus closed unexpectedly; report a stopped run so the caller can continue.
        Ok((RunOutcome::Stopped, session_id))
    }

    /// Concatenates the `Text` parts of the last message in the session — the final
    /// assistant reply for a `harness run` invocation to print.
    pub async fn final_text(&self, session_id: &SessionId) -> Result<String, AppError> {
        self.engine.final_text(session_id).await
    }
}

fn spawn_auto_approve_task(engine: &EngineHandle) -> JoinHandle<()> {
    let bus = engine.bus();
    let permissions = engine.permissions();
    tokio::spawn(async move {
        let mut rx = bus.subscribe();
        while let Ok(event) = rx.recv().await {
            if let AppEvent::PermissionAsked { request } = event {
                permissions.reply(&request.id, PermissionReply::Once);
            }
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    use async_trait::async_trait;
    use harness_core::llm::{
        FinishReason, LlmEvent, LlmEventStream, LlmRequest, Provider, ProviderError,
    };
    use harness_core::types::{ModelInfo, TokenUsage};

    /// Emits a single final-text turn for every stream call — enough to drive a child
    /// subagent session to completion deterministically.
    struct MockProvider;

    #[async_trait]
    impl Provider for MockProvider {
        fn id(&self) -> &str {
            "mock"
        }
        async fn list_models(&self) -> Result<Vec<ModelInfo>, ProviderError> {
            Ok(vec![])
        }
        async fn stream(
            &self,
            _req: LlmRequest,
            _cancel: CancellationToken,
        ) -> Result<LlmEventStream, ProviderError> {
            let events = vec![
                Ok(LlmEvent::TextStart { id: "t".into() }),
                Ok(LlmEvent::TextDelta {
                    id: "t".into(),
                    text: "explored the code".into(),
                }),
                Ok(LlmEvent::TextEnd { id: "t".into() }),
                Ok(LlmEvent::Finish {
                    reason: FinishReason::Stop,
                    usage: TokenUsage::default(),
                }),
            ];
            Ok(Box::pin(futures::stream::iter(events)))
        }
    }

    fn mock_engine(cwd: PathBuf) -> EngineHandle {
        let mut providers = ProviderRegistry::new();
        providers.register(Arc::new(MockProvider));
        EngineHandle::build(
            cwd,
            Store::open_in_memory().unwrap(),
            Config::default(),
            providers,
        )
        .unwrap()
    }

    /// Drives a subagent through one `read` tool call, then a final answer.
    struct ReadThenAnswerProvider {
        steps: std::sync::Mutex<std::collections::VecDeque<Vec<Result<LlmEvent, ProviderError>>>>,
    }

    #[async_trait]
    impl Provider for ReadThenAnswerProvider {
        fn id(&self) -> &str {
            "mock"
        }
        async fn list_models(&self) -> Result<Vec<ModelInfo>, ProviderError> {
            Ok(vec![])
        }
        async fn stream(
            &self,
            _req: LlmRequest,
            _cancel: CancellationToken,
        ) -> Result<LlmEventStream, ProviderError> {
            let events = self.steps.lock().unwrap().pop_front().unwrap_or_else(|| {
                vec![
                    Ok(LlmEvent::TextStart { id: "d".into() }),
                    Ok(LlmEvent::TextDelta {
                        id: "d".into(),
                        text: "done".into(),
                    }),
                    Ok(LlmEvent::TextEnd { id: "d".into() }),
                    Ok(LlmEvent::Finish {
                        reason: FinishReason::Stop,
                        usage: TokenUsage::default(),
                    }),
                ]
            });
            Ok(Box::pin(futures::stream::iter(events)))
        }
    }

    #[tokio::test]
    async fn subagent_file_reads_land_on_the_job_board() {
        let dir = tempfile::tempdir().unwrap();
        // A file with ≥10 lines so the read clears the board's reporting threshold.
        let body = (1..=15)
            .map(|n| format!("line {n}"))
            .collect::<Vec<_>>()
            .join("\n");
        std::fs::write(dir.path().join("auth.rs"), body).unwrap();

        let provider = Arc::new(ReadThenAnswerProvider {
            steps: std::sync::Mutex::new(
                vec![
                    // Turn 1: call the read tool.
                    vec![
                        Ok(LlmEvent::ToolCall {
                            call_id: "c1".into(),
                            name: "read".into(),
                            input: serde_json::json!({"file_path": "auth.rs"}),
                        }),
                        Ok(LlmEvent::Finish {
                            reason: FinishReason::ToolCalls,
                            usage: TokenUsage::default(),
                        }),
                    ],
                    // Turn 2: final answer.
                    vec![
                        Ok(LlmEvent::TextStart { id: "t".into() }),
                        Ok(LlmEvent::TextDelta {
                            id: "t".into(),
                            text: "found it".into(),
                        }),
                        Ok(LlmEvent::TextEnd { id: "t".into() }),
                        Ok(LlmEvent::Finish {
                            reason: FinishReason::Stop,
                            usage: TokenUsage::default(),
                        }),
                    ],
                ]
                .into(),
            ),
        });
        let mut providers = ProviderRegistry::new();
        providers.register(provider);
        let engine = EngineHandle::build(
            dir.path().to_path_buf(),
            Store::open_in_memory().unwrap(),
            Config::default(),
            providers,
        )
        .unwrap();
        // Auto-approve permission asks (the child's read tool gates on `read`).
        spawn_auto_approve_task(&engine);

        let root = engine
            .new_session("orchestrator", "mock/mock-model")
            .await
            .unwrap();
        let out = engine
            .inner
            .spawn(SpawnRequest {
                parent_session_id: root.clone(),
                parent_message_id: MessageId::new(),
                agent: "explorer".into(),
                description: "read auth".into(),
                prompt: "read auth.rs".into(),
                reuse_task_id: None,
                background: false,
                cancel: CancellationToken::new(),
            })
            .await
            .unwrap();

        let board = engine.inner.board_for(&root).await.unwrap();
        let job = board
            .snapshot()
            .into_iter()
            .find(|j| j.child_session == out.child_session_id)
            .expect("job on board");
        assert!(
            job.context_files.iter().any(|f| f.path == "auth.rs"),
            "expected auth.rs on the board, got {:?}",
            job.context_files
        );
    }

    #[tokio::test]
    async fn foreground_task_runs_a_subagent_and_alias_reuse_continues_the_same_session() {
        let dir = tempfile::tempdir().unwrap();
        let engine = mock_engine(dir.path().to_path_buf());
        let root = engine
            .new_session("orchestrator", "mock/mock-model")
            .await
            .unwrap();

        // Foreground spawn of the explorer subagent.
        let out = engine
            .inner
            .spawn(SpawnRequest {
                parent_session_id: root.clone(),
                parent_message_id: MessageId::new(),
                agent: "explorer".into(),
                description: "map the auth flow".into(),
                prompt: "find the login handler".into(),
                reuse_task_id: None,
                background: false,
                cancel: CancellationToken::new(),
            })
            .await
            .unwrap();
        assert!(!out.background);
        assert_eq!(out.final_text.as_deref(), Some("explored the code"));
        let child_id = out.child_session_id.clone();

        // The completed job is on the parent board, reusable by alias.
        let board = engine.inner.board_for(&root).await.unwrap();
        let job = board
            .snapshot()
            .into_iter()
            .find(|j| j.child_session == child_id)
            .expect("job registered");
        assert_eq!(job.alias, "exp-1");
        assert_eq!(job.state, JobState::Completed);

        let child_messages_before = engine
            .inner
            .store
            .messages(child_id.clone())
            .await
            .unwrap()
            .len();

        // Reuse by alias → same child session, with the new prompt appended.
        let reused = engine
            .inner
            .spawn(SpawnRequest {
                parent_session_id: root.clone(),
                parent_message_id: MessageId::new(),
                agent: "explorer".into(),
                description: "follow up".into(),
                prompt: "now find the logout handler".into(),
                reuse_task_id: Some("exp-1".into()),
                background: false,
                cancel: CancellationToken::new(),
            })
            .await
            .unwrap();
        assert_eq!(
            reused.child_session_id, child_id,
            "reuse must continue the same child session"
        );

        let child_messages_after = engine.inner.store.messages(child_id).await.unwrap().len();
        assert!(
            child_messages_after > child_messages_before,
            "reused session should have the follow-up turn appended ({child_messages_before} -> {child_messages_after})"
        );
    }

    #[tokio::test]
    async fn depth_limit_is_enforced() {
        let dir = tempfile::tempdir().unwrap();
        let engine = mock_engine(dir.path().to_path_buf());
        // Build a session already at the depth limit so any child would exceed it.
        let now = now_ms();
        let mut deep = Session::new_root("explorer", ModelRef::new("mock", "mock-model"), now);
        deep.depth = engine.inner.config.orchestration.depth_limit as u8;
        engine
            .inner
            .store
            .upsert_session(deep.clone())
            .await
            .unwrap();

        let err = engine
            .inner
            .spawn(SpawnRequest {
                parent_session_id: deep.id.clone(),
                parent_message_id: MessageId::new(),
                agent: "explorer".into(),
                description: "too deep".into(),
                prompt: "go deeper".into(),
                reuse_task_id: None,
                background: false,
                cancel: CancellationToken::new(),
            })
            .await
            .unwrap_err();
        assert!(matches!(err, SpawnError::DepthExceeded));
    }

    #[tokio::test]
    async fn unknown_subagent_is_rejected() {
        let dir = tempfile::tempdir().unwrap();
        let engine = mock_engine(dir.path().to_path_buf());
        let root = engine
            .new_session("orchestrator", "mock/mock-model")
            .await
            .unwrap();
        let err = engine
            .inner
            .spawn(SpawnRequest {
                parent_session_id: root,
                parent_message_id: MessageId::new(),
                agent: "nonexistent".into(),
                description: "x".into(),
                prompt: "y".into(),
                reuse_task_id: None,
                background: false,
                cancel: CancellationToken::new(),
            })
            .await
            .unwrap_err();
        assert!(matches!(err, SpawnError::InvalidAgent(_)));
    }

    #[tokio::test]
    async fn init_loads_defaults_with_no_providers_configured() {
        let dir = tempfile::tempdir().unwrap();
        std::env::remove_var("ANTHROPIC_API_KEY");
        let app = App::init_in_memory(dir.path().to_path_buf()).unwrap();
        assert!(app.engine.inner.providers.get("anthropic").is_none());
        // 6 built-ins + the multiagent `task` tool.
        assert_eq!(app.engine.inner.tools.all().len(), 7);
    }

    #[tokio::test]
    async fn run_prompt_errors_on_unregistered_provider() {
        let dir = tempfile::tempdir().unwrap();
        std::env::remove_var("ANTHROPIC_API_KEY");
        let app = App::init_in_memory(dir.path().to_path_buf()).unwrap();
        let err = app
            .run_prompt("hi".into(), "anthropic/claude-sonnet-4-5")
            .await
            .unwrap_err();
        assert!(matches!(err, AppError::UnknownProvider(_)));
    }

    #[tokio::test]
    async fn run_prompt_errors_on_malformed_model_ref() {
        let dir = tempfile::tempdir().unwrap();
        let app = App::init_in_memory(dir.path().to_path_buf()).unwrap();
        let err = app.run_prompt("hi".into(), "no-slash").await.unwrap_err();
        assert!(matches!(err, AppError::InvalidModelRef(_)));
    }

    #[tokio::test]
    async fn final_text_is_empty_for_unknown_session() {
        let dir = tempfile::tempdir().unwrap();
        let app = App::init_in_memory(dir.path().to_path_buf()).unwrap();
        let text = app.final_text(&SessionId::new()).await.unwrap();
        assert_eq!(text, "");
    }

    #[tokio::test]
    async fn engine_init_in_memory_creates_handle() {
        let dir = tempfile::tempdir().unwrap();
        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
        assert!(!engine.config().providers.contains_key("anthropic"));
    }

    #[tokio::test]
    async fn engine_new_session_persists_session() {
        let dir = tempfile::tempdir().unwrap();
        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
        let id = engine
            .new_session("orchestrator", "anthropic/claude")
            .await
            .unwrap();
        let sessions = engine.list_sessions().await.unwrap();
        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions[0].id, id);
        assert_eq!(sessions[0].agent, "orchestrator");
    }

    #[tokio::test]
    async fn engine_list_sessions_returns_persisted_sessions() {
        let dir = tempfile::tempdir().unwrap();
        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
        let id1 = engine.new_session("a", "anthropic/claude").await.unwrap();
        let id2 = engine.new_session("b", "anthropic/claude").await.unwrap();
        let sessions = engine.list_sessions().await.unwrap();
        assert_eq!(sessions.len(), 2);
        assert!(sessions.iter().any(|s| s.id == id1));
        assert!(sessions.iter().any(|s| s.id == id2));
    }

    #[tokio::test]
    async fn engine_prompt_errors_on_unregistered_provider() {
        let dir = tempfile::tempdir().unwrap();
        std::env::remove_var("ANTHROPIC_API_KEY");
        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
        let id = engine
            .new_session("orchestrator", "anthropic/claude")
            .await
            .unwrap();
        let err = engine
            .prompt(id, "hi".into(), "anthropic/claude")
            .await
            .unwrap_err();
        assert!(matches!(err, AppError::UnknownProvider(_)));
    }

    #[tokio::test]
    async fn engine_prompt_errors_on_malformed_model_ref() {
        let dir = tempfile::tempdir().unwrap();
        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
        let id = engine
            .new_session("orchestrator", "anthropic/claude")
            .await
            .unwrap();
        let err = engine
            .prompt(id, "hi".into(), "no-slash")
            .await
            .unwrap_err();
        assert!(matches!(err, AppError::InvalidModelRef(_)));
    }

    #[tokio::test]
    async fn engine_is_running_false_for_inactive_session() {
        let dir = tempfile::tempdir().unwrap();
        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
        let id = engine
            .new_session("orchestrator", "anthropic/claude")
            .await
            .unwrap();
        assert!(!engine.is_running(&id));
    }
}