supercode-harness 0.4.20

//! Controlled-tier scheduled jobs (Domain 11, concept 6) — the FIRST
//! controlled-tier noun, and the shape the rest of wave 2 copies.
//!
//! Charter (`docs/plans/orchestration-domain-11-2026-09-02.md` §0.4):
//! **supercode never runs a cron engine.** Every mutation here is the
//! harness's OWN verb, executed as a subprocess, with supercode acting as the
//! uniform client:
//!
//! * **Hermes** — `hermes cron create | edit | pause | resume | run | remove`
//!   with `HERMES_HOME` in the environment (a profile IS a HERMES_HOME:
//!   upstream `hermes_cli/profiles.py` spawns profile work with
//!   `HERMES_HOME=<home>/profiles/<name>`).
//! * **OpenClaw** — `openclaw cron add | edit | disable | enable | run | rm`.
//!   Every one of these goes through the Gateway websocket, so the endpoint
//!   and credential are resolved from OPENCLAW's OWN config
//!   (`<state dir>/openclaw.json`, pointers `/gateway/remote/url`,
//!   `/gateway/port`, `/gateway/auth/token`) through the very same
//!   [`crate::RuntimeConnectLaunch`] the connect descriptor uses — never from
//!   supercode's own config, and never from an inherited environment variable.
//! * **The orchestrator** — its own package (ORC-13). The write door is the
//!   daemon's local socket while it is up and `node bin/orchestrator.mjs
//!   <op> …` when it is down, both landing in the SAME `applyOperator` →
//!   reducer → `save()` path inside `sdk/orchestrator`, which owns the
//!   folder's byte-stability and its residue rules
//!   (`docs/ORCHESTRATOR-IR.md` §4.6, §6). supercode writes no file of that
//!   folder itself; [`crate::orchestrator_door`] is the uniform client.
//! * **Claude Code** — refused. Its jobs are session-scoped runtime state
//!   created by the model inside a session (`CronCreate`); the harness
//!   publishes no verb a client can call.
//!
//! Three rules the whole tier inherits:
//!
//! 1. **The harness's answer is the answer.** After the verb exits 0 the row
//!    is re-read through the ORCH-7 loader ([`crate::jobs`]) and returned. A
//!    non-zero exit surfaces the harness's own stderr as the error — never a
//!    silent success, never a supercode-invented row.
//! 2. **The command is narrated.** Every outcome carries `ran`: the exact
//!    argv that was executed, with any credential rendered as `<redacted>`.
//!    Tokens are never printed, logged, or stored.
//! 3. **A field the harness has no verb for is refused**
//!    ([`JobControlError::Unsupported`] → `UnsupportedAction`), never dropped.

use std::collections::BTreeSet;
use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};
use serde_json::Value;

pub(crate) use crate::harness_command::shell_quote;
pub(crate) use crate::harness_command::HarnessCommand;
use crate::{HarnessHomes, HarnessId, ScheduledJob};

pub use crate::harness_command::{HERMES_BIN_ENV, OPENCLAW_BIN_ENV};

/// Harnesses whose scheduled jobs supercode can MUTATE through their own CLI
/// verb. Strictly narrower than [`crate::jobs::JOB_HARNESSES`]: Claude Code is
/// readable but not controllable.
pub const CONTROLLED_JOB_HARNESSES: &[&str] = &[
    HarnessId::HERMES,
    HarnessId::OPENCLAW,
    HarnessId::ORCHESTRATOR,
];

/// Why Claude Code refuses every mutating job verb.
pub const CLAUDE_CODE_REFUSAL: &str =
    "claude-code scheduled jobs are session-scoped runtime state: they are created by the model \
     inside a session (`CronCreate`) and restored on resume. Claude Code publishes no harness verb \
     a client can call, so supercode refuses rather than inventing one";

/// One uniform mutating verb.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum JobVerb {
    /// Create a new scheduled job.
    Create,
    /// Patch an existing job's fields.
    Update,
    /// Stop the scheduler from firing a job.
    Pause,
    /// Let the scheduler fire a job again.
    Resume,
    /// Fire a job now, out of schedule.
    Run,
    /// Remove a job.
    Delete,
}

impl JobVerb {
    /// Uniform spelling used in the RPC method and in outcomes.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Create => "create",
            Self::Update => "update",
            Self::Pause => "pause",
            Self::Resume => "resume",
            Self::Run => "run",
            Self::Delete => "delete",
        }
    }

    /// Whether the verb needs an existing job id.
    const fn needs_id(self) -> bool {
        !matches!(self, Self::Create)
    }
}

/// Uniform firing rule for a create/update.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct JobScheduleSpec {
    /// `interval` | `cron` | `once`.
    pub kind: String,
    /// Interval length, for `kind = "interval"`.
    pub minutes: Option<f64>,
    /// Cron expression, for `kind = "cron"`.
    pub expr: Option<String>,
    /// Absolute instant, for `kind = "once"`.
    pub run_at: Option<String>,
}

/// Uniform payload for a create/update.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct JobPayloadSpec {
    /// `prompt` | `system_event` | `command` | `script`.
    pub kind: String,
    /// The prompt, event, command line, or script the fire carries.
    pub text: Option<String>,
}

