supercode_harness/

jobs.rs

//! Observed-tier, READ-ONLY inventory of scheduled jobs across the harnesses
//! that have them (Domain 11, concept 6).
//!
//! Three harnesses keep scheduled jobs and they keep them in three different
//! places, at two different scopes:
//!
//! * **Claude Code** — SESSION-scoped runtime state. `CronCreate` and
//!   `ScheduleWakeup` records live in the session's own JSONL, and
//!   [`crate::ClaudeRuntimeManifest`] already folds them into `active_crons` /
//!   `pending_wakeups`. There is no other store: Claude's own success text
//!   calls these jobs "session-only".
//! * **Hermes** — INSTALL-scoped `cron/jobs.json` under `HERMES_HOME`, plus one
//!   per profile under `profiles/<name>/cron/`.
//! * **The orchestrator** — the same `cron/jobs.json`, in the same two
//!   places, under `SUPERCODE_ORCHESTRATOR_HOME`: its folder IS a Hermes home
//!   (`docs/ORCHESTRATOR-IR.md` §6), so the Hermes store walk and the Hermes
//!   record projection below are pointed at it unchanged. Adding it is a home
//!   and an id, not a second reader.
//! * **OpenClaw** — INSTALL-scoped `cron/jobs.json` under the OpenClaw state
//!   dir at the pinned version (2026.7.1-2). Upstream `main` has since migrated
//!   the store into the shared SQLite state DB; when the JSON file is gone this
//!   module reports the store as `absent_store` instead of failing, so a newer
//!   install produces an honest empty answer rather than an error.
//!
//! Nothing here writes, claims a fire, or starts a timer. Every field is read
//! from the harness's own file; the uniform row below is a projection, and
//! [`get_job`] returns the verbatim native record beside it so nothing is lost.
//!
//! Field provenance for the two JSON stores is
//! `docs/HERMES-IDEAL-SUPPORT-DESIGN.md` §1a/§1b, which lists them from
//! upstream `cron/jobs.py` and `docs/automation/cron-jobs.md`. Keys that
//! document names but not spellings (a job's paused flag) are read tolerantly
//! in both plausible spellings rather than guessed at in one.

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

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

use crate::{
    ClaudeCronJob, ClaudeRuntimeManifest, ClaudeWakeup, DiscoveryQuery, HarnessCatalog,
    HarnessHomes, HarnessId, Result, Session, SessionLocator,
};

/// Harnesses that have a scheduled-job concept at all. Every other harness
/// answers `jobs.list` / `jobs.get` with `UnsupportedAction`, never an empty
/// list — an absent verb and an empty inventory are different answers.
pub const JOB_HARNESSES: &[&str] = &[
    HarnessId::CLAUDE_CODE,
    HarnessId::HERMES,
    HarnessId::OPENCLAW,
    HarnessId::ORCHESTRATOR,
];

/// Newest-first cap on Claude Code sessions examined when no `session` filter
/// is given. Claude's jobs are session state, so an unfiltered listing would
/// otherwise walk the whole history; the scan is reported in
/// [`JobsListing::sources`] so a truncated answer is never silent.
pub const CLAUDE_SESSION_SCAN_LIMIT: usize = 200;

/// Tool names whose presence in a Claude Code transcript makes the (expensive)
/// manifest derivation worth doing. A cheap substring pre-filter over the raw
/// JSONL keeps an unfiltered listing bounded.
const CLAUDE_JOB_MARKERS: &[&str] = &["CronCreate", "ScheduleWakeup"];

/// When a job's schedule kind cannot be read from the store.
const UNKNOWN_SCHEDULE: &str = "unknown";

/// One harness's scheduled job, projected onto the uniform Domain 11 row.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ScheduledJob {
    /// Harness-native job id. For a Claude Code wakeup — which the harness
    /// never names — this is the `ScheduleWakeup` tool-use id.
    pub id: String,
    /// Owning harness.
    pub harness: String,
    /// `session` for Claude Code, `install` for Hermes and OpenClaw.
    pub scope: JobScope,
    /// Hermes profile name / OpenClaw agent id, when the job belongs to one.
    pub profile: Option<String>,
    /// Claude Code session this job is runtime state of. `None` for the
    /// install-scoped harnesses.
    pub session_id: Option<String>,
    /// When the job fires.
    pub schedule: JobSchedule,
    /// What fires.
    pub payload: JobPayload,
    /// OpenClaw's `sessionTarget` (`main` | `isolated` | `current` |
    /// `session:<id>`). `None` for Hermes (whose fires always open their own
    /// `platform="cron"` session) and for Claude Code.
    pub session_target: Option<String>,
    /// Where the run's output goes.
    pub deliver: JobDeliver,
    /// Whether the scheduler will fire this job.
    pub enabled: bool,
    /// Harness-facing state word (`active`, `paused`, `pending`).
    pub state: String,
    /// Next fire, when the store records one.
    pub next_run_at: Option<String>,
    /// Last fire, when the store records one.
    pub last_run_at: Option<String>,
    /// Outcome of the last fire, when the store records one.
    pub last_status: Option<String>,
    /// Creation timestamp, when the store records one.
    pub created_at: Option<String>,
    /// Whether the job repeats.
    pub recurring: bool,
}

