supercode-harness 0.4.16

//! 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 rusqlite::Connection;
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.
const COMPRESSION_CHAIN_LIMIT: usize = 32;

/// 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 world [`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 world never disagree about a fire.
    pub fn from_fire(
        harness: &str,
        fire: &supercode_interchange::world::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)
}

/// Open a harness store strictly read-only. These are live databases owned by
/// a running scheduler; no connection here may ever be handed a write API.
///
/// Same fallback as `jobs::openclaw_sqlite_records`: a WAL-mode store whose
/// `-shm` sidecar is missing refuses a plain read-only open, so the immutable
/// URI form is tried second.
fn open_read_only(path: &Path) -> std::result::Result<Connection, rusqlite::Error> {
    Connection::open_with_flags(
        path,
        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
    )
    .or_else(|_| {
        Connection::open_with_flags(
            format!("file:{}?immutable=1", path.display()),
            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY
                | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX
                | rusqlite::OpenFlags::SQLITE_OPEN_URI,
        )
    })
}

fn table_exists(conn: &Connection, table: &str) -> bool {
    conn.query_row(
        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1",
        [table],
        |row| row.get::<_, i64>(0),
    )
    .is_ok()
}

/// Render one SQLite value the way the store wrote it, for the native record.
fn native_value(value: rusqlite::types::ValueRef<'_>) -> Value {
    match value {
        rusqlite::types::ValueRef::Null => Value::Null,
        rusqlite::types::ValueRef::Integer(i) => Value::from(i),
        rusqlite::types::ValueRef::Real(f) => serde_json::Number::from_f64(f)
            .map(Value::Number)
            .unwrap_or(Value::Null),
        rusqlite::types::ValueRef::Text(t) => Value::from(String::from_utf8_lossy(t).into_owned()),
        rusqlite::types::ValueRef::Blob(_) => Value::Null,
    }
}

/// Every column of one row, keyed by the store's own column names.
fn native_row(row: &rusqlite::Row<'_>, columns: &[&str]) -> Value {
    let mut map = Map::new();
    for (index, name) in columns.iter().enumerate() {
        let value = row
            .get_ref(index)
            .map_or(Value::Null, |value| native_value(value));
        map.insert((*name).to_string(), value);
    }
    Value::Object(map)
}

// ---------------------------------------------------------------------------
// 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,
    sessions: PathBuf,
    /// The job store beside this ledger. A fire's delivery SURFACE is
    /// declared on the job, not on the execution row, so the obligation
    /// match (ORCH-13) needs it.
    jobs: 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"),
        sessions: homes.hermes.clone(),
        jobs: root.join("cron/jobs.json"),
        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();
                let own = home.join("state.db");
                HermesLedger {
                    executions: home.join("cron/executions.db"),
                    sessions: if own.is_file() {
                        own
                    } else {
                        homes.hermes.clone()
                    },
                    jobs: home.join("cron/jobs.json"),
                    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"),
            sessions: dir.join("state.db"),
            jobs: dir.join("cron/jobs.json"),
            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,
    );
}

