supercode-harness 0.4.20

//! Observed-tier, READ-ONLY inventory of scheduled-job FIRES across the
//! harnesses that keep a run store (Domain 11, concept 7).
//!
//! A run is one execution of a [`crate::jobs::ScheduledJob`], with its
//! outcome and — where the harness leaves enough behind to recover it — the
//! session the fire opened.
//!
//! * **Hermes** — `HERMES_HOME/cron/executions.db`, a profile-local SQLite
//!   audit ledger (`cron/executions.py`: "the ledger records what is known
//!   about each attempt; it is not a retry queue"). One `executions` row per
//!   attempt: `id, job_id, source, process_id, pid, process_started_at,
//!   status, claimed_at, started_at, finished_at, error`, status one of
//!   `claimed | running | completed | failed | unknown`. A Hermes profile home
//!   is a full HERMES_HOME, so each `profiles/<name>/` has its own ledger.
//! * **OpenClaw** — `cron_run_logs` in the shared state database
//!   (`<state dir>/state/openclaw.sqlite`) at the pinned 2026.7.1-2:
//!   `store_key, job_id, seq, ts, status, error, …, session_id, session_key,
//!   run_id, run_at_ms, duration_ms, …, entry_json`, status one of
//!   `ok | error | skipped`. `store_key` is `path.resolve(cron.store)` — the
//!   legacy `cron/jobs.json` path used purely as a per-store partition key;
//!   at the pin no such file exists, and neither does a `cron/runs/*.jsonl`
//!   run log (that shape is what `openclaw doctor --fix` imports FROM).
//! * **Claude Code** — has no run store at all. A `CronCreate` fire is an
//!   ordinary turn inside the session that created the job, so `runs.list` /
//!   `runs.get` refuse for `claude-code` rather than inventing a fire record
//!   from turns. See [`RUN_HARNESSES`].
//!
//! Nothing here writes, claims, retries, or prunes. Every store is opened
//! `SQLITE_OPEN_READ_ONLY` — these are live databases owned by a running
//! scheduler.
//!
//! **Status words are the harness's own.** Hermes says `completed`/`failed`,
//! OpenClaw says `ok`/`error`; renaming either onto a shared vocabulary would
//! discard the distinction Hermes draws between `failed` (a terminal result
//! it wrote) and `unknown` (an attempt whose owner died before writing one).
//!
//! **Delivery (ORCH-13).** Where a fire's output went is read from each
//! harness's own delivery record:
//!
//! * **OpenClaw** writes it onto the run-log row itself — `delivery_status`,
//!   `delivery_error`, `delivered` — and declares the destination on the job
//!   (`cron_jobs.delivery_channel` / `delivery_to`), so the row's `target` is
//!   joined from there: the run log records the OUTCOME, never the address.
//! * **Hermes** keeps a separate `delivery_obligations` ledger inside
//!   `state.db` (`gateway/delivery_ledger.py`), keyed by the CONVERSATION's
//!   `session_key` and the platform surface — not by job or fire. So a fire is
//!   matched to an obligation the way [`join_hermes_session`] matches a
//!   session: by the fire's own `[claimed_at, finished_at]` window, on the
//!   fire's own surface. See [`hermes_delivery`] for the two questions asked,
//!   in order, and for why an unanchored ledger instant answers `None`.
//!
//! `None` stays honest: a fire whose delivery nothing recorded says so rather
//! than borrowing a neighbouring fire's outcome.

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

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

use crate::{HarnessHomes, HarnessId, Result};

/// Harnesses that keep a run store at all. Every other harness answers
/// `runs.list` / `runs.get` with `UnsupportedAction`, never an empty list —
/// an absent store and an empty history are different answers.
pub const RUN_HARNESSES: &[&str] = &[
    HarnessId::HERMES,
    HarnessId::OPENCLAW,
    HarnessId::ORCHESTRATOR,
];

