supercode-harness 0.4.14

//! 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.
fn collect_hermes_shaped_jobs(
    harness: &str,
    query: &JobsQuery,
    jobs: &mut Vec<ScheduledJob>,
    sources: &mut Vec<JobSource>,
) {
    for store in job_store_paths(harness, &query.homes) {
        if !store.path.exists() {
            sources.push(JobSource::store(
                harness,
                store.path.clone(),
                "absent_store",
                store.profile.clone(),
            ));
            continue;
        }
        let records = read_job_array(&store.path);
        sources.push(JobSource::store(
            harness,
            store.path.clone(),
            "read",
            store.profile.clone(),
        ));
        for record in records {
            if let Some(job) = hermes_row(harness, &record, store.profile.clone()) {
                jobs.push(job);
            }
        }
    }
}

fn collect_openclaw_jobs(
    query: &JobsQuery,
    jobs: &mut Vec<ScheduledJob>,
    sources: &mut Vec<JobSource>,
) {
    let mut seen: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
    for store in job_store_paths(HarnessId::OPENCLAW, &query.homes) {
        if !store.path.exists() {
            sources.push(JobSource::store(
                HarnessId::OPENCLAW,
                store.path.clone(),
                "absent_store",
                None,
            ));
            continue;
        }
        let is_sqlite = store.path.extension().is_some_and(|ext| ext == "sqlite");
        let records = if is_sqlite {
            match openclaw_sqlite_records(&store.path) {
                Ok(records) => records,
                Err(error) => {
                    let mut source = JobSource::store(
                        HarnessId::OPENCLAW,
                        store.path.clone(),
                        "unreadable",
                        None,
                    );
                    source.detail = Some(error);
                    sources.push(source);
                    continue;
                }
            }
        } else {
            read_job_array(&store.path)
        };
        sources.push(JobSource::store(
            HarnessId::OPENCLAW,
            store.path.clone(),
            "read",
            None,
        ));
        for record in records {
            if let Some(job) = openclaw_row(&record) {
                if seen.insert(job.id.clone()) {
                    jobs.push(job);
                }
            }
        }
    }
}

/// Read `cron_jobs` from OpenClaw's shared SQLite state (read-only). Each
/// row's `job_json` is the same object `openclaw cron list --json` prints;
/// the runtime columns (`state_json`, `next_run_at_ms`, `last_run_at_ms`,
/// `last_run_status`) are folded in so [`openclaw_row`] sees one record.
/// A WAL-mode store without its `-shm` sidecar refuses a plain read-only
/// open, so the immutable URI form is the fallback.
fn openclaw_sqlite_records(path: &Path) -> std::result::Result<Vec<Value>, String> {
    use rusqlite::{Connection, OpenFlags};
    let plain = Connection::open_with_flags(
        path,
        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
    );
    let conn = match plain {
        Ok(conn) => conn,
        Err(_) => Connection::open_with_flags(
            format!("file:{}?immutable=1", path.display()),
            OpenFlags::SQLITE_OPEN_READ_ONLY
                | OpenFlags::SQLITE_OPEN_NO_MUTEX
                | OpenFlags::SQLITE_OPEN_URI,
        )
        .map_err(|error| error.to_string())?,
    };
    let mut statement = conn
        .prepare(
            "SELECT job_id, job_json, state_json, next_run_at_ms, last_run_at_ms, \
             last_run_status, created_at_ms, delivery_mode, delivery_channel, delivery_to, \
             delivery_thread_id, delivery_account_id \
             FROM cron_jobs ORDER BY sort_order, created_at_ms",
        )
        .map_err(|error| error.to_string())?;
    let rows = statement
        .query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, Option<String>>(1)?,
                row.get::<_, Option<String>>(2)?,
                row.get::<_, Option<i64>>(3)?,
                row.get::<_, Option<i64>>(4)?,
                row.get::<_, Option<String>>(5)?,
                row.get::<_, Option<i64>>(6)?,
                [
                    ("mode", row.get::<_, Option<String>>(7)?),
                    ("channel", row.get::<_, Option<String>>(8)?),
                    ("to", row.get::<_, Option<String>>(9)?),
                    ("threadId", row.get::<_, Option<String>>(10)?),
                    ("accountId", row.get::<_, Option<String>>(11)?),
                ],
            ))
        })
        .map_err(|error| error.to_string())?;
    let mut records = Vec::new();
    for row in rows.flatten() {
        let (job_id, job_json, state_json, next_ms, last_ms, last_status, created_ms, delivery) =
            row;
        let mut record: Value = job_json
            .as_deref()
            .and_then(|text| serde_json::from_str(text).ok())
            .unwrap_or_else(|| serde_json::json!({}));
        if !record.is_object() {
            record = serde_json::json!({});
        }
        let object = record.as_object_mut().expect("object");
        object.entry("id").or_insert(Value::String(job_id));
        if let Some(state) = state_json
            .as_deref()
            .and_then(|text| serde_json::from_str::<Value>(text).ok())
        {
            object.entry("state").or_insert(state);
        }
        if let Some(ms) = next_ms {
            object
                .entry("nextRunAt")
                .or_insert(Value::String(iso_from_ms(ms)));
        }
        if let Some(ms) = last_ms {
            object
                .entry("lastRunAt")
                .or_insert(Value::String(iso_from_ms(ms)));
        }
        if let Some(status) = last_status {
            object.entry("lastStatus").or_insert(Value::String(status));
        }
        if let Some(ms) = created_ms {
            object
                .entry("createdAt")
                .or_insert(Value::String(iso_from_ms(ms)));
        }
        // The store denormalizes the job's delivery target into its own
        // columns beside `job_json`. Fold them into one `delivery` object so
        // `openclaw_row` reads a single shape; a key `job_json` already
        // carries wins, since that is the record the gateway wrote.
        if delivery.iter().any(|(_, value)| value.is_some()) {
            let mut merged = object
                .get("delivery")
                .and_then(Value::as_object)
                .cloned()
                .unwrap_or_default();
            for (key, value) in delivery {
                if let Some(value) = value.filter(|value| !value.is_empty()) {
                    merged.entry(key).or_insert(Value::String(value));
                }
            }
            object.insert("delivery".into(), Value::Object(merged));
        }
        records.push(record);
    }
    Ok(records)
}