/// Whether a job belongs to one conversation or to the whole install.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum JobScope {
    /// Claude Code: the job is runtime state of one session.
    Session,
    /// Hermes / OpenClaw: the job outlives every conversation.
    Install,
}

/// A job's firing rule, with the native expression preserved.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct JobSchedule {
    /// `cron` | `interval` | `once` | `unknown`.
    pub kind: String,
    /// Cron expression, for `kind = "cron"`.
    pub expr: Option<String>,
    /// Interval in minutes, for `kind = "interval"`.
    pub minutes: Option<f64>,
    /// Absolute instant, for `kind = "once"`.
    pub run_at: Option<String>,
    /// One-line human rendering of whichever of the three above is set.
    pub display: String,
}

/// What a fire actually does.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JobPayload {
    /// `prompt` | `system_event` | `command` | `script` | `wakeup`.
    pub kind: String,
    /// The prompt, event, command line, or script the fire carries.
    pub text: Option<String>,
}

/// Where a fire's output is delivered.
///
/// `target` is WHERE and `mode` is HOW, for the one harness that separates
/// them. Hermes's `deliver` word names a destination (`origin` | `local` |
/// `home` | `<platform>` | `<platform>:<chat>[:<thread>]`) and it has no mode;
/// OpenClaw declares both, a mode (`announce` | `webhook` | `none`) and a
/// channel/`to`/account address. Folding OpenClaw's mode into `target` — as
/// this row did before ORCH-13 — left the channel it actually announces on
/// nowhere to go, so the mode moved to its own field.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JobDeliver {
    /// Hermes `deliver` (`origin` | `local` | `home` | `<platform>` |
    /// `<platform>:<chat_id>`), OpenClaw's delivery channel (`--channel`), or
    /// `session` for a Claude Code job, whose fire is a turn injected back
    /// into its own session.
    pub target: Option<String>,
    /// Chat the delivery is addressed to: Hermes `origin.chat_id` (or the
    /// chat in an explicit `<platform>:<chat>` target), OpenClaw `--to`.
    pub chat_id: Option<String>,
    /// Thread inside that chat, when the store names one: Hermes
    /// `origin.thread_id`, OpenClaw `delivery_thread_id`.
    pub thread_id: Option<String>,
    /// Channel account the delivery goes out through (OpenClaw
    /// `delivery_account_id`). Hermes routes by adapter profile, not account,
    /// so it is empty there.
    pub account: Option<String>,
    /// OpenClaw's delivery mode (`announce` | `webhook` | `none`). Empty for
    /// Hermes and Claude Code, neither of which has a mode word.
    pub mode: Option<String>,
}

impl JobDeliver {
    /// A delivery whose only fact is where it goes — the Claude Code case,
    /// whose fire is a turn injected back into its own session.
    fn to(target: &str) -> Self {
        Self {
            target: Some(target.to_string()),
            chat_id: None,
            thread_id: None,
            account: None,
            mode: None,
        }
    }
}

/// One store the listing consulted, and what it found there.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JobSource {
    /// Harness the store belongs to.
    pub harness: String,
    /// Absolute path consulted.
    pub path: PathBuf,
    /// `read` | `absent_store` | `scanned` | `unreadable`.
    pub state: String,
    /// Hermes profile / OpenClaw agent home this store belongs to.
    pub profile: Option<String>,
    /// Claude Code only: sessions examined by the pre-filter.
    pub sessions_scanned: Option<usize>,
    /// Claude Code only: the cap the scan ran under.
    pub scan_limit: Option<usize>,
    /// Why a store is `unreadable`.
    pub detail: Option<String>,
}

impl JobSource {
    fn store(harness: &str, path: PathBuf, state: &str, profile: Option<String>) -> Self {
        Self {
            harness: harness.to_string(),
            path,
            state: state.to_string(),
            profile,
            sessions_scanned: None,
            scan_limit: None,
            detail: None,
        }
    }
}

/// Result of a `jobs.list`: the rows plus every store that was consulted.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct JobsListing {
    /// Uniform rows, harness-major then store order.
    pub jobs: Vec<ScheduledJob>,
    /// Stores consulted, including the ones that were absent.
    pub sources: Vec<JobSource>,
}

/// Filters for a scheduled-job read.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct JobsQuery {
    /// Restrict to one harness. Absent means every harness in
    /// [`JOB_HARNESSES`].
    pub harness: Option<String>,
    /// Restrict to jobs belonging to one session (Claude Code).
    pub session: Option<String>,
    /// Restrict to one Hermes profile / OpenClaw agent.
    pub profile: Option<String>,
    /// Storage roots to read.
    pub homes: HarnessHomes,
}

