kowalski-cli 2.2.0

//! Local **markdown-staged app** CLI (`app.md` / `horde.md` + `agents/*.md`).
//! Execution rules live in [`kowalski_core::markdown_pipeline`]; this module wires HTTP chat,
//! federation helpers, and filesystem paths.

use kowalski_core::{
    has_conditional_outbound, is_loop_back_step, loop_edge_key,
    next_ready_step_conditional, parse_stage_status_from_artifact, resolve_execution_graph,
    retry_span, select_next_from_outcome, single_forward_predecessor, StageStatus,
    validate_horde_tree_report,
};
use kowalski_core::markdown_pipeline::{
    maybe_normalize_markdown, parse_app_manifest, parse_stage_agent, render_context_attachments,
    resolve_manifest_path, AppManifestMeta, StageAgentMeta,
};
use kowalski_core::source_bundle::{extract_project_path_from_source, ingest_assets_markdown};
use kowalski_core::{
    apply_patches_dry_run, format_apply_artifact, format_verify_artifact, resolve_verify_cwd,
    run_verify_command, DEFAULT_VERIFY_MAX_OUTPUT_BYTES, DEFAULT_VERIFY_TIMEOUT_SECS,
};
use kowalski_core::{
    execute_isolated_request, IsolatedStepEvent, IsolatedStepRequest, IsolatedStepResponse,
    LlmStepHandler, StepEventSink, StepHandlerRegistry, StepSpec, LLM_STEP_KINDS,
};
use kowalski_core::template::agent::TemplateAgent;
use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use reqwest::blocking as reqwest_blocking;
use std::collections::BTreeMap;
use std::fs;
use std::io::BufRead;
use std::path::{Path, PathBuf};

/// Intermediate artifacts (ingest captures, per-stage LLM outputs) live here — for monitoring / debugging only.
const ARTIFACTS_DEBUG_DIR: &str = "debug";
/// Single operator-facing markdown at the `workdir` (or app run) root.
const PASTE_ME_FILENAME: &str = "PASTE_ME.md";

#[inline]
fn work_debug(workdir: &Path) -> PathBuf {
    workdir.join(ARTIFACTS_DEBUG_DIR)
}

/// Default workdir for local `agent-app run` when `horde.md` uses a relative `workdir` (often `output/`).
const LOCAL_AGENT_APP_WORKDIR: &str = "output";

#[inline]
fn local_agent_app_workdir(app_root: &Path) -> PathBuf {
    app_root.join(LOCAL_AGENT_APP_WORKDIR)
}

#[derive(Default, Clone)]
struct RunArtifacts {
    summary: Option<PathBuf>,
    report: Option<PathBuf>,
    handoff: Option<PathBuf>,
    log: Option<PathBuf>,
}

#[derive(Debug)]
struct AgentDoc<T> {
    meta: T,
    path: PathBuf,
}

type AgentSpecMap = BTreeMap<String, AgentDoc<StageAgentMeta>>;

fn app_root(path: Option<&str>) -> PathBuf {
    path.map(PathBuf::from)
        .or_else(|| std::env::var("KOWALSKI_AGENT_APP_ROOT").ok().map(PathBuf::from))
        .unwrap_or_else(|| PathBuf::from("examples/knowledge-compiler"))
}

fn agents_dir(root: &Path) -> PathBuf {
    root.join("agents")
}

fn load_spec(
    root: &Path,
) -> Result<(AgentDoc<AppManifestMeta>, AgentSpecMap), Box<dyn std::error::Error>> {
    let mpath = resolve_manifest_path(root);
    if !mpath.is_file() {
        return Err(format!(
            "missing manifest (tried {} and {}), expected next to agents/",
            root.join("app.md").display(),
            root.join("horde.md").display()
        )
        .into());
    }
    let meta = parse_app_manifest(&mpath).map_err(|e| e.to_string())?;
    let main = AgentDoc {
        meta,
        path: mpath,
    };
    let mut map = BTreeMap::new();
    for entry in fs::read_dir(agents_dir(root))? {
        let path = entry?.path();
        if path.extension().and_then(|x| x.to_str()) != Some("md") {
            continue;
        }
        let sm = parse_stage_agent(&path).map_err(|e| e.to_string())?;
        map.insert(
            sm.name.clone(),
            AgentDoc {
                meta: sm,
                path: path.to_path_buf(),
            },
        );
    }
    Ok((main, map))
}

pub fn list_agents(path: Option<&str>) -> Result<(), Box<dyn std::error::Error>> {
    let root = app_root(path);
    let (main, agents) = load_spec(&root)?;
    let title = main
        .meta
        .display_name
        .as_deref()
        .unwrap_or(&main.meta.id);
    println!("App: {} ({})", title, main.meta.id);
    println!("Pipeline: {}", main.meta.pipeline.join(" -> "));
    println!("Pipeline agents:");
    for step in &main.meta.pipeline {
        if let Some(agent) = agents.get(step) {
            println!("- {} ({})", step, agent.meta.kind);
        } else {
            println!("- {} (missing agents/{step}.md)", step);
        }
    }
    for name in agents.keys() {
        if !main.meta.pipeline.contains(name)
            && let Some(agent) = agents.get(name)
        {
            println!(
                "- {} ({}) [not in manifest pipeline — remove or add to pipeline]",
                name, agent.meta.kind
            );
        }
    }
    Ok(())
}

pub fn validate(path: Option<&str>) -> Result<(), Box<dyn std::error::Error>> {
    let root = app_root(path);
    let warnings = validate_horde_tree_report(&root).map_err(|e| e.to_string())?;
    println!("OK - manifest + agents/ definitions are valid");
    let meta = parse_app_manifest(&resolve_manifest_path(&root))?;
    if !meta.triggers.is_empty() {
        println!("triggers:");
        for t in &meta.triggers {
            println!("- {}", t.summary());
        }
    }
    for w in &warnings {
        println!("warning: {w}");
    }
    Ok(())
}

#[derive(Debug, Clone, Default)]
struct StageChatOptions {
    use_tools: bool,
    tool_ids: Vec<String>,
    project_path: Option<String>,
    conversation_id: Option<String>,
}

fn chat_stage(
    api: &str,
    prompt: &str,
    opts: &StageChatOptions,
) -> Result<String, Box<dyn std::error::Error>> {
    let client = reqwest_blocking::Client::builder()
        .timeout(std::time::Duration::from_secs(300))
        .build()?;
    let route = "/api/chat";
    let url = format!("{}{}", api.trim_end_matches('/'), route);
    let use_tools = opts.use_tools && !opts.tool_ids.is_empty();
    let mut body = serde_json::json!({
        "message": prompt,
        "use_memory": false,
        "use_tools": use_tools,
    });
    if use_tools {
        body["tool_ids"] = serde_json::json!(opts.tool_ids);
        if let Some(root) = opts.project_path.as_ref().filter(|s| !s.trim().is_empty()) {
            body["sandbox_root"] = serde_json::json!(root);
        }
    }
    if let Some(cid) = opts.conversation_id.as_ref().filter(|s| !s.trim().is_empty()) {
        body["conversation_id"] = serde_json::json!(cid);
    }
    let resp = with_bearer(client.post(&url))
        .json(&body)
        .send()
        .map_err(|e| friendly_http_error(api, route, &url, &e))?;
    let status = resp.status();
    let body_text = resp.text().unwrap_or_default();
    if !status.is_success() {
        let detail = body_text.trim();
        let body_json = if detail.is_empty() {
            None
        } else {
            serde_json::from_str::<serde_json::Value>(detail)
                .ok()
                .or_else(|| Some(serde_json::json!({ "detail": detail })))
        };
        return Err(
            friendly_http_status_error(api, route, &url, status.as_u16(), body_json.as_ref())
                .into(),
        );
    }
    let v: serde_json::Value = serde_json::from_str(&body_text).map_err(|e| {
        format!(
            "response for {} ({}) was HTTP {} but not valid JSON: {}; body prefix: {:.200}",
            route, url, status.as_u16(), e, body_text
        )
    })?;
    Ok(v.get("reply")
        .and_then(|x| x.as_str())
        .unwrap_or("")
        .trim()
        .to_string())
}

