supercode_harness/

jobs_apply.rs

//! `jobs.apply`: an author's declared jobs reach a harness through the
//! harness's own verbs (`docs/architecture/content-spec-status.md`).
//!
//! A spec is a set of [`JobSpec`]s keyed by name. `apply` reads what the
//! harness has in the one store the request addresses (the root home, or the
//! named profile's), keyed by each job's native `name`, then runs the
//! controlled-tier verbs in [`crate::jobs_control`]: `create` for a name the
//! store lacks, `update` with only the declared fields that differ for a name it
//! has. It writes no harness file; every outcome is a verb's own re-read row.
//!
//! What `apply` never does:
//!
//! * delete — a job the spec does not name (the agent's, an operator's, one with
//!   no name) stays, and is reported as `unnamed`;
//! * guess — a declared name two existing jobs share is refused;
//! * override an operator — `enabled` is the value at creation; a job whose
//!   state differs from the spec is reported as `drift`, never paused or
//!   resumed;
//! * declare a one-shot — a `once` schedule is an act, not a standing
//!   declaration, and is refused.
//!
//! Declared values are normalized the way Hermes stores them (skills trimmed
//! and de-duplicated in order, model and provider trimmed, `~` expanded and
//! paths canonical in `workdir`), so a spec that already holds is `unchanged`
//! on every re-apply. A value Hermes would store differently in a way that
//! cannot be normalized is refused, never re-edited forever.

use std::collections::BTreeMap;
use std::path::PathBuf;

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

use crate::jobs::{list_jobs, JobsQuery, ScheduledJob};
use crate::jobs_control::{
    hermes_deliver, mutate, supports_job_control, JobControlError, JobDeliverSpec, JobMutation,
    JobMutationOutcome, JobPayloadSpec, JobScheduleSpec, JobVerb,
};
use crate::{HarnessHomes, HarnessId};

/// One declared job: what the author wants to exist, never what the harness
/// has made of it.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct JobSpec {
    /// The author's key. On Hermes it is the job's `name`.
    pub name: String,
    /// When it fires: `interval` or `cron`.
    pub schedule: JobScheduleSpec,
    /// The turn a fire sends to the agent.
    #[serde(default)]
    pub prompt: Option<String>,
    /// Skills the fire loads.
    #[serde(default)]
    pub skills: Option<Vec<String>>,
    /// Where the fire's output goes.
    #[serde(default)]
    pub deliver: Option<JobDeliverSpec>,
    /// Model the job is pinned to.
    #[serde(default)]
    pub model: Option<String>,
    /// Provider serving that model.
    #[serde(default)]
    pub provider: Option<String>,
    /// Directory the fire runs from.
    #[serde(default)]
    pub workdir: Option<String>,
    /// Fire this many times, then stop (at least 1; omit to fire forever).
    #[serde(default)]
    pub repeat: Option<u32>,
    /// Whether the job fires once created. `false` pauses it right after the
    /// create; an existing job's state is the operator's.
    #[serde(default)]
    pub enabled: Option<bool>,
}

/// One `jobs.apply` request.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct JobsApply {
    /// Harness the spec is applied to.
    pub harness: String,
    /// Hermes profile / OpenClaw agent the jobs belong to. Absent means the
    /// root home only, never every profile.
    pub profile: Option<String>,
    /// The declared jobs.
    pub jobs: Vec<JobSpec>,
    /// Report what would run without running it.
    pub plan: bool,
    /// Storage roots.
    pub homes: HarnessHomes,
}

/// A job named in an apply report.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AppliedJob {
    /// The declared or existing name (the id, for a job with no name).
    pub name: String,
    /// The harness's id, when a job exists.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub id: Option<String>,
    /// What was (or would be) done, or why not.
    pub detail: String,
}

/// What one `jobs.apply` did, every outcome the harness's own answer.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct JobsApplyOutcome {
    /// Harness the spec was applied to.
    pub harness: String,
    /// Jobs created, each with the row the harness reports after the last verb
    /// that touched it (the pause, for a job declared `enabled: false`).
    pub created: Vec<JobMutationOutcome>,
    /// Jobs edited, each with the row the harness reports afterwards.
    pub updated: Vec<JobMutationOutcome>,
    /// Declared jobs that already match.
    pub unchanged: Vec<AppliedJob>,
    /// Existing jobs the spec does not name, left as they are.
    pub unnamed: Vec<AppliedJob>,
    /// Existing jobs whose operator-owned state differs from the spec.
    pub drift: Vec<AppliedJob>,
    /// Declared jobs `apply` would not or could not act on, and why.
    pub refused: Vec<AppliedJob>,
    /// With `plan`, the verbs that would run.
    pub planned: Vec<AppliedJob>,
}