/// Whether `harness` has a scheduled-job concept.
pub fn supports_jobs(harness: &str) -> bool {
    JOB_HARNESSES.contains(&harness)
}

/// Read every scheduled job the query selects.
///
/// Read-only: no store is opened for writing, no fire is claimed, no scheduler
/// is activated.
pub fn list_jobs(query: &JobsQuery) -> Result<JobsListing> {
    let mut jobs = Vec::new();
    let mut sources = Vec::new();
    let wanted = query.harness.as_deref();
    if wanted.is_none_or(|harness| harness == HarnessId::CLAUDE_CODE) {
        collect_claude_jobs(query, &mut jobs, &mut sources)?;
    }
    if wanted.is_none_or(|harness| harness == HarnessId::HERMES) {
        collect_hermes_jobs(query, &mut jobs, &mut sources);
    }
    if wanted.is_none_or(|harness| harness == HarnessId::OPENCLAW) {
        collect_openclaw_jobs(query, &mut jobs, &mut sources);
    }
    if wanted.is_none_or(|harness| harness == HarnessId::ORCHESTRATOR) {
        collect_hermes_shaped_jobs(HarnessId::ORCHESTRATOR, query, &mut jobs, &mut sources);
    }
    jobs.retain(|job| {
        query
            .session
            .as_deref()
            .is_none_or(|session| job.session_id.as_deref() == Some(session))
            && query
                .profile
                .as_deref()
                .is_none_or(|profile| job.profile.as_deref() == Some(profile))
    });
    Ok(JobsListing { jobs, sources })
}

/// Read one job by harness and id, with the verbatim native record beside the
/// uniform row. `Ok(None)` means the harness has no such job.
pub fn get_job(
    harness: &str,
    id: &str,
    homes: &HarnessHomes,
) -> Result<Option<(ScheduledJob, Value)>> {
    let listing = list_jobs(&JobsQuery {
        harness: Some(harness.to_string()),
        homes: homes.clone(),
        ..JobsQuery::default()
    })?;
    let Some(job) = listing.jobs.into_iter().find(|job| job.id == id) else {
        return Ok(None);
    };
    let source = native_record(&job, homes)?;
    Ok(Some((job, source)))
}

/// Re-read the harness's own record for one already-projected row, so `get`
/// answers with the native fields as well as the uniform ones.
fn native_record(job: &ScheduledJob, homes: &HarnessHomes) -> Result<Value> {
    match job.harness.as_str() {
        HarnessId::CLAUDE_CODE => claude_native_record(job, homes),
        HarnessId::HERMES | HarnessId::OPENCLAW | HarnessId::ORCHESTRATOR => {
            for store in job_store_paths(&job.harness, homes) {
                for record in read_job_array(&store.path) {
                    if record_id(&record).as_deref() == Some(job.id.as_str()) {
                        return Ok(record);
                    }
                }
            }
            Ok(Value::Null)
        }
        _ => Ok(Value::Null),
    }
}

fn claude_native_record(job: &ScheduledJob, homes: &HarnessHomes) -> Result<Value> {
    let Some(session_id) = job.session_id.as_deref() else {
        return Ok(Value::Null);
    };
    for locator in claude_locators(homes, Some(session_id), usize::MAX)? {
        let Ok(session) = Session::load(locator.storage.path()) else {
            continue;
        };
        let Ok(manifest) = ClaudeRuntimeManifest::from_session(&session) else {
            continue;
        };
        if let Some(cron) = manifest
            .active_crons
            .iter()
            .find(|cron| cron.id == job.id)
            .cloned()
        {
            return Ok(serde_json::to_value(cron)?);
        }
        if let Some(wakeup) = manifest
            .pending_wakeups
            .iter()
            .find(|wakeup| wakeup.tool_use_id == job.id)
            .cloned()
        {
            return Ok(serde_json::to_value(wakeup)?);
        }
    }
    Ok(Value::Null)
}

// ---------------------------------------------------------------------------
// Claude Code — session-scoped runtime state
// ---------------------------------------------------------------------------

/// Claude Code session locators to consider, newest first. With a `session`
/// filter the answer is that one session; without one the scan is capped.
fn claude_locators(
    homes: &HarnessHomes,
    session: Option<&str>,
    limit: usize,
) -> Result<Vec<SessionLocator>> {
    let query = DiscoveryQuery {
        harnesses: vec![HarnessId::new(HarnessId::CLAUDE_CODE)],
        homes: homes.clone(),
        limit: (limit != usize::MAX).then_some(limit),
        ..DiscoveryQuery::default()
    };
    let mut found = HarnessCatalog::new().discover(&query)?;
    if let Some(session) = session {
        found.retain(|descriptor| descriptor.locator.session_id == session);
    }
    Ok(found
        .into_iter()
        .map(|descriptor| descriptor.locator)
        .collect())
}