/// Read every fire from a set of Hermes-SHAPED ledgers.
///
/// Hermes and the orchestrator keep the same `executions` table, the same
/// `delivery_obligations` ledger and the same job store beside them, so the
/// harness id and the ledger list are parameters and the reader is one
/// implementation (ORC-7). ORCH-13's delivery join runs unchanged for both.
fn collect_hermes_shaped(
    harness: &str,
    ledgers: Vec<HermesLedger>,
    query: &RunsQuery,
    rows: &mut Vec<(HarnessRun, Value)>,
    sources: &mut Vec<RunSource>,
) {
    for ledger in ledgers {
        if !ledger.executions.is_file() {
            sources.push(RunSource::store(
                harness,
                ledger.executions.clone(),
                "absent_store",
                ledger.profile.clone(),
            ));
            continue;
        }
        let connection = match open_read_only(&ledger.executions) {
            Ok(connection) => connection,
            Err(error) => {
                sources.push(RunSource {
                    detail: Some(error.to_string()),
                    ..RunSource::store(
                        harness,
                        ledger.executions.clone(),
                        "unreadable",
                        ledger.profile.clone(),
                    )
                });
                continue;
            }
        };
        if !table_exists(&connection, "executions") {
            sources.push(RunSource {
                detail: Some("no `executions` table — not a Hermes cron ledger".into()),
                ..RunSource::store(
                    harness,
                    ledger.executions.clone(),
                    "unreadable",
                    ledger.profile.clone(),
                )
            });
            continue;
        }
        match read_hermes_ledger(harness, &connection, &ledger, query) {
            Ok(found) => {
                sources.push(RunSource::store(
                    harness,
                    ledger.executions.clone(),
                    "read",
                    ledger.profile.clone(),
                ));
                rows.extend(found);
            }
            Err(error) => sources.push(RunSource {
                detail: Some(error.to_string()),
                ..RunSource::store(
                    harness,
                    ledger.executions.clone(),
                    "unreadable",
                    ledger.profile.clone(),
                )
            }),
        }
    }
}

fn read_hermes_ledger(
    harness: &str,
    connection: &Connection,
    ledger: &HermesLedger,
    query: &RunsQuery,
) -> std::result::Result<Vec<(HarnessRun, Value)>, rusqlite::Error> {
    // The ledger's own ordering index is `(job_id, claimed_at DESC, id DESC)`;
    // read newest-claimed first so a `limit` keeps the recent fires.
    let sql = format!(
        "SELECT {} FROM executions {} ORDER BY claimed_at DESC, id DESC {}",
        HERMES_EXECUTION_COLUMNS.join(", "),
        if query.job.is_some() {
            "WHERE job_id = ?1"
        } else {
            ""
        },
        query
            .limit
            .map_or_else(String::new, |limit| format!("LIMIT {limit}")),
    );
    let mut statement = connection.prepare(&sql)?;
    let read = |row: &rusqlite::Row<'_>| -> rusqlite::Result<(HermesExecution, Value)> {
        Ok((
            HermesExecution {
                id: row.get::<_, Option<String>>(0)?.unwrap_or_default(),
                job_id: row.get::<_, Option<String>>(1)?.unwrap_or_default(),
                status: row.get::<_, Option<String>>(6)?.unwrap_or_default(),
                claimed_at: row.get(7)?,
                started_at: row.get(8)?,
                finished_at: row.get(9)?,
                error: row.get(10)?,
            },
            native_row(row, HERMES_EXECUTION_COLUMNS),
        ))
    };
    let executions: Vec<(HermesExecution, Value)> = match query.job.as_deref() {
        Some(job) => statement
            .query_map([job], read)?
            .collect::<rusqlite::Result<_>>()?,
        None => statement
            .query_map([], read)?
            .collect::<rusqlite::Result<_>>()?,
    };
    // Sessions are joined from the ledger's own home, once per read; the job
    // store beside the ledger is read once for the delivery surfaces.
    let sessions = open_read_only(&ledger.sessions).ok();
    let surfaces = hermes_delivery_surfaces(&ledger.jobs);
    Ok(executions
        .into_iter()
        .map(|(execution, native)| {
            let session_id = sessions
                .as_ref()
                .and_then(|connection| join_hermes_session(connection, &execution));
            let delivery = sessions.as_ref().and_then(|connection| {
                hermes_delivery(
                    connection,
                    &execution,
                    session_id.as_deref(),
                    surfaces.get(execution.job_id.as_str()),
                )
            });
            (
                HarnessRun {
                    id: execution.id,
                    harness: harness.into(),
                    job_id: execution.job_id,
                    status: execution.status,
                    claimed_at: execution.claimed_at,
                    started_at: execution.started_at,
                    finished_at: execution.finished_at,
                    error: execution.error,
                    session_id,
                    delivery,
                },
                native,
            )
        })
        .collect())
}