/// Uniform delivery for a create/update.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct JobDeliverSpec {
    /// Hermes `deliver` grammar (`origin` | `local` | `<platform>`), or an
    /// OpenClaw delivery mode (`announce` | `webhook` | `none`).
    pub target: Option<String>,
    /// Chat / destination the delivery is addressed to.
    pub chat_id: Option<String>,
}

/// One mutating request, in the uniform Domain 11 vocabulary.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct JobMutation {
    /// Harness that owns the job.
    pub harness: String,
    /// Job id, for every verb but `create`.
    #[serde(default)]
    pub id: Option<String>,
    /// Human-friendly job name.
    #[serde(default)]
    pub name: Option<String>,
    /// When the job fires.
    #[serde(default)]
    pub schedule: Option<JobScheduleSpec>,
    /// What fires.
    #[serde(default)]
    pub payload: Option<JobPayloadSpec>,
    /// OpenClaw `sessionTarget` (`main` | `isolated`).
    #[serde(default)]
    pub session_target: Option<String>,
    /// Where the fire's output goes.
    #[serde(default)]
    pub deliver: Option<JobDeliverSpec>,
    /// Hermes profile name / OpenClaw agent id.
    #[serde(default)]
    pub profile: Option<String>,
    /// Storage roots, so an isolated home is addressed the same way the
    /// read side addresses it.
    #[serde(default)]
    pub homes: HarnessHomes,
}

/// What one mutation did, with the harness's own row read back afterwards.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct JobMutationOutcome {
    /// Harness that ran the verb.
    pub harness: String,
    /// Uniform verb that was asked for.
    pub verb: String,
    /// The exact harness command that ran, credentials redacted.
    pub ran: String,
    /// Affected job id.
    pub id: String,
    /// The job as the harness's own store reports it AFTER the verb.
    /// Absent for `delete`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub job: Option<ScheduledJob>,
    /// `true` on a successful `delete`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub deleted: Option<bool>,
}

/// Why a mutation could not be performed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum JobControlError {
    /// The harness has no verb for what was asked (refused, never faked).
    Unsupported(String),
    /// The request itself is incoherent.
    Invalid(String),
    /// The harness verb ran and failed; the message carries its stderr.
    Failed(String),
}

impl std::fmt::Display for JobControlError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Unsupported(message) | Self::Invalid(message) | Self::Failed(message) => {
                formatter.write_str(message)
            }
        }
    }
}

impl std::error::Error for JobControlError {}

type Result<T> = std::result::Result<T, JobControlError>;

/// Whether `harness` can have its scheduled jobs mutated at all.
pub fn supports_job_control(harness: &str) -> bool {
    CONTROLLED_JOB_HARNESSES.contains(&harness)
}

pub fn harness_program(harness: &str) -> Result<String> {
    crate::harness_command::harness_program(harness).map_err(|detail| {
        JobControlError::Unsupported(detail.unwrap_or_else(|| unsupported_harness(harness)))
    })
}

fn unsupported_harness(harness: &str) -> String {
    if harness == HarnessId::CLAUDE_CODE {
        return CLAUDE_CODE_REFUSAL.to_string();
    }
    format!(
        "`{harness}` has no mutable scheduled jobs; mutating job verbs are supported for: {}",
        CONTROLLED_JOB_HARNESSES.join(", ")
    )
}

/// `HERMES_HOME` for this request: the profile's own home when one is named
/// (upstream treats a profile as a full HERMES_HOME), else the install root.
fn hermes_home(mutation: &JobMutation) -> PathBuf {
    // `HarnessHomes::hermes` addresses `state.db`; HERMES_HOME is its parent,
    // the same derivation the read side uses.
    let root = mutation
        .homes
        .hermes
        .parent()
        .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
    match mutation.profile.as_deref() {
        Some(profile) => root.join("profiles").join(profile),
        None => root,
    }
}

/// Resolve OpenClaw's gateway endpoint and credential from OPENCLAW's own
/// config, through the registry's connect descriptor.
///
/// The pointers are never re-spelled here: the descriptor
/// (`/gateway/remote/url` → `/gateway/port` → the documented default, auth at
/// `/gateway/auth/token`) is taken from the compiled registry and only its
/// `config_path` is re-anchored onto the state dir the caller addressed, so an
/// isolated home resolves ITS token and the default home resolves the real
/// one. supercode's own config is never consulted.
fn openclaw_connection(homes: &HarnessHomes) -> Result<crate::ResolvedRuntimeConnection> {
    let registry = crate::harness_support_registry();
    let descriptor = registry
        .harnesses
        .iter()
        .find(|descriptor| descriptor.id.as_str() == HarnessId::OPENCLAW)
        .ok_or_else(|| {
            JobControlError::Unsupported("the registry has no openclaw descriptor".into())
        })?;
    let connect = descriptor.runtime.connect_launch.as_ref().ok_or_else(|| {
        JobControlError::Unsupported(
            "openclaw has no registered connect-mode launch, so its gateway cannot be located"
                .into(),
        )
    })?;
    let mut connect = connect.clone();
    connect.config_path = homes
        .openclaw
        .join("openclaw.json")
        .to_string_lossy()
        .into_owned();
    connect.resolve(Path::new("/")).map_err(|error| {
        JobControlError::Unsupported(format!(
            "openclaw's gateway endpoint could not be resolved from its own config: {error}"
        ))
    })
}