/// Cheap pre-filter: does this transcript mention a scheduling tool at all?
/// Streamed line by line and abandoned at the first hit, so a large history
/// costs a scan, not a parse.
fn mentions_a_scheduling_tool(path: &Path) -> bool {
    use std::io::BufRead;
    let Ok(file) = std::fs::File::open(path) else {
        return false;
    };
    for line in std::io::BufReader::new(file)
        .lines()
        .map_while(std::result::Result::ok)
    {
        if CLAUDE_JOB_MARKERS
            .iter()
            .any(|marker| line.contains(marker))
        {
            return true;
        }
    }
    false
}

fn collect_claude_jobs(
    query: &JobsQuery,
    jobs: &mut Vec<ScheduledJob>,
    sources: &mut Vec<JobSource>,
) -> Result<()> {
    let session = query.session.as_deref();
    let limit = if session.is_some() {
        usize::MAX
    } else {
        CLAUDE_SESSION_SCAN_LIMIT
    };
    let locators = claude_locators(&query.homes, session, limit)?;
    let mut scanned = 0usize;
    for locator in locators {
        scanned += 1;
        if !mentions_a_scheduling_tool(locator.storage.path()) {
            continue;
        }
        let Ok(loaded) = Session::load(locator.storage.path()) else {
            sources.push(JobSource {
                detail: Some("session could not be loaded".into()),
                ..JobSource::store(
                    HarnessId::CLAUDE_CODE,
                    locator.storage.path().to_path_buf(),
                    "unreadable",
                    None,
                )
            });
            continue;
        };
        let manifest = ClaudeRuntimeManifest::from_session(&loaded)?;
        for cron in &manifest.active_crons {
            jobs.push(claude_cron_row(&locator.session_id, cron));
        }
        for wakeup in &manifest.pending_wakeups {
            jobs.push(claude_wakeup_row(&locator.session_id, wakeup));
        }
    }
    sources.push(JobSource {
        sessions_scanned: Some(scanned),
        scan_limit: (session.is_none()).then_some(CLAUDE_SESSION_SCAN_LIMIT),
        ..JobSource::store(
            HarnessId::CLAUDE_CODE,
            query.homes.claude_code.clone(),
            "scanned",
            None,
        )
    });
    Ok(())
}

fn claude_cron_row(session_id: &str, cron: &ClaudeCronJob) -> ScheduledJob {
    ScheduledJob {
        id: cron.id.clone(),
        harness: HarnessId::CLAUDE_CODE.into(),
        scope: JobScope::Session,
        profile: None,
        session_id: Some(session_id.to_string()),
        schedule: JobSchedule {
            kind: "cron".into(),
            expr: Some(cron.schedule.clone()),
            minutes: None,
            run_at: None,
            display: cron.schedule.clone(),
        },
        payload: JobPayload {
            kind: "prompt".into(),
            text: Some(cron.prompt.clone()),
        },
        session_target: None,
        // A Claude Code fire is a prompt injected back into the session that
        // created it, never an external channel.
        deliver: JobDeliver::to("session"),
        enabled: true,
        state: "active".into(),
        // Claude Code persists no next/last fire, and supercode fires none of
        // these jobs. Reporting a computed instant here would be supercode's
        // arithmetic, not the harness's record.
        next_run_at: None,
        last_run_at: None,
        last_status: None,
        created_at: cron.created_at.clone(),
        recurring: cron.recurring,
    }
}

fn claude_wakeup_row(session_id: &str, wakeup: &ClaudeWakeup) -> ScheduledJob {
    ScheduledJob {
        id: wakeup.tool_use_id.clone(),
        harness: HarnessId::CLAUDE_CODE.into(),
        scope: JobScope::Session,
        profile: None,
        session_id: Some(session_id.to_string()),
        schedule: JobSchedule {
            kind: "once".into(),
            expr: None,
            minutes: None,
            run_at: wakeup.scheduled_for.clone(),
            display: format!("once, +{}s", wakeup.delay_seconds),
        },
        payload: JobPayload {
            kind: "wakeup".into(),
            text: wakeup.prompt.clone().or_else(|| wakeup.reason.clone()),
        },
        session_target: None,
        deliver: JobDeliver::to("session"),
        enabled: true,
        state: "pending".into(),
        next_run_at: wakeup.scheduled_for.clone(),
        last_run_at: None,
        last_status: None,
        created_at: wakeup.created_at.clone(),
        recurring: false,
    }
}

// ---------------------------------------------------------------------------
// Hermes and OpenClaw — install-scoped JSON job stores
// ---------------------------------------------------------------------------

/// One `cron/jobs.json` to read, with the profile it belongs to.
struct JobStore {
    path: PathBuf,
    profile: Option<String>,
}