fn read_or_empty(path: &Path) -> String {
    fs::read_to_string(path).unwrap_or_default()
}

fn ensure_dirs(root: &Path) -> Result<(), Box<dyn std::error::Error>> {
    fs::create_dir_all(root)?;
    let b = work_debug(root);
    for rel in [
        "raw",
        "raw/images",
        "reports",
        "slides",
        "lint",
        "scratch",
    ] {
        fs::create_dir_all(b.join(rel))?;
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn run_llm_stage(
    api: &str,
    app_root: &Path,
    workdir: &Path,
    agent: &AgentDoc<StageAgentMeta>,
    step: &str,
    extra_user_block: &str,
    step_paths: &BTreeMap<String, PathBuf>,
    previous_artifact: Option<&Path>,
    chat_opts: &StageChatOptions,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
    let rel = agent
        .meta
        .output
        .as_deref()
        .ok_or_else(|| format!("stage `{step}` missing `output` in {}", agent.path.display()))?;
    // A trailing-slash `output` declares a directory: write `<dir>/<step>.md` inside it.
    let out_path = if rel.ends_with('/') {
        let dir = workdir.join(rel);
        fs::create_dir_all(&dir)?;
        dir.join(format!("{step}.md"))
    } else {
        let path = workdir.join(rel);
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent)?;
        }
        path
    };
    let prompt_path = app_root.join(
        agent
            .meta
            .prompt_file
            .as_deref()
            .unwrap_or("prompts/stage.md"),
    );
    let prompt = read_or_empty(&prompt_path);
    let ctx = render_context_attachments(
        workdir,
        &agent.meta.context_paths,
        step_paths,
        previous_artifact,
    )
    .map_err(|e| -> Box<dyn std::error::Error> { e.to_string().into() })?;
    let msg = if extra_user_block.trim().is_empty() {
        format!("{prompt}\n\n{ctx}")
    } else {
        format!("{prompt}\n\n{extra_user_block}\n\n{ctx}")
    };
    let reply = chat_stage(api, &msg, chat_opts)?;
    let reply = maybe_normalize_markdown(&agent.meta, &reply);
    fs::write(&out_path, reply)?;
    Ok(out_path)
}

fn resolve_stage_project_path(
    project_path: Option<String>,
    source: Option<&str>,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
    if let Some(p) = project_path {
        let path = PathBuf::from(p.trim());
        if path.is_dir() {
            return Ok(path);
        }
    }
    if let Some(s) = source
        && let Some(p) = extract_project_path_from_source(s)
    {
        return Ok(p);
    }
    Err("verify/apply: missing operator project_path (set on ingest form)".into())
}

fn run_verify_stage(
    _workspace_root: &Path,
    workdir: &Path,
    agent: &AgentDoc<StageAgentMeta>,
    step: &str,
    project_path: Option<String>,
    source: Option<&str>,
    _previous_artifact: Option<&Path>,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
    ensure_dirs(workdir)?;
    let command = agent
        .meta
        .verify_command
        .as_deref()
        .filter(|s| !s.trim().is_empty())
        .ok_or_else(|| format!("verify stage `{step}` missing `verify_command` in {}", agent.path.display()))?;
    let project = resolve_stage_project_path(project_path, source)?;
    let cwd = resolve_verify_cwd(&project, agent.meta.verify_cwd.as_deref())
        .map_err(|e| e.to_string())?;
    let rel = agent
        .meta
        .output
        .as_deref()
        .ok_or_else(|| format!("verify stage `{step}` missing `output`"))?;
    let out_path = workdir.join(rel);
    if let Some(parent) = out_path.parent() {
        fs::create_dir_all(parent)?;
    }
    let result = run_verify_command(
        command,
        &cwd,
        DEFAULT_VERIFY_MAX_OUTPUT_BYTES,
        Duration::from_secs(DEFAULT_VERIFY_TIMEOUT_SECS),
    );
    let doc = format_verify_artifact(&result);
    fs::write(&out_path, doc)?;
    Ok(out_path)
}

fn run_apply_stage(
    _workspace_root: &Path,
    workdir: &Path,
    agent: &AgentDoc<StageAgentMeta>,
    step: &str,
    project_path: Option<String>,
    source: Option<&str>,
    previous_artifact: Option<&Path>,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
    ensure_dirs(workdir)?;
    let prev = previous_artifact
        .ok_or("apply: missing previous_artifact from orchestrator")?;
    let prev_body = fs::read_to_string(prev)?;
    let project = resolve_stage_project_path(project_path, source)?;
    let mode = agent
        .meta
        .apply_mode
        .as_deref()
        .unwrap_or("dry-run");
    let execute = mode.eq_ignore_ascii_case("execute")
        && std::env::var("KOWALSKI_HORDE_APPLY")
            .ok()
            .is_some_and(|v| v == "1" || v.eq_ignore_ascii_case("true"));
    let dry = apply_patches_dry_run(&project, &prev_body);
    let rel = agent
        .meta
        .output
        .as_deref()
        .ok_or_else(|| format!("apply stage `{step}` missing `output`"))?;
    let out_path = workdir.join(rel);
    if let Some(parent) = out_path.parent() {
        fs::create_dir_all(parent)?;
    }
    let doc = format_apply_artifact(mode, &dry, execute);
    fs::write(&out_path, doc)?;
    Ok(out_path)
}

fn stage_chat_options(
    agent: &StageAgentMeta,
    project_path: Option<String>,
    run_id: Option<&str>,
    step: &str,
    instr_tool_ids: &[String],
) -> StageChatOptions {
    let tool_ids = if agent.tool_ids.is_empty() {
        instr_tool_ids.to_vec()
    } else {
        agent.tool_ids.clone()
    };
    StageChatOptions {
        use_tools: !tool_ids.is_empty(),
        tool_ids,
        project_path,
        conversation_id: run_id.map(|r| format!("horde-{r}-{step}")),
    }
}