/// Hard stop on how far a compression chain is followed from a fire's own
/// session to the readable tip. Hermes chains are short; a cycle in a
/// corrupted store must not spin.
/// One fire of one scheduled job, projected onto the uniform Domain 11 row.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct HarnessRun {
    /// Harness-native run id: Hermes's `executions.id` (a uuid hex),
    /// OpenClaw's `run_id` — or `<job_id>#<seq>` when a run-log row predates
    /// run ids, since the `(job_id, seq)` pair is that store's own key.
    pub id: String,
    /// Owning harness.
    pub harness: String,
    /// The scheduled job this fire belongs to.
    pub job_id: String,
    /// The harness's own outcome word: Hermes `claimed | running | completed
    /// | failed | unknown`, OpenClaw `ok | error | skipped`.
    pub status: String,
    /// When the scheduler claimed the fire. Hermes only — OpenClaw's run log
    /// is written once, at finish, and records no claim.
    pub claimed_at: Option<String>,
    /// When the fire began executing.
    pub started_at: Option<String>,
    /// When the fire reached a terminal state.
    pub finished_at: Option<String>,
    /// The failure the harness recorded, verbatim.
    pub error: Option<String>,
    /// The session this fire opened, when it is recoverable: OpenClaw records
    /// it on the row; Hermes does not, so it is recovered by matching
    /// `cron_<job_id>_<YYYYMMDD_HHMMSS>` session ids inside the fire's own
    /// window (see [`join_hermes_session`]). `None` means no session is
    /// recoverable — never a guess.
    pub session_id: Option<String>,
    /// Where this fire's output went, when the harness recorded a delivery
    /// for it. `None` means nothing in the harness's delivery record matches
    /// this fire — never that the delivery failed.
    pub delivery: Option<RunDelivery>,
}

/// A fire's delivery outcome (ORCH-13).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunDelivery {
    /// Where the output was addressed: Hermes's obligation surface
    /// `<platform>:<chat_id>[:<thread_id>]`, or OpenClaw's job-declared
    /// `<channel>[:<to>]`.
    pub target: Option<String>,
    /// The harness's own state word: Hermes `pending | attempting |
    /// delivered | failed | abandoned`, OpenClaw `delivery_status`.
    pub state: Option<String>,
    /// Delivery attempts recorded. Hermes only — OpenClaw's run log counts
    /// no attempts.
    pub attempts: Option<u64>,
    /// The last delivery failure, verbatim.
    pub last_error: Option<String>,
    /// When the harness stamped the delivery as done. Hermes only: it is the
    /// obligation's `updated_at` on a `delivered` row (the ledger writes no
    /// separate delivered-at column). OpenClaw's run log records `delivered`
    /// as a flag with no instant of its own, so it stays empty there.
    pub delivered_at: Option<String>,
}

impl HarnessRun {
    /// The observed row of a typed orchestration [`Fire`] (`docs/ONTOLOGY.md` §2.7):
    /// the join to its session and delivery is the caller's, as it is for the
    /// ledger read, so `runs list` and the orchestration never disagree about a fire.
    pub fn from_fire(
        harness: &str,
        fire: &supercode_interchange::orchestration::Fire,
        session_id: Option<String>,
        delivery: Option<RunDelivery>,
    ) -> Self {
        Self {
            id: fire.id.clone(),
            harness: harness.into(),
            job_id: fire.job_id.clone(),
            status: fire.status.hermes_word().to_string(),
            claimed_at: Some(fire.claimed_at.clone()),
            started_at: fire.started_at.clone(),
            finished_at: fire.finished_at.clone(),
            error: fire.error.clone(),
            session_id: session_id.or_else(|| fire.session_id.clone()),
            delivery,
        }
    }
}

/// One store the listing consulted, and what it found there.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunSource {
    /// Harness the store belongs to.
    pub harness: String,
    /// Absolute path consulted.
    pub path: PathBuf,
    /// `read` | `absent_store` | `unreadable`.
    pub state: String,
    /// Hermes profile home this ledger belongs to.
    pub profile: Option<String>,
    /// Why a store is `unreadable`.
    pub detail: Option<String>,
}

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

/// Result of a `runs.list`: the rows plus every store that was consulted.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunsListing {
    /// Uniform rows, harness-major then store order, newest fire first
    /// within a store.
    pub runs: Vec<HarnessRun>,
    /// Stores consulted, including the ones that were absent.
    pub sources: Vec<RunSource>,
}

/// Filters for a run-history read.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct RunsQuery {
    /// Restrict to one harness. Absent means every harness in
    /// [`RUN_HARNESSES`].
    pub harness: Option<String>,
    /// Restrict to one job's fires.
    pub job: Option<String>,
    /// Cap on rows. Applied per store as the read's own `LIMIT` (so a long
    /// history is never fully materialized) and again to the merged listing.
    pub limit: Option<usize>,
    /// Storage roots to read.
    pub homes: HarnessHomes,
}

/// Whether `harness` keeps a run store.
pub fn supports_runs(harness: &str) -> bool {
    RUN_HARNESSES.contains(&harness)
}