/// Every `cron/jobs.json` an install can hold.
///
/// Hermes keeps one under `HERMES_HOME` and one under each
/// `profiles/<name>/cron/`; OpenClaw keeps one under its state dir. The
/// segments named here are what the orchestration ledger's `store` evidence
/// cites.
fn job_store_paths(harness: &str, homes: &HarnessHomes) -> Vec<JobStore> {
    match harness {
        HarnessId::HERMES => {
            // `HarnessHomes::hermes` addresses `state.db`; the cron store is
            // its sibling under the same HERMES_HOME.
            let home = homes
                .hermes
                .parent()
                .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
            let mut stores = vec![JobStore {
                path: home.join("cron/jobs.json"),
                profile: None,
            }];
            let profiles = home.join("profiles");
            if let Ok(entries) = std::fs::read_dir(&profiles) {
                let mut found: Vec<JobStore> = entries
                    .flatten()
                    .filter(|entry| entry.path().is_dir())
                    .map(|entry| JobStore {
                        path: entry.path().join("cron/jobs.json"),
                        profile: entry.file_name().to_string_lossy().into_owned().into(),
                    })
                    .collect();
                found.sort_by(|left, right| left.profile.cmp(&right.profile));
                stores.extend(found);
            }
            stores
        }
        // The orchestrator's folder layout is Hermes's: the root is the
        // `default` profile, `profiles/<name>/` are the named ones
        // (`docs/ORCHESTRATOR-IR.md` §6). One helper states that layout for
        // every reader.
        HarnessId::ORCHESTRATOR => crate::orchestrator_profile_dirs(&homes.orchestrator)
            .into_iter()
            .map(|(name, dir)| JobStore {
                path: dir.join("cron/jobs.json"),
                profile: (name != "default").then_some(name),
            })
            .collect(),
        HarnessId::OPENCLAW => vec![
            // Pinned 2026.7.1-2 (measured on an isolated gateway, receipt
            // `docs/interop/research/orch7-openclaw-jobs-receipt-2026-09-03.json`):
            // jobs live in the shared SQLite state DB, table `cron_jobs`;
            // `cron/jobs.json` survives only as that table's `store_key`.
            JobStore {
                path: homes.openclaw.join("state/openclaw.sqlite"),
                profile: None,
            },
            // Legacy file store (pre-SQLite installs); rows already seen in
            // the SQLite store are not repeated.
            JobStore {
                path: homes.openclaw.join("cron/jobs.json"),
                profile: None,
            },
        ],
        _ => Vec::new(),
    }
}

/// Read a `cron/jobs.json` into its job records.
///
/// Both harnesses have shipped the file as a bare array and as an object with
/// a `jobs` key; accept either and treat anything else as no jobs.
pub(crate) fn read_job_array(path: &Path) -> Vec<Value> {
    let Ok(text) = std::fs::read_to_string(path) else {
        return Vec::new();
    };
    let Ok(value) = serde_json::from_str::<Value>(&text) else {
        return Vec::new();
    };
    match value {
        Value::Array(items) => items,
        Value::Object(map) => map
            .get("jobs")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default(),
        _ => Vec::new(),
    }
}

pub(crate) fn record_id(record: &Value) -> Option<String> {
    ["id", "job_id", "jobId"]
        .iter()
        .find_map(|key| record.get(*key).and_then(Value::as_str))
        .map(str::to_string)
}

fn collect_hermes_jobs(
    query: &JobsQuery,
    jobs: &mut Vec<ScheduledJob>,
    sources: &mut Vec<JobSource>,
) {
    collect_hermes_shaped_jobs(HarnessId::HERMES, query, jobs, sources);
}