/// Perform one mutation: translate to the harness's own verb, run it, then
/// re-read the row through the ORCH-7 loader.
pub fn mutate(verb: JobVerb, mutation: &JobMutation) -> Result<JobMutationOutcome> {
    if !supports_job_control(&mutation.harness) {
        return Err(JobControlError::Unsupported(unsupported_harness(
            &mutation.harness,
        )));
    }
    if verb.needs_id() && mutation.id.as_deref().unwrap_or("").trim().is_empty() {
        return Err(JobControlError::Invalid(format!(
            "`jobs.{}` needs the job id to act on",
            verb.as_str()
        )));
    }
    if matches!(verb, JobVerb::Create) && mutation.schedule.is_none() {
        return Err(JobControlError::Invalid(
            "`jobs.create` needs a schedule (interval, cron, or once)".into(),
        ));
    }
    // ORC-13: the orchestrator's verb is not a CLI subprocess but its own
    // package's operator door, so it branches before the command table.
    if mutation.harness == HarnessId::ORCHESTRATOR {
        return orchestrator_mutate(verb, mutation);
    }
    let command = match mutation.harness.as_str() {
        HarnessId::HERMES => hermes_command(verb, mutation)?,
        HarnessId::OPENCLAW => openclaw_command(verb, mutation)?,
        other => return Err(JobControlError::Unsupported(unsupported_harness(other))),
    };
    let ran = command.narrate();
    let before = matches!(verb, JobVerb::Create).then(|| known_ids(mutation));
    let stdout = command.run().map_err(JobControlError::Failed)?;
    let id = match (verb, before) {
        (JobVerb::Create, Some(before)) => created_id(mutation, &before, &stdout, &ran)?,
        _ => mutation.id.clone().unwrap_or_default(),
    };
    // The harness's own store is the answer: re-read, never echo the request.
    let read = crate::jobs::get_job(&mutation.harness, &id, &mutation.homes).map_err(|error| {
        JobControlError::Failed(format!(
            "`{ran}` succeeded but the job store could not be re-read: {error}"
        ))
    })?;
    match verb {
        JobVerb::Delete => {
            if read.is_some() {
                return Err(JobControlError::Failed(format!(
                    "`{ran}` reported success but `{id}` is still in {}'s job store",
                    mutation.harness
                )));
            }
            Ok(JobMutationOutcome {
                harness: mutation.harness.clone(),
                verb: verb.as_str().to_string(),
                ran,
                id,
                job: None,
                deleted: Some(true),
            })
        }
        _ => {
            let (job, _) = read.ok_or_else(|| {
                JobControlError::Failed(format!(
                    "`{ran}` reported success but `{}` has no job `{id}` afterwards",
                    mutation.harness
                ))
            })?;
            Ok(JobMutationOutcome {
                harness: mutation.harness.clone(),
                verb: verb.as_str().to_string(),
                ran,
                id,
                job: Some(job),
                deleted: None,
            })
        }
    }
}

// ---------------------------------------------------------------------------
// The orchestrator — its own package's operator door (ORC-13)
// ---------------------------------------------------------------------------

/// The orchestrator profile this mutation acts in: `--profile`, else the
/// root folder, which IS the `default` profile (`docs/ORCHESTRATOR-IR.md` §6).
fn orchestrator_profile(mutation: &JobMutation) -> &str {
    mutation
        .profile
        .as_deref()
        .map(str::trim)
        .filter(|profile| !profile.is_empty())
        .unwrap_or("default")
}

/// The uniform row translated onto the orchestrator's OWN job vocabulary
/// (`docs/ORCHESTRATOR-IR.md` §2.6) — the same words its MCP tools and its
/// chat commands use. A field the model has no home for is refused by name,
/// never dropped.
fn orchestrator_args(verb: JobVerb, mutation: &JobMutation) -> Result<Value> {
    let mut args = serde_json::Map::new();
    if let Some(id) = mutation.id.as_deref().filter(|id| !id.trim().is_empty()) {
        args.insert("id".into(), Value::String(id.trim().to_string()));
    }
    if matches!(verb, JobVerb::Create | JobVerb::Update) {
        if mutation.session_target.is_some() {
            return Err(JobControlError::Unsupported(
                "an orchestrator cron fire opens its own binding on the job's origin surface \
                 (`docs/ORCHESTRATOR-IR.md` §4.3); the model has no session-target field, so \
                 supercode refuses rather than dropping it"
                    .into(),
            ));
        }
        if let Some(name) = &mutation.name {
            args.insert("name".into(), Value::String(name.clone()));
        }
        if let Some(schedule) = &mutation.schedule {
            args.insert("schedule".into(), orchestrator_schedule(schedule)?);
        }
        if let Some(payload) = &mutation.payload {
            match payload_kind(payload) {
                "prompt" => {
                    args.insert(
                        "prompt".into(),
                        Value::String(payload_text(payload)?.to_string()),
                    );
                }
                other => {
                    return Err(JobControlError::Unsupported(format!(
                        "an orchestrator job carries a `prompt` — the fire opens a worker session \
                         and sends it (§4.3); there is no `{other}` payload, so supercode refuses \
                         rather than inventing one"
                    )))
                }
            }
        }
        if let Some(deliver) = &mutation.deliver {
            if let Some(target) = hermes_deliver(deliver) {
                // The orchestrator's `deliver` grammar IS Hermes's
                // (`origin | local | home | <platform>[:<chat_id>]`, §2.6).
                args.insert("deliver".into(), Value::String(target));
            }
        }
    } else if mutation.name.is_some()
        || mutation.schedule.is_some()
        || mutation.payload.is_some()
        || mutation.deliver.is_some()
        || mutation.session_target.is_some()
    {
        return Err(JobControlError::Invalid(format!(
            "`jobs.{}` changes no fields; pass definition fields to `jobs.update`",
            verb.as_str()
        )));
    }
    Ok(Value::Object(args))
}

