kanade-backend 0.43.53

use async_nats::jetstream::kv::Config as KvConfig;
use axum::Json;
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::http::header::HeaderMap;
use futures::TryStreamExt;
use kanade_shared::kv::{
    BUCKET_SCHEDULES, BUCKET_SCHEDULES_YAML, BUCKET_SCRIPT_STATUS, SCRIPT_STATUS_REVOKED,
};
use kanade_shared::manifest::Schedule;
use serde::{Deserialize, Serialize};
use tracing::{info, warn};

use crate::api::AppState;
use crate::api::yaml_body::{YamlOrJson, mirror_yaml, yaml_headers};
use crate::audit;
use crate::audit::Caller;

#[derive(Serialize)]
pub struct ScheduleSummary {
    pub id: String,
    /// Operator-facing one-liner (`When`'s Display): `per_pc once`,
    /// `per_pc every 6h`, `cron: 0 0 9 * * mon-fri`, …
    pub when: String,
    pub enabled: bool,
    pub job_id: String,
}

/// GET /api/schedules — full Schedule list, KV-backed.
pub async fn list(State(s): State<AppState>) -> Result<Json<Vec<Schedule>>, (StatusCode, String)> {
    let kv = match s.jetstream.get_key_value(BUCKET_SCHEDULES).await {
        Ok(k) => k,
        Err(_) => return Ok(Json(Vec::new())),
    };
    let keys_stream = kv
        .keys()
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("kv keys: {e}")))?;
    let keys: Vec<String> = keys_stream
        .try_collect()
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("kv keys: {e}")))?;
    let mut out = Vec::with_capacity(keys.len());
    for k in keys {
        if let Ok(Some(bytes)) = kv.get(&k).await
            && let Ok(sched) = serde_json::from_slice::<Schedule>(&bytes)
        {
            out.push(sched);
        }
    }
    Ok(Json(out))
}

/// Query params for [`preview`].
#[derive(Deserialize, Debug)]
pub struct PreviewQuery {
    /// How many upcoming fires to list (calendar schedules only).
    /// Defaults to 5; clamped to `1..=50` so a huge `count` can't make
    /// the backend walk croner thousands of times per request.
    #[serde(default = "default_preview_count")]
    pub count: usize,
}

fn default_preview_count() -> usize {
    5
}

/// Dry-run result for [`preview`].
#[derive(Serialize)]
pub struct PreviewResponse {
    pub id: String,
    /// `When`'s Display — `at 09:00 [mon-fri]`, `per_pc every 6h`, …
    pub when: String,
    /// `local` / `utc` — the tz the fire times are resolved in.
    pub tz: String,
    /// The schedule's `enabled` flag. The fire times are computed from
    /// the cron regardless, so a disabled schedule still previews its
    /// *would-be* fires — surface the flag so callers don't mistake a
    /// dormant schedule for an active one (claude #578 review).
    pub enabled: bool,
    /// Upcoming fire instants (RFC3339 UTC), soonest first. Empty for
    /// reconcile shapes and for calendars that can never fire — see
    /// `note`.
    pub fires: Vec<String>,
    /// Present only when `fires` is empty, explaining why.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub note: Option<String>,
}

/// GET /api/schedules/{id}/preview?count=N — dry-run the next N fire
/// times of a schedule (#418 "ドライラン / プレビュー"). Calendar
/// schedules return discrete tz-resolved instants (honoring the
/// `active` window + `constraints.window`); reconcile shapes have no
/// discrete fire times, so `fires` is empty and `note` describes the
/// cadence. Read-only: never touches KV state.
pub async fn preview(
    State(s): State<AppState>,
    Path(id): Path<String>,
    Query(q): Query<PreviewQuery>,
) -> Result<Json<PreviewResponse>, (StatusCode, String)> {
    let kv = s
        .jetstream
        .get_key_value(BUCKET_SCHEDULES)
        .await
        .map_err(|e| {
            (
                StatusCode::SERVICE_UNAVAILABLE,
                format!("schedules bucket missing: {e}"),
            )
        })?;
    let bytes = kv
        .get(&id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("KV get: {e}")))?
        .ok_or_else(|| (StatusCode::NOT_FOUND, format!("schedule '{id}' not found")))?;
    let schedule: Schedule = serde_json::from_slice(&bytes).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("deserialize stored schedule: {e}"),
        )
    })?;

    let count = q.count.clamp(1, 50);
    let fires = schedule.preview_fires(chrono::Utc::now(), count);
    let note = if !fires.is_empty() {
        None
    } else if matches!(schedule.when, kanade_shared::manifest::When::Calendar(_)) {
        Some(
            "no upcoming fires — a past one-shot, or the fire time is excluded by the \
             active window / constraints.window"
                .to_string(),
        )
    } else {
        Some(format!(
            "reconcile cadence ({}) polls every minute gated by cooldown — no discrete \
             fire times to preview",
            schedule.when
        ))
    };

    Ok(Json(PreviewResponse {
        id: schedule.id.clone(),
        when: schedule.when.to_string(),
        tz: schedule.tz.as_str().to_string(),
        enabled: schedule.enabled,
        fires: fires.iter().map(|t| t.to_rfc3339()).collect(),
        note,
    }))
}

