//! Background worker that drains durable topology run jobs.
//!
//! A `POST /api/topologies/run` enqueues a job (`topology_runs` row, status
//! `queued`); this loop claims it, drives the topology turn-by-turn via the
//! ZeroClaw runtime, and checkpoints the [`RunProgress`] after every step. If
//! the worker (or the whole server) dies mid-run, the row is left `running`;
//! the stale sweep requeues it and the next claim resumes it from the last
//! checkpointed step — so long-horizon runs survive restarts.
//!
//! This reuses the agent-run durability pattern (claim CAS, checkpoint, resume
//! sweep) without coupling topology runs to the chat-session schema.

use std::sync::Arc;
use std::time::Duration;

use cm_domain::WorkspaceId;
use cm_orchestrator::{execute_resumable, OrchestratorError, RunProgress, RunRecord, TurnExecutor};
use cm_topology::TopologyGraph;
use sqlx::PgPool;
use uuid::Uuid;

use crate::recursive_exec::{SubTopologyExecutor, Tier};
use crate::topology_exec::ZeroClawDriveExecutor;

/// Requeue a `running` job whose worker hasn't checkpointed within this window.
const STALE_AFTER_SECS: f64 = 180.0;

/// Maximum age a `running` run may spend WITHOUT journaling any step
/// records before the reaper kills its container and fails it.
///
/// **This must stay LONGER than the runtime's per-turn timeout.** A run
/// journals its first record when its first step COMPLETES, so any turn still
/// legitimately in flight looks identical to a wedged container. The runtime
/// grants a turn `timeout_secs = 3000` (50 min), so a shorter reaper window
/// does not detect stuck runs — it kills healthy slow ones.
///
/// This was 15 minutes, chosen when "healthy first-step latency is typically
/// 5–60s" was true of the model in use. It was, on haiku. Moving the mission
/// agents to sonnet-5 made first turns longer than the window, and mission
/// 01a00c41's research phase was reaped at 900s having already written 402
/// lines across 13 files — work the delivery path then captured and pushed,
/// which is the only reason we could tell the run was healthy at all.
///
/// The lesson generalises past this constant: a liveness timeout calibrated
/// against one model silently becomes a correctness bug when the model changes.
const REAP_STUCK_AFTER_SECS: i64 = 60 * 60;

/// Spawn the durable topology job worker. Polls for queued jobs every `poll`
/// interval; runs each to completion (or failure), checkpointing per step.
pub fn spawn(
    pool: PgPool,
    runtime: cm_runtime::Runtime,
    hub: Arc<crate::fleet::NodeHub>,
    poll: Duration,
) {
    // Fire the stuck-container reaper on its own cadence — checking
    // once a minute is plenty and keeps this off the hot claim loop.
    let reaper_pool = pool.clone();
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(Duration::from_secs(60));
        loop {
            ticker.tick().await;
            if let Err(e) = reap_stuck_runs(&reaper_pool).await {
                eprintln!("topology_worker::reaper: reap failed: {e}");
            }
        }
    });

    tokio::spawn(async move {
        loop {
            // Recover jobs orphaned by a dead worker before claiming new ones.
            if let Err(e) = cm_db::repo::topology_runs::requeue_stale(&pool, STALE_AFTER_SECS).await
            {
                eprintln!("topology_worker: requeue_stale failed: {e}");
            }
            match cm_db::repo::topology_runs::claim_next_queued(&pool).await {
                Ok(Some(job)) => run_job(&pool, &runtime, &hub, job).await,
                Ok(None) => tokio::time::sleep(poll).await,
                Err(e) => {
                    eprintln!("topology_worker: claim failed: {e}");
                    tokio::time::sleep(poll).await;
                }
            }
        }
    });
}

