clawmates/crates/cm-api/src/topology_worker.rs

//! 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;

/// 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, poll: Duration) {
    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, job).await,
                Ok(None) => tokio::time::sleep(poll).await,
                Err(e) => {
                    eprintln!("topology_worker: claim failed: {e}");
                    tokio::time::sleep(poll).await;
                }
            }
        }
    });
}

/// Drive one claimed job to a terminal state, persisting checkpoints as it goes.
async fn run_job(
    pool: &PgPool,
    runtime: &cm_runtime::Runtime,
    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_transition_research_topic(pool, 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
        .and_then(|c| serde_json::from_value(c).ok())
        .unwrap_or_default();

    // If this run belongs to a research topic OR a loop with a per-team
    // ZeroClaw container spawned, point the executor at THAT container's
    // gateway URL so the run's turns hit its isolated daemon instead of
    // the workspace-wide one. Falls back to the env-derived executor when
    // there's no per-topic/loop container (chat sessions, or research/loop
    // runs where spawn failed and we recorded no URL).
    let per_topic_url = match cm_db::repo::topology_runs::research_topic_id(pool, id).await {
        Ok(Some(topic_id)) => {
            match cm_db::repo::research_topics::get(pool, topic_id, job.workspace_id).await {
                Ok(Some(t)) => t.zeroclaw_gateway_url,
                _ => None,
            }
        }
        _ => None,
    };
    // Loop lookup runs only when the research lookup didn't hit — a run
    // is bound to at most one of {topic, loop}. This preserves the
    // existing research fast path unchanged.
    let per_topic_url = if per_topic_url.is_some() {
        per_topic_url
    } else {
        match cm_db::repo::topology_runs::loop_id_for_run(pool, id).await {
            Ok(Some(loop_id)) => cm_db::repo::loops::zeroclaw_gateway_url(pool, loop_id)
                .await
                .unwrap_or(None),
            _ => None,
        }
    };
    if let Some(url) = &per_topic_url {
        // Best-effort readiness gate — a freshly-spawned team container may
        // still be starting when the worker claims the run. Cap the wait so
        // a broken image can't hang the worker.
        if let Err(e) =
            crate::research_container::wait_ready(url, std::time::Duration::from_secs(30)).await
        {
            eprintln!("topology_worker: research team {url} readiness: {e} — proceeding anyway");
        }
    }
    let leaf_result = match &per_topic_url {
        Some(url) => ZeroClawDriveExecutor::from_env_for_gateway(url.clone()),
        None => 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;
        }
    };

    // 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, &graph, &job.task, progress, &exec).await
        }
        _ => drive(pool, id, &graph, &job.task, progress, &leaf).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: complete({id}) failed: {e}");
            }
            freeze_research_outcome(pool, id, &record.final_output).await;
            advance_loop_after_completion(pool, id, &record.final_output).await;
            continue_initial_burst(pool, id).await;
        }
        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;
            }
        }
    }
    maybe_transition_research_topic(pool, id).await;
}

/// If this run belongs to a research topic, snapshot the orchestrator's
/// final synthesis as a versioned `research_outcomes` row. The frontend
/// canvas reads `latest_outcome` for anything past `standby` so reviewers
/// see the produced draft rather than the original prompt. Best-effort:
/// a failure here logs but doesn't fail the run.
async fn freeze_research_outcome(pool: &PgPool, run_id: Uuid, final_output: &str) {
    let topic_id = match cm_db::repo::topology_runs::research_topic_id(pool, run_id).await {
        Ok(Some(id)) => id,
        Ok(None) => return,
        Err(e) => {
            eprintln!("topology_worker: research_topic_id({run_id}) failed: {e}");
            return;
        }
    };
    if final_output.trim().is_empty() {
        return;
    }
    if let Err(e) =
        cm_db::repo::research_outcomes::insert(pool, topic_id, final_output, Some(run_id)).await
    {
        eprintln!("topology_worker: research_outcomes::insert({run_id}) failed: {e}");
        return;
    }
    // Fan-out: any exec-kind loop bound to this topic with the
    // on_artifact_update trigger enabled wakes up now. Coalesced —
    // if a loop already has a queued/running run we skip (D3 fallback:
    // the coordinator sees the fresh artifact on its next iteration
    // anyway). Best-effort per loop; one loop's Docker/DB hiccup
    // doesn't affect the others.
    let awakened = cm_db::repo::loops::loops_awaiting_topic(pool, topic_id)
        .await
        .unwrap_or_default();
    for (loop_id, workspace_id, task_template, graph) in awakened {
        if cm_db::repo::loops::has_active_run(pool, loop_id)
            .await
            .unwrap_or(false)
        {
            continue; // Coalesce.
        }
        let iter = cm_db::repo::loops::next_iteration(pool, loop_id)
            .await
            .unwrap_or(0);
        // Build the enriched task with the freshly-inserted artifact
        // (compose_iteration_task reads latest, which is what we just
        // wrote).
        let task =
            crate::routes::loops::compose_iteration_task(pool, loop_id, &task_template).await;
        if let Err(e) = cm_db::repo::loops::enqueue_iteration(
            pool,
            loop_id,
            workspace_id,
            &task,
            &graph,
            iter,
            Some(run_id),
        )
        .await
        {
            eprintln!("topology_worker: on_artifact_update enqueue({loop_id}) failed: {e}");
        }
    }
}