/// Trailing window for the success/fail tally in [`status`].
const STATUS_WINDOW_HOURS: i64 = 24;

/// The most recent run of a schedule's job (`exit_code` / `finished_at`
/// `null` = still in flight).
#[derive(Serialize, Default)]
pub struct LastRun {
    pub pc_id: String,
    pub exit_code: Option<i64>,
    pub started_at: Option<String>,
    pub finished_at: Option<String>,
}

/// Finished-run tally over the trailing [`STATUS_WINDOW_HOURS`].
#[derive(Serialize, Default)]
pub struct RecentCounts {
    pub window_hours: i64,
    pub ok: i64,
    pub fail: i64,
}

/// Coverage view for [`status`].
#[derive(Serialize)]
pub struct StatusResponse {
    pub id: String,
    pub when: String,
    pub tz: String,
    pub enabled: bool,
    /// Soonest upcoming fire (RFC3339 UTC) — calendar schedules only;
    /// `null` for reconcile shapes or a schedule that can never fire.
    pub next_run: Option<String>,
    /// Most recent run, or `null` if this schedule's job has never run.
    pub last_run: Option<LastRun>,
    pub recent: RecentCounts,
}

/// Last run + trailing success/fail tally for a job. Keyed by `job_id`
/// (a schedule references a job; `execution_results` has no
/// `schedule_id`), so two schedules sharing a job share these numbers —
/// accurate for the common 1:1 case, an over-count otherwise. Pure DB
/// read, factored out so it's unit-testable against an in-memory pool.
async fn schedule_run_stats(
    pool: &sqlx::SqlitePool,
    job_id: &str,
    since: chrono::DateTime<chrono::Utc>,
) -> Result<(Option<LastRun>, RecentCounts), sqlx::Error> {
    use sqlx::Row;
    let last = sqlx::query(
        "SELECT pc_id, exit_code, started_at, finished_at
           FROM execution_results
          WHERE job_id = ?
          ORDER BY recorded_at DESC
          LIMIT 1",
    )
    .bind(job_id)
    .fetch_optional(pool)
    .await?
    .map(|r| LastRun {
        pc_id: r.try_get("pc_id").unwrap_or_default(),
        // Read the nullable columns as `Option<_>` explicitly:
        // sqlx-sqlite decodes a NULL via `try_get::<i64>` / `<String>`
        // into `0` / `""` rather than erroring, so `try_get(..).ok()`
        // would mislabel a still-running row (NULL exit_code +
        // finished_at) as "exit 0, finished at ''". The Option form maps
        // NULL → None. `started_at` is NOT NULL so it stays a plain read.
        exit_code: r.try_get::<Option<i64>, _>("exit_code").unwrap_or(None),
        started_at: r.try_get("started_at").ok(),
        finished_at: r
            .try_get::<Option<String>, _>("finished_at")
            .unwrap_or(None),
    });
    let counts = sqlx::query(
        "SELECT
             COALESCE(SUM(CASE WHEN exit_code = 0 THEN 1 ELSE 0 END), 0) AS ok,
             COALESCE(SUM(CASE WHEN exit_code IS NOT NULL AND exit_code <> 0 THEN 1 ELSE 0 END), 0) AS fail
           FROM execution_results
          WHERE job_id = ? AND finished_at IS NOT NULL AND recorded_at >= ?",
    )
    .bind(job_id)
    .bind(since)
    .fetch_one(pool)
    .await?;
    let recent = RecentCounts {
        window_hours: STATUS_WINDOW_HOURS,
        ok: counts.try_get("ok").unwrap_or(0),
        fail: counts.try_get("fail").unwrap_or(0),
    };
    Ok((last, recent))
}

