tachyon_types/

commands.rs

use std::collections::HashMap;

use myko_macros::myko_command;
use myko::command::{CommandContext, CommandError, CommandHandler};
use serde_json::Value;

use crate::{
    derive_resources, GetAllInventorys, GetPinsByIds, GetPlaybooksByIds, GetRunsByIds,
    HostStats, OutputLine, OutputLineId, OutputStream, Pin, PinId, Playbook, PlaybookId, Run, RunId,
    RunStatus,
};

/// Run an Ansible playbook. Creates a new `Run` entity with status `Queued`;
/// the admission step in tachyon-server promotes it to `Running` once it holds
/// a lease on every resource it targets, and the run watcher then executes it.
#[myko_command]
pub struct RunPlaybook {
    pub playbook_id: PlaybookId,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub extra_vars: Option<Value>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<String>,
}

impl CommandHandler for RunPlaybook {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        // Confirm the playbook exists.
        ctx.exec_query_first(GetPlaybooksByIds {
            ids: vec![self.playbook_id.clone()],
        })?
        .ok_or_else(|| CommandError {
            tx: ctx.tx().to_string(),
            command_id: ctx.command_id.to_string(),
            message: format!("Playbook {} not found", self.playbook_id.0),
        })?;

        // Derive the resource set from the limit against the current inventory,
        // so the queue can serialize on shared clusters. No inventory (e.g. it
        // failed to discover) means no derivable resources → the run leases
        // nothing and is admitted immediately.
        let resources = match ctx.exec_query(GetAllInventorys {})?.into_iter().next() {
            Some(inv) => derive_resources(self.limit.as_deref(), &inv),
            None => Vec::new(),
        };

        let run_id = RunId::from(uuid::Uuid::new_v4().to_string());
        let now = chrono::Utc::now().to_rfc3339();
        let run = Run {
            playbook_id: self.playbook_id,
            status: RunStatus::Queued,
            started_at: now,
            finished_at: None,
            exit_code: None,
            extra_vars: self.extra_vars,
            limit: self.limit,
            resources,
            host_stats: Default::default(),
            // Attribution of the LAUNCHING connection. Stamped here from the command's
            // request context — #[myko_client_id] emit-stamping only covers direct client
            // SET events, never a handler's server-side emit (lv-67c8). None = an
            // internal dispatch (saga/startup) with no originating connection.
            client_id: ctx.client_id().map(|c| c.to_string()),
            claimed_by: None, // set by ClaimRun when a runner wins the run
            id: run_id,
        };
        ctx.emit_set(&run)?;
        Ok(())
    }
}

/// Cancel a playbook execution. Works on a `Queued` run (drops it from the
/// queue before it ever starts) or a `Running` one (signals the child process).
/// No-op if the run has already finished.
#[myko_command]
pub struct CancelRun {
    pub run_id: RunId,
}

impl CommandHandler for CancelRun {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        let current = ctx
            .exec_query_first(GetRunsByIds {
                ids: vec![self.run_id.clone()],
            })?
            .ok_or_else(|| CommandError {
                tx: ctx.tx().to_string(),
                command_id: ctx.command_id.to_string(),
                message: format!("Run {} not found", self.run_id.0),
            })?;

        // Only an in-flight run (queued or running) can be cancelled; a
        // finished run stays as it is. A queued run carries no PID, so the
        // watcher's cancel handler simply finds nothing to signal.
        if !matches!(current.status, RunStatus::Queued | RunStatus::Running) {
            return Ok(());
        }

        let mut updated = (*current).clone();
        updated.status = RunStatus::Cancelled;
        ctx.emit_set(&updated)?;
        Ok(())
    }
}

/// Claim a run for execution — the compare-and-swap that makes exactly ONE runner
/// own each run. Succeeds only while the run is `Running` and UNCLAIMED; commands
/// execute serialized in the cell, so the check-and-set is atomic and two racing
/// runners cannot both win. A loser treats the error as "not mine, skip" — the
/// claim is the execution gate, not a failure.
///
/// Idempotent for the winner: re-claiming a run you already hold succeeds (a
/// subscription re-fire after winning must not error).
#[myko_command(RunId)]
pub struct ClaimRun {
    pub run_id: RunId,
    /// Stable runner identity (one per runner instance/host). A restarted runner
    /// with the same id uses it to recognize — and fail out — runs it left behind.
    pub runner_id: String,
}