/// Mark `running` mission-bound topology_runs that have been alive past
/// [`REAP_STUCK_AFTER_SECS`] without journaling a single step record as
/// `failed`, with a diagnostic error so the user sees WHY instead of an
/// infinitely-spinning pipeline.
///
/// Only reaps runs bound to a mission — non-mission runs (raw API-driven
/// topology runs) are left to the existing stale-checkpoint requeuer.
async fn reap_stuck_runs(pool: &PgPool) -> Result<(), sqlx::Error> {
    use sqlx::Row;
    let rows: Vec<sqlx::postgres::PgRow> = sqlx::query(
        "SELECT id, mission_id
           FROM topology_runs
          WHERE status = 'running'
            AND mission_id IS NOT NULL
            -- Only jobs this worker drives. mission_id IS NOT NULL used to mean
            -- the same thing as orchestrator-driven, and the microvm and session
            -- tiers broke that: their checkpoint is NULL for life BY DESIGN, so the
            -- zero-step-records test below is true of a perfectly healthy run.
            AND tier = ANY($2)
            AND created_at < now() - make_interval(secs => $1::float)
            AND coalesce(jsonb_array_length(coalesce(checkpoint->'records', '[]'::jsonb)), 0) = 0",
    )
    .bind(REAP_STUCK_AFTER_SECS as f64)
    .bind(
        // REAPABLE, not worker-driven: `microvm_graph` is driven by this worker
        // and must NOT be reaped — one of its steps is a whole agent session in a
        // VM, so "no step records in 15 minutes" describes a healthy composed run
        // as readily as a wedged one.
        cm_db::repo::topology_runs::REAPABLE_TIERS
            .iter()
            .map(|s| (*s).to_string())
            .collect::<Vec<_>>(),
    )
    .fetch_all(pool)
    .await?;

    for row in rows {
        let id: Uuid = row.get("id");
        let mission_id: Uuid = row.get("mission_id");
        eprintln!(
            "topology_worker::reaper: reaping stuck run run_id={} mission_id={} (no step records after {}s)",
            id, mission_id, REAP_STUCK_AFTER_SECS,
        );
        let _ = cm_db::repo::topology_runs::fail(
            pool,
            id,
            &format!("reaped: no step records after {}s", REAP_STUCK_AFTER_SECS),
        )
        .await;
    }
    Ok(())
}