/// Read every fire the query selects.
///
/// Read-only: every store is opened `SQLITE_OPEN_READ_ONLY`, and no fire is
/// claimed, retried, or pruned.
pub fn list_runs(query: &RunsQuery) -> Result<RunsListing> {
    let (rows, sources) = collect(query);
    let mut runs: Vec<HarnessRun> = rows.into_iter().map(|(run, _)| run).collect();
    if let Some(limit) = query.limit {
        runs.truncate(limit);
    }
    Ok(RunsListing { runs, sources })
}

/// Read one fire by harness and id, with the verbatim native record beside
/// the uniform row. `Ok(None)` means the harness's stores hold no such run.
pub fn get_run(
    harness: &str,
    id: &str,
    homes: &HarnessHomes,
) -> Result<Option<(HarnessRun, Value)>> {
    let (rows, _) = collect(&RunsQuery {
        harness: Some(harness.to_string()),
        homes: homes.clone(),
        ..RunsQuery::default()
    });
    Ok(rows.into_iter().find(|(run, _)| run.id == id))
}

/// Every store the query selects, in harness-major order, each row paired
/// with the harness's own record so `get` never re-reads (and so the two
/// verbs can never disagree about a fire).
fn collect(query: &RunsQuery) -> (Vec<(HarnessRun, Value)>, Vec<RunSource>) {
    let mut rows = Vec::new();
    let mut sources = Vec::new();
    let wanted = query.harness.as_deref();
    if wanted.is_none_or(|harness| harness == HarnessId::HERMES) {
        collect_hermes(query, &mut rows, &mut sources);
    }
    if wanted.is_none_or(|harness| harness == HarnessId::OPENCLAW) {
        collect_openclaw(query, &mut rows, &mut sources);
    }
    if wanted.is_none_or(|harness| harness == HarnessId::ORCHESTRATOR) {
        collect_hermes_shaped(
            HarnessId::ORCHESTRATOR,
            orchestrator_ledgers(&query.homes),
            query,
            &mut rows,
            &mut sources,
        );
    }
    (rows, sources)
}

// ---------------------------------------------------------------------------
// Hermes — `cron/executions.db`, joined to `state.db` sessions
// ---------------------------------------------------------------------------

/// One Hermes execution ledger, with the profile home it belongs to and the
/// `state.db` whose sessions its fires opened.
struct HermesLedger {
    executions: PathBuf,
    profile: Option<String>,
}

/// Every `cron/executions.db` a Hermes install can hold.
///
/// A Hermes profile home IS a full HERMES_HOME (`hermes_constants.get_hermes_home`
/// resolves the context-local profile override first), so the root home and
/// every `profiles/<name>/` carry their own ledger AND their own `state.db`.
/// A profile that has no `state.db` of its own falls back to the root store,
/// where its rows carry `profile_name = <name>`.
fn hermes_ledgers(homes: &HarnessHomes) -> Vec<HermesLedger> {
    // `HarnessHomes::hermes` addresses `state.db`; the cron store is its
    // sibling under the same HERMES_HOME.
    let root = homes
        .hermes
        .parent()
        .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
    let mut ledgers = vec![HermesLedger {
        executions: root.join("cron/executions.db"),
        profile: None,
    }];
    if let Ok(entries) = std::fs::read_dir(root.join("profiles")) {
        let mut found: Vec<HermesLedger> = entries
            .flatten()
            .filter(|entry| entry.path().is_dir())
            .map(|entry| {
                let home = entry.path();
                HermesLedger {
                    executions: home.join("cron/executions.db"),
                    profile: Some(entry.file_name().to_string_lossy().into_owned()),
                }
            })
            .collect();
        found.sort_by(|left, right| left.profile.cmp(&right.profile));
        ledgers.extend(found);
    }
    ledgers
}

/// Every ledger an orchestrator home holds.
///
/// Identical in shape to [`hermes_ledgers`] because the folder is: each
/// profile folder is a complete home with its own `cron/executions.db`,
/// `cron/jobs.json` and `state.db` (`docs/ORCHESTRATOR-IR.md` §6). Bindings
/// and obligations live in the profile's own store, so — unlike Hermes's
/// multiplexed gateway — there is no fallback to a root store.
fn orchestrator_ledgers(homes: &HarnessHomes) -> Vec<HermesLedger> {
    crate::orchestrator_profile_dirs(&homes.orchestrator)
        .into_iter()
        .map(|(name, dir)| HermesLedger {
            executions: dir.join("cron/executions.db"),
            profile: (name != "default").then_some(name),
        })
        .collect()
}