/// The uniform schedule in the orchestrator's typed form (§2.6).
fn orchestrator_schedule(schedule: &JobScheduleSpec) -> Result<Value> {
    match schedule.kind.as_str() {
        "interval" => schedule
            .minutes
            .map(|minutes| serde_json::json!({"kind": "interval", "minutes": minutes}))
            .ok_or_else(|| JobControlError::Invalid("an interval schedule needs `minutes`".into())),
        "cron" => schedule
            .expr
            .as_deref()
            .map(|expr| serde_json::json!({"kind": "cron", "expr": expr}))
            .ok_or_else(|| JobControlError::Invalid("a cron schedule needs `expr`".into())),
        "once" => schedule
            .run_at
            .as_deref()
            .map(|run_at| serde_json::json!({"kind": "once", "run_at": run_at}))
            .ok_or_else(|| JobControlError::Invalid("a once schedule needs `run_at`".into())),
        other => Err(JobControlError::Invalid(format!(
            "unknown schedule kind `{other}`; use interval, cron, or once"
        ))),
    }
}

/// One orchestrator job mutation: through the package's door, then re-read
/// through the ORC-7 loader like every other harness's row.
fn orchestrator_mutate(verb: JobVerb, mutation: &JobMutation) -> Result<JobMutationOutcome> {
    let args = orchestrator_args(verb, mutation)?;
    let root = mutation.homes.orchestrator.clone();
    let profile = orchestrator_profile(mutation);
    let op = format!("jobs.{}", verb.as_str());
    let answer = crate::orchestrator_door::call(&root, &op, &args, profile).map_err(|error| {
        match error {
            // The package refused: its sentence is the answer, in the same
            // shape a harness's stderr takes for the other two.
            crate::orchestrator_door::DoorError::Refused(message) => {
                JobControlError::Failed(message)
            }
            crate::orchestrator_door::DoorError::Failed(message) => {
                JobControlError::Failed(message)
            }
        }
    })?;
    let ran = format!("{} [{}]", answer.ran, answer.door.as_str());
    let id = answer
        .result
        .pointer("/job_id")
        .and_then(Value::as_str)
        .map(str::to_string)
        .or_else(|| mutation.id.clone())
        .ok_or_else(|| JobControlError::Failed(format!("`{ran}` succeeded but named no job id")))?;
    // The FOLDER is the answer, re-read through the same loader `jobs list`
    // uses — never the door's echo of what it wrote.
    let read = crate::jobs::get_job(&mutation.harness, &id, &mutation.homes).map_err(|error| {
        JobControlError::Failed(format!(
            "`{ran}` succeeded but the job store could not be re-read: {error}"
        ))
    })?;
    match verb {
        JobVerb::Delete => {
            if read.is_some() {
                return Err(JobControlError::Failed(format!(
                    "`{ran}` reported success but `{id}` is still in the orchestrator's job store"
                )));
            }
            Ok(JobMutationOutcome {
                harness: mutation.harness.clone(),
                verb: verb.as_str().to_string(),
                ran,
                id,
                job: None,
                deleted: Some(true),
            })
        }
        _ => {
            let (job, _) = read.ok_or_else(|| {
                JobControlError::Failed(format!(
                    "`{ran}` reported success but the orchestrator has no job `{id}` afterwards"
                ))
            })?;
            Ok(JobMutationOutcome {
                harness: mutation.harness.clone(),
                verb: verb.as_str().to_string(),
                ran,
                id,
                job: Some(job),
                deleted: None,
            })
        }
    }
}

/// Every job id the harness's store holds right now.
fn known_ids(mutation: &JobMutation) -> BTreeSet<String> {
    crate::jobs::list_jobs(&crate::jobs::JobsQuery {
        harness: Some(mutation.harness.clone()),
        homes: mutation.homes.clone(),
        ..crate::jobs::JobsQuery::default()
    })
    .map(|listing| listing.jobs.into_iter().map(|job| job.id).collect())
    .unwrap_or_default()
}