/// 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}")
    }
}

/// Project one Hermes `cron/jobs.json` record.
///
/// Fields per `docs/HERMES-IDEAL-SUPPORT-DESIGN.md` §1a: `id`,
/// `schedule{kind: once|interval|cron, run_at|minutes|expr}`, `prompt`,
/// `script`, `no_agent`, `deliver`, `origin{platform, chat_id, thread_id}`,
/// `next_run_at`, `last_run_at`, `last_status`.
fn hermes_row(harness: &str, record: &Value, profile: Option<String>) -> Option<ScheduledJob> {
    let id = record_id(record)?;
    let schedule = record.get("schedule").cloned().unwrap_or(Value::Null);
    let kind = schedule
        .get("kind")
        .and_then(Value::as_str)
        .unwrap_or(UNKNOWN_SCHEDULE)
        .to_string();
    let expr = schedule
        .get("expr")
        .and_then(Value::as_str)
        .map(str::to_string);
    let minutes = schedule.get("minutes").and_then(Value::as_f64);
    let run_at = schedule
        .get("run_at")
        .and_then(Value::as_str)
        .map(str::to_string);
    let script = record.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: text_field(record, &["prompt"]),
        },
    };
    let deliver = text_field(record, &["deliver"]);
    // `origin` is the creating session's channel; an explicit target is
    // written the way `hermes send --to` spells one,
    // `<platform>[:<chat>[:<thread>]]`.
    let explicit: Vec<&str> = deliver
        .as_deref()
        .map(|deliver| deliver.split(':').collect())
        .unwrap_or_default();
    let chat_id = record
        .pointer("/origin/chat_id")
        .and_then(Value::as_str)
        .map(str::to_string)
        .or_else(|| explicit.get(1).map(|chat| (*chat).to_string()));
    let thread_id = record
        .pointer("/origin/thread_id")
        .and_then(Value::as_str)
        .map(str::to_string)
        .or_else(|| explicit.get(2).map(|thread| (*thread).to_string()));
    let enabled = enabled_flag(record);
    let recurring = record
        .get("repeat")
        .and_then(Value::as_bool)
        .unwrap_or(kind != "once");
    Some(ScheduledJob {
        id,
        harness: harness.into(),
        scope: JobScope::Install,
        profile,
        session_id: None,
        schedule: JobSchedule {
            display: schedule_display(&kind, &expr, minutes, &run_at),
            kind,
            expr,
            minutes,
            run_at,
        },
        payload,
        // Hermes fires always open their own `platform="cron"` session; the
        // harness offers no target choice, so the column is honestly empty.
        session_target: None,
        deliver: JobDeliver {
            target: deliver,
            chat_id,
            thread_id,
            // Hermes routes a delivery by adapter profile, and has no mode
            // word: its `deliver` value already names the destination.
            account: None,
            mode: None,
        },
        enabled,
        state: if enabled { "active" } else { "paused" }.into(),
        next_run_at: text_field(record, &["next_run_at"]),
        last_run_at: text_field(record, &["last_run_at"]),
        last_status: text_field(record, &["last_status"]),
        created_at: text_field(record, &["created_at"]),
        recurring,
    })
}