fn store_error(error: impl std::fmt::Display) -> JobControlError {
    JobControlError::Failed(format!("the job store could not be read: {error}"))
}

/// Apply a declared job set through the harness's own verbs.
pub fn apply(request: &JobsApply) -> Result<JobsApplyOutcome, JobControlError> {
    if !supports_job_control(&request.harness) {
        return Err(JobControlError::Unsupported(format!(
            "`{}` has no mutable scheduled jobs, so a job spec cannot be applied to it",
            request.harness
        )));
    }
    let mut seen = BTreeMap::new();
    for spec in &request.jobs {
        if spec.name.trim().is_empty() {
            return Err(JobControlError::Invalid(
                "every declared job needs a name".into(),
            ));
        }
        if seen.insert(spec.name.as_str(), ()).is_some() {
            return Err(JobControlError::Invalid(format!(
                "the spec declares `{}` twice",
                spec.name
            )));
        }
    }
    // The addressed store, as each harness names it: the orchestrator's root
    // folder IS its `default` profile; an OpenClaw job always belongs to an
    // agent, so which one must be said rather than guessed.
    let mut request = request.clone();
    if request.harness == HarnessId::ORCHESTRATOR && request.profile.as_deref() == Some("default") {
        request.profile = None;
    }
    if request.harness == HarnessId::OPENCLAW && request.profile.is_none() {
        return Err(JobControlError::Invalid(
            "an OpenClaw job belongs to an agent; name it with `profile`".into(),
        ));
    }
    let request = &request;
    let listing = list_jobs(&JobsQuery {
        harness: Some(request.harness.clone()),
        profile: request.profile.clone(),
        homes: request.homes.clone(),
        ..JobsQuery::default()
    })
    .map_err(store_error)?;
    let mut by_name: BTreeMap<String, Vec<(ScheduledJob, Value)>> = BTreeMap::new();
    let mut outcome = JobsApplyOutcome {
        harness: request.harness.clone(),
        ..JobsApplyOutcome::default()
    };
    // Only the addressed store: with no profile, the root home's jobs, never a
    // profile's (whose verbs would run against another HERMES_HOME).
    for job in listing
        .jobs
        .into_iter()
        .filter(|job| job.profile == request.profile)
    {
        let (_, native) = crate::jobs::get_job(&request.harness, &job.id, &request.homes)
            .map_err(store_error)?
            .ok_or_else(|| store_error(format!("job `{}` vanished while reading", job.id)))?;
        match native.get("name").and_then(Value::as_str) {
            Some(name) => by_name
                .entry(name.to_string())
                .or_default()
                .push((job, native)),
            None => outcome.unnamed.push(AppliedJob {
                name: job.id.clone(),
                id: Some(job.id),
                detail: "has no name, so no declaration can address it; left as it is".into(),
            }),
        }
    }
    for spec in &request.jobs {
        let spec = match normalize(&request.harness, spec) {
            Ok(spec) => spec,
            Err(reason) => {
                outcome.refused.push(AppliedJob {
                    name: spec.name.clone(),
                    id: None,
                    detail: reason,
                });
                continue;
            }
        };
        match by_name.remove(&spec.name).unwrap_or_default().as_slice() {
            [] => create(request, &spec, &mut outcome),
            [(job, native)] => update(request, &spec, job, native, &mut outcome),
            several => outcome.refused.push(AppliedJob {
                name: spec.name.clone(),
                id: None,
                detail: format!(
                    "{} existing jobs share this name ({}); apply refuses rather than guess which one the spec means",
                    several.len(),
                    several
                        .iter()
                        .map(|(job, _)| job.id.as_str())
                        .collect::<Vec<_>>()
                        .join(", ")
                ),
            }),
        }
    }
    for (name, jobs) in by_name {
        for (job, _) in jobs {
            outcome.unnamed.push(AppliedJob {
                name: name.clone(),
                id: Some(job.id),
                detail: "not in the spec; left as it is".into(),
            });
        }
    }
    Ok(outcome)
}