/// Identify the job the create verb just made: the id the harness's own store
/// gained. When several appeared (a concurrent writer), the harness's stdout
/// decides between them.
fn created_id(
    mutation: &JobMutation,
    before: &BTreeSet<String>,
    stdout: &str,
    ran: &str,
) -> Result<String> {
    let after = known_ids(mutation);
    let mut fresh: Vec<String> = after.difference(before).cloned().collect();
    if fresh.len() == 1 {
        return Ok(fresh.remove(0));
    }
    if let Some(named) = fresh.iter().find(|id| stdout.contains(id.as_str())) {
        return Ok(named.clone());
    }
    // Last resort: an id the harness printed that the store now holds (a
    // store that reuses an existing id, e.g. an idempotent declaration key).
    if let Some(id) = stdout_id(stdout).filter(|id| after.contains(id)) {
        return Ok(id);
    }
    Err(JobControlError::Failed(format!(
        "`{ran}` reported success but {} gained {} job(s), so the new job cannot be identified",
        mutation.harness,
        fresh.len()
    )))
}

/// An `id` field from a harness's JSON stdout, when it prints one.
fn stdout_id(stdout: &str) -> Option<String> {
    let value: Value = serde_json::from_str(stdout.trim()).ok()?;
    for pointer in ["/id", "/job/id", "/jobId", "/job_id", "/result/id"] {
        if let Some(id) = value.pointer(pointer).and_then(Value::as_str) {
            return Some(id.to_string());
        }
    }
    None
}

// ---------------------------------------------------------------------------
// Hermes — `hermes cron …` over HERMES_HOME
// ---------------------------------------------------------------------------

/// Hermes's schedule argument: one positional string its own parser reads
/// (`cron/jobs.py::parse_schedule` — `every 10m`, a cron expression, or an
/// ISO instant for a one-shot).
fn hermes_schedule(schedule: &JobScheduleSpec) -> Result<String> {
    match schedule.kind.as_str() {
        "interval" => schedule
            .minutes
            .map(|minutes| format!("every {}m", trim_float(minutes)))
            .ok_or_else(|| JobControlError::Invalid("an interval schedule needs `minutes`".into())),
        "cron" => schedule
            .expr
            .clone()
            .ok_or_else(|| JobControlError::Invalid("a cron schedule needs `expr`".into())),
        "once" => schedule
            .run_at
            .clone()
            .ok_or_else(|| JobControlError::Invalid("a once schedule needs `run_at`".into())),
        other => Err(JobControlError::Invalid(format!(
            "unknown schedule kind `{other}`; use interval, cron, or once"
        ))),
    }
}

fn trim_float(value: f64) -> String {
    if value.fract().abs() < f64::EPSILON {
        format!("{}", value as i64)
    } else {
        format!("{value}")
    }
}

/// Hermes's delivery argument, in hermes's own grammar
/// (`origin | local | <platform> | <platform>:<chat_id>`).
fn hermes_deliver(deliver: &JobDeliverSpec) -> Option<String> {
    let target = deliver.target.as_deref()?.trim().to_string();
    match deliver.chat_id.as_deref() {
        Some(chat) if !target.contains(':') && !chat.trim().is_empty() => {
            Some(format!("{target}:{}", chat.trim()))
        }
        _ => Some(target),
    }
}

fn hermes_command(verb: JobVerb, mutation: &JobMutation) -> Result<HarnessCommand> {
    if mutation.session_target.is_some() {
        return Err(JobControlError::Unsupported(
            "hermes cron fires always open their own `platform=cron` session; hermes has no \
             session-target verb, so supercode refuses rather than dropping the field"
                .into(),
        ));
    }
    let mut command = HarnessCommand::new(harness_program(HarnessId::HERMES)?);
    command.env("HERMES_HOME", hermes_home(mutation).to_string_lossy());
    command.args(["cron"]);
    let id = mutation.id.clone().unwrap_or_default();
    match verb {
        JobVerb::Create => {
            command.arg("create");
            if let Some(name) = &mutation.name {
                command.args(["--name", name]);
            }
            if let Some(deliver) = mutation.deliver.as_ref().and_then(hermes_deliver) {
                command.args(["--deliver", &deliver]);
            }
            if let Some(payload) = &mutation.payload {
                if payload_kind(payload) == "script" {
                    command.args(["--script", payload_text(payload)?]);
                }
            }
            // The schedule is positional and must precede the prompt.
            let schedule = hermes_schedule(
                mutation
                    .schedule
                    .as_ref()
                    .expect("create validates a schedule"),
            )?;
            command.arg(schedule);
            if let Some(payload) = &mutation.payload {
                match payload_kind(payload) {
                    "prompt" => {
                        command.arg(payload_text(payload)?);
                    }
                    "script" => {}
                    other => return Err(hermes_payload_refusal(other)),
                }
            }
        }
        JobVerb::Update => {
            command.args(["edit", &id]);
            if let Some(schedule) = &mutation.schedule {
                command.args(["--schedule", &hermes_schedule(schedule)?]);
            }
            if let Some(name) = &mutation.name {
                command.args(["--name", name]);
            }
            if let Some(deliver) = mutation.deliver.as_ref().and_then(hermes_deliver) {
                command.args(["--deliver", &deliver]);
            }
            if let Some(payload) = &mutation.payload {
                match payload_kind(payload) {
                    "prompt" => {
                        command.args(["--prompt", payload_text(payload)?]);
                    }
                    "script" => {
                        command.args(["--script", payload_text(payload)?]);
                    }
                    other => return Err(hermes_payload_refusal(other)),
                }
            }
        }
        JobVerb::Pause => {
            command.args(["pause", &id]);
        }
        JobVerb::Resume => {
            command.args(["resume", &id]);
        }
        JobVerb::Run => {
            command.args(["run", &id]);
        }
        JobVerb::Delete => {
            command.args(["remove", &id]);
        }
    }
    Ok(command)
}