fn run_with_progress<F>(
    path: Option<&str>,
    source: &str,
    question: Option<&str>,
    api_url: Option<&str>,
    mut on_step: F,
) -> Result<RunArtifacts, Box<dyn std::error::Error>>
where
    F: FnMut(&str, &str, &Path),
{
    validate(path)?;
    let root = app_root(path);
    let work = local_agent_app_workdir(&root);
    let (main, agents) = load_spec(&root)?;
    let api = api_base(api_url);
    let api = api.as_str();
    ensure_dirs(&work)?;
    let q = question
        .map(ToString::to_string)
        .or(main.meta.default_question.clone())
        .unwrap_or_else(|| "What changed?".to_string());

    let run_stamp = Utc::now().format("%Y%m%d-%H%M%S");
    let log_file = work_debug(&work)
        .join("scratch")
        .join(format!("orchestration-{run_stamp}.md"));
    let mut log = String::new();
    let mut task_outputs: Vec<(String, PathBuf)> = Vec::new();
    let mut artifacts = RunArtifacts::default();
    let edge_slice = if main.meta.edges.is_empty() {
        None
    } else {
        Some(main.meta.edges.as_slice())
    };
    let graph = resolve_execution_graph(&main.meta.pipeline, edge_slice)
        .map_err(|e| e.to_string())?;
    let last_step = main.meta.pipeline.last().cloned();
    let run_title = main
        .meta
        .display_name
        .as_deref()
        .unwrap_or(&main.meta.id);
    log.push_str("# Agent App Run\n\n");
    log.push_str(&format!(
        "- App: {} ({})\n- Source: {}\n- Question: {}\n\n",
        run_title, main.meta.id, source, q
    ));
    println!("Starting staged app: {} ({})", run_title, main.meta.id);

    let mut step_paths: BTreeMap<String, PathBuf> = BTreeMap::new();
    let mut step_status: BTreeMap<String, String> = main
        .meta
        .pipeline
        .iter()
        .map(|s| (s.clone(), "pending".to_string()))
        .collect();
    let mut step_outcome: BTreeMap<String, StageStatus> = BTreeMap::new();
    let mut loop_counts: BTreeMap<String, u32> = BTreeMap::new();
    let project_path = kowalski_core::source_bundle::extract_project_path_from_source(source)
        .map(|p| p.display().to_string());
    let run_id = run_stamp.to_string();
    let max_steps = main.meta.pipeline.len().saturating_mul(4).max(1);
    let mut steps_run = 0usize;

    while steps_run < max_steps {
        let status_view: BTreeMap<String, &str> = step_status
            .iter()
            .map(|(k, v)| (k.clone(), v.as_str()))
            .collect();
        let step = match next_ready_step_conditional(
            &main.meta.pipeline,
            &graph,
            &status_view,
            &step_outcome,
        ) {
            Some(s) => s,
            None => break,
        };
        steps_run += 1;
        let agent = agents
            .get(&step)
            .ok_or_else(|| format!("missing step agent: {step}"))?;
        let prev_path = single_forward_predecessor(&main.meta.pipeline, &graph, &step)
            .and_then(|pred| step_paths.get(&pred).cloned());
        println!(
            "[{}] {} ({})",
            steps_run,
            step,
            agent.meta.kind
        );
        log.push_str(&format!("## Step: {} ({})\n\n", step, agent.meta.kind));

        let out_path = if agent.meta.kind == "ingest" {
            let p = ingest_assets_markdown(&work_debug(&work), source)?;
            log.push_str(&format!("- output: {}\n\n", p.display()));
            task_outputs.push((step.clone(), p.clone()));
            on_step(&step, agent.meta.kind.as_str(), p.as_path());
            println!("  -> {}", p.display());
            p
        } else if agent.meta.kind == "verify" {
            let op = run_verify_stage(
                &root,
                &work,
                agent,
                &step,
                project_path.clone(),
                Some(source),
                prev_path.as_deref(),
            )?;
            log.push_str(&format!("- output: {}\n\n", op.display()));
            task_outputs.push((step.clone(), op.clone()));
            on_step(&step, agent.meta.kind.as_str(), op.as_path());
            println!("  -> {}", op.display());
            op
        } else if agent.meta.kind == "apply" {
            let op = run_apply_stage(
                &root,
                &work,
                agent,
                &step,
                project_path.clone(),
                Some(source),
                prev_path.as_deref(),
            )?;
            log.push_str(&format!("- output: {}\n\n", op.display()));
            task_outputs.push((step.clone(), op.clone()));
            on_step(&step, agent.meta.kind.as_str(), op.as_path());
            println!("  -> {}", op.display());
            op
        } else {
            let extra = if agent.meta.kind == "ask" {
                format!("User question:\n{q}\n")
            } else {
                String::new()
            };
            let chat_opts = stage_chat_options(
                &agent.meta,
                project_path.clone(),
                Some(&run_id),
                &step,
                &[],
            );
            let op = run_llm_stage(
                api,
                &root,
                &work,
                agent,
                &step,
                &extra,
                &step_paths,
                prev_path.as_deref(),
                &chat_opts,
            )?;
            log.push_str(&format!("- output: {}\n\n", op.display()));
            task_outputs.push((step.clone(), op.clone()));
            let is_last = Some(step.as_str()) == last_step.as_deref();
            if is_last {
                artifacts.handoff = Some(op.clone());
            } else if agent.meta.kind != "ingest" {
                if artifacts.summary.is_none() {
                    artifacts.summary = Some(op.clone());
                } else if artifacts.report.is_none() {
                    artifacts.report = Some(op.clone());
                }
            }
            on_step(&step, agent.meta.kind.as_str(), op.as_path());
            println!("  -> {}", op.display());
            op
        };

        step_status.insert(step.clone(), "success".to_string());
        let outcome = if agent.meta.kind == "verify" || agent.meta.kind == "apply" {
            fs::read_to_string(&out_path)
                .ok()
                .and_then(|b| parse_stage_status_from_artifact(&b))
                .unwrap_or(StageStatus::Fail)
        } else {
            StageStatus::Pass
        };
        step_outcome.insert(step.clone(), outcome);
        step_paths.insert(step.clone(), out_path);

        if has_conditional_outbound(&graph.edges, &step)
            && let Some(next) = select_next_from_outcome(
                &main.meta.pipeline,
                &graph.edges,
                &step,
                outcome,
                &loop_counts,
            )
            && is_loop_back_step(&main.meta.pipeline, &step, &next)
        {
            let key = loop_edge_key(&step, &next);
            let count = {
                let c = loop_counts.entry(key).or_insert(0);
                *c += 1;
                *c
            };
            for s in retry_span(&main.meta.pipeline, &next, &step) {
                step_status.insert(s.clone(), "pending".to_string());
                step_outcome.remove(&s);
                step_paths.remove(&s);
            }
            log.push_str(&format!(
                "- loop retry → `{}` (count {})\n\n",
                next, count
            ));
        }
    }

    fs::write(&log_file, log)?;
    println!("\nSub-agent execution trace:");
    for (task, out) in &task_outputs {
        println!("- {} -> {}", task, out.display());
    }

    println!("\nFinal output artifacts:");
    if let Some(p) = &artifacts.summary {
        println!("- summary stage: {}", p.display());
    }
    if let Some(p) = &artifacts.report {
        println!("- report stage: {}", p.display());
    }
    if let Some(p) = &artifacts.handoff {
        println!("- final handoff: {}", p.display());
    }
    println!("Agent app run complete. Log: {}", log_file.display());
    artifacts.log = Some(log_file);
    Ok(artifacts)
}