/// GET /api/schedules/{id}/status — coverage view: enabled, next fire
/// (via `preview_fires`), the schedule's most recent run, and a 24h
/// ok/fail tally (#418 "カバレッジ可視化"). Read-only. The run figures
/// are `job_id`-keyed — see [`schedule_run_stats`].
pub async fn status(
    State(s): State<AppState>,
    Path(id): Path<String>,
) -> Result<Json<StatusResponse>, (StatusCode, String)> {
    let kv = s
        .jetstream
        .get_key_value(BUCKET_SCHEDULES)
        .await
        .map_err(|e| {
            (
                StatusCode::SERVICE_UNAVAILABLE,
                format!("schedules bucket missing: {e}"),
            )
        })?;
    let bytes = kv
        .get(&id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("KV get: {e}")))?
        .ok_or_else(|| (StatusCode::NOT_FOUND, format!("schedule '{id}' not found")))?;
    let schedule: Schedule = serde_json::from_slice(&bytes).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("deserialize stored schedule: {e}"),
        )
    })?;

    let now = chrono::Utc::now();
    let next_run = schedule
        .preview_fires(now, 1)
        .first()
        .map(chrono::DateTime::to_rfc3339);
    let since = now - chrono::Duration::hours(STATUS_WINDOW_HOURS);
    let (last_run, recent) = schedule_run_stats(&s.pool, &schedule.job_id, since)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("run stats: {e}")))?;

    Ok(Json(StatusResponse {
        id: schedule.id.clone(),
        when: schedule.when.to_string(),
        tz: schedule.tz.as_str().to_string(),
        enabled: schedule.enabled,
        next_run,
        last_run,
        recent,
    }))
}

/// POST /api/schedules — upsert.
///
/// Accepts JSON (`application/json`, default) or YAML
/// (`application/yaml`, `text/yaml`). YAML callers also populate the
/// parallel `BUCKET_SCHEDULES_YAML` so the SPA editor preserves
/// comments + formatting across edits. JSON callers fall back to a
/// `serde_yaml::to_string` mirror — best-effort, warn-logged on
/// failure.
pub async fn create(
    State(s): State<AppState>,
    caller: Caller,
    body: YamlOrJson<Schedule>,
) -> Result<Json<ScheduleSummary>, (StatusCode, String)> {
    let YamlOrJson {
        value: schedule,
        raw_yaml,
    } = body;

    // #418 decision F: reject broken schedules at create time
    // instead of letting them sit in KV and warn-skip every tick.
    // validate() covers the pure cross-field rules; the job_id
    // existence check needs the JOBS KV, so it lives here.
    if let Err(e) = schedule.validate() {
        return Err((StatusCode::BAD_REQUEST, format!("invalid schedule: {e}")));
    }
    match crate::api::jobs::fetch(&s.jetstream, &schedule.job_id).await {
        Ok(Some(_)) => {}
        Ok(None) => {
            return Err((
                StatusCode::BAD_REQUEST,
                format!(
                    "unknown job_id '{}' — register the job first (kanade job create)",
                    schedule.job_id
                ),
            ));
        }
        Err(e) => {
            return Err((
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("job catalog lookup: {e}"),
            ));
        }
    }

    // Make sure the KV bucket exists (idempotent).
    let kv = s
        .jetstream
        .create_key_value(KvConfig {
            bucket: BUCKET_SCHEDULES.into(),
            history: 5,
            ..Default::default()
        })
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("ensure KV: {e}")))?;

    let body_bytes = serde_json::to_vec(&schedule)
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("serialize: {e}")))?;
    kv.put(&schedule.id, body_bytes.into())
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("KV put: {e}")))?;

    // Operator-facing YAML mirror — best-effort, same reasoning as
    // jobs::create. Scheduler/agent read the JSON catalog; the YAML
    // store only feeds the SPA editor.
    let yaml_source = raw_yaml.unwrap_or_else(|| {
        serde_yaml::to_string(&schedule)
            .unwrap_or_else(|_| String::from("# YAML mirror unavailable for this entry"))
    });
    if let Err(e) = mirror_yaml(&s, BUCKET_SCHEDULES_YAML, &schedule.id, &yaml_source).await {
        warn!(
            error = %e,
            schedule_id = %schedule.id,
            "schedules: YAML mirror put failed; JSON catalog is current",
        );
    }

    info!(
        schedule_id = %schedule.id,
        when = %schedule.when,
        job_id = %schedule.job_id,
        "schedule upserted",
    );
    audit::record(
        &s.nats,
        "operator",
        "schedule_upsert",
        Some(&schedule.id),
        Some(&caller),
        serde_json::json!({
            "when": schedule.when.to_string(),
            "job_id": schedule.job_id,
            "enabled": schedule.enabled,
        }),
    )
    .await;
    Ok(Json(ScheduleSummary {
        id: schedule.id.clone(),
        when: schedule.when.to_string(),
        enabled: schedule.enabled,
        job_id: schedule.job_id.clone(),
    }))
}