impl CommandHandler for ClaimRun {
    // Returns the RunId (not unit) so the CALLER CAN AWAIT the result — a unit
    // command's response never populates the client's result cell, which made the
    // runner's claim-await time out on every claim that actually landed.
    fn execute(self, ctx: CommandContext) -> Result<RunId, CommandError> {
        let current = ctx
            .exec_query_first(GetRunsByIds {
                ids: vec![self.run_id.clone()],
            })?
            .ok_or_else(|| CommandError {
                tx: ctx.tx().to_string(),
                command_id: ctx.command_id.to_string(),
                message: format!("Run {} not found", self.run_id.0),
            })?;

        if current.status != RunStatus::Running {
            return Err(CommandError {
                tx: ctx.tx().to_string(),
                command_id: ctx.command_id.to_string(),
                message: format!(
                    "ClaimRun: run {} is {:?}, not Running — nothing to execute",
                    self.run_id.0, current.status
                ),
            });
        }
        match &current.claimed_by {
            Some(holder) if *holder == self.runner_id => return Ok(self.run_id), // already ours
            Some(holder) => {
                return Err(CommandError {
                    tx: ctx.tx().to_string(),
                    command_id: ctx.command_id.to_string(),
                    message: format!(
                        "ClaimRun: run {} already claimed by '{holder}'",
                        self.run_id.0
                    ),
                });
            }
            None => {}
        }

        let mut updated = (*current).clone();
        updated.claimed_by = Some(self.runner_id);
        ctx.emit_set(&updated)?;
        Ok(self.run_id)
    }
}

/// Pin a playbook to the sidebar's "Pinned" section. Idempotent: re-pinning
/// an already-pinned playbook is a no-op. Refuses to overwrite a locked
/// (config-defined) pin.
#[myko_command]
pub struct PinPlaybook {
    pub playbook_id: PlaybookId,
}

impl CommandHandler for PinPlaybook {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        ctx.exec_query_first(GetPlaybooksByIds {
            ids: vec![self.playbook_id.clone()],
        })?
        .ok_or_else(|| CommandError {
            tx: ctx.tx().to_string(),
            command_id: ctx.command_id.to_string(),
            message: format!("Playbook {} not found", self.playbook_id.0),
        })?;

        let pin_id = PinId::from(self.playbook_id.0.clone());
        if let Some(existing) = ctx.exec_query_first(GetPinsByIds {
            ids: vec![pin_id.clone()],
        })? {
            if existing.locked {
                return Ok(());
            }
        }

        let pin = Pin {
            playbook_id: self.playbook_id,
            locked: false,
            id: pin_id,
        };
        ctx.emit_set(&pin)?;
        Ok(())
    }
}

/// Append one line of playbook output to a run. Issued by the out-of-process
/// runner as it streams `ansible-playbook` output back to the cell (the
/// in-process runner writes `OutputLine`s directly via `ctx`). `seq` is the
/// runner's monotonic per-run counter across both stdout+stderr, so the UI can
/// order lines and jump to a parsed failure's source line.
#[myko_command]
pub struct AppendRunOutput {
    pub run_id: RunId,
    pub stream: OutputStream,
    pub line: String,
    pub seq: u64,
}

impl CommandHandler for AppendRunOutput {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        let entity = OutputLine {
            id: OutputLineId::from(format!("{}:{}", self.run_id.0, self.seq)),
            run_id: self.run_id,
            stream: self.stream,
            line: self.line,
            seq: self.seq,
        };
        ctx.emit_set(&entity)?;
        Ok(())
    }
}

/// Mark a run finished, reported by the runner when `ansible-playbook` exits.
/// The cell decides the terminal status: a run already `Cancelled` (a cancel
/// landed mid-run) stays cancelled; otherwise exit code 0 -> `Completed`,
/// anything else -> `Failed`. `exit_code` is `None` when the process was killed
/// by a signal or never spawned.
#[myko_command]
pub struct FinishRun {
    pub run_id: RunId,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_code: Option<i32>,
}