/// The platform surface each job in `store` delivers to, when the job names
/// one that a `delivery_obligations` row could carry.
///
/// Only two of Hermes's `deliver` values name a chat: `origin` (the creating
/// conversation, recorded on the job as `origin{platform, chat_id}`) and the
/// explicit `<platform>:<chat>[:<thread>]` form `hermes send --to` spells.
/// `local` and `home` deliver nowhere a platform ledger would see, so those
/// jobs get no surface and their fires answer `None` — matching them on the
/// job's origin anyway would attribute a chat delivery to a run that never
/// made one.
fn hermes_delivery_surfaces(store: &Path) -> std::collections::BTreeMap<String, (String, String)> {
    let mut surfaces = std::collections::BTreeMap::new();
    for record in crate::jobs::read_job_array(store) {
        let Some(job_id) = crate::jobs::record_id(&record) else {
            continue;
        };
        let deliver = record
            .get("deliver")
            .and_then(Value::as_str)
            .unwrap_or_default();
        let surface = if deliver == "origin" {
            let platform = record.pointer("/origin/platform").and_then(Value::as_str);
            let chat = record.pointer("/origin/chat_id").and_then(Value::as_str);
            platform.zip(chat)
        } else {
            let mut parts = deliver.splitn(3, ':');
            parts.next().zip(parts.next())
        };
        if let Some((platform, chat)) = surface {
            if !platform.is_empty() && !chat.is_empty() {
                surfaces.insert(job_id, (platform.to_string(), chat.to_string()));
            }
        }
    }
    surfaces
}

/// Read the delivery Hermes recorded for one fire, from the
/// `delivery_obligations` ledger inside the same `state.db`.
///
/// The ledger is the GATEWAY's, not the scheduler's: `gateway/delivery_ledger.py`
/// records one row per outbound final response, keyed by the conversation's
/// `session_key` and the platform surface it was addressed to. Nothing in it
/// names a job or a fire. So the fire's own `[claimed_at, finished_at]` window
/// does the work, and two questions are asked in order:
///
/// 1. **By session key** — when the fire opened a session that carries one,
///    the obligations recorded for that key inside the window are this fire's.
/// 2. **By surface** — otherwise, the obligations addressed to the job's own
///    delivery surface (`<platform>, <chat_id>`) inside the window. A Hermes
///    cron session carries no `session_key` at all (the scheduler clears the
///    routing vars before the run), so this is the usual path for a fire.
///
/// The newest obligation in the window wins: a fire that retried its send
/// wrote more than one, and the last is its outcome.
///
/// **Both instants must be anchored.** Executions are written as
/// `hermes_time.now().isoformat()` — an offset-bearing instant — while
/// obligations are written as `time.time()`, UTC epoch seconds. An execution
/// string with no offset cannot be placed on that epoch line without assuming
/// a timezone supercode has no business guessing, so such a fire answers
/// `None` rather than matching on an invented instant.
fn hermes_delivery(
    connection: &Connection,
    execution: &HermesExecution,
    session_id: Option<&str>,
    surface: Option<&(String, String)>,
) -> Option<RunDelivery> {
    if !table_exists(connection, "delivery_obligations") {
        return None;
    }
    let from = execution.claimed_at.as_deref().and_then(hermes_epoch)?;
    let to = execution
        .finished_at
        .as_deref()
        .and_then(hermes_epoch)
        .unwrap_or(f64::MAX);
    let session_key = session_id.and_then(|session_id| {
        connection
            .query_row(
                "SELECT session_key FROM sessions WHERE id = ?1",
                [session_id],
                |row| row.get::<_, Option<String>>(0),
            )
            .ok()
            .flatten()
            .filter(|key| !key.is_empty())
    });
    let by_key = session_key.and_then(|key| {
        read_obligation(
            connection,
            "session_key = ?1",
            rusqlite::params![key, from, to],
        )
    });
    by_key.or_else(|| {
        let (platform, chat_id) = surface?;
        read_obligation(
            connection,
            "platform = ?1 AND chat_id = ?4",
            rusqlite::params![platform, from, to, chat_id],
        )
    })
}