/// `GET /api/schedules/{id}/yaml` — fetch the operator's YAML source
/// for the schedule. Falls back to a `serde_yaml::to_string` of the
/// JSON catalog row when the YAML mirror is missing (legacy entries
/// from before this endpoint).
pub async fn get_yaml(
    State(s): State<AppState>,
    Path(id): Path<String>,
) -> Result<(StatusCode, HeaderMap, String), (StatusCode, String)> {
    if let Ok(kv) = s.jetstream.get_key_value(BUCKET_SCHEDULES_YAML).await
        && let Ok(Some(bytes)) = kv.get(&id).await
        && let Ok(text) = String::from_utf8(bytes.to_vec())
    {
        return Ok((StatusCode::OK, yaml_headers(), text));
    }

    let kv = s
        .jetstream
        .get_key_value(BUCKET_SCHEDULES)
        .await
        .map_err(|_| (StatusCode::NOT_FOUND, format!("schedule '{id}' not found")))?;
    let bytes = kv
        .get(&id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("KV get: {e}")))?
        .ok_or_else(|| (StatusCode::NOT_FOUND, format!("schedule '{id}' not found")))?;
    let schedule: Schedule = serde_json::from_slice(&bytes)
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("decode: {e}")))?;
    let yaml = serde_yaml::to_string(&schedule).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("encode YAML: {e}"),
        )
    })?;
    Ok((StatusCode::OK, yaml_headers(), yaml))
}

/// Patch the top-level `enabled:` line in an operator's YAML source,
/// preserving every other line — comments and block-scalar formatting
/// are the reason the YAML mirror exists, so a full
/// `serde_yaml::to_string` re-render is not an option here. Column-0
/// match only, so an indented `enabled:` inside a nested map is left
/// alone. Appends the line when missing (legacy YAML that relied on
/// the serde default).
fn patch_yaml_enabled(yaml: &str, enabled: bool) -> String {
    let mut out = String::with_capacity(yaml.len() + 16);
    let mut found = false;
    for line in yaml.lines() {
        if !found && line.starts_with("enabled:") {
            // Replace the whole line. An inline comment here (e.g.
            // "# stopped during incident") describes the old value,
            // so dropping it alongside the flip is the lesser evil.
            out.push_str(&format!("enabled: {enabled}\n"));
            found = true;
        } else {
            out.push_str(line);
            out.push('\n');
        }
    }
    if !found {
        out.push_str(&format!("enabled: {enabled}\n"));
    }
    out
}

/// Best-effort sync of the YAML mirror after an enable/disable flip.
/// Without this, the SPA editor (which prefers `BUCKET_SCHEDULES_YAML`
/// over the JSON catalog, see [`get_yaml`]) would load the stale
/// `enabled` state and silently clobber the flip back on the next
/// save (gemini #400 review). Missing bucket / missing entry are
/// fine — `get_yaml` then falls back to rendering the (correct) JSON
/// catalog row. Write failures are warn-logged, same contract as
/// `mirror_yaml`: the JSON catalog the scheduler reads is already
/// current.
async fn sync_yaml_mirror_enabled(s: &AppState, id: &str, enabled: bool) {
    let Ok(kv) = s.jetstream.get_key_value(BUCKET_SCHEDULES_YAML).await else {
        return;
    };
    let Ok(Some(bytes)) = kv.get(id).await else {
        return;
    };
    let Ok(text) = String::from_utf8(bytes.to_vec()) else {
        warn!(schedule_id = %id, "YAML mirror is not UTF-8; skipping enabled sync");
        return;
    };
    let patched = patch_yaml_enabled(&text, enabled);
    if let Err(e) = kv.put(id, patched.into_bytes().into()).await {
        warn!(
            error = %e,
            schedule_id = %id,
            enabled,
            "YAML mirror enabled-flag sync failed; JSON catalog is current",
        );
    }
}