/// Read every `cron/jobs.json` a Hermes-SHAPED install keeps, projecting each
/// record onto the uniform row.
///
/// Two harnesses are Hermes-shaped here: Hermes itself, and the orchestrator,
/// whose folder is a Hermes home by construction (`docs/ORCHESTRATOR-IR.md`
/// §6) and whose job records are written in Hermes's own schema. The only
/// difference between them is the home the stores are enumerated from, so the
/// harness id is a parameter and there is exactly one implementation.
/// The Hermes-shaped homes (Hermes, the orchestrator) as the orchestration codec
/// reads them: one compile, then every profile's jobs projected through
/// [`ScheduledJob::from_job`] — the root profile first, named profiles in
/// name order, jobs by id (the orchestration keeps a profile's jobs by id).
fn collect_hermes_shaped_jobs(
    harness: &str,
    query: &JobsQuery,
    jobs: &mut Vec<ScheduledJob>,
    sources: &mut Vec<JobSource>,
) {
    use supercode_interchange::orchestration::codec::{from_hermes, load_home, Flavor};
    let loaded = match harness {
        HarnessId::HERMES => {
            let home = query
                .homes
                .hermes
                .parent()
                .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
            from_hermes(&home)
        }
        _ => load_home(&query.homes.orchestrator, Flavor::Orchestrator),
    };
    let loaded = match loaded {
        Ok(loaded) => loaded,
        Err(error) => {
            for store in job_store_paths(harness, &query.homes) {
                let mut source = JobSource::store(
                    harness,
                    store.path.clone(),
                    if store.path.exists() {
                        "unreadable"
                    } else {
                        "absent_store"
                    },
                    store.profile.clone(),
                );
                source.detail = Some(error.to_string());
                sources.push(source);
            }
            return;
        }
    };
    let mut names: Vec<&String> = loaded.orchestration.profiles.keys().collect();
    names.sort_by_key(|name| (name.as_str() != "default", name.as_str()));
    for name in names {
        let profile = &loaded.orchestration.profiles[name];
        let profile_name = (name != "default").then(|| name.clone());
        let path = profile.dir.join("cron/jobs.json");
        sources.push(JobSource::store(
            harness,
            path.clone(),
            if path.exists() {
                "read"
            } else {
                "absent_store"
            },
            profile_name.clone(),
        ));
        for job in profile.jobs.values() {
            jobs.push(ScheduledJob::from_job(harness, profile_name.clone(), job));
        }
    }
}

/// OpenClaw: the store's jobs as the orchestration codec reads them (projected
/// through [`ScheduledJob::from_job`]), plus the legacy `cron/jobs.json`
/// the reader has always merged — the store's rows win on a shared id.
fn collect_openclaw_jobs(
    query: &JobsQuery,
    jobs: &mut Vec<ScheduledJob>,
    sources: &mut Vec<JobSource>,
) {
    use supercode_interchange::orchestration::codec::from_openclaw;
    let mut seen: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
    let store = query.homes.openclaw.join("state/openclaw.sqlite");
    if !store.exists() {
        sources.push(JobSource::store(
            HarnessId::OPENCLAW,
            store.clone(),
            "absent_store",
            None,
        ));
    } else {
        match from_openclaw(&query.homes.openclaw) {
            Ok(loaded) => {
                sources.push(JobSource::store(
                    HarnessId::OPENCLAW,
                    store.clone(),
                    "read",
                    None,
                ));
                let mut names: Vec<&String> = loaded.orchestration.profiles.keys().collect();
                names.sort_by_key(|name| (name.as_str() != "default", name.as_str()));
                for name in names {
                    let profile_name = (name != "default").then(|| name.clone());
                    for job in loaded.orchestration.profiles[name].jobs.values() {
                        if seen.insert(job.id.clone()) {
                            jobs.push(ScheduledJob::from_job(
                                HarnessId::OPENCLAW,
                                profile_name.clone(),
                                job,
                            ));
                        }
                    }
                }
            }
            Err(error) => {
                let mut source =
                    JobSource::store(HarnessId::OPENCLAW, store.clone(), "unreadable", None);
                source.detail = Some(error.to_string());
                sources.push(source);
            }
        }
    }
    let legacy = query.homes.openclaw.join("cron/jobs.json");
    if !legacy.exists() {
        sources.push(JobSource::store(
            HarnessId::OPENCLAW,
            legacy,
            "absent_store",
            None,
        ));
        return;
    }
    sources.push(JobSource::store(
        HarnessId::OPENCLAW,
        legacy.clone(),
        "read",
        None,
    ));
    for record in read_job_array(&legacy) {
        if let Some(job) = openclaw_row(&record) {
            if seen.insert(job.id.clone()) {
                jobs.push(job);
            }
        }
    }
}

/// Unix milliseconds → RFC 3339 (UTC, second precision). Proleptic Gregorian
/// civil-from-days; no calendar dependency.
fn iso_from_ms(ms: i64) -> String {
    let secs = ms.div_euclid(1000);
    let days = secs.div_euclid(86_400);
    let sod = secs.rem_euclid(86_400);
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z - era * 146_097;
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    format!(
        "{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}Z",
        sod / 3600,
        (sod % 3600) / 60,
        sod % 60
    )
}

fn text_field(record: &Value, keys: &[&str]) -> Option<String> {
    keys.iter()
        .find_map(|key| record.get(*key).and_then(Value::as_str))
        .map(str::to_string)
}

/// A job is enabled unless the store says otherwise, in either of the two
/// spellings a paused job has been written with.
fn enabled_flag(record: &Value) -> bool {
    if let Some(enabled) = record.get("enabled").and_then(Value::as_bool) {
        return enabled;
    }
    if let Some(paused) = record
        .get("paused")
        .or_else(|| record.get("is_paused"))
        .and_then(Value::as_bool)
    {
        return !paused;
    }
    true
}