/// The newest obligation matching `predicate` inside `[?2, ?3]`.
fn read_obligation(
    connection: &Connection,
    predicate: &str,
    params: &[&dyn rusqlite::ToSql],
) -> Option<RunDelivery> {
    let sql = format!(
        "SELECT platform, chat_id, thread_id, state, attempts, last_error, updated_at \
         FROM delivery_obligations \
         WHERE {predicate} AND created_at >= ?2 AND created_at <= ?3 \
         ORDER BY created_at DESC LIMIT 1"
    );
    connection
        .query_row(&sql, params, |row| {
            let platform: String = row.get(0)?;
            let chat_id: String = row.get(1)?;
            let thread_id: Option<String> = row.get(2)?;
            let state: Option<String> = row.get(3)?;
            let updated_at: Option<f64> = row.get(6)?;
            Ok(RunDelivery {
                target: Some(match thread_id.filter(|thread| !thread.is_empty()) {
                    Some(thread) => format!("{platform}:{chat_id}:{thread}"),
                    None => format!("{platform}:{chat_id}"),
                }),
                // Only a `delivered` row carries an instant of delivery: the
                // ledger stamps `updated_at` on every transition, so reading
                // it on a `failed` row would date the failure, not a send.
                delivered_at: updated_at
                    .filter(|_| state.as_deref() == Some("delivered"))
                    .map(|seconds| crate::sidecar::ms_to_rfc3339((seconds * 1000.0) as i64)),
                state,
                attempts: row
                    .get::<_, Option<i64>>(4)?
                    .map(|attempts| attempts as u64),
                last_error: row
                    .get::<_, Option<String>>(5)?
                    .filter(|error| !error.is_empty()),
            })
        })
        .ok()
}

/// An offset-bearing Hermes instant as UTC epoch seconds, or `None` when the
/// string carries no offset (see [`hermes_delivery`]).
fn hermes_epoch(iso: &str) -> Option<f64> {
    let (instant, offset) = split_offset(iso)?;
    let (date, time) = instant.split_once('T')?;
    let mut date = date.splitn(3, '-');
    let year: i64 = date.next()?.parse().ok()?;
    let month: i64 = date.next()?.parse().ok()?;
    let day: i64 = date.next()?.parse().ok()?;
    let mut clock = time.splitn(3, ':');
    let hour: i64 = clock.next()?.parse().ok()?;
    let minute: i64 = clock.next()?.parse().ok()?;
    let seconds: f64 = clock.next()?.parse().ok()?;
    // Days from the civil date (Howard Hinnant's `days_from_civil`), the
    // inverse of `jobs::iso_from_ms`'s civil-from-days.
    let year = year - i64::from(month <= 2);
    let era = year.div_euclid(400);
    let yoe = year - era * 400;
    let doy = (153 * (if month > 2 { month - 3 } else { month + 9 }) + 2) / 5 + day - 1;
    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
    let days = era * 146_097 + doe - 719_468;
    Some((days * 86_400 + hour * 3_600 + minute * 60) as f64 + seconds - offset)
}

/// Split `<instant><offset>` into the naive part and the offset in seconds.
/// `Z` is zero; a bare instant has no offset and cannot be placed on the
/// epoch line at all.
fn split_offset(iso: &str) -> Option<(&str, f64)> {
    if let Some(instant) = iso.strip_suffix('Z') {
        return Some((instant, 0.0));
    }
    // The sign is only an offset after the time part, never the date's own
    // separators, so search from the `T`.
    let time_at = iso.find('T')?;
    let sign_at = iso[time_at..]
        .find(['+', '-'])
        .map(|index| index + time_at)?;
    let (instant, offset) = iso.split_at(sign_at);
    let (hours, minutes) = offset[1..].split_once(':')?;
    let seconds = hours.parse::<f64>().ok()? * 3_600.0 + minutes.parse::<f64>().ok()? * 60.0;
    Some((
        instant,
        if offset.starts_with('-') {
            -seconds
        } else {
            seconds
        },
    ))
}