/// v0.27 — query params for [`disable`].
#[derive(Deserialize, Debug, Default)]
pub struct DisableQuery {
    /// When `true`, also Layer 2 cascade-revoke the underlying Job
    /// so any in-flight Command for `schedule.job_id` gets skipped
    /// by the agent's `handle_command` KV check. SPEC §2.6.4 (c)
    /// "hard disable". Default `false` = soft disable (cron stops,
    /// in-flight runs to completion).
    #[serde(default)]
    pub cascade: bool,
    /// When `true`, also Layer 3 cascade-KILL — publish `kill.{exec_id}`
    /// for every still-running exec of `schedule.job_id` so currently-
    /// executing child processes are terminated now (SPEC §2.6.4 (c)).
    /// Orthogonal to `cascade`: kill stops *running* work, revoke stops
    /// *queued/future* work — combine both for a full hard-disable.
    /// Online-only (a kill can't reach an offline agent's child).
    /// Default `false` — killing in-flight work is a deliberate,
    /// destructive opt-in.
    #[serde(default)]
    pub cascade_kill: bool,
}

/// POST /api/schedules/{id}/disable
///
/// Two flavours, controlled by the `?cascade=` query param:
///
/// * **soft disable** (default, `cascade=false`): flip `enabled =
///   false` on the schedule in `BUCKET_SCHEDULES`. The cron loop
///   stops firing on the next watch tick (backend `scheduler.rs` +
///   agent `local_scheduler.rs` both watch this bucket). Already-
///   fired Commands are left alone — they run to completion or fail
///   on their own merits.
///
/// * **hard disable** (`cascade=true`): SPEC §2.6.4 (c). Soft disable
///   PLUS Layer 2 cascade — write
///   `script_status.{schedule.job_id} = REVOKED` so any Command
///   already in flight (live core sub delivery in progress or
///   sitting in `STREAM_EXEC` awaiting a reconnecting agent) gets
///   caught by the agent's Layer 2 KV check and skipped. The kill
///   cascade of *currently-running children* (Layer 3) is **not**
///   part of this PR — operators can follow up with
///   `kanade kill <job_id>` per execution. Tracked for v0.28 (needs
///   `executions.schedule_id` to find the in-flight job_ids the
///   schedule produced).
pub async fn disable(
    State(s): State<AppState>,
    Path(id): Path<String>,
    Query(q): Query<DisableQuery>,
    caller: Caller,
) -> Result<StatusCode, (StatusCode, String)> {
    let schedules_kv = s
        .jetstream
        .get_key_value(BUCKET_SCHEDULES)
        .await
        .map_err(|e| {
            warn!(error = %e, "schedules KV missing on disable");
            (
                StatusCode::SERVICE_UNAVAILABLE,
                format!("schedules bucket missing: {e}"),
            )
        })?;

    // Fetch the full Entry (not just the value) so we can use its
    // revision for an optimistic-concurrency `update` instead of a
    // blind `put`. Without that, a concurrent edit (operator changing
    // the cron expression while we're racing to disable) would be
    // silently clobbered — gemini #37 review flagged this as a
    // priority bug, and it lines up with the PR's "stop the rollout"
    // story where two operators reaching for the brake at once is a
    // realistic scenario.
    let entry = schedules_kv
        .entry(&id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("KV entry: {e}")))?
        .ok_or_else(|| (StatusCode::NOT_FOUND, format!("schedule '{id}' not found")))?;
    let mut schedule: Schedule = serde_json::from_slice(&entry.value).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("deserialize stored schedule: {e}"),
        )
    })?;

    // Only write back if there's something to change. Skipping the
    // already-disabled case avoids a redundant watch event for the
    // backend / agent scheduler loops.
    if schedule.enabled {
        schedule.enabled = false;
        let body = serde_json::to_vec(&schedule).map_err(|e| {
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("serialize schedule: {e}"),
            )
        })?;
        schedules_kv
            .update(&id, body.into(), entry.revision)
            .await
            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("KV update: {e}")))?;
        sync_yaml_mirror_enabled(&s, &id, false).await;
    } else {
        info!(schedule_id = %id, "schedule already disabled; revoke-only path");
    }

    // Cascade Layer 2: revoke the underlying Manifest so already-
    // published Commands get caught at agent fire time. Same pattern
    // as `jobs::delete`: revoke is idempotent, status KV missing in
    // dev is a 503 so callers can `kanade jetstream setup` and retry.
    let cascade_applied = if q.cascade {
        let status_kv = s
            .jetstream
            .get_key_value(BUCKET_SCRIPT_STATUS)
            .await
            .map_err(|e| {
                warn!(
                    error = %e,
                    bucket = BUCKET_SCRIPT_STATUS,
                    "schedule_disable cascade: status KV unavailable",
                );
                (
                    StatusCode::SERVICE_UNAVAILABLE,
                    format!("script_status bucket missing: {e}"),
                )
            })?;
        status_kv
            .put(&schedule.job_id, bytes::Bytes::from(SCRIPT_STATUS_REVOKED))
            .await
            .map_err(|e| {
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    format!("script_status put: {e}"),
                )
            })?;
        info!(
            schedule_id = %id,
            job_id = %schedule.job_id,
            "schedule disabled with cascade revoke",
        );
        true
    } else {
        info!(schedule_id = %id, "schedule soft-disabled");
        false
    };

    // Cascade Layer 3: kill currently-running children of this
    // schedule's job (SPEC §2.6.4 (c)). Enumerate the still-in-flight
    // execs (a `finished_at IS NULL` row in `execution_results` for
    // this `job_id`) and publish `kill.{exec_id}` for each — the agent
    // that's running it terminates the child via
    // `run_command_with_kill`. Orthogonal to the revoke above: this
    // stops *running* work, revoke stops *queued/future* work. Best-
    // effort + online-only: a kill can't reach an offline agent's
    // child, and a DB hiccup degrades to "no kills" (warn) rather than
    // failing the disable, which already took effect on the KV above.
    let killed_execs = if q.cascade_kill {
        match sqlx::query_scalar::<_, String>(
            "SELECT DISTINCT exec_id FROM execution_results \
             WHERE job_id = ? AND finished_at IS NULL AND exec_id IS NOT NULL",
        )
        .bind(&schedule.job_id)
        .fetch_all(&s.pool)
        .await
        {
            Ok(exec_ids) => {
                // Publish the kills concurrently rather than awaiting
                // each in series (gemini #480) — with many in-flight
                // execs the per-publish round-trips would otherwise add
                // up. Each publish is independent; failures are logged,
                // never abort the others.
                futures::future::join_all(exec_ids.iter().map(|eid| {
                    let nats = s.nats.clone();
                    let eid = eid.clone();
                    async move {
                        if let Err(e) = nats
                            .publish(kanade_shared::subject::kill(&eid), bytes::Bytes::new())
                            .await
                        {
                            warn!(error = %e, exec_id = %eid, "schedule_disable cascade-kill: publish failed");
                        }
                    }
                }))
                .await;
                // Flush so the kills actually leave before we return —
                // otherwise a fast caller could disconnect first.
                if let Err(e) = s.nats.flush().await {
                    warn!(error = %e, "schedule_disable cascade-kill: flush failed");
                }
                info!(
                    schedule_id = %id,
                    job_id = %schedule.job_id,
                    count = exec_ids.len(),
                    "schedule disabled with cascade kill (in-flight execs signalled)",
                );
                exec_ids.len()
            }
            Err(e) => {
                warn!(error = %e, job_id = %schedule.job_id, "schedule_disable cascade-kill: in-flight exec query failed; no kills sent");
                0
            }
        }
    } else {
        0
    };

    audit::record(
        &s.nats,
        "operator",
        "schedule_disable",
        Some(&id),
        Some(&caller),
        serde_json::json!({
            "cascade": cascade_applied,
            "cascade_kill": q.cascade_kill,
            "killed_execs": killed_execs,
            "job_id": schedule.job_id,
        }),
    )
    .await;
    Ok(StatusCode::NO_CONTENT)
}