/// The declaration as the harness would store it, or why it cannot be applied.
/// Everything here is predictable, so `plan` refuses exactly what apply would.
fn normalize(harness: &str, spec: &JobSpec) -> Result<JobSpec, String> {
    let mut spec = spec.clone();
    match spec.schedule.kind.as_str() {
        "interval"
            if harness == HarnessId::HERMES
                && spec
                    .schedule
                    .minutes
                    .is_some_and(|minutes| minutes.fract() != 0.0) =>
        {
            return Err("Hermes schedules whole minutes; `minutes` must be a whole number".into())
        }
        "interval" if spec.schedule.minutes.is_some_and(|minutes| minutes > 0.0) => {}
        "interval" => return Err("an interval schedule needs `minutes` above zero".into()),
        "cron"
            if spec
                .schedule
                .expr
                .as_deref()
                .is_some_and(|e| !e.trim().is_empty()) => {}
        "cron" => return Err("a cron schedule needs `expr`".into()),
        "once" => return Err(
            "a `once` schedule is an act, not a standing declaration; create it with `jobs create`"
                .into(),
        ),
        other => {
            return Err(format!(
                "unknown schedule kind `{other}`; use interval or cron"
            ))
        }
    }
    if let Some(deliver) = &spec.deliver {
        if deliver
            .target
            .as_deref()
            .is_none_or(|t| t.trim().is_empty())
        {
            return Err("`deliver` needs a target; a chat id alone names no destination".into());
        }
    }
    if spec.repeat == Some(0) {
        return Err("`repeat` must be at least 1; omit it to fire forever".into());
    }
    let trim = |value: &mut Option<String>| {
        *value = value
            .as_deref()
            .map(str::trim)
            .filter(|v| !v.is_empty())
            .map(String::from);
    };
    trim(&mut spec.model);
    trim(&mut spec.provider);
    if let Some(skills) = &spec.skills {
        let mut kept: Vec<String> = Vec::new();
        for skill in skills.iter().map(|s| s.trim()).filter(|s| !s.is_empty()) {
            if !kept.iter().any(|k| k == skill) {
                kept.push(skill.to_string());
            }
        }
        spec.skills = Some(kept);
    }
    if harness != HarnessId::HERMES {
        let hermes_only = [
            (
                "skills",
                spec.skills.as_ref().is_some_and(|s| !s.is_empty()),
            ),
            ("workdir", spec.workdir.is_some()),
            ("model", spec.model.is_some()),
            ("provider", spec.provider.is_some()),
            ("repeat", spec.repeat.is_some()),
        ]
        .into_iter()
        .find_map(|(field, set)| set.then_some(field));
        if let Some(field) = hermes_only {
            return Err(format!(
                "{harness} has no job verb that sets `{field}`, so supercode refuses rather than dropping it"
            ));
        }
        spec.skills = None;
    }
    Ok(spec)
}

fn base_mutation(request: &JobsApply) -> JobMutation {
    JobMutation {
        harness: request.harness.clone(),
        profile: request.profile.clone(),
        homes: request.homes.clone(),
        ..JobMutation::default()
    }
}

fn create(request: &JobsApply, spec: &JobSpec, outcome: &mut JobsApplyOutcome) {
    let mutation = JobMutation {
        name: Some(spec.name.clone()),
        schedule: Some(spec.schedule.clone()),
        payload: spec.prompt.as_ref().map(|text| JobPayloadSpec {
            kind: "prompt".into(),
            text: Some(text.clone()),
        }),
        deliver: spec.deliver.clone(),
        skills: spec.skills.clone().filter(|skills| !skills.is_empty()),
        model: spec.model.clone(),
        provider: spec.provider.clone(),
        workdir: spec.workdir.clone(),
        repeat: spec.repeat,
        ..base_mutation(request)
    };
    let pause = spec.enabled == Some(false);
    if request.plan {
        outcome.planned.push(AppliedJob {
            name: spec.name.clone(),
            id: None,
            detail: if pause {
                "create, then pause"
            } else {
                "create"
            }
            .into(),
        });
        return;
    }
    let created = match mutate(JobVerb::Create, &mutation) {
        Ok(created) => created,
        Err(error) => {
            outcome.refused.push(AppliedJob {
                name: spec.name.clone(),
                id: None,
                detail: error.to_string(),
            });
            return;
        }
    };
    if !pause {
        outcome.created.push(created);
        return;
    }
    let id = created.id.clone();
    let pause = JobMutation {
        id: Some(id.clone()),
        ..base_mutation(request)
    };
    match mutate(JobVerb::Pause, &pause) {
        // The pause's re-read row is the job as it now stands; the command
        // shown is the create that made it and the pause that followed.
        Ok(paused) => outcome.created.push(JobMutationOutcome {
            verb: created.verb,
            ran: format!("{}; {}", created.ran, paused.ran),
            ..paused
        }),
        Err(error) => {
            outcome.created.push(created);
            outcome.refused.push(AppliedJob {
                name: spec.name.clone(),
                id: Some(id),
                detail: format!("created, but the pause the spec declares failed: {error}"),
            });
        }
    }
}