fn schedule_display(
    kind: &str,
    expr: &Option<String>,
    minutes: Option<f64>,
    run_at: &Option<String>,
) -> String {
    match kind {
        "cron" => expr.clone().unwrap_or_else(|| UNKNOWN_SCHEDULE.into()),
        "interval" => minutes.map_or_else(
            || UNKNOWN_SCHEDULE.to_string(),
            |minutes| format!("every {} min", trim_float(minutes)),
        ),
        "once" => run_at
            .clone()
            .map_or_else(|| "once".to_string(), |run_at| format!("once @{run_at}")),
        _ => UNKNOWN_SCHEDULE.into(),
    }
}

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

impl ScheduledJob {
    /// The observed row of a typed orchestration [`Job`] (`docs/ONTOLOGY.md` §2.7):
    /// `jobs list` projects the record the orchestration codec decoded, so the
    /// observed view and the orchestration never disagree about a job. OpenClaw's
    /// delivery object, payload object and session target ride as residue
    /// and are shown as the store spells them.
    pub fn from_job(
        harness: &str,
        profile: Option<String>,
        job: &supercode_interchange::orchestration::Job,
    ) -> Self {
        use supercode_interchange::orchestration::{Schedule, Target};
        let (kind, expr, minutes, run_at) = match &job.schedule {
            Schedule::Once { run_at } => ("once", None, None, Some(run_at.clone())),
            Schedule::Interval { minutes } => ("interval", None, Some(*minutes), None),
            Schedule::Cron { expr, .. } => ("cron", Some(expr.clone()), None, None),
        };
        let script = job.residue.0.get("script").and_then(Value::as_str);
        let payload = match script {
            Some(script) => JobPayload {
                kind: "script".into(),
                text: Some(script.to_string()),
            },
            None => JobPayload {
                kind: "prompt".into(),
                text: job.prompt.clone(),
            },
        };
        // OpenClaw's payload object rides as residue; its kind word is the
        // store's, read by the same rule as the legacy `cron/jobs.json` rows
        let payload = match job.residue.0.get("__payload") {
            Some(native) => openclaw_payload(&serde_json::json!({ "payload": native })),
            None => payload,
        };
        let deliver = Some(job.deliver.render());
        let (explicit_chat, explicit_thread) = match &job.deliver {
            Target::Explicit {
                chat_id, thread_id, ..
            } => (chat_id.clone(), thread_id.clone()),
            _ => (None, None),
        };
        let chat_id = job
            .origin
            .as_ref()
            .and_then(|o| o.chat_id.clone())
            .or(explicit_chat);
        let thread_id = job
            .origin
            .as_ref()
            .and_then(|o| o.thread_id.clone())
            .or(explicit_thread);
        let enabled = job.enabled;
        // OpenClaw keeps its delivery object and session target beside the
        // record; the codec carries them as residue, and the observed view
        // shows them as the store spells them
        let oc_delivery = job.residue.0.get("__delivery").and_then(Value::as_object);
        let oc_text = |key: &str| {
            oc_delivery
                .and_then(|d| d.get(key))
                .and_then(Value::as_str)
                .map(str::to_string)
        };
        // an OpenClaw row's target is the store's channel word; none means none
        let deliver = if harness == HarnessId::OPENCLAW {
            oc_text("channel")
        } else {
            deliver
        };
        let chat_id = oc_text("to").or(chat_id);
        let thread_id = oc_text("threadId").or(thread_id);
        let session_target = job
            .residue
            .0
            .get("__session_target")
            .and_then(Value::as_str)
            .map(str::to_string);
        Self {
            id: job.id.clone(),
            harness: harness.into(),
            scope: JobScope::Install,
            profile,
            session_id: None,
            schedule: JobSchedule {
                display: schedule_display(kind, &expr, minutes, &run_at),
                kind: kind.into(),
                expr,
                minutes,
                run_at,
            },
            payload,
            session_target,
            deliver: JobDeliver {
                target: deliver,
                chat_id,
                thread_id,
                account: oc_text("accountId"),
                mode: oc_text("mode"),
            },
            enabled,
            state: if enabled { "active" } else { "paused" }.into(),
            next_run_at: job.next_run_at.clone(),
            last_run_at: job.last_run_at.clone(),
            last_status: job.last_status.clone(),
            created_at: job.created_at.clone(),
            recurring: kind != "once",
        }
    }
}