pub fn run(
    path: Option<&str>,
    source: &str,
    question: Option<&str>,
    api_url: Option<&str>,
) -> Result<(), Box<dyn std::error::Error>> {
    let _ = run_with_progress(path, source, question, api_url, |_step, _kind, _output| {})?;
    Ok(())
}

/// Single source of truth for the server base URL: explicit `--api` flag, else the
/// `KOWALSKI_API` env var (exported by the server to spawned workers), else
/// `http://<DEFAULT_API_BIND>`. Every CLI `/api/*` call site resolves through here.
fn api_base(api_url: Option<&str>) -> String {
    if let Some(flag) = api_url {
        let flag = flag.trim();
        if !flag.is_empty() {
            return flag.to_string();
        }
    }
    if let Ok(env) = std::env::var(kowalski_core::config::API_URL_ENV) {
        let env = env.trim().to_string();
        if !env.is_empty() {
            return env;
        }
    }
    format!("http://{}", kowalski_core::config::DEFAULT_API_BIND)
}

/// Server API token (`KOWALSKI_API_TOKEN`) — set by the operator, or inherited from the
/// `kowalski` server when it spawns this worker. The server requires it on `/api/*`.
fn api_token() -> Option<String> {
    std::env::var(kowalski_core::config::API_TOKEN_ENV)
        .ok()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
}

fn with_bearer(rb: reqwest_blocking::RequestBuilder) -> reqwest_blocking::RequestBuilder {
    match api_token() {
        Some(token) => rb.bearer_auth(token),
        None => rb,
    }
}

fn post_json(
    api: &str,
    route: &str,
    payload: serde_json::Value,
) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
    let client = reqwest_blocking::Client::builder()
        .timeout(std::time::Duration::from_secs(60))
        .build()?;
    let resp = with_bearer(client.post(format!("{}{}", api.trim_end_matches('/'), route)))
        .json(&payload)
        .send()
        .map_err(|e| {
            friendly_http_error(
                api,
                route,
                &format!("{}{}", api.trim_end_matches('/'), route),
                &e,
            )
        })?;
    let status = resp.status();
    let v: serde_json::Value = resp.json().unwrap_or_else(|_| serde_json::json!({}));
    if !status.is_success() {
        return Err(friendly_http_status_error(
            api,
            route,
            &format!("{}{}", api.trim_end_matches('/'), route),
            status.as_u16(),
            Some(&v),
        )
        .into());
    }
    Ok(v)
}

fn friendly_http_error(api: &str, route: &str, url: &str, err: &reqwest::Error) -> String {
    let mut msg = format!("request failed for {} ({}): {}", route, url, err);
    msg.push_str("\nPossible root causes:");
    if err.is_connect() {
        msg.push_str("\n- Kowalski server is not running or not reachable.");
        msg.push_str("\n- API URL is wrong.");
    } else if err.is_timeout() {
        msg.push_str("\n- Server is running but timed out.");
        msg.push_str("\n- LLM/provider backend is slow or blocked.");
    } else {
        msg.push_str("\n- Network or server-side error.");
    }
    msg.push_str("\nHow to fix:");
    msg.push_str("\n- Start server: cargo run -p kowalski --bin kowalski");
    msg.push_str(&format!(
        "\n- Verify health: curl {}/api/health",
        api.trim_end_matches('/')
    ));
    msg.push_str(&format!(
        "\n- If using custom API, set KOWALSKI_API and retry (current: {}).",
        api
    ));
    msg
}

fn friendly_http_status_error(
    api: &str,
    route: &str,
    url: &str,
    status_code: u16,
    body: Option<&serde_json::Value>,
) -> String {
    let mut msg = format!(
        "request failed for {} ({}): HTTP {}",
        route, url, status_code
    );
    if let Some(v) = body {
        if let Some(d) = v.get("detail").and_then(|x| x.as_str()).filter(|s| !s.is_empty()) {
            msg.push_str("\nServer detail:\n");
            msg.push_str(d);
        } else if *v != serde_json::json!({}) {
            msg.push_str(&format!("\nResponse body: {}", v));
        }
    }
    msg.push_str("\nPossible root causes:");
    if status_code == 404 {
        msg.push_str("\n- Endpoint is missing (version mismatch or wrong API URL).");
    } else if status_code >= 500 {
        msg.push_str("\n- Kowalski server hit an error while handling the request (see \"Server detail\" above if present).");
        msg.push_str("\n- Upstream LLM (Ollama/OpenAI) unreachable, wrong host/port, or model missing.");
        if route == "/api/chat" {
            msg.push_str("\n- For Ollama: run `ollama serve` (or the desktop app), then `curl -s http://127.0.0.1:11434/api/tags` and `ollama pull <model>` for the model in the server's `config.toml` `[llm]` / `[ollama]`.");
        }
    } else if status_code == 401 || status_code == 403 {
        msg.push_str("\n- Authentication/authorization problem.");
        msg.push_str(
            "\n- The server requires a bearer token on /api/* by default: set KOWALSKI_API_TOKEN to the token from the server's `db/api_token` file (path is logged at server startup).",
        );
    } else {
        msg.push_str("\n- Request rejected by server configuration.");
    }
    msg.push_str("\nHow to fix:");
    msg.push_str("\n- Ensure server is running: cargo run -p kowalski --bin kowalski");
    msg.push_str(&format!(
        "\n- Verify health: curl {}/api/health",
        api.trim_end_matches('/')
    ));
    msg.push_str(&format!(
        "\n- Confirm API URL and route availability (current API: {}, route: {}).",
        api, route
    ));
    msg
}

fn latest_md_in(dir: &Path) -> Option<PathBuf> {
    let mut files: Vec<PathBuf> = fs::read_dir(dir)
        .ok()?
        .filter_map(|e| e.ok().map(|x| x.path()))
        .filter(|p| p.extension().and_then(|x| x.to_str()) == Some("md"))
        .collect();
    files.sort();
    files.pop()
}

pub fn federate_delegate(
    api_url: Option<&str>,
    capability: &str,
    source: &str,
    question: Option<&str>,
) -> Result<(), Box<dyn std::error::Error>> {
    let api = api_base(api_url);
    let api = api.as_str();
    let task_id = format!("kc-{}", Utc::now().timestamp());
    let instruction = format!(
        "kc.run:{}|{}",
        source,
        question.unwrap_or("What changed in the latest source?")
    );
    let body = serde_json::json!({
        "task_id": task_id,
        "instruction": instruction,
        "capability": capability,
    });
    let out = post_json(api, "/api/federation/delegate", body)?;
    println!("{}", serde_json::to_string_pretty(&out)?);
    Ok(())
}

/// Build the same [`StepHandlerRegistry`] the server uses for one step kind:
/// deterministic handlers always; the LLM handlers only when `kind` needs them
/// (loads the CLI config and talks to the provider directly — an isolated
/// executor never calls the server API).
async fn isolated_step_registry(
    kind: &str,
    config_path: Option<&str>,
) -> Result<StepHandlerRegistry, Box<dyn std::error::Error>> {
    let mut registry = StepHandlerRegistry::with_builtin_deterministic();
    if LLM_STEP_KINDS.contains(&kind) {
        let path = crate::ops::mcp_config_path(config_path);
        let cfg = crate::ops::load_kowalski_config_for_serve(&path)?;
        kowalski_core::db::run_memory_migrations_if_configured(&cfg).await?;
        let model = kowalski_core::config::default_model(&cfg);
        let agent = TemplateAgent::new(cfg).await?;
        let agent = Arc::new(tokio::sync::Mutex::new(agent));
        LlmStepHandler::register_all(&mut registry, agent, &model);
    }
    Ok(registry)
}