/// Drive one claimed job to a terminal state, persisting checkpoints as it goes.
async fn run_job(
    pool: &PgPool,
    runtime: &cm_runtime::Runtime,
    hub: &Arc<crate::fleet::NodeHub>,
    job: cm_db::repo::topology_runs::ClaimedTopologyRun,
) {
    let id = job.id;

    // Swarm runs aren't graph topologies — the `graph` JSONB holds the swarm
    // config. Branch before the graph parse and run the self-verifying loop.
    if job.tier == "swarm" {
        let cfg = job.graph.clone().unwrap_or(serde_json::Value::Null);
        let swarm_job: crate::swarm::SwarmJob = serde_json::from_value(cfg).unwrap_or_default();
        let result = crate::swarm::run_swarm_job(pool, runtime, id, swarm_job, &job.task).await;
        match result {
            Ok(record) => {
                let value = serde_json::to_value(&record).unwrap_or(serde_json::Value::Null);
                if let Err(e) = cm_db::repo::topology_runs::complete(pool, id, &value).await {
                    eprintln!("topology_worker: swarm complete({id}) failed: {e}");
                }
            }
            Err(e) => {
                let _ = cm_db::repo::topology_runs::fail(pool, id, &e).await;
            }
        }
        maybe_teardown_ephemeral_team(pool, runtime, id).await;
        return;
    }

    let Some(graph) = job
        .graph
        .as_ref()
        .and_then(|g| serde_json::from_value::<TopologyGraph>(g.clone()).ok())
    else {
        let _ = cm_db::repo::topology_runs::fail(pool, id, "missing or invalid graph").await;
        return;
    };

    // Resume from the last checkpoint, or start fresh.
    let progress: RunProgress = job
        .checkpoint
        .clone()
        .and_then(|c| serde_json::from_value(c).ok())
        .unwrap_or_default();

    // The composed engines (Slice 4): this graph's nodes are not claws, they are
    // Claude-Code-in-a-microVM sessions. Branched BEFORE the leaf executor is
    // built, because that build reads the ZeroClaw gateway config — a composed
    // run must not fail for want of a runtime it never dials.
    if job.tier == "microvm_graph" {
        let result = run_composed(pool, hub, &job, &graph, progress).await;
        finish(pool, id, result).await;
        maybe_teardown_ephemeral_team(pool, runtime, id).await;
        return;
    }

    // C3: prefer the mission's per-run runtime endpoint when set on
    // the missions row; else fall back to the shared env-derived
    // gateway (pre-C3 missions + non-mission runs). This is what
    // isolates agents' workspace filesystem to that mission's repo.
    type MissionBinding = (
        Option<String>,
        Option<String>,
        Uuid,
        Option<Uuid>,
        Option<String>,
    );
    let mission_binding: Option<MissionBinding> = sqlx::query_as::<_, MissionBinding>(
        "SELECT m.runtime_endpoint, m.runtime_pairing_code, m.id, r.mission_phase_id,
                m.runtime_token
           FROM topology_runs r
           JOIN missions m ON m.id = r.mission_id
          WHERE r.id = $1",
    )
    .bind(id)
    .fetch_optional(pool)
    .await
    .ok()
    .flatten();
    // What this run's turns will be attributed to. `None` when the run belongs
    // to no mission — a bare topology run has no phase to hang tool calls on.
    let tap = mission_binding
        .as_ref()
        .map(
            |(_, _, mission_id, phase_id, _)| crate::topology_exec::MissionTap {
                pool: pool.clone(),
                workspace_id: job.workspace_id,
                mission_id: *mission_id,
                phase_id: *phase_id,
                run_id: Some(id),
            },
        );
    let leaf_result = match mission_binding {
        // Seed the cached bearer from `runtime_token` when we have one: the
        // pairing code is single-use, so after a restart it is the only way in.
        Some((Some(url), Some(code), _, _, tok)) => {
            ZeroClawDriveExecutor::from_env_for_gateway_with_code(url, code)
                .map(|e| e.with_token(tok))
        }
        // No pairing code (pre-C3 missions): the persisted token is the only
        // credential, so seed it here too.
        Some((Some(url), None, _, _, tok)) => {
            ZeroClawDriveExecutor::from_env_for_gateway(url).map(|e| e.with_token(tok))
        }
        _ => ZeroClawDriveExecutor::from_env(),
    };
    let leaf = match leaf_result {
        Ok(e) => e,
        Err(e) => {
            let _ = cm_db::repo::topology_runs::fail(pool, id, &e).await;
            return;
        }
    };
    // The tap rides on the leaf executor, so the recursive tiers get it too:
    // they drive the same leaf all the way down, and a company-tier mission's
    // tool calls belong to its phase exactly as a team-tier one's do.
    let leaf = match tap {
        Some(t) => leaf.with_tap(t),
        None => leaf,
    };

    // Select the executor by deploy tier: `team` drives claws directly; the
    // upper tiers drive the recursive sub-topology executor (which runs each
    // child tier's graph, all the way down to the same leaf claw executor).
    let result = match job.tier.as_str() {
        "company" | "org" => {
            let tier = if job.tier == "org" {
                Tier::Org
            } else {
                Tier::Company
            };
            let exec = SubTopologyExecutor::new(
                pool.clone(),
                WorkspaceId::from(job.workspace_id),
                tier,
                id,
                Arc::new(leaf),
            );
            drive(
                pool,
                id,
                job.workspace_id,
                &graph,
                &job.task,
                progress,
                &exec,
            )
            .await
        }
        _ => {
            drive(
                pool,
                id,
                job.workspace_id,
                &graph,
                &job.task,
                progress,
                &leaf,
            )
            .await
        }
    };

    finish(pool, id, result).await;
    maybe_teardown_ephemeral_team(pool, runtime, id).await;
}