const HERMES_EXECUTION_COLUMNS: &[&str] = &[
    "id",
    "job_id",
    "source",
    "process_id",
    "pid",
    "process_started_at",
    "status",
    "claimed_at",
    "started_at",
    "finished_at",
    "error",
];

fn collect_hermes(
    query: &RunsQuery,
    rows: &mut Vec<(HarnessRun, Value)>,
    sources: &mut Vec<RunSource>,
) {
    collect_hermes_shaped(
        HarnessId::HERMES,
        hermes_ledgers(&query.homes),
        query,
        rows,
        sources,
    );
}

/// Every fire of a set of Hermes-SHAPED ledgers, as the orchestration codec
/// reads the home: Hermes and the orchestrator keep the same `executions`
/// table and `delivery_obligations` ledger, so one compile serves both and
/// each ledger's fires are projected through [`HarnessRun::from_fire`] —
/// newest-claimed first, the ledger's own order. The fire's session and
/// delivery are the codec's derived links (`Fire.session_id`,
/// `Fire.obligation_id`).
fn collect_hermes_shaped(
    harness: &str,
    ledgers: Vec<HermesLedger>,
    query: &RunsQuery,
    rows: &mut Vec<(HarnessRun, Value)>,
    sources: &mut Vec<RunSource>,
) {
    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 ledger in ledgers {
                let state = if ledger.executions.is_file() {
                    "unreadable"
                } else {
                    "absent_store"
                };
                sources.push(RunSource {
                    detail: (state == "unreadable").then(|| error.to_string()),
                    ..RunSource::store(harness, ledger.executions, state, ledger.profile)
                });
            }
            return;
        }
    };
    for ledger in ledgers {
        if !ledger.executions.is_file() {
            sources.push(RunSource::store(
                harness,
                ledger.executions,
                "absent_store",
                ledger.profile,
            ));
            continue;
        }
        let name = ledger.profile.clone().unwrap_or_else(|| "default".into());
        let Some(profile) = loaded.orchestration.profiles.get(&name) else {
            sources.push(RunSource::store(
                harness,
                ledger.executions,
                "absent_store",
                ledger.profile,
            ));
            continue;
        };
        sources.push(RunSource::store(
            harness,
            ledger.executions,
            "read",
            ledger.profile,
        ));
        let mut fires: Vec<_> = profile
            .fires
            .iter()
            .filter(|fire| query.job.as_deref().is_none_or(|job| job == fire.job_id))
            .collect();
        fires.sort_by(|a, b| {
            b.claimed_at
                .cmp(&a.claimed_at)
                .then_with(|| b.id.cmp(&a.id))
        });
        if let Some(limit) = query.limit {
            fires.truncate(limit);
        }
        for fire in fires {
            let delivery = fire
                .obligation_id
                .as_deref()
                .and_then(|id| obligation_delivery(&loaded, id));
            let native: Map<String, Value> = HERMES_EXECUTION_COLUMNS
                .iter()
                .map(|c| (*c).to_string())
                .zip(supercode_interchange::orchestration::codec::decode::encode_fire_row(fire))
                .collect();
            rows.push((
                HarnessRun::from_fire(harness, fire, None, delivery),
                Value::Object(native),
            ));
        }
    }
}

/// The delivery an obligation records, as the run shows it.
fn obligation_delivery(
    loaded: &supercode_interchange::orchestration::codec::LoadedHome,
    id: &str,
) -> Option<RunDelivery> {
    let obligation = loaded
        .orchestration
        .profiles
        .values()
        .flat_map(|p| p.obligations.iter())
        .find(|o| o.id == id)?;
    let platform = obligation.target.platform.clone().unwrap_or_default();
    let chat_id = obligation.target.chat_id.clone().unwrap_or_default();
    Some(RunDelivery {
        target: Some(
            match obligation
                .target
                .thread_id
                .as_deref()
                .filter(|t| !t.is_empty())
            {
                Some(thread) => format!("{platform}:{chat_id}:{thread}"),
                None => format!("{platform}:{chat_id}"),
            },
        ),
        state: Some(obligation.state.hermes_word().to_string()),
        attempts: Some(obligation.attempts),
        last_error: obligation.last_error.clone().filter(|e| !e.is_empty()),
        delivered_at: obligation
            .delivered_at
            .as_deref()
            .and_then(|at| at.parse::<f64>().ok())
            .map(|seconds| crate::sidecar::ms_to_rfc3339((seconds * 1000.0) as i64)),
    })
}