/// Write one [`IsolatedStepEvent`] protocol line to stdout (flushed — the
/// parent consumes the stream live).
fn emit_isolated_event(event: &IsolatedStepEvent) -> Result<(), Box<dyn std::error::Error>> {
    use std::io::Write;
    let mut out = std::io::stdout().lock();
    serde_json::to_writer(&mut out, event)?;
    out.write_all(b"\n")?;
    out.flush()?;
    Ok(())
}

/// Sink turning handler feed messages into `message` protocol lines.
struct StdoutEventSink;

#[async_trait::async_trait]
impl StepEventSink for StdoutEventSink {
    async fn message(&self, text: &str) {
        let _ = emit_isolated_event(&IsolatedStepEvent::Message {
            text: text.to_string(),
        });
    }
}

/// One-shot process-isolated step executor (`agent-app exec-step`), spawned by
/// the server for steps with `isolation = "process"`: reads one
/// [`IsolatedStepRequest`] JSON document from stdin, executes it through the
/// same step handlers the server runs in-process, streams
/// [`IsolatedStepEvent`] lines on stdout, and exits. Diagnostics (including
/// the pid the parent logs as isolation evidence) go to stderr.
pub async fn exec_step(config_path: Option<&str>) -> Result<(), Box<dyn std::error::Error>> {
    use std::io::Read;
    let mut input = String::new();
    std::io::stdin().read_to_string(&mut input)?;
    let request: IsolatedStepRequest = serde_json::from_str(input.trim())
        .map_err(|e| format!("exec-step: invalid IsolatedStepRequest on stdin: {e}"))?;
    eprintln!(
        "exec-step pid={} run={} step={} kind={}",
        std::process::id(),
        request.run_id,
        request.step.name,
        request.step.kind
    );
    let registry = isolated_step_registry(&request.step.kind, config_path).await?;
    let response = execute_isolated_request(&request, &registry, &StdoutEventSink).await;
    emit_isolated_event(&IsolatedStepEvent::Outcome(response))
}

pub fn federate_worker(
    path: Option<&str>,
    api_url: Option<&str>,
    agent_id: &str,
    topic: Option<&str>,
    role: Option<&str>,
    capability_override: Option<&str>,
) -> Result<(), Box<dyn std::error::Error>> {
    let api = api_base(api_url);
    let api = api.as_str();
    let root = app_root(path);
    let topic = topic.unwrap_or("federation");

    let role = role.map(|s| s.to_string());
    let capabilities: Vec<String> = if let Some(cap) = capability_override {
        vec![cap.to_string()]
    } else if let Some(r) = role.as_deref() {
        vec![format!("kc.{}", r)]
    } else {
        vec!["knowledge-compiler".to_string(), "kc.run".to_string()]
    };

    // Role workers execute steps through the same StepHandler registry the
    // server uses in-process; LLM kinds talk to the configured provider
    // directly (no `/api/chat` loopback), so the worker needs no server-side
    // chat access at all.
    let runtime = tokio::runtime::Runtime::new()?;
    let registry = match role.as_deref() {
        Some(kind) => Some(runtime.block_on(isolated_step_registry(kind, None))?),
        None => None,
    };

    let reg = serde_json::json!({
        "id": agent_id,
        "capabilities": capabilities.clone(),
    });
    let _ = post_json(api, "/api/federation/register", reg)?;
    println!(
        "Registered worker `{}` (role={}, capabilities={}). Listening on topic `{}`.",
        agent_id,
        role.as_deref().unwrap_or("(legacy: kc.run)"),
        capabilities.join(","),
        topic
    );

    // SSE worker must keep the HTTP connection open indefinitely.
    // reqwest blocking `Client` defaults to a 30s **overall** timeout; idle SSE
    // reads then fail with `error decoding response body` (Decode kind) — disable it here.
    let client = reqwest_blocking::Client::builder()
        .timeout(None::<std::time::Duration>)
        .build()?;
    let stream_url = format!(
        "{}/api/federation/stream?topic={}",
        api.trim_end_matches('/'),
        topic
    );
    let resp = with_bearer(client.get(stream_url)).send()?;
    if !resp.status().is_success() {
        return Err(format!("stream failed: HTTP {}", resp.status()).into());
    }
    let reader = std::io::BufReader::new(resp);

    for line in reader.lines() {
        let line = match line {
            Ok(v) => v,
            Err(e) => {
                eprintln!("federation stream decode warning (ignored): {}", e);
                continue;
            }
        };
        if !line.starts_with("data: ") {
            continue;
        }
        let data = &line[6..];
        let env: serde_json::Value = match serde_json::from_str(data) {
            Ok(v) => v,
            Err(_) => continue,
        };
        let payload = env
            .get("payload")
            .cloned()
            .unwrap_or_else(|| serde_json::json!({}));
        if payload.get("kind").and_then(|x| x.as_str()) != Some("task_delegate") {
            continue;
        }
        if payload.get("to_agent").and_then(|x| x.as_str()) != Some(agent_id) {
            continue;
        }
        let task_id = payload
            .get("task_id")
            .and_then(|x| x.as_str())
            .unwrap_or("unknown-task")
            .to_string();
        let instruction = payload
            .get("instruction")
            .and_then(|x| x.as_str())
            .unwrap_or("")
            .to_string();

        if let Some(role_kind) = role.as_deref() {
            let registry = registry
                .as_ref()
                .expect("step handler registry built for role workers");
            handle_role_delegate(
                api,
                topic,
                agent_id,
                role_kind,
                &root,
                &task_id,
                &instruction,
                &runtime,
                registry,
            );
        } else {
            handle_legacy_run_delegate(api, topic, agent_id, &root, &task_id, &instruction);
        }
        let _ = post_json(
            api,
            "/api/federation/heartbeat",
            serde_json::json!({ "agent_id": agent_id }),
        );
    }
    Ok(())
}

fn parse_horde_instruction(instruction: &str) -> Option<HordeInstruction> {
    let v: serde_json::Value = serde_json::from_str(instruction).ok()?;
    let horde = v
        .get("horde")
        .and_then(|x| x.as_str())
        .unwrap_or("")
        .to_string();
    let run_id = v
        .get("run_id")
        .and_then(|x| x.as_str())
        .unwrap_or("")
        .to_string();
    let step = v
        .get("step")
        .and_then(|x| x.as_str())
        .unwrap_or("")
        .to_string();
    let kind = v
        .get("kind")
        .and_then(|x| x.as_str())
        .unwrap_or(&step)
        .to_string();
    if horde.is_empty() || run_id.is_empty() || step.is_empty() {
        return None;
    }
    Some(HordeInstruction {
        horde,
        run_id,
        step,
        kind,
        source: v
            .get("source")
            .and_then(|x| x.as_str())
            .map(ToString::to_string),
        question: v
            .get("question")
            .and_then(|x| x.as_str())
            .map(ToString::to_string),
        previous_artifact: v
            .get("previous_artifact")
            .and_then(|x| x.as_str())
            .map(ToString::to_string),
        horde_root: v
            .get("horde_root")
            .and_then(|x| x.as_str())
            .map(ToString::to_string),
        workdir: v
            .get("workdir")
            .and_then(|x| x.as_str())
            .map(ToString::to_string),
        project_path: v
            .get("project_path")
            .and_then(|x| x.as_str())
            .map(ToString::to_string),
        tool_ids: v
            .get("tool_ids")
            .and_then(|x| x.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|x| x.as_str().map(ToString::to_string))
                    .collect()
            })
            .unwrap_or_default(),
    })
}