impl ScheduledJob {
    /// The observed row of a typed world [`Job`] (`docs/ONTOLOGY.md` §2.7):
    /// the same projection [`hermes_row`] makes from the raw record, made from
    /// the record the world codec already decoded, so `jobs list` and the
    /// world never disagree about a job.
    pub fn from_job(
        harness: &str,
        profile: Option<String>,
        job: &supercode_interchange::world::Job,
    ) -> Self {
        use supercode_interchange::world::{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(),
            },
        };
        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;
        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: None,
            deliver: JobDeliver {
                target: deliver,
                chat_id,
                thread_id,
                account: None,
                mode: None,
            },
            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). [`openclaw_sqlite_records`] folds the store's own
    // `delivery_*` columns in beside it, so both spellings reach here.
    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()
}

#[cfg(test)]
mod world_projection_tests {
    use super::*;
    use supercode_interchange::world::{Job, JobOrigin, Schedule, Target};

    /// The typed projection and the raw-record projection agree row for row.
    #[test]
    fn from_job_matches_hermes_row() {
        let raw = serde_json::json!({
            "id": "coder-standup",
            "schedule": {"kind": "cron", "expr": "0 9 * * 1-5", "tz": "UTC"},
            "prompt": "Post the standup.",
            "deliver": "origin",
            "origin": {"platform": "telegram", "chat_id": "-100777", "thread_id": "55"},
            "enabled": true,
            "next_run_at": "2026-09-03T09:00:00Z",
            "last_run_at": "2026-09-02T10:00:00Z",
            "last_status": "ok",
            "created_at": "2026-08-28T10:00:00Z",
        });
        let job = Job {
            id: "coder-standup".into(),
            schedule: Schedule::Cron {
                expr: "0 9 * * 1-5".into(),
                tz: "UTC".into(),
            },
            prompt: Some("Post the standup.".into()),
            workdir: None,
            model: None,
            skills: Vec::new(),
            context_from: None,
            deliver: Target::Origin,
            failure_deliver: None,
            origin: Some(JobOrigin {
                platform: "telegram".into(),
                chat_type: None,
                chat_id: Some("-100777".into()),
                thread_id: Some("55".into()),
            }),
            attach_to_session: None,
            repeat: None,
            enabled: true,
            next_run_at: Some("2026-09-03T09:00:00Z".into()),
            last_run_at: Some("2026-09-02T10:00:00Z".into()),
            last_status: Some("ok".into()),
            created_at: Some("2026-08-28T10:00:00Z".into()),
            residue: Default::default(),
        };
        let from_raw = hermes_row("hermes", &raw, Some("coder".into())).unwrap();
        let from_typed = ScheduledJob::from_job("hermes", Some("coder".into()), &job);
        assert_eq!(from_typed, from_raw);

        let explicit_raw = serde_json::json!({
            "id": "digest", "schedule": {"kind": "interval", "minutes": 120}, "prompt": "Digest.",
            "deliver": "slack:C0FIXTURE:t1", "enabled": false, "repeat": {"times": null, "completed": 0},
        });
        let explicit = Job {
            id: "digest".into(),
            schedule: Schedule::Interval { minutes: 120.0 },
            prompt: Some("Digest.".into()),
            deliver: Target::Explicit {
                platform: "slack".into(),
                chat_id: Some("C0FIXTURE".into()),
                thread_id: Some("t1".into()),
            },
            enabled: false,
            ..job.clone()
        };
        let explicit = Job {
            origin: None,
            next_run_at: None,
            last_run_at: None,
            last_status: None,
            created_at: None,
            ..explicit
        };
        assert_eq!(
            ScheduledJob::from_job("hermes", None, &explicit),
            hermes_row("hermes", &explicit_raw, None).unwrap()
        );
    }
}