fn hermes_payload_refusal(kind: &str) -> JobControlError {
    JobControlError::Unsupported(format!(
        "hermes cron carries a `prompt` or a `--script` payload; it has no verb for a `{kind}` \
         payload"
    ))
}

fn payload_kind(payload: &JobPayloadSpec) -> &str {
    if payload.kind.trim().is_empty() {
        "prompt"
    } else {
        payload.kind.trim()
    }
}

fn payload_text(payload: &JobPayloadSpec) -> Result<&str> {
    payload
        .text
        .as_deref()
        .filter(|text| !text.trim().is_empty())
        .ok_or_else(|| {
            JobControlError::Invalid(format!(
                "a `{}` payload needs its text",
                payload_kind(payload)
            ))
        })
}

// ---------------------------------------------------------------------------
// OpenClaw — `openclaw cron …` through the Gateway
// ---------------------------------------------------------------------------

fn openclaw_command(verb: JobVerb, mutation: &JobMutation) -> Result<HarnessCommand> {
    let connection = openclaw_connection(&mutation.homes)?;
    let mut command = HarnessCommand::new(harness_program(HarnessId::OPENCLAW)?);
    // Point the spawned CLI at the SAME state the read side addresses, using
    // openclaw's own environment contract (`OPENCLAW_STATE_DIR` names the
    // state dir; `OPENCLAW_CONFIG_PATH` names the config file inside it).
    command.env(
        "OPENCLAW_STATE_DIR",
        mutation.homes.openclaw.to_string_lossy(),
    );
    command.env(
        "OPENCLAW_CONFIG_PATH",
        mutation
            .homes
            .openclaw
            .join("openclaw.json")
            .to_string_lossy(),
    );
    command.arg("cron");
    let id = mutation.id.clone().unwrap_or_default();
    match verb {
        JobVerb::Create => {
            command.arg("add");
        }
        JobVerb::Update => {
            command.args(["edit", &id]);
        }
        // OpenClaw spells pause/resume `disable`/`enable`.
        JobVerb::Pause => {
            command.args(["disable", &id]);
        }
        JobVerb::Resume => {
            command.args(["enable", &id]);
        }
        JobVerb::Run => {
            command.args(["run", &id]);
        }
        JobVerb::Delete => {
            command.args(["rm", &id]);
        }
    }
    command.args(["--url", &connection.address]);
    if let Some(token) = &connection.auth {
        command.arg("--token");
        command.secret(token.secret());
    }
    if matches!(verb, JobVerb::Create | JobVerb::Update) {
        if let Some(name) = &mutation.name {
            command.args(["--name", name]);
        }
        if let Some(schedule) = &mutation.schedule {
            match schedule.kind.as_str() {
                "interval" => {
                    let minutes = schedule.minutes.ok_or_else(|| {
                        JobControlError::Invalid("an interval schedule needs `minutes`".into())
                    })?;
                    command.args(["--every", &format!("{}m", trim_float(minutes))]);
                }
                "cron" => {
                    let expr = schedule.expr.as_deref().ok_or_else(|| {
                        JobControlError::Invalid("a cron schedule needs `expr`".into())
                    })?;
                    command.args(["--cron", expr]);
                }
                "once" => {
                    let run_at = schedule.run_at.as_deref().ok_or_else(|| {
                        JobControlError::Invalid("a once schedule needs `run_at`".into())
                    })?;
                    command.args(["--at", run_at]);
                }
                other => {
                    return Err(JobControlError::Invalid(format!(
                        "unknown schedule kind `{other}`; use interval, cron, or once"
                    )))
                }
            }
        }
        if let Some(payload) = &mutation.payload {
            match payload_kind(payload) {
                "prompt" => {
                    command.args(["--message", payload_text(payload)?]);
                }
                "system_event" => {
                    command.args(["--system-event", payload_text(payload)?]);
                }
                "command" => {
                    command.args(["--command", payload_text(payload)?]);
                }
                other => {
                    return Err(JobControlError::Unsupported(format!(
                        "openclaw cron carries `message`, `system-event` or `command` payloads; \
                         it has no verb for a `{other}` payload"
                    )))
                }
            }
        }
        if let Some(target) = &mutation.session_target {
            command.args(["--session", target]);
        }
        if let Some(profile) = &mutation.profile {
            command.args(["--agent", profile]);
        }
        if let Some(deliver) = &mutation.deliver {
            openclaw_deliver(deliver, &mut command)?;
        }
        // Measured against the pin (receipt orch18-openclaw-jobs-receipt):
        // `cron add|rm|list` accept `--json`, `cron edit` REJECTS it
        // ("OpenClaw does not recognize option \"--json\""). The flag is only
        // an id hint for create anyway — the answer always comes from the
        // re-read.
        if matches!(verb, JobVerb::Create) {
            command.arg("--json");
        }
    } else if mutation.profile.is_some()
        || mutation.session_target.is_some()
        || mutation.deliver.is_some()
        || mutation.name.is_some()
        || mutation.schedule.is_some()
        || mutation.payload.is_some()
    {
        return Err(JobControlError::Invalid(format!(
            "`jobs.{}` changes no fields; pass definition fields to `jobs.update`",
            verb.as_str()
        )));
    }
    Ok(command)
}