#[derive(Debug, Clone)]
struct HordeInstruction {
    horde: String,
    run_id: String,
    step: String,
    kind: String,
    source: Option<String>,
    question: Option<String>,
    previous_artifact: Option<String>,
    horde_root: Option<String>,
    workdir: Option<String>,
    project_path: Option<String>,
    tool_ids: Vec<String>,
}

fn publish_acl(api: &str, topic: &str, sender: &str, payload: serde_json::Value) {
    let body = serde_json::json!({
        "sender": sender,
        "topic": topic,
        "payload": payload,
    });
    if let Err(e) = post_json(api, "/api/federation/publish", body) {
        eprintln!("publish failed: {}", e);
    }
}

fn publish_task_started(
    api: &str,
    topic: &str,
    sender: &str,
    instr: &HordeInstruction,
    text: &str,
) {
    publish_acl(
        api,
        topic,
        sender,
        serde_json::json!({
            "kind": "task_started",
            "run_id": instr.run_id,
            "horde": instr.horde,
            "step": instr.step,
            "agent": sender,
            "text": text,
        }),
    );
}

#[allow(clippy::too_many_arguments)] // one event, many fields; a params struct is a later refactor
fn publish_task_finished(
    api: &str,
    topic: &str,
    sender: &str,
    instr: &HordeInstruction,
    success: bool,
    artifact: Option<&str>,
    summary: &str,
    outcome: Option<&str>,
) {
    publish_acl(
        api,
        topic,
        sender,
        serde_json::json!({
            "kind": "task_finished",
            "run_id": instr.run_id,
            "horde": instr.horde,
            "step": instr.step,
            "agent": sender,
            "success": success,
            "artifact": artifact,
            "summary": summary,
            "outcome": outcome,
        }),
    );
}

fn publish_task_result(
    api: &str,
    topic: &str,
    sender: &str,
    task_id: &str,
    outcome: &str,
    success: bool,
) {
    publish_acl(
        api,
        topic,
        sender,
        serde_json::json!({
            "kind": "task_result",
            "task_id": task_id,
            "from_agent": sender,
            "outcome": outcome,
            "success": success,
        }),
    );
}

/// `StepSpec` for a worker-executed step, hydrated from the local
/// `agents/<step>.md` frontmatter (the delegation instruction carries run
/// context, not the stage spec — same fields the orchestrator snapshots).
fn step_spec_from_agent_doc(
    workspace_root: &Path,
    step: &str,
    kind: &str,
    instr_tool_ids: &[String],
) -> StepSpec {
    let meta =
        parse_stage_agent(&workspace_root.join("agents").join(format!("{step}.md"))).ok();
    let meta_tool_ids = meta
        .as_ref()
        .map(|m| m.tool_ids.clone())
        .unwrap_or_default();
    StepSpec {
        name: step.to_string(),
        kind: kind.to_string(),
        output: meta.as_ref().and_then(|m| m.output.clone()),
        verify_command: meta.as_ref().and_then(|m| m.verify_command.clone()),
        verify_cwd: meta.as_ref().and_then(|m| m.verify_cwd.clone()),
        apply_mode: meta.as_ref().and_then(|m| m.apply_mode.clone()),
        tool_ids: if meta_tool_ids.is_empty() {
            instr_tool_ids.to_vec()
        } else {
            meta_tool_ids
        },
    }
}

/// Feed sink for the SSE worker: republishes handler progress messages as
/// federation `agent_message`s over the (bearer-authenticated) HTTP API.
struct WorkerEventSink {
    api: String,
    topic: String,
    sender: String,
    run_id: String,
    horde: String,
    step: String,
}

#[async_trait::async_trait]
impl StepEventSink for WorkerEventSink {
    async fn message(&self, text: &str) {
        let (api, topic, sender) = (self.api.clone(), self.topic.clone(), self.sender.clone());
        let payload = serde_json::json!({
            "kind": "agent_message",
            "run_id": self.run_id,
            "horde": self.horde,
            "from": self.sender,
            "step": self.step,
            "text": text,
        });
        // publish_acl uses the blocking HTTP client — hop off the async runtime.
        let _ = tokio::task::spawn_blocking(move || publish_acl(&api, &topic, &sender, payload))
            .await;
    }
}

/// Thin client of the shared step handlers: parse the horde instruction into
/// an [`IsolatedStepRequest`], execute it through the same registry the server
/// runs in-process (kind + context in, outcome out — no divergent logic), and
/// publish the federation envelopes for the result.
#[allow(clippy::too_many_arguments)]
fn handle_role_delegate(
    api: &str,
    topic: &str,
    agent_id: &str,
    role_kind: &str,
    fallback_root: &Path,
    task_id: &str,
    instruction: &str,
    runtime: &tokio::runtime::Runtime,
    registry: &StepHandlerRegistry,
) {
    let Some(instr) = parse_horde_instruction(instruction) else {
        let summary = format!(
            "rejected: instruction is not a horde JSON envelope (task_id={})",
            task_id
        );
        eprintln!("{}", summary);
        publish_task_result(api, topic, agent_id, task_id, &summary, false);
        return;
    };

    if instr.kind != role_kind {
        let summary = format!(
            "rejected: agent role `{}` does not match instruction kind `{}`",
            role_kind, instr.kind
        );
        publish_task_finished(api, topic, agent_id, &instr, false, None, &summary, None);
        publish_task_result(api, topic, agent_id, task_id, &summary, false);
        return;
    }

    let workspace_root = instr
        .horde_root
        .clone()
        .map(PathBuf::from)
        .filter(|p| p.exists())
        .unwrap_or_else(|| fallback_root.to_path_buf());
    let workdir = instr
        .workdir
        .clone()
        .map(PathBuf::from)
        .unwrap_or_else(|| workspace_root.clone());

    publish_task_started(
        api,
        topic,
        agent_id,
        &instr,
        &format!("{} agent starting", role_kind),
    );

    if let Err(e) = ensure_dirs(&workdir) {
        let summary = format!("{} step failed: {}", role_kind, e);
        publish_task_finished(api, topic, agent_id, &instr, false, None, &summary, None);
        publish_task_result(api, topic, agent_id, task_id, &summary, false);
        return;
    }

    let request = IsolatedStepRequest {
        run_id: instr.run_id.clone(),
        horde_id: instr.horde.clone(),
        step: step_spec_from_agent_doc(&workspace_root, &instr.step, role_kind, &instr.tool_ids),
        workdir: workdir.clone(),
        horde_root: workspace_root.clone(),
        source: instr.source.clone(),
        question: instr
            .question
            .clone()
            .unwrap_or_else(|| "What changed in the latest source?".to_string()),
        project_path: instr.project_path.as_deref().map(PathBuf::from),
        previous_artifact: instr.previous_artifact.as_deref().map(PathBuf::from),
    };
    let sink = WorkerEventSink {
        api: api.to_string(),
        topic: topic.to_string(),
        sender: agent_id.to_string(),
        run_id: instr.run_id.clone(),
        horde: instr.horde.clone(),
        step: instr.step.clone(),
    };
    let response: IsolatedStepResponse =
        runtime.block_on(execute_isolated_request(&request, registry, &sink));

    if response.success {
        publish_task_finished(
            api,
            topic,
            agent_id,
            &instr,
            true,
            response.artifact.as_deref(),
            &response.summary,
            response.outcome.as_deref(),
        );
        publish_task_result(
            api,
            topic,
            agent_id,
            task_id,
            &format!(
                "artifact={}; {}",
                response.artifact.as_deref().unwrap_or(""),
                response.summary
            ),
            true,
        );
    } else {
        publish_task_finished(
            api,
            topic,
            agent_id,
            &instr,
            false,
            None,
            &response.summary,
            None,
        );
        publish_task_result(api, topic, agent_id, task_id, &response.summary, false);
    }
}