/// Project one OpenClaw `cron/jobs.json` record.
///
/// Fields per `docs/HERMES-IDEAL-SUPPORT-DESIGN.md` §1b: payload kinds
/// `systemEvent` / `message` / `command` / `script`, `sessionTarget`
/// (`main` | `isolated` | `current` | `session:<id>`), delivery `announce` |
/// `webhook` | `none` with `--channel` / `--to`.
fn openclaw_row(record: &Value) -> Option<ScheduledJob> {
    let id = record_id(record)?;
    // Pinned shape: `schedule` is an object (`{kind: every, everyMs} |
    // {kind: cron, expr} | {kind: at, at}`); legacy files carried a bare
    // cron string plus `everyMinutes` / `runAt`.
    let schedule_obj = record.get("schedule").filter(|v| v.is_object());
    let expr = text_field(record, &["schedule", "cron"])
        .or_else(|| schedule_obj.and_then(|o| text_field(o, &["expr", "cron"])));
    let minutes = record
        .get("everyMinutes")
        .or_else(|| record.get("every_minutes"))
        .and_then(Value::as_f64)
        .or_else(|| {
            schedule_obj
                .and_then(|o| o.get("everyMs"))
                .and_then(Value::as_f64)
                .map(|ms| ms / 60_000.0)
        });
    let run_at = text_field(record, &["runAt", "run_at"])
        .or_else(|| schedule_obj.and_then(|o| text_field(o, &["at", "runAt"])));
    let state_obj = record.get("state").filter(|v| v.is_object());
    let state_ms = |key: &str| {
        state_obj
            .and_then(|o| o.get(key))
            .and_then(Value::as_i64)
            .map(iso_from_ms)
    };
    let kind = if minutes.is_some() {
        "interval"
    } else if expr.is_some() {
        "cron"
    } else if run_at.is_some() {
        "once"
    } else {
        UNKNOWN_SCHEDULE
    }
    .to_string();
    let payload = openclaw_payload(record);
    // `delivery` is `{mode, channel, to, threadId, accountId}` at the pin
    // (legacy files spell the mode `kind`, and a very old one wrote the mode
    // as a bare string); this reads the legacy `cron/jobs.json` rows only —
    // the store's rows come through the orchestration codec.
    let delivery = record.get("delivery").cloned().unwrap_or(Value::Null);
    let mode = delivery
        .as_str()
        .map(str::to_string)
        .or_else(|| text_field(&delivery, &["mode", "kind", "type"]));
    let target = text_field(&delivery, &["channel"]).or_else(|| text_field(record, &["channel"]));
    let chat_id = text_field(&delivery, &["to"]).or_else(|| text_field(record, &["to"]));
    let thread_id = text_field(&delivery, &["threadId", "thread_id"]);
    let account = text_field(&delivery, &["accountId", "account_id"]);
    let enabled = enabled_flag(record);
    let recurring = kind != "once";
    Some(ScheduledJob {
        id,
        harness: HarnessId::OPENCLAW.into(),
        scope: JobScope::Install,
        // OpenClaw's per-agent config home is the same noun as a Hermes
        // profile (inventory concept 2).
        profile: text_field(record, &["agentId", "agent_id"]),
        session_id: None,
        schedule: JobSchedule {
            display: schedule_display(&kind, &expr, minutes, &run_at),
            kind,
            expr,
            minutes,
            run_at,
        },
        payload,
        session_target: text_field(record, &["sessionTarget", "session_target"]),
        deliver: JobDeliver {
            target,
            chat_id,
            thread_id,
            account,
            mode,
        },
        enabled,
        state: if enabled { "active" } else { "paused" }.into(),
        next_run_at: text_field(record, &["nextRunAt", "next_run_at"])
            .or_else(|| state_ms("nextRunAtMs")),
        last_run_at: text_field(record, &["lastRunAt", "last_run_at"])
            .or_else(|| state_ms("lastRunAtMs")),
        last_status: text_field(record, &["lastStatus", "last_status"])
            .or_else(|| state_obj.and_then(|o| text_field(o, &["lastStatus", "lastRunStatus"]))),
        created_at: text_field(record, &["createdAt", "created_at"]).or_else(|| {
            record
                .get("createdAtMs")
                .and_then(Value::as_i64)
                .map(iso_from_ms)
        }),
        recurring,
    })
}

fn openclaw_payload(record: &Value) -> JobPayload {
    let payload = record.get("payload").cloned().unwrap_or(Value::Null);
    let native = text_field(&payload, &["kind", "type"])
        .or_else(|| text_field(record, &["payloadKind", "payload_kind"]));
    let text = text_field(&payload, &["text", "message", "command", "script"])
        .or_else(|| text_field(record, &["message", "command", "script"]));
    let kind = match native.as_deref() {
        Some("systemEvent" | "system_event") => "system_event",
        Some("message" | "prompt" | "agentTurn" | "agent_turn") => "prompt",
        Some("command") => "command",
        Some("script") => "script",
        Some(_) | None => {
            if payload_has(record, &payload, "systemEvent") {
                "system_event"
            } else if payload_has(record, &payload, "command") {
                "command"
            } else if payload_has(record, &payload, "script") {
                "script"
            } else {
                "prompt"
            }
        }
    };
    JobPayload {
        kind: kind.into(),
        text,
    }
}

fn payload_has(record: &Value, payload: &Value, key: &str) -> bool {
    payload.get(key).is_some() || record.get(key).is_some()
}