/// The fields of one `executions` row the uniform projection and the session
/// join need.
struct HermesExecution {
    id: String,
    job_id: String,
    status: String,
    claimed_at: Option<String>,
    started_at: Option<String>,
    finished_at: Option<String>,
    error: Option<String>,
}

/// Sortable `YYYYMMDDHHMMSS` key for a Hermes wall-clock instant.
///
/// The ledger writes `_hermes_now().isoformat()` and the scheduler mints the
/// fire session id with `_hermes_now().strftime('%Y%m%d_%H%M%S')` — the SAME
/// clock, so comparing the two as naive local instants needs no timezone
/// arithmetic and cannot be wrong by an offset. Sub-second precision is
/// dropped on both sides (the session id has none).
fn hermes_instant_key(iso: &str) -> Option<u64> {
    let digits: String = iso
        .chars()
        .take_while(|c| *c != '+' && *c != 'Z')
        .filter(char::is_ascii_digit)
        .collect();
    (digits.len() >= 14).then(|| digits[..14].parse().ok())?
}

/// `cron_<job>_<YYYYMMDD_HHMMSS>` → the same sortable key.
fn hermes_session_key(session_id: &str, job_id: &str) -> Option<u64> {
    if crate::session::hermes_cron_job_id(session_id).as_deref() != Some(job_id) {
        return None;
    }
    let stamp = session_id.rsplit_once('_')?;
    let date = stamp.0.rsplit_once('_')?.1;
    format!("{date}{}", stamp.1).parse().ok()
}

/// Recover the session a Hermes fire opened.
///
/// Hermes writes no link from an execution row to its session: the scheduler
/// mints `cron_<job_id>_<YYYYMMDD_HHMMSS>` inside the run body, after the
/// claim and before the terminal write. So the candidates are the sessions
/// whose id carries this job's prefix, and the fire's own window picks one:
///
/// * A terminated fire owns the candidates in `[claimed_at, finished_at]`,
///   and the NEWEST of those is the session that produced the outcome (a
///   retried script can open more than one).
/// * An unterminated fire (`claimed` / `running`) has an open-ended window,
///   so the OLDEST candidate at or after the claim is taken — the newest
///   there would be a later fire's session.
///
/// No candidate in the window means `None`: an honest "not recoverable"
/// rather than the nearest-looking session.
fn join_hermes_session(connection: &Connection, execution: &HermesExecution) -> Option<String> {
    let claimed = execution
        .claimed_at
        .as_deref()
        .and_then(hermes_instant_key)?;
    let finished = execution
        .finished_at
        .as_deref()
        .and_then(hermes_instant_key);
    let prefix = format!("cron_{}_", execution.job_id);
    let mut statement = connection
        .prepare("SELECT id FROM sessions WHERE substr(id, 1, ?1) = ?2")
        .ok()?;
    let candidates: Vec<(u64, String)> = statement
        .query_map(
            rusqlite::params![prefix.chars().count() as i64, prefix],
            |row| row.get::<_, String>(0),
        )
        .ok()?
        .flatten()
        .filter_map(|id| {
            let key = hermes_session_key(&id, &execution.job_id)?;
            (key >= claimed && finished.is_none_or(|finished| key <= finished)).then_some((key, id))
        })
        .collect();
    let chosen = match finished {
        Some(_) => candidates.into_iter().max_by_key(|(key, _)| *key),
        None => candidates.into_iter().min_by_key(|(key, _)| *key),
    }?;
    Some(compression_tip(connection, chosen.1))
}