/// If the just-completed run was a loop iteration with
/// initial_burst_remaining > 0, enqueue the next iteration and
/// decrement the counter (CAS-safe via take_initial_burst_slot).
/// No-op for non-loop runs and for loops whose burst is exhausted.
async fn continue_initial_burst(pool: &PgPool, run_id: Uuid) {
    let loop_id = match cm_db::repo::topology_runs::loop_id_for_run(pool, run_id).await {
        Ok(Some(id)) => id,
        _ => return,
    };
    // If another worker races us, only ONE gets the slot; the other
    // sees 0 (no-op).
    let prev = cm_db::repo::loops::take_initial_burst_slot(pool, loop_id)
        .await
        .unwrap_or(0);
    if prev == 0 {
        return;
    }
    // Coalesce with a concurrently-in-flight iteration (a webhook
    // arriving during a burst, say).
    if cm_db::repo::loops::has_active_run(pool, loop_id)
        .await
        .unwrap_or(false)
    {
        return;
    }
    // Fetch the loop so we have the workspace + graph + task_template
    // to enqueue the next iteration.
    let Ok(Some(l)) = cm_db::repo::loops::get_any_workspace(pool, loop_id).await else {
        return;
    };
    let iter = cm_db::repo::loops::next_iteration(pool, loop_id)
        .await
        .unwrap_or(0);
    let task = crate::routes::loops::compose_iteration_task(pool, loop_id, &l.task_template).await;
    if let Err(e) = cm_db::repo::loops::enqueue_iteration(
        pool,
        loop_id,
        l.workspace_id,
        &task,
        &l.graph,
        iter,
        Some(run_id),
    )
    .await
    {
        eprintln!("topology_worker: continue_initial_burst enqueue failed: {e}");
    }
}

/// Post-terminal hook for loop-bound runs. Parses `COMPLETED: INT-<NN>`
/// markers out of the run's final output and advances the loop's
/// `consumed_int_ids` + `current_int_index`. Only fires for runs that
/// belong to a loop AND that loop is bound to a source research topic
/// (the integrations flow). Standalone loops or unbound runs no-op.
///
/// The marker parser is deliberately forgiving — accepts INT-XX and
/// INT-XXX, optionally with surrounding backticks or dashes, so
/// coordinator prompts that emit slightly different formats still
/// advance the pointer.
async fn advance_loop_after_completion(pool: &PgPool, run_id: Uuid, final_output: &str) {
    let loop_id = match cm_db::repo::topology_runs::loop_id_for_run(pool, run_id).await {
        Ok(Some(id)) => id,
        _ => return,
    };
    let ctx = match cm_db::repo::loops::source_research_context(pool, loop_id).await {
        Ok(Some(c)) => c,
        _ => return, // Not a research-bound loop; nothing to advance.
    };
    let mut completed = parse_completed_int_ids(final_output);
    // Drop items already recorded so re-runs don't double-count.
    let (_topic, already, _idx) = ctx;
    completed.retain(|id| !already.contains(id));
    if !completed.is_empty() {
        if let Err(e) =
            cm_db::repo::loops::advance_after_completion(pool, loop_id, &completed).await
        {
            eprintln!("topology_worker: loops::advance_after_completion({loop_id}) failed: {e}");
        }
    }

    // Reorder rationale — coordinator emits "REORDER: <text>" when it
    // works on an INT-XX out of order (usually because a prereq was
    // unmet). Append each occurrence to the loop's reorder_events
    // array so a mini-timeline UI can surface the history. Iteration
    // number comes from topology_runs; -1 if the lookup fails (best-
    // effort — we still record the event with a sentinel).
    let iteration = cm_db::repo::topology_runs::iteration_for_run(pool, run_id)
        .await
        .unwrap_or(Some(-1))
        .unwrap_or(-1);
    for text in parse_reorder_rationale(final_output) {
        if let Err(e) =
            cm_db::repo::loops::append_reorder_event(pool, loop_id, run_id, iteration, &text).await
        {
            eprintln!("topology_worker: loops::append_reorder_event({loop_id}) failed: {e}");
        }
    }
}