/// Write a driven run's terminal state. The single place a run finishes, shared
/// by every tier — a second one would be a second completion path, which is where
/// every microVM bug this project has hit came from.
async fn finish(pool: &PgPool, id: Uuid, result: Result<RunRecord, OrchestratorError>) {
    match result {
        Ok(record) => {
            let value = serde_json::to_value(&record).unwrap_or(serde_json::Value::Null);
            if let Err(e) = cm_db::repo::topology_runs::complete(pool, id, &value).await {
                eprintln!("topology_worker: complete({id}) failed: {e}");
            }
        }
        Err(e) => {
            // Don't clobber a cancellation (or any already-terminal state) with `failed`.
            let terminal = matches!(
                cm_db::repo::topology_runs::current_status(pool, id).await,
                Ok(Some(ref s)) if s == "cancelled" || s == "completed" || s == "failed"
            );
            if !terminal {
                let _ = cm_db::repo::topology_runs::fail(pool, id, &format!("{e}")).await;
            }
        }
    }
}

/// Drive a composed run: the outer graph is Engine Z, every node is a
/// Claude-Code-in-a-microVM session (Engine C).
///
/// The mission columns are read here rather than carried on the run row so a
/// re-placed or re-backed mission takes effect on resume, and so the composed
/// path has exactly one source of truth for where a VM boots.
async fn run_composed(
    pool: &PgPool,
    hub: &Arc<crate::fleet::NodeHub>,
    job: &cm_db::repo::topology_runs::ClaimedTopologyRun,
    graph: &TopologyGraph,
    progress: RunProgress,
) -> Result<RunRecord, OrchestratorError> {
    let mission_id = job.mission_id.ok_or_else(|| {
        OrchestratorError::Executor(
            "a composed run has no mission, so there is no checkout for its nodes \
             to share"
                .into(),
        )
    })?;
    let phase_id = job.mission_phase_id.ok_or_else(|| {
        OrchestratorError::Executor("a composed run must belong to a mission phase".into())
    })?;

    let mission: (Option<Uuid>, Option<String>, Option<String>, bool) = sqlx::query_as(
        "SELECT target_node_id, backend, team_engine, (repo_id IS NOT NULL) \
             FROM missions WHERE id = $1",
    )
    .bind(mission_id)
    .fetch_one(pool)
    .await
    .map_err(|e| OrchestratorError::Executor(format!("load mission {mission_id}: {e}")))?;

    // The phase's completion gate, read here rather than carried on the run row
    // so an edited `done_when_check` takes effect on the next node instead of at
    // the next mission.
    let phase: (String, serde_json::Value) =
        sqlx::query_as("SELECT kind, config FROM mission_phases WHERE id = $1")
            .bind(phase_id)
            .fetch_one(pool)
            .await
            .map_err(|e| OrchestratorError::Executor(format!("load phase {phase_id}: {e}")))?;

    let exec = crate::microvm_turn_executor::for_fleet(
        hub.clone(),
        pool.clone(),
        crate::microvm_turn_executor::ComposedRun {
            run_id: job.id,
            mission_id,
            phase_id,
            iteration: job.iteration.unwrap_or(1),
            repo: crate::mission_workspace::checkout_path(mission_id),
            // A repo-less composed mission gets an empty shared workspace, the
            // same as a solo phase — the graph's whole property is that node 2
            // sees node 1's files, and that holds whether or not it is a git
            // checkout.
            has_repo: mission.3,
            target_node_id: mission.0,
            backend: mission.1,
            team_engine: mission.2,
            gate: crate::vm_stop_gate::StopGate::for_phase(&phase.0, &phase.1)
                .and_then(crate::vm_stop_gate::StopGate::per_node),
            // Resume continues the step numbering; restarting it would re-use a
            // finished node's vm id.
            completed_steps: progress.completed as u32,
        },
    );
    drive(
        pool,
        job.id,
        job.workspace_id,
        graph,
        &job.task,
        progress,
        &exec,
    )
    .await
}