/// POST /api/schedules/{id}/enable
///
/// Symmetrical to [`disable`]'s soft path: flip `enabled = true` on
/// the schedule in `BUCKET_SCHEDULES` so the cron loops (backend
/// `scheduler.rs` + agent `local_scheduler.rs`) pick it back up on
/// the next watch tick. Uses the same `kv.entry().revision` +
/// `update()` optimistic-concurrency pattern as `disable` so an
/// enable click can't clobber a concurrent cron/target edit.
///
/// Note: this only re-arms the cron. It does NOT touch
/// `script_status` — a job revoked by a hard disable stays REVOKED
/// until the operator runs `kanade unrevoke <job_id>` explicitly.
/// Silently un-revoking here would defeat the point of the Layer 2
/// brake.
///
/// History: the SPA has called this endpoint since PR #38, but the
/// backend handler was lost in that PR's squash merge — every Enable
/// click 404'd. Restored here with a `schedule_enable` audit record.
pub async fn enable(
    State(s): State<AppState>,
    Path(id): Path<String>,
    caller: Caller,
) -> Result<StatusCode, (StatusCode, String)> {
    let kv = s
        .jetstream
        .get_key_value(BUCKET_SCHEDULES)
        .await
        .map_err(|e| {
            warn!(error = %e, "schedules KV missing on enable");
            (
                StatusCode::SERVICE_UNAVAILABLE,
                format!("schedules bucket missing: {e}"),
            )
        })?;
    let entry = kv
        .entry(&id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("KV entry: {e}")))?
        .ok_or_else(|| (StatusCode::NOT_FOUND, format!("schedule '{id}' not found")))?;
    let mut schedule: Schedule = serde_json::from_slice(&entry.value).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("deserialize stored schedule: {e}"),
        )
    })?;

    if schedule.enabled {
        info!(schedule_id = %id, "schedule already enabled; no-op");
        return Ok(StatusCode::NO_CONTENT);
    }
    schedule.enabled = true;
    let body = serde_json::to_vec(&schedule).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("serialize schedule: {e}"),
        )
    })?;
    kv.update(&id, body.into(), entry.revision)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("KV update: {e}")))?;
    sync_yaml_mirror_enabled(&s, &id, true).await;

    info!(schedule_id = %id, "schedule enabled");
    audit::record(
        &s.nats,
        "operator",
        "schedule_enable",
        Some(&id),
        Some(&caller),
        serde_json::json!({
            "job_id": schedule.job_id,
        }),
    )
    .await;
    Ok(StatusCode::NO_CONTENT)
}