fn handle_legacy_run_delegate(
    api: &str,
    topic: &str,
    agent_id: &str,
    root: &Path,
    task_id: &str,
    instruction: &str,
) {
    let mut success = true;
    let outcome: String;
    if let Some(raw) = instruction.strip_prefix("kc.run:") {
        let mut parts = raw.splitn(2, '|');
        let source = parts.next().unwrap_or("").trim();
        let question = parts
            .next()
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .unwrap_or("What changed in the latest source?");
        if source.is_empty() {
            success = false;
            outcome = "missing source in instruction".to_string();
        } else {
            let root_s = root.to_string_lossy().to_string();
            let work = local_agent_app_workdir(root);
            let run_out = run_with_progress(
                Some(root_s.as_str()),
                source,
                Some(question),
                Some(api),
                |_step, _kind, _output| {},
            );
            match run_out {
                Err(e) => {
                    success = false;
                    outcome = format!("run failed: {}", e);
                }
                Ok(done) => {
                    let report = done
                        .report
                        .or_else(|| latest_md_in(&work_debug(&work).join("reports")))
                        .map(|p| p.display().to_string())
                        .unwrap_or_else(|| "(none)".to_string());
                    let handoff_disp = done
                        .handoff
                        .map(|p| p.display().to_string())
                        .unwrap_or_else(|| "(none)".to_string());
                    let summary = done
                        .summary
                        .map(|p| p.display().to_string())
                        .unwrap_or_else(|| "(none)".to_string());
                    let log = done
                        .log
                        .map(|p| p.display().to_string())
                        .unwrap_or_else(|| "(none)".to_string());
                    outcome = format!(
                        "run complete; summary={}; report={}; handoff={}; log={}",
                        summary, report, handoff_disp, log
                    );
                }
            }
        }
    } else {
        success = false;
        outcome = format!("unsupported instruction: {}", instruction);
    }
    publish_task_result(api, topic, agent_id, task_id, &outcome, success);
}

pub fn proof_check(
    path: Option<&str>,
    api_url: Option<&str>,
    agent_id: Option<&str>,
    capability: Option<&str>,
    source: Option<&str>,
    question: Option<&str>,
) -> Result<(), Box<dyn std::error::Error>> {
    let root = app_root(path);
    let work = local_agent_app_workdir(&root);
    let api = api_base(api_url);
    let api = api.as_str();
    let agent_id = agent_id.unwrap_or("kc-worker-1");
    let capability = capability.unwrap_or("kc.run");
    let source = source.unwrap_or("https://example.com/article");
    let question = question.unwrap_or("What changed?");

    // Preflight checks
    println!("Preflight:");
    validate(Some(root.to_string_lossy().as_ref()))?;
    let health = reqwest_blocking::get(format!("{}/api/health", api.trim_end_matches('/')));
    match health {
        Ok(r) if r.status().is_success() => println!("- API reachable at {}", api),
        Ok(r) => println!("- API responded with HTTP {} at {}", r.status(), api),
        Err(e) => println!("- API not reachable at {} ({})", api, e),
    }
    println!("- App path: {}", root.display());

    println!("\nProof-run checklist (3 terminals):");
    println!("1) Terminal A: start server");
    println!("   cargo run -p kowalski --bin kowalski");
    println!("2) Terminal B: start worker");
    println!(
        "   cargo run -p kowalski-cli -- agent-app worker --path \"{}\" --api \"{}\" \"{}\"",
        root.display(),
        api,
        agent_id
    );
    println!("3) Terminal C: delegate task");
    println!(
        "   cargo run -p kowalski-cli -- agent-app delegate --api \"{}\" --question \"{}\" \"{}\" \"{}\"",
        api, question, capability, source
    );
    println!("\nVerify artifacts (under {}):", work.display());
    println!(
        "- latest report: {}",
        latest_md_in(&work_debug(&work).join("reports"))
            .map(|p| p.display().to_string())
            .unwrap_or_else(|| "(none yet)".to_string())
    );
    println!(
        "- stage outputs (see manifest `agents/*.md` `output` paths): {}",
        work_debug(&work).display()
    );
    println!(
        "- latest run log: {}",
        latest_md_in(&work_debug(&work).join("scratch"))
            .map(|p| p.display().to_string())
            .unwrap_or_else(|| "(none yet)".to_string())
    );
    println!(
        "- paste pack: {}",
        work.join(PASTE_ME_FILENAME).display()
    );
    Ok(())
}

/// Resolve an export target: an existing horde directory path, else a horde id
/// looked up across the shared discovery roots (`kowalski_core::config::default_horde_roots`).
fn resolve_horde_dir(
    horde: &str,
    config_path: Option<&str>,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
    let as_path = PathBuf::from(horde);
    if as_path.is_dir() {
        return Ok(as_path);
    }
    let config = crate::ops::mcp_config_path(config_path);
    let roots = kowalski_core::config::default_horde_roots(config.parent());
    for root in &roots {
        let candidate = root.join(horde);
        if candidate.is_dir() {
            return Ok(candidate);
        }
    }
    Err(format!(
        "horde `{horde}` is neither a directory nor an id under the horde roots ({})",
        roots
            .iter()
            .map(|r| r.display().to_string())
            .collect::<Vec<_>>()
            .join(", ")
    )
    .into())
}