/// Follow a Hermes compression chain from a fire's own session to the session
/// that still holds the conversation.
///
/// A compressed session ends with `end_reason = 'compression'` and its
/// continuation is the row whose `parent_session_id` points back at it — the
/// same tri-semantic lineage `hermes_lineage_kind` classifies as `compaction`.
/// A fire whose session was compressed mid-run is therefore only readable at
/// the chain's tip, so that is what the row reports.
fn compression_tip(connection: &Connection, start: String) -> String {
    let mut current = start;
    for _ in 0..COMPRESSION_CHAIN_LIMIT {
        let compressed = connection
            .query_row(
                "SELECT end_reason FROM sessions WHERE id = ?1",
                [&current],
                |row| row.get::<_, Option<String>>(0),
            )
            .ok()
            .flatten()
            .is_some_and(|reason| reason == "compression");
        if !compressed {
            return current;
        }
        let next: Option<String> = connection
            .query_row(
                "SELECT id FROM sessions WHERE parent_session_id = ?1 \
                 ORDER BY started_at DESC, id DESC LIMIT 1",
                [&current],
                |row| row.get(0),
            )
            .ok();
        match next {
            // A compressed session whose continuation is missing is still the
            // best answer available: the fire did open it.
            None => return current,
            Some(next) => current = next,
        }
    }
    current
}

// ---------------------------------------------------------------------------
// OpenClaw — `cron_run_logs` in the shared state DB
// ---------------------------------------------------------------------------

/// The shared OpenClaw state database, at the pinned 2026.7.1-2 path
/// (`src/state/openclaw-state-db.paths.ts`: the state root — `OPENCLAW_STATE_DIR`
/// or `~/.openclaw` — plus `state/openclaw.sqlite`).
fn openclaw_state_db(homes: &HarnessHomes) -> PathBuf {
    homes.openclaw.join("state/openclaw.sqlite")
}

const OPENCLAW_RUN_LOG_COLUMNS: &[&str] = &[
    "store_key",
    "job_id",
    "seq",
    "ts",
    "status",
    "error",
    "summary",
    "delivery_status",
    "delivery_error",
    "delivered",
    "session_id",
    "session_key",
    "run_id",
    "run_at_ms",
    "duration_ms",
];

fn collect_openclaw(
    query: &RunsQuery,
    rows: &mut Vec<(HarnessRun, Value)>,
    sources: &mut Vec<RunSource>,
) {
    let state_db = openclaw_state_db(&query.homes);
    if !state_db.is_file() {
        sources.push(RunSource::store(
            HarnessId::OPENCLAW,
            state_db.clone(),
            "absent_store",
            None,
        ));
    } else {
        match open_read_only(&state_db).and_then(|connection| {
            if table_exists(&connection, "cron_run_logs") {
                let targets = openclaw_delivery_targets(&connection);
                read_openclaw_run_logs(&connection, query, &targets)
            } else {
                Ok(Vec::new())
            }
        }) {
            Ok(found) => {
                sources.push(RunSource::store(
                    HarnessId::OPENCLAW,
                    state_db.clone(),
                    "read",
                    None,
                ));
                rows.extend(found);
            }
            Err(error) => sources.push(RunSource {
                detail: Some(error.to_string()),
                ..RunSource::store(HarnessId::OPENCLAW, state_db.clone(), "unreadable", None)
            }),
        }
    }
}

/// Where each job announces, from the store's own `cron_jobs` delivery
/// columns. A run-log row records whether a delivery happened, never where it
/// went, so the address comes from the job the fire belongs to.
fn openclaw_delivery_targets(
    connection: &Connection,
) -> std::collections::BTreeMap<String, String> {
    let mut targets = std::collections::BTreeMap::new();
    if !table_exists(connection, "cron_jobs") {
        return targets;
    }
    let Ok(mut statement) =
        connection.prepare("SELECT job_id, delivery_channel, delivery_to FROM cron_jobs")
    else {
        return targets;
    };
    let Ok(rows) = statement.query_map([], |row| {
        Ok((
            row.get::<_, String>(0)?,
            row.get::<_, Option<String>>(1)?,
            row.get::<_, Option<String>>(2)?,
        ))
    }) else {
        return targets;
    };
    for (job_id, channel, to) in rows.flatten() {
        let channel = channel.filter(|value| !value.is_empty());
        let to = to.filter(|value| !value.is_empty());
        let target = match (channel, to) {
            (Some(channel), Some(to)) => Some(format!("{channel}:{to}")),
            (Some(only), None) | (None, Some(only)) => Some(only),
            (None, None) => None,
        };
        if let Some(target) = target {
            targets.insert(job_id, target);
        }
    }
    targets
}