/// DELETE /api/schedules/{id}
pub async fn delete(
    State(s): State<AppState>,
    Path(id): Path<String>,
    caller: Caller,
) -> Result<StatusCode, (StatusCode, String)> {
    let kv = match s.jetstream.get_key_value(BUCKET_SCHEDULES).await {
        Ok(k) => k,
        Err(e) => {
            warn!(error = %e, "schedules KV missing on delete");
            return Err((StatusCode::NOT_FOUND, "schedules bucket missing".into()));
        }
    };
    kv.delete(&id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("kv delete: {e}")))?;
    info!(schedule_id = %id, "schedule deleted");
    audit::record(
        &s.nats,
        "operator",
        "schedule_delete",
        Some(&id),
        Some(&caller),
        serde_json::json!({}),
    )
    .await;
    Ok(StatusCode::NO_CONTENT)
}

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

    #[test]
    fn flips_existing_flag_and_preserves_comments() {
        let yaml = "# nightly inventory sweep\nid: inv-hw\ncron: \"0 3 * * *\"\nenabled: false\njob_id: inventory-hw\n";
        let out = patch_yaml_enabled(yaml, true);
        assert!(out.contains("enabled: true\n"));
        assert!(!out.contains("enabled: false"));
        // Comments and the other keys survive untouched.
        assert!(out.starts_with("# nightly inventory sweep\n"));
        assert!(out.contains("cron: \"0 3 * * *\"\n"));
    }

    #[test]
    fn drops_inline_comment_on_the_flipped_line_only() {
        let yaml = "id: s1\nenabled: true # stopped during incident\ncron: \"* * * * *\"\n";
        let out = patch_yaml_enabled(yaml, false);
        assert!(out.contains("enabled: false\n"));
        assert!(!out.contains("stopped during incident"));
        assert!(out.contains("cron: \"* * * * *\"\n"));
    }

    #[test]
    fn ignores_indented_enabled_in_nested_maps() {
        let yaml = "id: s1\ntarget:\n  enabled: false\nenabled: false\n";
        let out = patch_yaml_enabled(yaml, true);
        // Top-level flipped, nested left alone.
        assert!(out.contains("\nenabled: true\n"));
        assert!(out.contains("  enabled: false\n"));
    }

    #[test]
    fn appends_when_missing() {
        let yaml = "id: s1\ncron: \"* * * * *\"\n";
        let out = patch_yaml_enabled(yaml, false);
        assert!(out.ends_with("enabled: false\n"));
        assert!(out.starts_with("id: s1\n"));
    }

    #[test]
    fn only_first_top_level_occurrence_is_patched() {
        // Duplicate top-level keys are invalid YAML, but the patcher
        // shouldn't multiply writes if one sneaks in.
        let yaml = "enabled: false\nenabled: false\n";
        let out = patch_yaml_enabled(yaml, true);
        assert_eq!(out.matches("enabled: true").count(), 1);
    }

    // ---- schedule_run_stats (#418 coverage view) ----

    use super::schedule_run_stats;
    use chrono::{Duration, Utc};
    use sqlx::SqlitePool;

    async fn fresh_pool() -> SqlitePool {
        let pool = sqlx::sqlite::SqlitePoolOptions::new()
            .connect("sqlite::memory:")
            .await
            .unwrap();
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();
        pool
    }

    /// Insert one `execution_results` row. `exit_code: None` +
    /// `finished: false` models an in-flight run. `recorded_ago_min`
    /// bounds it relative to "now". Every timestamp is bound via chrono
    /// (RFC 3339), matching how the projector writes — the `since`
    /// comparison breaks otherwise (#390).
    async fn insert_exec(
        pool: &SqlitePool,
        result_id: &str,
        job_id: &str,
        exit_code: Option<i64>,
        finished: bool,
        recorded_ago_min: i64,
    ) {
        let now = Utc::now();
        let recorded = now - Duration::minutes(recorded_ago_min);
        let started = recorded - Duration::minutes(1);
        let finished_at = finished.then_some(recorded);
        sqlx::query(
            "INSERT INTO execution_results
                (result_id, request_id, pc_id, exit_code, stdout, stderr,
                 started_at, finished_at, recorded_at, job_id)
             VALUES (?, 'req', 'pc-1', ?, '', '', ?, ?, ?, ?)",
        )
        .bind(result_id)
        .bind(exit_code)
        .bind(started)
        .bind(finished_at)
        .bind(recorded)
        .bind(job_id)
        .execute(pool)
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn run_stats_tallies_recent_and_picks_latest() {
        let pool = fresh_pool().await;
        // job j1: an old ok (outside 24h), a fail, an ok, then a still-
        // running row that is the most recent. job j2 must not bleed in.
        insert_exec(&pool, "old", "j1", Some(0), true, 60 * 30).await; // 30h ago
        insert_exec(&pool, "fail", "j1", Some(1), true, 120).await;
        insert_exec(&pool, "ok", "j1", Some(0), true, 60).await;
        insert_exec(&pool, "running", "j1", None, false, 10).await; // latest
        insert_exec(&pool, "other", "j2", Some(1), true, 60).await;

        let since = Utc::now() - Duration::hours(24);
        let (last, recent) = schedule_run_stats(&pool, "j1", since).await.unwrap();

        // Most recent row wins, in-flight (no exit / finished) surfaced.
        let last = last.expect("j1 has runs");
        assert_eq!(last.exit_code, None);
        assert!(last.finished_at.is_none());
        // 24h tally: the ok and the fail; the 30h-old ok and the still-
        // running row are excluded; j2 is a different job.
        assert_eq!(recent.ok, 1);
        assert_eq!(recent.fail, 1);
        assert_eq!(recent.window_hours, 24);
    }

    #[tokio::test]
    async fn run_stats_empty_for_unknown_job() {
        let pool = fresh_pool().await;
        insert_exec(&pool, "x", "j1", Some(0), true, 10).await;
        let since = Utc::now() - Duration::hours(24);
        let (last, recent) = schedule_run_stats(&pool, "no-such-job", since)
            .await
            .unwrap();
        assert!(last.is_none());
        assert_eq!((recent.ok, recent.fail), (0, 0));
    }
}