/// Post-terminal hook: if this run's team is `ephemeral` and no siblings are
/// still in flight, deprovision every bound claw on the ZeroClaw daemon,
/// delete the claw rows, and delete the team row. Best-effort — a failure to
/// tear down leaves the team intact and logs; a future sweep can retry.
async fn maybe_teardown_ephemeral_team(pool: &PgPool, runtime: &cm_runtime::Runtime, id: Uuid) {
    let teardown = match cm_db::repo::topology_runs::check_ephemeral_teardown(pool, id).await {
        Ok(Some(t)) => t,
        Ok(None) => return,
        Err(e) => {
            eprintln!("topology_worker: check_ephemeral_teardown({id}) failed: {e}");
            return;
        }
    };
    // Deprovision each claw on the daemon before deleting rows — if the daemon
    // side fails we still delete our rows (the daemon can be swept for orphans
    // by the fleet-reconcile timer). This is the trade cm-api owns everywhere:
    // Postgres is authoritative, the daemon config is a cache.
    //
    // Goes through the shared reaper so an ephemeral team's claws also get
    // their sandbox containers and `.brain` files removed — this path used to
    // do the daemon + DB halves only, leaking a container per ephemeral run.
    let provisioner = crate::runtime_provision::RuntimeProvisioner::from_env();
    for cid in &teardown.claw_ids {
        let report = crate::routes::claws::purge_agent(
            pool,
            runtime,
            provisioner.as_ref(),
            cm_domain::AgentId::from(*cid),
        )
        .await;
        if let Err(e) = report.counts {
            eprintln!("topology_worker: agents::hard_purge({cid}) failed: {e}");
        }
    }
    if let Err(e) = cm_db::repo::teams::delete_team(
        pool,
        teardown.team_id,
        cm_domain::WorkspaceId::from(teardown.workspace_id),
    )
    .await
    {
        eprintln!(
            "topology_worker: teams::delete_team({}) failed: {e}",
            teardown.team_id
        );
    }
}