fn read_openclaw_run_logs(
    connection: &Connection,
    query: &RunsQuery,
    targets: &std::collections::BTreeMap<String, String>,
) -> std::result::Result<Vec<(HarnessRun, Value)>, rusqlite::Error> {
    // `idx_cron_run_logs_job_status` orders by `(ts DESC, seq DESC)`; the
    // store's own newest-first order.
    let sql = format!(
        "SELECT {} FROM cron_run_logs {} ORDER BY ts DESC, seq DESC {}",
        OPENCLAW_RUN_LOG_COLUMNS.join(", "),
        if query.job.is_some() {
            "WHERE job_id = ?1"
        } else {
            ""
        },
        query
            .limit
            .map_or_else(String::new, |limit| format!("LIMIT {limit}")),
    );
    let mut statement = connection.prepare(&sql)?;
    let read = |row: &rusqlite::Row<'_>| -> rusqlite::Result<(HarnessRun, Value)> {
        let native = native_row(row, OPENCLAW_RUN_LOG_COLUMNS);
        let job_id = row.get::<_, Option<String>>(1)?.unwrap_or_default();
        let delivery = openclaw_delivery(
            targets.get(job_id.as_str()).cloned(),
            row.get(7)?,
            row.get(8)?,
            row.get(9)?,
        );
        Ok((
            openclaw_row(
                job_id,
                row.get(12)?,
                row.get::<_, Option<i64>>(2)?,
                row.get(4)?,
                row.get(5)?,
                row.get(13)?,
                row.get(3)?,
                row.get(10)?,
                delivery,
            ),
            native,
        ))
    };
    match query.job.as_deref() {
        Some(job) => statement.query_map([job], read)?.collect(),
        None => statement.query_map([], read)?.collect(),
    }
}

/// Project one OpenClaw run-log entry, from either of its two stores.
///
/// `run_at_ms` is when the fire began and `ts` is the entry's own timestamp,
/// written once the run finished (`parseCronRunLogEntryObject` accepts only
/// `action: "finished"` records), so those are the row's start and finish.
/// OpenClaw records no claim, so `claimed_at` is honestly empty.
#[allow(clippy::too_many_arguments)]
fn openclaw_row(
    job_id: String,
    run_id: Option<String>,
    seq: Option<i64>,
    status: Option<String>,
    error: Option<String>,
    run_at_ms: Option<i64>,
    ts: Option<i64>,
    session_id: Option<String>,
    delivery: Option<RunDelivery>,
) -> HarnessRun {
    let id = run_id
        .filter(|run_id| !run_id.is_empty())
        .unwrap_or_else(|| match seq {
            Some(seq) => format!("{job_id}#{seq}"),
            None => job_id.clone(),
        });
    HarnessRun {
        id,
        harness: HarnessId::OPENCLAW.into(),
        job_id,
        status: status.unwrap_or_default(),
        claimed_at: None,
        started_at: run_at_ms.map(crate::sidecar::ms_to_rfc3339),
        finished_at: ts.map(crate::sidecar::ms_to_rfc3339),
        error: error.filter(|error| !error.is_empty()),
        session_id: session_id.filter(|session| !session.is_empty()),
        delivery,
    }
}

/// Project one run-log row's delivery columns.
///
/// `delivery_status` is the harness's own word; when a row carries only the
/// `delivered` flag, that flag becomes the state, so the fact is never lost
/// for want of a status string. A row with none of the three recorded no
/// delivery at all and answers `None` — the job's declared target alone is
/// not evidence that anything was sent.
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,
    })
}