/// OpenClaw's delivery flags. `target` is the delivery MODE the observed row
/// reports (`announce` | `webhook` | `none`); `chat_id` is the destination.
fn openclaw_deliver(deliver: &JobDeliverSpec, command: &mut HarnessCommand) -> Result<()> {
    let Some(target) = deliver.target.as_deref().map(str::trim) else {
        if let Some(chat) = deliver.chat_id.as_deref() {
            command.args(["--to", chat]);
        }
        return Ok(());
    };
    match target {
        "announce" => {
            command.arg("--announce");
            if let Some(chat) = deliver.chat_id.as_deref() {
                command.args(["--to", chat]);
            }
        }
        "webhook" => {
            let url = deliver.chat_id.as_deref().ok_or_else(|| {
                JobControlError::Invalid(
                    "an openclaw `webhook` delivery needs the URL in `chat_id`".into(),
                )
            })?;
            command.args(["--webhook", url]);
        }
        "none" => {
            command.arg("--no-deliver");
        }
        other => {
            return Err(JobControlError::Unsupported(format!(
                "openclaw delivers `announce`, `webhook`, or `none`; it has no `{other}` delivery \
                 mode"
            )))
        }
    }
    Ok(())
}

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

    fn homes(root: &Path) -> HarnessHomes {
        HarnessHomes {
            hermes: root.join("hermes_home/state.db"),
            openclaw: root.join("openclaw_home"),
            ..HarnessHomes::default()
        }
    }

    #[test]
    fn the_program_comes_from_the_registry_launch() {
        // Guard: the registry's hermes launch is the ACP BRIDGE (`hermes-acp`);
        // the cron verb lives on the base CLI.
        assert_eq!(harness_program(HarnessId::HERMES).unwrap(), "hermes");
        assert_eq!(harness_program(HarnessId::OPENCLAW).unwrap(), "openclaw");
    }

    #[test]
    fn claude_code_refuses_every_mutating_verb() {
        let error = mutate(
            JobVerb::Pause,
            &JobMutation {
                harness: HarnessId::CLAUDE_CODE.into(),
                id: Some("release-watch".into()),
                ..JobMutation::default()
            },
        )
        .unwrap_err();
        assert!(matches!(error, JobControlError::Unsupported(_)), "{error}");
        assert!(error.to_string().contains("CronCreate"), "{error}");
    }

    #[test]
    fn a_harness_without_jobs_refuses() {
        let error = mutate(
            JobVerb::Delete,
            &JobMutation {
                harness: "codex".into(),
                id: Some("x".into()),
                ..JobMutation::default()
            },
        )
        .unwrap_err();
        assert!(matches!(error, JobControlError::Unsupported(_)), "{error}");
    }

    #[test]
    fn hermes_translates_the_uniform_row_onto_its_own_verb() {
        let root = PathBuf::from("/tmp/orch18-unit");
        let command = hermes_command(
            JobVerb::Create,
            &JobMutation {
                harness: HarnessId::HERMES.into(),
                name: Some("health".into()),
                schedule: Some(JobScheduleSpec {
                    kind: "interval".into(),
                    minutes: Some(10.0),
                    ..JobScheduleSpec::default()
                }),
                payload: Some(JobPayloadSpec {
                    kind: "prompt".into(),
                    text: Some("nightly health check".into()),
                }),
                deliver: Some(JobDeliverSpec {
                    target: Some("local".into()),
                    chat_id: None,
                }),
                homes: homes(&root),
                ..JobMutation::default()
            },
        )
        .unwrap();
        assert_eq!(
            command.narrate(),
            "hermes cron create --name health --deliver local 'every 10m' 'nightly health check'"
        );
        assert_eq!(
            command.env,
            vec![(
                "HERMES_HOME".to_string(),
                root.join("hermes_home").to_string_lossy().into_owned()
            )]
        );
    }

    #[test]
    fn a_hermes_profile_is_its_own_home() {
        let root = PathBuf::from("/tmp/orch18-unit");
        let command = hermes_command(
            JobVerb::Pause,
            &JobMutation {
                harness: HarnessId::HERMES.into(),
                id: Some("abc".into()),
                profile: Some("ops".into()),
                homes: homes(&root),
                ..JobMutation::default()
            },
        )
        .unwrap();
        assert_eq!(command.narrate(), "hermes cron pause abc");
        assert_eq!(
            command.env[0].1,
            root.join("hermes_home/profiles/ops")
                .to_string_lossy()
                .into_owned()
        );
    }

    #[test]
    fn hermes_refuses_a_field_it_has_no_verb_for() {
        let error = hermes_command(
            JobVerb::Update,
            &JobMutation {
                harness: HarnessId::HERMES.into(),
                id: Some("abc".into()),
                session_target: Some("isolated".into()),
                ..JobMutation::default()
            },
        )
        .unwrap_err();
        assert!(matches!(error, JobControlError::Unsupported(_)), "{error}");
    }

    /// ORC-13: the uniform row translated onto the ORCHESTRATOR's own job
    /// vocabulary (`docs/ORCHESTRATOR-IR.md` §2.6) — the typed schedule, the
    /// prompt, and Hermes's own `deliver` grammar.
    #[test]
    fn the_orchestrator_translates_the_uniform_row_onto_its_own_operator_args() {
        let args = orchestrator_args(
            JobVerb::Create,
            &JobMutation {
                harness: HarnessId::ORCHESTRATOR.into(),
                name: Some("health".into()),
                schedule: Some(JobScheduleSpec {
                    kind: "interval".into(),
                    minutes: Some(10.0),
                    ..JobScheduleSpec::default()
                }),
                payload: Some(JobPayloadSpec {
                    kind: "prompt".into(),
                    text: Some("nightly health check".into()),
                }),
                deliver: Some(JobDeliverSpec {
                    target: Some("loopback".into()),
                    chat_id: Some("ops-room".into()),
                }),
                ..JobMutation::default()
            },
        )
        .unwrap();
        assert_eq!(
            args,
            serde_json::json!({
                "name": "health",
                "schedule": {"kind": "interval", "minutes": 10.0},
                "prompt": "nightly health check",
                "deliver": "loopback:ops-room",
            })
        );
    }

    /// A field the orchestrator's model has no home for is REFUSED, never
    /// dropped — the same rule the other two harnesses inherit.
    #[test]
    fn the_orchestrator_refuses_a_field_its_model_does_not_have() {
        for (mutation, needle) in [
            (
                JobMutation {
                    harness: HarnessId::ORCHESTRATOR.into(),
                    session_target: Some("isolated".into()),
                    ..JobMutation::default()
                },
                "no session-target field",
            ),
            (
                JobMutation {
                    harness: HarnessId::ORCHESTRATOR.into(),
                    payload: Some(JobPayloadSpec {
                        kind: "command".into(),
                        text: Some("ls".into()),
                    }),
                    ..JobMutation::default()
                },
                "there is no `command` payload",
            ),
        ] {
            let error = orchestrator_args(JobVerb::Create, &mutation).unwrap_err();
            assert!(matches!(error, JobControlError::Unsupported(_)), "{error}");
            assert!(error.to_string().contains(needle), "{error}");
        }
        // A verb that sets no fields refuses definition fields outright.
        let error = orchestrator_args(
            JobVerb::Pause,
            &JobMutation {
                harness: HarnessId::ORCHESTRATOR.into(),
                id: Some("job_x".into()),
                name: Some("renamed".into()),
                ..JobMutation::default()
            },
        )
        .unwrap_err();
        assert!(matches!(error, JobControlError::Invalid(_)), "{error}");
    }

    /// The root folder IS the `default` profile (§6), so an unnamed profile
    /// addresses it rather than defaulting to nothing.
    #[test]
    fn the_orchestrators_unnamed_profile_is_the_root_folder() {
        assert_eq!(
            orchestrator_profile(&JobMutation {
                harness: HarnessId::ORCHESTRATOR.into(),
                ..JobMutation::default()
            }),
            "default"
        );
        assert_eq!(
            orchestrator_profile(&JobMutation {
                harness: HarnessId::ORCHESTRATOR.into(),
                profile: Some("  coder ".into()),
                ..JobMutation::default()
            }),
            "coder"
        );
    }

    #[test]
    fn openclaw_carries_the_gateway_endpoint_and_never_prints_the_token() {
        let root = std::env::temp_dir().join(format!(
            "supercode-orch18-unit-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let state = root.join("openclaw_home");
        std::fs::create_dir_all(&state).unwrap();
        std::fs::write(
            state.join("openclaw.json"),
            r#"{"gateway": {"port": 18999, "auth": {"token": "super-secret-token"}}}"#,
        )
        .unwrap();
        let command = openclaw_command(
            JobVerb::Create,
            &JobMutation {
                harness: HarnessId::OPENCLAW.into(),
                name: Some("digest".into()),
                schedule: Some(JobScheduleSpec {
                    kind: "cron".into(),
                    expr: Some("0 9 * * 1".into()),
                    ..JobScheduleSpec::default()
                }),
                payload: Some(JobPayloadSpec {
                    kind: "system_event".into(),
                    text: Some("weekly digest".into()),
                }),
                session_target: Some("main".into()),
                homes: homes(&root),
                ..JobMutation::default()
            },
        )
        .unwrap();
        assert_eq!(
            command.narrate(),
            "openclaw cron add --url ws://127.0.0.1:18999 --token <redacted> --name digest --cron \
             '0 9 * * 1' --system-event 'weekly digest' --session main --json"
        );
        assert_eq!(command.secrets, vec!["super-secret-token".to_string()]);
        assert!(
            !command.narrate().contains("super-secret-token"),
            "the credential must never be narrated"
        );
        std::fs::remove_dir_all(&root).ok();
    }
}