/// `agent-app export <horde> [-o out] [--json]`: write `<id>-<version>.kwf.zip`.
/// `out` ending in `.zip` is the bundle file path; otherwise it is the directory
/// the canonically named bundle is written into (default: current directory).
pub fn export_horde(
    horde: &str,
    output: Option<&str>,
    json: bool,
    config_path: Option<&str>,
) -> Result<(), Box<dyn std::error::Error>> {
    let horde_root = resolve_horde_dir(horde, config_path)?;
    let (dest_dir, file_target) = match output {
        Some(o) if o.ends_with(".zip") => {
            let target = PathBuf::from(o);
            let parent = target
                .parent()
                .filter(|p| !p.as_os_str().is_empty())
                .map(Path::to_path_buf)
                .unwrap_or_else(|| PathBuf::from("."));
            (parent, Some(target))
        }
        Some(o) => (PathBuf::from(o), None),
        None => (PathBuf::from("."), None),
    };
    let written = kowalski_core::manifest::export_horde_dir_bundle(&horde_root, &dest_dir)?;
    let bundle_path = match file_target {
        Some(target) if target != written => {
            fs::rename(&written, &target)?;
            target
        }
        _ => written,
    };
    if json {
        println!(
            "{}",
            serde_json::json!({ "ok": true, "bundle": bundle_path.display().to_string() })
        );
    } else {
        println!("Exported {}", bundle_path.display());
    }
    Ok(())
}

/// What this deployment knows for the import portability report: the tool ids a
/// `TemplateAgent` built from the CLI config registers, and the models its Ollama
/// endpoint serves. Either list degrades to "check skipped" when unavailable.
async fn cli_portability_context(
    config_path: Option<&str>,
) -> kowalski_core::manifest::PortabilityContext {
    let path = crate::ops::mcp_config_path(config_path);
    let mut context = kowalski_core::manifest::PortabilityContext::default();
    let Ok(cfg) = crate::ops::load_kowalski_config_for_serve(&path) else {
        return context;
    };
    match TemplateAgent::new(cfg.clone()).await {
        Ok(agent) => {
            context.known_tool_ids = Some(
                agent
                    .list_tools()
                    .await
                    .into_iter()
                    .map(|(name, _)| name)
                    .collect(),
            );
        }
        Err(e) => log::warn!("import: tool check skipped (agent init failed: {e})"),
    }
    let base_url = kowalski_core::config::ollama_base_url(&cfg);
    match kowalski_core::model::ModelManager::new(base_url) {
        Ok(manager) => match manager.list_models().await {
            Ok(response) => {
                context.available_models =
                    Some(response.models.into_iter().map(|m| m.name).collect());
            }
            Err(e) => log::warn!("import: model check skipped (Ollama unreachable: {e})"),
        },
        Err(e) => log::warn!("import: model check skipped ({e})"),
    }
    context
}

/// Render a [`PortabilityReport`] for operators; empty gaps collapse to one line.
fn render_portability_report(report: &kowalski_core::manifest::PortabilityReport) -> String {
    let mut lines = Vec::new();
    if report.is_empty() {
        lines.push("Portability: no gaps — fully runnable on this deployment.".to_string());
    } else {
        lines.push("Portability gaps (the horde still lands as a draft):".to_string());
        let mut gap = |label: &str, values: &[String]| {
            if !values.is_empty() {
                lines.push(format!("  {label}: {}", values.join(", ")));
            }
        };
        gap("unknown step kinds", &report.unknown_step_kinds);
        gap("unknown tool providers", &report.unknown_tool_providers);
        gap("missing builtin tools", &report.unknown_tool_ids);
        gap("unresolved pinned models", &report.unresolved_models);
        gap("warnings", &report.warnings);
    }
    for note in &report.migrations {
        lines.push(format!("  migrated: {note}"));
    }
    if report.triggers_disabled > 0 {
        lines.push(format!(
            "  {} trigger(s) imported disabled — re-enable deliberately",
            report.triggers_disabled
        ));
    }
    if report.steps_isolated > 0 {
        lines.push(format!(
            "  {} step(s) set to run in a separate process (isolation = \"process\")",
            report.steps_isolated
        ));
    }
    lines.join("\n")
}

/// `agent-app import <bundle> [--dir <hordes-root>] [--json]`: land a `.kwf.zip`
/// (or `.bbwf.zip`) bundle as a draft horde and print the portability report.
/// Default destination is the user hordes root, never `examples/`.
pub async fn import_horde(
    bundle: &str,
    dir: Option<&str>,
    json: bool,
    config_path: Option<&str>,
) -> Result<(), Box<dyn std::error::Error>> {
    let config = crate::ops::mcp_config_path(config_path);
    let dest_root = match dir {
        Some(d) => PathBuf::from(d),
        None => kowalski_core::config::user_hordes_root(config.parent())
            .ok_or("cannot resolve the user hordes root; pass --dir")?,
    };
    let context = cli_portability_context(config_path).await;
    let import =
        kowalski_core::manifest::import_bundle(Path::new(bundle), &dest_root, false, &context)?;
    if json {
        println!(
            "{}",
            serde_json::json!({
                "ok": true,
                "horde_id": import.manifest.identifier,
                "horde_root": import.horde_root.display().to_string(),
                "report": import.report,
            })
        );
    } else {
        println!(
            "Imported `{}` -> {}",
            import.manifest.identifier,
            import.horde_root.display()
        );
        println!("{}", render_portability_report(&import.report));
    }
    Ok(())
}

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

    fn coder_example() -> PathBuf {
        Path::new(env!("CARGO_MANIFEST_DIR")).join("../examples/coder")
    }

    #[tokio::test]
    async fn export_then_import_across_roots_reports_json() {
        let staging = tempfile::tempdir().unwrap();
        let export_dir = staging.path().join("bundles");
        let bundle =
            kowalski_core::manifest::export_horde_dir_bundle(&coder_example(), &export_dir)
                .unwrap();
        let name = bundle.file_name().unwrap().to_string_lossy().to_string();
        assert!(name.ends_with(".kwf.zip"), "canonical name, got {name}");

        let dest_root = staging.path().join("hordes");
        let import = kowalski_core::manifest::import_bundle(
            &bundle,
            &dest_root,
            false,
            &kowalski_core::manifest::PortabilityContext::default(),
        )
        .unwrap();
        assert!(import.horde_root.starts_with(&dest_root));
        assert!(import.horde_root.join("horde.md").is_file());

        let report = serde_json::to_value(&import.report).unwrap();
        assert_eq!(report["unknown_step_kinds"], serde_json::json!([]));
        assert_eq!(report["unknown_tool_providers"], serde_json::json!([]));
        assert_eq!(report["unresolved_models"], serde_json::json!([]));
        assert!(report["triggers_disabled"].is_u64());
    }

    #[test]
    fn export_horde_writes_to_explicit_zip_target() {
        let staging = tempfile::tempdir().unwrap();
        let target = staging.path().join("handoff.kwf.zip");
        export_horde(
            coder_example().to_str().unwrap(),
            Some(target.to_str().unwrap()),
            true,
            None,
        )
        .unwrap();
        assert!(target.is_file(), "bundle renamed to the -o target");
    }

    #[test]
    fn export_rejects_unknown_horde_id() {
        let err = resolve_horde_dir("definitely-not-a-horde-id", Some("/nonexistent/config.toml"))
            .unwrap_err();
        assert!(err.to_string().contains("definitely-not-a-horde-id"));
    }

    #[test]
    fn render_report_flags_gaps_and_disabled_triggers() {
        let report = kowalski_core::manifest::PortabilityReport {
            unknown_tool_ids: vec!["quantum_tool".into()],
            triggers_disabled: 2,
            ..Default::default()
        };
        let text = render_portability_report(&report);
        assert!(text.contains("missing builtin tools: quantum_tool"));
        assert!(text.contains("2 trigger(s) imported disabled"));
    }
}