/// Drive a graph to completion with the durable per-step checkpoint +
/// cancellation closure, generic over the executor so the team (leaf) and
/// company/org (recursive) tiers share the same outer durability logic. The
/// checkpoint here is parent-node-level (coarse resume); the recursive executor
/// additionally touches `updated_at` from each inner leaf step to stay alive.
async fn drive<E: TurnExecutor>(
    pool: &PgPool,
    id: Uuid,
    workspace_id: Uuid,
    graph: &TopologyGraph,
    task: &str,
    progress: RunProgress,
    executor: &E,
) -> Result<RunRecord, OrchestratorError> {
    let pool_cb = pool.clone();
    // node_id -> agent id, resolved once. The binding lives in the node's
    // attrs (`agent = claw_<uuid>`), which is also what the runtime dispatches
    // on — so usage is attributed to exactly the claw that did the work.
    let agent_of: std::sync::Arc<std::collections::HashMap<String, Uuid>> = std::sync::Arc::new(
        graph
            .nodes
            .iter()
            .filter_map(|n| {
                let alias = n.attrs.get("agent")?;
                let uuid = alias.strip_prefix("claw_")?;
                Some((n.id.clone(), Uuid::parse_str(uuid).ok()?))
            })
            .collect(),
    );
    execute_resumable(graph, task, executor, progress, move |snap| {
        let pool = pool_cb.clone();
        let agent_of = agent_of.clone();
        async move {
            // 2026-07-15: verbose per-step trace so `docker logs
            // clawmates_server_1` shows which topology node just fired,
            // its role, output size, tokens, and gated-action count.
            // Cheap (one info! per step) and gives us the missing
            // "topology is flowing" signal without needing to open the
            // canvas.
            if let Some(last) = snap.records.last() {
                let phase = match &last.phase {
                    cm_orchestrator::StepPhase::Plan => "plan",
                    cm_orchestrator::StepPhase::Work => "work",
                    cm_orchestrator::StepPhase::Synth => "synth",
                    cm_orchestrator::StepPhase::Aggregate => "aggregate",
                };
                eprintln!(
                    "topology_worker::step run_id={} step={} node={} role={} phase={} output_bytes={} tokens={} gated={}",
                    id,
                    snap.completed,
                    last.node_id,
                    last.role,
                    phase,
                    last.output.len(),
                    last.tokens,
                    last.gated.len(),
                );
                // Per-agent usage. Without this the command centre's SPEND,
                // ACTIVITY and THROUGHPUT cards read `usage_events`, which
                // nothing on the mission path ever wrote — so they showed 0 for
                // an agent that had just burned 15k tokens.
                //
                // `charge` also decrements credit lots, which is the point: a
                // mission turn costs what it costs. It clamps at the available
                // balance and still records the full obligation, so an empty
                // wallet cannot fail a turn.
                // The agent's own words, for the REASONING STREAM card. The
                // world feed is a DB poll, not a push bus, so a live card can
                // only show what was persisted — this is the step output the
                // worker already has in hand, attributed to the claw that
                // produced it. Truncated because the card renders a tail, not a
                // transcript, and mission_events is capped per phase.
                if let Some(agent_id) = agent_of.get(&last.node_id).copied() {
                    let text: String = last.output.chars().take(600).collect();
                    if !text.trim().is_empty() {
                        if let Some(mission_id) = sqlx::query_scalar::<_, Option<Uuid>>(
                            "SELECT mission_id FROM topology_runs WHERE id = $1",
                        )
                        .bind(id)
                        .fetch_optional(&pool)
                        .await
                        .ok()
                        .flatten()
                        .flatten()
                        {
                            let mut ev = crate::mission_events::MissionEvent::new(
                                mission_id,
                                "reasoning",
                            );
                            ev.agent_id = Some(agent_id);
                            ev.run_id = Some(id);
                            ev.target = Some(last.role.clone());
                            ev.detail = serde_json::json!({ "text": text });
                            crate::mission_events::record(&pool, ev).await;
                        }
                    }
                }
                if let Some(agent_id) = agent_of.get(&last.node_id).copied() {
                    if last.tokens > 0 {
                        // The split and the provider come from the runtime's
                        // `done` frame via `StepRecord.spend`. An executor
                        // that reports only a total leaves the split at 0/0
                        // and the total goes on the output side, as before.
                        let (tin, tout) = if last.spend.input_tokens + last.spend.output_tokens > 0
                        {
                            (last.spend.input_tokens, last.spend.output_tokens)
                        } else {
                            (0, last.tokens as u64)
                        };
                        let mission_id: Option<Uuid> = sqlx::query_scalar::<_, Option<Uuid>>(
                            "SELECT mission_id FROM topology_runs WHERE id = $1",
                        )
                        .bind(id)
                        .fetch_optional(&pool)
                        .await
                        .ok()
                        .flatten()
                        .flatten();
                        if let Err(e) = cm_billing::charge(
                            &pool,
                            cm_domain::WorkspaceId::from(workspace_id),
                            cm_domain::AgentId::from(agent_id),
                            None,
                            tin,
                            tout,
                            last.spend.provider.as_deref(),
                            last.spend.model.as_deref(),
                            mission_id,
                        )
                        .await
                        {
                            eprintln!("topology_worker: usage for {agent_id} failed: {e}");
                        }
                    }
                }
            }
            // Best-effort checkpoint: a failed write just means we re-run the
            // step on resume (idempotent — topology turns are pure reads here).
            if let Ok(v) = serde_json::to_value(&snap) {
                let _ =
                    cm_db::repo::topology_runs::checkpoint(&pool, id, &v, snap.completed as i64)
                        .await;
            }
            // Honor cancellation at the step boundary: stop before the next turn.
            if matches!(
                cm_db::repo::topology_runs::current_status(&pool, id).await,
                Ok(Some(ref s)) if s == "cancelled"
            ) {
                return Err(OrchestratorError::Executor("run cancelled".into()));
            }
            Ok(())
        }
    })
    .await
}