impl CommandHandler for FinishRun {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        let current = ctx
            .exec_query_first(GetRunsByIds {
                ids: vec![self.run_id.clone()],
            })?
            .ok_or_else(|| CommandError {
                tx: ctx.tx().to_string(),
                command_id: ctx.command_id.to_string(),
                message: format!("Run {} not found", self.run_id.0),
            })?;

        let mut updated = (*current).clone();
        // A concurrent cancel wins; otherwise derive terminal status from exit.
        if updated.status != RunStatus::Cancelled {
            updated.status = if self.exit_code == Some(0) {
                RunStatus::Completed
            } else {
                RunStatus::Failed
            };
        }
        updated.exit_code = self.exit_code;
        updated.finished_at = Some(chrono::Utc::now().to_rfc3339());
        ctx.emit_set(&updated)?;
        Ok(())
    }
}

/// Declare the playbooks a runner has available. The runner scans its
/// `playbook_dir` and reports each as a `Playbook` entity (upsert by id).
/// Discovery lives where the playbook files are - the runner - not the cell,
/// so `RunPlaybook`'s existence check has something to resolve against.
#[myko_command]
pub struct DeclarePlaybooks {
    pub playbooks: Vec<Playbook>,
}

impl CommandHandler for DeclarePlaybooks {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        for pb in self.playbooks {
            ctx.emit_set(&pb)?;
        }
        Ok(())
    }
}

/// Record an EXTERNALLY-executed (CLI) op as a terminal `Run`.
///
/// Mutating cluster ops are run via `ansible-playbook` on the control node, so they
/// never dispatch through the cell and the runs wall never sees them. `RecordRun`
/// lets the deploy-side callback self-report the op at completion: it creates a
/// `Run` ALREADY TERMINAL (Completed/Failed/Cancelled). The runner executes every
/// `Running` run (it keys purely on `status == Running`, with no executor/owner
/// field to tell a self-declared run from a dispatched one), so a terminal run is
/// invisible to it — no double-execution, no `RunStatus` change. The wall renders
/// it like any other run (it reads `GetAllRuns`, source-agnostic); captured stdout
/// streams on via `AppendRunOutput(run_id, …)`.
///
/// Terminal-only BY CONTRACT: `Running`/`Queued` are rejected fail-loud — a
/// non-terminal self-declared run WOULD be picked up and re-executed by the runner.
/// Records hold no lease (`resources: []`): a completed op is a record, not a work
/// item. Lives here (not the cell) so every myko consumer — the cell handler AND
/// the deploy-side recorder client — shares ONE definition; no hand-synced contract.
#[myko_command(RunId)]
pub struct RecordRun {
    /// The playbook that was run (e.g. `ue-cluster-redeploy`).
    pub playbook_id: String,
    /// Terminal outcome ONLY — `completed` | `failed` | `cancelled`. `running` /
    /// `queued` are rejected: a non-terminal self-declared run would be executed
    /// by the runner.
    pub status: RunStatus,
    /// RFC3339 — when the external op started.
    pub started_at: String,
    /// RFC3339 — when it finished (a record is created post-hoc, so always known).
    pub finished_at: String,
    /// Process exit code, when captured.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_code: Option<i32>,
    /// The op's vars (put operator / cluster-def context here for the "ran via CLI"
    /// attribution; the actor's WS identity is separately server-stamped as client_id).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub extra_vars: Option<Value>,
    /// The ansible `--limit`, when scoped.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<String>,
    /// Per-host ok/changed/failed/… recap, when the caller parsed the PLAY RECAP.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub host_stats: Option<HashMap<String, HostStats>>,
    /// ATTACH mode: the id of an EXISTING dispatched Run to enrich with this
    /// execution report (host_stats, limit) instead of minting a new record.
    ///
    /// This is the recap-decoupling fix: the Run LIFECYCLE (status, exit_code,
    /// finished_at) is owned by the runner via `FinishRun`; the EXECUTION DETAIL
    /// (per-host recap) is owned by whoever ran ansible and holds the stats object —
    /// the deploy-side callback. Before this field the callback could only mint a
    /// duplicate record or stay silent, so every cell-dispatched run recorded an
    /// empty recap. The runner advertises the id to the playbook process as
    /// `TACHYON_RUN_ID`; the callback passes it back here.
    ///
    /// Attach does NOT touch lifecycle fields and is accepted while the Run is
    /// still `Running` (ansible stats fire before the process exits).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub run_id: Option<RunId>,
}