fn openclaw_state_db(homes: &HarnessHomes) -> PathBuf {
    homes.openclaw.join("state/openclaw.sqlite")
}

/// OpenClaw's run logs as the orchestration codec reads the store: every
/// profile's fires, newest first (`ts`, then id), the store's own status word
/// and delivery columns from the fire's residue, the job's delivery channel
/// as the target.
fn collect_openclaw(
    query: &RunsQuery,
    rows: &mut Vec<(HarnessRun, Value)>,
    sources: &mut Vec<RunSource>,
) {
    use supercode_interchange::orchestration::codec::{from_openclaw, openclaw::encode_fire_row};
    let state_db = openclaw_state_db(&query.homes);
    if !state_db.is_file() {
        sources.push(RunSource::store(
            HarnessId::OPENCLAW,
            state_db,
            "absent_store",
            None,
        ));
        return;
    }
    let loaded = match from_openclaw(&query.homes.openclaw) {
        Ok(loaded) => loaded,
        Err(error) => {
            sources.push(RunSource {
                detail: Some(error.to_string()),
                ..RunSource::store(HarnessId::OPENCLAW, state_db, "unreadable", None)
            });
            return;
        }
    };
    sources.push(RunSource::store(
        HarnessId::OPENCLAW,
        state_db,
        "read",
        None,
    ));
    let text = |fire: &supercode_interchange::orchestration::Fire, key: &str| {
        fire.residue
            .0
            .get(key)
            .and_then(Value::as_str)
            .map(str::to_string)
    };
    let mut fires: Vec<_> = loaded
        .orchestration
        .profiles
        .values()
        .flat_map(|profile| profile.fires.iter().map(move |fire| (profile, fire)))
        .filter(|(_, fire)| query.job.as_deref().is_none_or(|job| job == fire.job_id))
        .collect();
    fires.sort_by(|(_, a), (_, b)| {
        b.finished_at
            .cmp(&a.finished_at)
            .then_with(|| b.id.cmp(&a.id))
    });
    if let Some(limit) = query.limit {
        fires.truncate(limit);
    }
    for (profile, fire) in fires {
        let target = profile.jobs.get(&fire.job_id).and_then(|job| {
            let delivery = job.residue.0.get("__delivery")?.as_object()?;
            let word = |k: &str| {
                delivery
                    .get(k)
                    .and_then(Value::as_str)
                    .filter(|v| !v.is_empty())
                    .map(str::to_string)
            };
            match (word("channel"), word("to")) {
                (Some(channel), Some(to)) => Some(format!("{channel}:{to}")),
                (Some(only), None) | (None, Some(only)) => Some(only),
                (None, None) => None,
            }
        });
        let delivery = openclaw_delivery(
            target,
            text(fire, "delivery_status"),
            text(fire, "delivery_error"),
            fire.residue.0.get("delivered").and_then(Value::as_i64),
        );
        let store_key = text(fire, "store_key").unwrap_or_default();
        rows.push((
            HarnessRun {
                id: fire.id.clone(),
                harness: HarnessId::OPENCLAW.into(),
                job_id: fire.job_id.clone(),
                status: text(fire, "status").unwrap_or_default(),
                claimed_at: None,
                started_at: fire.started_at.clone(),
                finished_at: fire.finished_at.clone(),
                error: fire.error.clone().filter(|e| !e.is_empty()),
                session_id: fire.session_id.clone().filter(|s| !s.is_empty()),
                delivery,
            },
            Value::Object(encode_fire_row(fire, None, &store_key)),
        ));
    }
}

fn openclaw_delivery(
    target: Option<String>,
    status: Option<String>,
    error: Option<String>,
    delivered: Option<i64>,
) -> Option<RunDelivery> {
    let status = status.filter(|status| !status.is_empty());
    let error = error.filter(|error| !error.is_empty());
    if status.is_none() && error.is_none() && delivered.is_none() {
        return None;
    }
    Some(RunDelivery {
        target,
        state: status.or_else(|| {
            delivered.map(|delivered| {
                if delivered == 0 {
                    "not-delivered".to_string()
                } else {
                    "delivered".to_string()
                }
            })
        }),
        // OpenClaw's run log counts no delivery attempts, and stamps no
        // instant on `delivered` — the flag rides the finish record.
        attempts: None,
        last_error: error,
        delivered_at: None,
    })
}