/// `~` expanded and the path canonical where it exists: how Hermes stores it.
fn canonical_dir(path: &str) -> String {
    let expanded = match path.strip_prefix("~") {
        Some(rest) if rest.is_empty() || rest.starts_with('/') => std::env::var_os("HOME")
            .map(|home| format!("{}{rest}", PathBuf::from(home).display()))
            .unwrap_or_else(|| path.to_string()),
        _ => path.to_string(),
    };
    std::fs::canonicalize(&expanded)
        .map(|p| p.to_string_lossy().into_owned())
        .unwrap_or(expanded)
}

fn update(
    request: &JobsApply,
    spec: &JobSpec,
    job: &ScheduledJob,
    native: &Value,
    outcome: &mut JobsApplyOutcome,
) {
    let mut mutation = JobMutation {
        id: Some(job.id.clone()),
        ..base_mutation(request)
    };
    let mut changed = Vec::new();
    if !same_schedule(&spec.schedule, job) {
        mutation.schedule = Some(spec.schedule.clone());
        changed.push("schedule");
    }
    if let Some(prompt) = &spec.prompt {
        // The prompt lives in its own field; a script job's payload text is
        // the script, not the prompt.
        let have = native.get("prompt").and_then(Value::as_str).or_else(|| {
            (job.payload.kind == "prompt")
                .then_some(job.payload.text.as_deref())
                .flatten()
        });
        if have != Some(prompt.as_str()) {
            mutation.payload = Some(JobPayloadSpec {
                kind: "prompt".into(),
                text: Some(prompt.clone()),
            });
            changed.push("prompt");
        }
    }
    if let Some(deliver) = &spec.deliver {
        // Compare the destination as the harness spells it (`telegram:123`).
        let declared = if request.harness == HarnessId::HERMES {
            hermes_deliver(deliver)
        } else {
            deliver.target.clone()
        };
        if declared != job.deliver.target {
            mutation.deliver = Some(deliver.clone());
            changed.push("deliver");
        }
    }
    if let Some(skills) = &spec.skills {
        let have: Vec<String> = native
            .get("skills")
            .and_then(Value::as_array)
            .map(|skills| {
                skills
                    .iter()
                    .filter_map(|s| s.as_str().map(String::from))
                    .collect()
            })
            .unwrap_or_default();
        if &have != skills {
            mutation.skills = Some(skills.clone());
            changed.push("skills");
        }
    }
    if let Some(workdir) = &spec.workdir {
        let have = native
            .get("workdir")
            .and_then(Value::as_str)
            .map(canonical_dir);
        if have.as_deref() != Some(canonical_dir(workdir).as_str()) {
            mutation.workdir = Some(workdir.clone());
            changed.push("workdir");
        }
    }
    for (field, declared, slot) in [
        ("model", &spec.model, &mut mutation.model),
        ("provider", &spec.provider, &mut mutation.provider),
    ] {
        if let Some(declared) = declared {
            if native.get(field).and_then(Value::as_str).map(str::trim) != Some(declared.as_str()) {
                *slot = Some(declared.clone());
                changed.push(field);
            }
        }
    }
    if let Some(repeat) = spec.repeat {
        let have = native
            .get("repeat")
            .and_then(|repeat| repeat.get("times").or(Some(repeat)))
            .and_then(Value::as_u64);
        if have != Some(u64::from(repeat)) {
            mutation.repeat = Some(repeat);
            changed.push("repeat");
        }
    }
    if let Some(enabled) = spec.enabled {
        if enabled != job.enabled {
            outcome.drift.push(AppliedJob {
                name: spec.name.clone(),
                id: Some(job.id.clone()),
                detail: format!(
                    "the spec declares enabled={enabled}; the job is {} (an operator's state stands)",
                    job.state
                ),
            });
        }
    }
    if changed.is_empty() {
        outcome.unchanged.push(AppliedJob {
            name: spec.name.clone(),
            id: Some(job.id.clone()),
            detail: "matches the spec".into(),
        });
        return;
    }
    if request.plan {
        outcome.planned.push(AppliedJob {
            name: spec.name.clone(),
            id: Some(job.id.clone()),
            detail: format!("update {}", changed.join(", ")),
        });
        return;
    }
    match mutate(JobVerb::Update, &mutation) {
        Ok(updated) => outcome.updated.push(updated),
        Err(error) => outcome.refused.push(AppliedJob {
            name: spec.name.clone(),
            id: Some(job.id.clone()),
            detail: error.to_string(),
        }),
    }
}

/// Whether the harness's parsed schedule is the declared one.
fn same_schedule(spec: &JobScheduleSpec, job: &ScheduledJob) -> bool {
    let have = &job.schedule;
    if spec.kind != have.kind {
        return false;
    }
    match spec.kind.as_str() {
        "interval" => spec.minutes == have.minutes,
        "cron" => spec.expr.as_deref().map(str::trim) == have.expr.as_deref().map(str::trim),
        _ => false,
    }
}