impl CommandHandler for RecordRun {
    fn execute(self, ctx: CommandContext) -> Result<RunId, CommandError> {
        // ATTACH mode: enrich the named EXISTING run with the execution report.
        // Lifecycle fields (status / exit_code / finished_at) belong to the runner's
        // `FinishRun` and are deliberately untouched, so the two writers never race
        // over the same fact. Accepted while the run is still Running — ansible's
        // stats callback fires before the process exits.
        if let Some(rid) = self.run_id {
            let current = ctx
                .exec_query_first(GetRunsByIds { ids: vec![rid.clone()] })?
                .ok_or_else(|| CommandError {
                    tx: ctx.tx().to_string(),
                    command_id: ctx.command_id.to_string(),
                    message: format!(
                        "RecordRun attach: run {} not found — TACHYON_RUN_ID names a run this cell does not have",
                        rid.0
                    ),
                })?;
            let mut updated = (*current).clone();
            if let Some(hs) = self.host_stats {
                updated.host_stats = hs;
            }
            if updated.limit.is_none() {
                updated.limit = self.limit;
            }
            ctx.emit_set(&updated)?;
            return Ok(rid);
        }

        // MINT mode (no run_id): a deploy-side self-report of an operator-run op.
        // Terminal-only. A Running/Queued self-declared run would be grabbed and
        // re-executed by the runner (it watches status == Running); refuse it so a
        // record can never trigger a real op.
        if !matches!(
            self.status,
            RunStatus::Completed | RunStatus::Failed | RunStatus::Cancelled
        ) {
            return Err(CommandError {
                tx: ctx.tx().to_string(),
                command_id: ctx.command_id.to_string(),
                message: format!(
                    "RecordRun records COMPLETED external ops — status must be completed/failed/cancelled, not {:?} (a non-terminal run would be double-executed by the runner)",
                    self.status
                ),
            });
        }

        let id = RunId::from(uuid::Uuid::new_v4().to_string());
        let run = Run {
            id: id.clone(),
            playbook_id: PlaybookId::from(self.playbook_id),
            status: self.status,
            started_at: self.started_at,
            finished_at: Some(self.finished_at),
            exit_code: self.exit_code,
            extra_vars: self.extra_vars,
            limit: self.limit,
            // A completed record holds no lease — it is not a work item.
            resources: Vec::new(),
            host_stats: self.host_stats.unwrap_or_default(),
            // The RECORDING connection (callback/CLI), stamped from the request context —
            // handler emits are never #[myko_client_id]-stamped (lv-67c8).
            client_id: ctx.client_id().map(|c| c.to_string()),
            claimed_by: None, // a terminal record is not a work item; nothing claims it
        };
        ctx.emit_set(&run)?;
        Ok(id)
    }
}

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

    /// Wire compat: a pre-0.6 payload (no runId) must still deserialize with
    /// run_id None — the mint path — so old callers keep working unchanged.
    #[test]
    fn record_run_without_run_id_is_mint_shape() {
        let old: RecordRun = serde_json::from_str(
            r#"{"playbookId":"ue-cluster-up","status":"completed",
                "startedAt":"2026-07-24T00:00:00Z","finishedAt":"2026-07-24T00:01:00Z"}"#,
        )
        .unwrap();
        assert!(old.run_id.is_none());
        assert!(old.host_stats.is_none());
    }

    #[test]
    fn record_run_with_run_id_is_attach_shape() {
        let attach: RecordRun = serde_json::from_str(
            r#"{"playbookId":"ue-cluster-up","status":"completed",
                "startedAt":"2026-07-24T00:00:00Z","finishedAt":"2026-07-24T00:01:00Z",
                "runId":"abc-123",
                "hostStats":{"render-01":{"ok":5,"changed":1,"unreachable":0,"failed":0,"skipped":0,"rescued":0,"ignored":0}}}"#,
        )
        .unwrap();
        assert_eq!(attach.run_id.as_ref().map(|r| r.0.as_ref()), Some("abc-123"));
        assert_eq!(attach.host_stats.unwrap()["render-01"].ok, 5);
    }
}