/// Extract "REORDER: <text>" rationales — one per line the coordinator
/// emits when it works out of order. Same permissive line matcher as
/// the completed-marker parser (list dashes, backticks, emphasis).
/// Returns the text after the colon, trimmed. Skips empty rationales.
fn parse_reorder_rationale(text: &str) -> Vec<String> {
    let mut out = Vec::new();
    for line in text.lines() {
        let normalized = line.trim_start_matches(|c: char| {
            c.is_whitespace() || c == '-' || c == '*' || c == '#' || c == '>'
        });
        let upper = normalized.to_ascii_uppercase();
        if !upper.starts_with("REORDER:") {
            continue;
        }
        // Preserve original case of the rationale text — only the
        // marker matched case-insensitively.
        let colon = normalized.find(':').map(|i| i + 1).unwrap_or(0);
        let rationale = normalized[colon..].trim();
        if !rationale.is_empty() {
            out.push(rationale.to_string());
        }
    }
    out
}

/// Extract stable INT-XX ids from a completion line. Matches
/// `COMPLETED: INT-01`, `COMPLETED: INT-01, INT-02`, or `- COMPLETED: `INT-01``.
/// De-duplicates within a single output.
fn parse_completed_int_ids(text: &str) -> Vec<String> {
    let mut out = Vec::new();
    let mut seen = std::collections::HashSet::new();
    for line in text.lines() {
        // Case-insensitive, tolerates surrounding whitespace, list dashes,
        // markdown emphasis, and backticks.
        let normalized = line.trim_start_matches(|c: char| {
            c.is_whitespace() || c == '-' || c == '*' || c == '#' || c == '>'
        });
        let upper = normalized.to_ascii_uppercase();
        if !upper.starts_with("COMPLETED:") {
            continue;
        }
        for token in upper
            .trim_start_matches("COMPLETED:")
            .split(|c: char| c == ',' || c == ';' || c.is_whitespace())
        {
            let stripped = token.trim_matches(|c: char| c == '`' || c == '*' || c == '.');
            if stripped.starts_with("INT-")
                && stripped.len() >= 5
                && seen.insert(stripped.to_string())
            {
                out.push(stripped.to_string());
            }
        }
    }
    out
}

/// Post-terminal hook: if this run belongs to a research topic and it was
/// the last sibling in flight, transition the topic `processing → reviewing`.
/// Best-effort — a DB hiccup here logs but doesn't fail the run.
async fn maybe_transition_research_topic(pool: &PgPool, id: Uuid) {
    match cm_db::repo::topology_runs::notify_run_completed(pool, id).await {
        Ok(true) => {
            // Left intentionally quiet on success; the UI polls the topic
            // status. Future: emit a run_event so live viewers see it flip.
        }
        Ok(false) => {}
        Err(e) => eprintln!("topology_worker: notify_run_completed({id}) failed: {e}"),
    }
    maybe_teardown_ephemeral_team(pool, id).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, 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.
    if let Some(prov) = crate::runtime_provision::RuntimeProvisioner::from_env() {
        for cid in &teardown.claw_ids {
            if let Err(e) = prov.deprovision_claw(*cid).await {
                eprintln!("topology_worker: deprovision_claw({cid}) failed: {e}");
            }
        }
    }
    for cid in &teardown.claw_ids {
        if let Err(e) = cm_db::repo::agents::hard_purge(pool, cm_domain::AgentId::from(*cid)).await
        {
            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,
    graph: &TopologyGraph,
    task: &str,
    progress: RunProgress,
    executor: &E,
) -> Result<RunRecord, OrchestratorError> {
    let pool_cb = pool.clone();
    execute_resumable(graph, task, executor, progress, move |snap| {
        let pool = pool_cb.clone();
        async move {
            // 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
}