symbi-runtime 1.21.1

//! Persistent job store backed by SQLite.

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use uuid::Uuid;

use super::cron_types::*;

/// Abstract job store for cron job persistence.
#[async_trait]
pub trait JobStore: Send + Sync {
    /// Persist a job definition (insert or update).
    async fn save_job(&self, job: &CronJobDefinition) -> Result<(), JobStoreError>;

    /// Retrieve a job by its ID.
    async fn get_job(&self, job_id: CronJobId) -> Result<Option<CronJobDefinition>, JobStoreError>;

    /// Delete a job.
    async fn delete_job(&self, job_id: CronJobId) -> Result<bool, JobStoreError>;

    /// List all jobs, optionally filtered by status.
    async fn list_jobs(
        &self,
        status_filter: Option<CronJobStatus>,
    ) -> Result<Vec<CronJobDefinition>, JobStoreError>;

    /// Return all enabled, active jobs whose `next_run <= now`.
    async fn get_due_jobs(
        &self,
        now: DateTime<Utc>,
    ) -> Result<Vec<CronJobDefinition>, JobStoreError>;

    /// Update the run-state fields after a tick (last_run, next_run, run_count, status, etc.).
    async fn update_run_state(
        &self,
        job_id: CronJobId,
        last_run: DateTime<Utc>,
        next_run: Option<DateTime<Utc>>,
        run_count: u64,
        status: CronJobStatus,
        enabled: bool,
    ) -> Result<(), JobStoreError>;

    /// Increment `failure_count` for a job.
    async fn record_failure(
        &self,
        job_id: CronJobId,
        failure_count: u64,
        status: CronJobStatus,
    ) -> Result<(), JobStoreError>;

    /// Insert a run record or finish its existing running record.
    async fn save_run_record(&self, record: &JobRunRecord) -> Result<(), JobStoreError>;

    /// Query run history for a job, newest first.
    async fn get_run_history(
        &self,
        job_id: CronJobId,
        limit: usize,
    ) -> Result<Vec<JobRunRecord>, JobStoreError>;
}

/// Errors produced by the job store.
#[derive(Debug, thiserror::Error)]
pub enum JobStoreError {
    #[error("invocation ID conflicts with a different request")]
    InvocationConflict,
    #[error("SQLite error: {0}")]
    Sqlite(String),
    #[error("Serialization error: {0}")]
    Serialization(String),
    #[error("Job not found: {0}")]
    NotFound(CronJobId),
}

/// SQLite-backed persistent store for cron jobs.
pub struct SqliteJobStore {
    pub(super) conn: tokio::sync::Mutex<rusqlite::Connection>,
}

impl SqliteJobStore {
    pub(super) async fn persist_job(
        &self,
        job: &CronJobDefinition,
        replace: bool,
    ) -> Result<(), JobStoreError> {
        let agent_json = serde_json::to_string(&job.agent_config)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let policy_json = serde_json::to_string(&job.policy_ids)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let delivery_json = job
            .delivery_config
            .as_ref()
            .map(serde_json::to_string)
            .transpose()
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let audit_str = serde_json::to_string(&job.audit_level)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let status_str = serde_json::to_string(&job.status)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;

        let session_mode_str = serde_json::to_string(&job.session_mode)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;

        let conn = self.conn.lock().await;
        conn.execute(
            "INSERT INTO cron_jobs
                (job_id, name, cron_expr, timezone, agent_json, policy_ids,
                 audit_level, status, enabled, one_shot, created_at, updated_at,
                 last_run, next_run, run_count, failure_count, max_retries,
                 max_concurrent, delivery_json, jitter_max_secs, session_mode, agentpin_jwt, input_json)
             VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22,?23)
             ON CONFLICT(job_id) DO UPDATE SET
                name=excluded.name,
                cron_expr=excluded.cron_expr,
                timezone=excluded.timezone,
                agent_json=excluded.agent_json,
                policy_ids=excluded.policy_ids,
                audit_level=excluded.audit_level,
                status=excluded.status,
                enabled=excluded.enabled,
                one_shot=excluded.one_shot,
                created_at=excluded.created_at,
                updated_at=excluded.updated_at,
                last_run=excluded.last_run,
                next_run=excluded.next_run,
                run_count=excluded.run_count,
                failure_count=excluded.failure_count,
                max_retries=excluded.max_retries,
                max_concurrent=excluded.max_concurrent,
                delivery_json=excluded.delivery_json,
                jitter_max_secs=excluded.jitter_max_secs,
                session_mode=excluded.session_mode,
                agentpin_jwt=excluded.agentpin_jwt,
                input_json=excluded.input_json WHERE ?24",
            rusqlite::params![
                job.job_id.to_string(),
                job.name,
                job.cron_expression,
                job.timezone,
                agent_json,
                policy_json,
                audit_str,
                status_str,
                job.enabled as i32,
                job.one_shot as i32,
                job.created_at.to_rfc3339(),
                job.updated_at.to_rfc3339(),
                job.last_run.map(|t| t.to_rfc3339()),
                job.next_run.map(|t| t.to_rfc3339()),
                job.run_count as i64,
                job.failure_count as i64,
                job.max_retries as i32,
                job.max_concurrent as i32,
                delivery_json,
                job.jitter_max_secs as i32,
                session_mode_str,
                job.agentpin_jwt,
                job.input.to_string(),
                replace,
            ],
        )
        .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        Ok(())
    }

    /// Open (or create) the store at the given path.
    pub fn open(path: &std::path::Path) -> Result<Self, JobStoreError> {
        #[cfg(unix)]
        Self::protect_store(path)?;
        let conn = rusqlite::Connection::open_with_flags(
            path,
            rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE
                | rusqlite::OpenFlags::SQLITE_OPEN_CREATE
                | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX
                | rusqlite::OpenFlags::SQLITE_OPEN_NOFOLLOW,
        )
        .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        // WAL mode for concurrent access.
        conn.pragma_update(None, "journal_mode", "WAL")
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        // Create tables before wrapping in Mutex (avoids blocking_lock in async context).
        Self::init_schema(&conn)?;

        Ok(Self {
            conn: tokio::sync::Mutex::new(conn),
        })
    }

    #[cfg(unix)]
    fn protect_store(path: &std::path::Path) -> Result<(), JobStoreError> {
        use std::os::unix::fs::{MetadataExt, OpenOptionsExt};
        let invalid = JobStoreError::Sqlite;
        let parent = path
            .parent()
            .ok_or_else(|| invalid("cron store needs a private parent directory".into()))?;
        crate::reasoning::protected_journal::private_directory(parent).map_err(invalid)?;
        // Check existing sidecars before SQLite can follow them. The private
        // parent is runtime authority and must never be granted to workers.
        for suffix in ["", "-wal", "-shm", "-journal"] {
            let mut name = path.as_os_str().to_os_string();
            name.push(suffix);
            let candidate = std::path::PathBuf::from(name);
            let metadata = match std::fs::symlink_metadata(&candidate) {
                Ok(metadata) => metadata,
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
                Err(e) => return Err(invalid(e.to_string())),
            };
            // SAFETY: geteuid has no memory-safety preconditions.
            if !metadata.is_file()
                || metadata.uid() != unsafe { libc::geteuid() }
                || metadata.mode() & 0o077 != 0
                || metadata.nlink() != 1
            {
                return Err(invalid(
                    "cron store and sidecars must be private runtime-owned regular files".into(),
                ));
            }
        }
        let file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .mode(0o600)
            .custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC)
            .open(path)
            .map_err(|e| invalid(e.to_string()))?;
        file.sync_all()
            .and_then(|_| std::fs::File::open(parent)?.sync_all())
            .map_err(|e| invalid(e.to_string()))
    }

    /// A persisted schedule must execute under the project that owns its store.
    pub async fn bind_project(&self, project: &std::path::Path) -> Result<(), JobStoreError> {
        let project = project
            .canonicalize()
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        let project = project.to_str().ok_or_else(|| {
            JobStoreError::Serialization("cron project path must be UTF-8".into())
        })?;
        let mut conn = self.conn.lock().await;
        let tx = conn
            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        tx.execute_batch("CREATE TABLE IF NOT EXISTS cron_store_owner (singleton INTEGER PRIMARY KEY CHECK(singleton=1),project TEXT NOT NULL)")
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        let saved: Option<String> = tx
            .query_row(
                "SELECT project FROM cron_store_owner WHERE singleton=1",
                [],
                |r| r.get(0),
            )
            .optional()
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        if let Some(saved) = saved {
            if saved != project {
                return Err(JobStoreError::Serialization(
                    "cron store belongs to a different execution project".into(),
                ));
            }
        } else {
            tx.execute(
                "INSERT INTO cron_store_owner(singleton,project) VALUES(1,?1)",
                [project],
            )
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        }
        tx.commit()
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))
    }

    /// Open an in-memory store (useful for tests).
    pub fn open_in_memory() -> Result<Self, JobStoreError> {
        let conn = rusqlite::Connection::open_in_memory()
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        Self::init_schema(&conn)?;

        Ok(Self {
            conn: tokio::sync::Mutex::new(conn),
        })
    }

    /// Legacy global path, retained only for explicit operator migration.
    /// New schedulers use their execution project's protected store.
    pub fn default_path() -> std::path::PathBuf {
        let base = dirs::data_dir().unwrap_or_else(|| std::path::PathBuf::from("."));
        base.join("symbi").join("cron_jobs.db")
    }

    fn init_schema(conn: &rusqlite::Connection) -> Result<(), JobStoreError> {
        conn.busy_timeout(std::time::Duration::from_secs(2))
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        conn.pragma_update(None, "synchronous", "FULL")
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS schema_version (
                version INTEGER PRIMARY KEY
            );

            INSERT OR IGNORE INTO schema_version (version) VALUES (1);

            CREATE TABLE IF NOT EXISTS cron_jobs (
                job_id       TEXT PRIMARY KEY,
                name         TEXT NOT NULL,
                cron_expr    TEXT NOT NULL,
                timezone     TEXT NOT NULL,
                agent_json   TEXT NOT NULL,
                policy_ids   TEXT NOT NULL DEFAULT '[]',
                audit_level  TEXT NOT NULL DEFAULT 'None',
                status       TEXT NOT NULL DEFAULT 'Active',
                enabled      INTEGER NOT NULL DEFAULT 1,
                one_shot     INTEGER NOT NULL DEFAULT 0,
                created_at   TEXT NOT NULL,
                updated_at   TEXT NOT NULL,
                last_run     TEXT,
                next_run     TEXT,
                run_count    INTEGER NOT NULL DEFAULT 0,
                failure_count INTEGER NOT NULL DEFAULT 0,
                max_retries  INTEGER NOT NULL DEFAULT 3,
                max_concurrent INTEGER NOT NULL DEFAULT 1,
                delivery_json TEXT,
                jitter_max_secs INTEGER NOT NULL DEFAULT 0,
                session_mode TEXT NOT NULL DEFAULT '\"EphemeralWithSummary\"',
                agentpin_jwt TEXT
            );

            CREATE TABLE IF NOT EXISTS job_run_log (
                run_id        TEXT PRIMARY KEY,
                job_id        TEXT NOT NULL,
                agent_id      TEXT NOT NULL,
                started_at    TEXT NOT NULL,
                completed_at  TEXT,
                status        TEXT NOT NULL,
                error         TEXT,
                exec_time_ms  INTEGER,
                FOREIGN KEY (job_id) REFERENCES cron_jobs(job_id) ON DELETE CASCADE
            );

            CREATE TABLE IF NOT EXISTS cron_occurrences (
                invocation_id TEXT PRIMARY KEY,
                job_id TEXT NOT NULL,
                request_hash TEXT NOT NULL,
                request_json TEXT NOT NULL,
                state TEXT NOT NULL,
                created_at TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_cron_occurrences_state ON cron_occurrences(state,created_at);
            CREATE INDEX IF NOT EXISTS idx_cron_occurrences_job ON cron_occurrences(job_id,state);

            CREATE INDEX IF NOT EXISTS idx_cron_jobs_next_run ON cron_jobs(next_run);
            CREATE INDEX IF NOT EXISTS idx_job_run_log_job_id ON job_run_log(job_id);
            CREATE INDEX IF NOT EXISTS idx_job_run_log_started ON job_run_log(started_at);",
        )
        .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        // Additive migration preserves existing jobs and run history.
        for (table, column, declaration) in [
            ("cron_jobs", "input_json", "TEXT NOT NULL DEFAULT 'null'"),
            ("job_run_log", "execution_json", "TEXT"),
            ("job_run_log", "admission_audit_json", "TEXT"),
            ("job_run_log", "resolution_json", "TEXT"),
        ] {
            let mut statement = conn
                .prepare(&format!("PRAGMA table_info({table})"))
                .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
            let names = statement
                .query_map([], |row| row.get::<_, String>(1))
                .map_err(|e| JobStoreError::Sqlite(e.to_string()))?
                .collect::<Result<Vec<_>, _>>()
                .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
            if !names.iter().any(|name| name == column) {
                conn.execute_batch(&format!(
                    "ALTER TABLE {table} ADD COLUMN {column} {declaration}"
                ))
                .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
            }
        }
        // Legacy running records have no durable caller identity. Preserve them
        // as unresolved and stop their jobs instead of manufacturing a retry.
        conn.execute_batch("UPDATE cron_jobs SET status='\"DeadLetter\"',enabled=0 WHERE job_id IN
            (SELECT job_id FROM job_run_log WHERE status IN ('pending','running') AND NOT EXISTS
                (SELECT 1 FROM cron_occurrences WHERE invocation_id=job_run_log.run_id));
            UPDATE job_run_log SET status='unresolved',error='legacy run has no durable invocation identity; inspect original effects before retry'
                WHERE status IN ('pending','running') AND NOT EXISTS
                (SELECT 1 FROM cron_occurrences WHERE invocation_id=job_run_log.run_id);")
            .map_err(|e|JobStoreError::Sqlite(e.to_string()))?;
        Ok(())
    }
}

#[async_trait]
impl JobStore for SqliteJobStore {
    async fn save_job(&self, job: &CronJobDefinition) -> Result<(), JobStoreError> {
        self.persist_job(job, true).await
    }

    async fn get_job(&self, job_id: CronJobId) -> Result<Option<CronJobDefinition>, JobStoreError> {
        let conn = self.conn.lock().await;
        let mut stmt = conn
            .prepare(
                "SELECT job_id, name, cron_expr, timezone, agent_json, policy_ids,
                        audit_level, status, enabled, one_shot, created_at, updated_at,
                        last_run, next_run, run_count, failure_count, max_retries,
                        max_concurrent, delivery_json, jitter_max_secs, session_mode, agentpin_jwt, input_json
                 FROM cron_jobs WHERE job_id = ?1",
            )
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        let result = stmt
            .query_row(rusqlite::params![job_id.to_string()], row_to_job)
            .optional()
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        match result {
            Some(Ok(job)) => Ok(Some(job)),
            Some(Err(e)) => Err(e),
            None => Ok(None),
        }
    }

    async fn delete_job(&self, job_id: CronJobId) -> Result<bool, JobStoreError> {
        let conn = self.conn.lock().await;
        let rows = conn
            .execute(
                "DELETE FROM cron_jobs WHERE job_id = ?1",
                rusqlite::params![job_id.to_string()],
            )
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        Ok(rows > 0)
    }

    async fn list_jobs(
        &self,
        status_filter: Option<CronJobStatus>,
    ) -> Result<Vec<CronJobDefinition>, JobStoreError> {
        let conn = self.conn.lock().await;
        let (sql, params): (&str, Vec<Box<dyn rusqlite::types::ToSql>>) = match status_filter {
            Some(s) => {
                let status_str = serde_json::to_string(&s)
                    .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
                (
                    "SELECT job_id, name, cron_expr, timezone, agent_json, policy_ids,
                            audit_level, status, enabled, one_shot, created_at, updated_at,
                            last_run, next_run, run_count, failure_count, max_retries,
                            max_concurrent, delivery_json, jitter_max_secs, session_mode, agentpin_jwt, input_json
                     FROM cron_jobs WHERE status = ?1 ORDER BY created_at",
                    vec![Box::new(status_str)],
                )
            }
            None => (
                "SELECT job_id, name, cron_expr, timezone, agent_json, policy_ids,
                        audit_level, status, enabled, one_shot, created_at, updated_at,
                        last_run, next_run, run_count, failure_count, max_retries,
                        max_concurrent, delivery_json, jitter_max_secs, session_mode, agentpin_jwt, input_json
                 FROM cron_jobs ORDER BY created_at",
                vec![],
            ),
        };

        let mut stmt = conn
            .prepare(sql)
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
            params.iter().map(|p| p.as_ref()).collect();
        let rows = stmt
            .query_map(param_refs.as_slice(), row_to_job)
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        let mut jobs = Vec::new();
        for row_result in rows {
            let inner = row_result.map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
            jobs.push(inner?);
        }
        Ok(jobs)
    }

    async fn get_due_jobs(
        &self,
        now: DateTime<Utc>,
    ) -> Result<Vec<CronJobDefinition>, JobStoreError> {
        let conn = self.conn.lock().await;
        let mut stmt = conn
            .prepare(
                "SELECT job_id, name, cron_expr, timezone, agent_json, policy_ids,
                        audit_level, status, enabled, one_shot, created_at, updated_at,
                        last_run, next_run, run_count, failure_count, max_retries,
                        max_concurrent, delivery_json, jitter_max_secs, session_mode, agentpin_jwt, input_json
                 FROM cron_jobs
                 WHERE enabled = 1
                   AND status = '\"Active\"'
                   AND next_run IS NOT NULL
                   AND next_run <= ?1
                   AND NOT EXISTS(SELECT 1 FROM job_run_log WHERE job_run_log.job_id=cron_jobs.job_id AND job_run_log.status='unresolved')
                   AND NOT EXISTS(SELECT 1 FROM cron_occurrences WHERE cron_occurrences.job_id=cron_jobs.job_id AND cron_occurrences.state='unresolved')
                 ORDER BY next_run",
            )
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        let rows = stmt
            .query_map(rusqlite::params![now.to_rfc3339()], row_to_job)
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        let mut jobs = Vec::new();
        for row_result in rows {
            let inner = row_result.map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
            jobs.push(inner?);
        }
        Ok(jobs)
    }

    async fn update_run_state(
        &self,
        job_id: CronJobId,
        last_run: DateTime<Utc>,
        next_run: Option<DateTime<Utc>>,
        run_count: u64,
        status: CronJobStatus,
        enabled: bool,
    ) -> Result<(), JobStoreError> {
        let status_str = serde_json::to_string(&status)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let conn = self.conn.lock().await;
        let rows = conn
            .execute(
                "UPDATE cron_jobs
                 SET last_run = ?1, next_run = ?2, run_count = ?3,
                     status = ?4, enabled = ?5, updated_at = ?6
                 WHERE job_id = ?7",
                rusqlite::params![
                    last_run.to_rfc3339(),
                    next_run.map(|t| t.to_rfc3339()),
                    run_count as i64,
                    status_str,
                    enabled as i32,
                    Utc::now().to_rfc3339(),
                    job_id.to_string(),
                ],
            )
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        if rows == 0 {
            return Err(JobStoreError::NotFound(job_id));
        }
        Ok(())
    }

    async fn record_failure(
        &self,
        job_id: CronJobId,
        failure_count: u64,
        status: CronJobStatus,
    ) -> Result<(), JobStoreError> {
        let status_str = serde_json::to_string(&status)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let conn = self.conn.lock().await;
        let rows = conn
            .execute(
                "UPDATE cron_jobs SET failure_count = ?1, status = ?2, updated_at = ?3 WHERE job_id = ?4",
                rusqlite::params![
                    failure_count as i64,
                    status_str,
                    Utc::now().to_rfc3339(),
                    job_id.to_string(),
                ],
            )
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        if rows == 0 {
            return Err(JobStoreError::NotFound(job_id));
        }
        Ok(())
    }

    async fn save_run_record(&self, record: &JobRunRecord) -> Result<(), JobStoreError> {
        let conn = self.conn.lock().await;
        let changed = conn.execute(
            "INSERT INTO job_run_log
                (run_id, job_id, agent_id, started_at, completed_at, status, error, exec_time_ms, execution_json, admission_audit_json)
             VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)
             ON CONFLICT(run_id) DO UPDATE SET completed_at=excluded.completed_at,
                status=excluded.status, error=excluded.error, exec_time_ms=excluded.exec_time_ms,
                execution_json=excluded.execution_json, admission_audit_json=excluded.admission_audit_json
             WHERE job_run_log.job_id=excluded.job_id AND job_run_log.agent_id=excluded.agent_id
                AND job_run_log.status='running' AND job_run_log.started_at=excluded.started_at",
            rusqlite::params![
                record.run_id.to_string(),
                record.job_id.to_string(),
                record.agent_id.to_string(),
                record.started_at.to_rfc3339(),
                record.completed_at.map(|t| t.to_rfc3339()),
                record.status.to_string(),
                record.error,
                record.execution_time_ms.map(|v| v as i64),
                record.execution.as_ref().map(serde_json::to_string).transpose()
                    .map_err(|e| JobStoreError::Serialization(e.to_string()))?,
                record.admission_audit.as_ref().map(serde_json::to_string).transpose()
                    .map_err(|e| JobStoreError::Serialization(e.to_string()))?,
            ],
        )
        .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
        if changed != 1 {
            return Err(JobStoreError::Serialization(
                "run record is already terminal or its identity does not match".into(),
            ));
        }
        Ok(())
    }

    async fn get_run_history(
        &self,
        job_id: CronJobId,
        limit: usize,
    ) -> Result<Vec<JobRunRecord>, JobStoreError> {
        let conn = self.conn.lock().await;
        let mut stmt = conn
            .prepare(
                "SELECT run_id, job_id, agent_id, started_at, completed_at, status, error, exec_time_ms, execution_json, admission_audit_json, resolution_json
                 FROM job_run_log
                 WHERE job_id = ?1
                 ORDER BY started_at DESC
                 LIMIT ?2",
            )
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        let rows = stmt
            .query_map(
                rusqlite::params![job_id.to_string(), limit as i64],
                row_to_run_record,
            )
            .map_err(|e| JobStoreError::Sqlite(e.to_string()))?;

        let mut records = Vec::new();
        for row_result in rows {
            let inner = row_result.map_err(|e| JobStoreError::Sqlite(e.to_string()))?;
            records.push(inner?);
        }
        Ok(records)
    }
}

// ── Row-mapping helpers ───────────────────────────────────────────────

fn row_to_job(
    row: &rusqlite::Row<'_>,
) -> rusqlite::Result<Result<CronJobDefinition, JobStoreError>> {
    let job_id_str: String = row.get(0)?;
    let name: String = row.get(1)?;
    let cron_expr: String = row.get(2)?;
    let timezone: String = row.get(3)?;
    let agent_json: String = row.get(4)?;
    let policy_json: String = row.get(5)?;
    let audit_str: String = row.get(6)?;
    let status_str: String = row.get(7)?;
    let enabled: i32 = row.get(8)?;
    let one_shot: i32 = row.get(9)?;
    let created_str: String = row.get(10)?;
    let updated_str: String = row.get(11)?;
    let last_run_str: Option<String> = row.get(12)?;
    let next_run_str: Option<String> = row.get(13)?;
    let run_count: i64 = row.get(14)?;
    let failure_count: i64 = row.get(15)?;
    let max_retries: i32 = row.get(16)?;
    let max_concurrent: i32 = row.get(17)?;
    let delivery_json: Option<String> = row.get(18)?;
    let jitter_max_secs: i32 = row.get(19)?;
    let session_mode_str: String = row.get(20)?;
    let agentpin_jwt: Option<String> = row.get(21)?;
    let input_json: String = row.get(22)?;

    Ok((|| -> Result<CronJobDefinition, JobStoreError> {
        let job_id: CronJobId = job_id_str
            .parse()
            .map_err(|e: uuid::Error| JobStoreError::Serialization(e.to_string()))?;
        let agent_config = serde_json::from_str(&agent_json)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let policy_ids = serde_json::from_str(&policy_json)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let audit_level = serde_json::from_str(&audit_str)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let status = serde_json::from_str(&status_str)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let created_at = DateTime::parse_from_rfc3339(&created_str)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?
            .with_timezone(&Utc);
        let updated_at = DateTime::parse_from_rfc3339(&updated_str)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?
            .with_timezone(&Utc);
        let last_run = last_run_str
            .map(|s| {
                DateTime::parse_from_rfc3339(&s)
                    .map(|dt| dt.with_timezone(&Utc))
                    .map_err(|e| JobStoreError::Serialization(e.to_string()))
            })
            .transpose()?;
        let next_run = next_run_str
            .map(|s| {
                DateTime::parse_from_rfc3339(&s)
                    .map(|dt| dt.with_timezone(&Utc))
                    .map_err(|e| JobStoreError::Serialization(e.to_string()))
            })
            .transpose()?;
        let delivery_config = delivery_json
            .map(|s| serde_json::from_str(&s))
            .transpose()
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let session_mode = serde_json::from_str(&session_mode_str)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;

        Ok(CronJobDefinition {
            job_id,
            name,
            cron_expression: cron_expr,
            timezone,
            agent_config,
            policy_ids,
            audit_level,
            status,
            enabled: enabled != 0,
            one_shot: one_shot != 0,
            created_at,
            updated_at,
            last_run,
            next_run,
            run_count: run_count as u64,
            failure_count: failure_count as u64,
            max_retries: max_retries as u32,
            max_concurrent: max_concurrent as u32,
            delivery_config,
            jitter_max_secs: jitter_max_secs as u32,
            session_mode,
            agentpin_jwt,
            input: serde_json::from_str(&input_json)
                .map_err(|e| JobStoreError::Serialization(e.to_string()))?,
        })
    })())
}

fn row_to_run_record(
    row: &rusqlite::Row<'_>,
) -> rusqlite::Result<Result<JobRunRecord, JobStoreError>> {
    let run_id_str: String = row.get(0)?;
    let job_id_str: String = row.get(1)?;
    let agent_id_str: String = row.get(2)?;
    let started_str: String = row.get(3)?;
    let completed_str: Option<String> = row.get(4)?;
    let status_str: String = row.get(5)?;
    let error: Option<String> = row.get(6)?;
    let exec_time: Option<i64> = row.get(7)?;
    let execution_json: Option<String> = row.get(8)?;
    let admission_audit_json: Option<String> = row.get(9)?;
    let resolution_json: Option<String> = row.get(10)?;

    Ok((|| -> Result<JobRunRecord, JobStoreError> {
        let run_id = Uuid::parse_str(&run_id_str)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?;
        let job_id: CronJobId = job_id_str
            .parse()
            .map_err(|e: uuid::Error| JobStoreError::Serialization(e.to_string()))?;
        let agent_id: crate::types::AgentId = agent_id_str
            .parse()
            .map_err(|e| JobStoreError::Serialization(format!("agent_id: {e}")))?;
        let started_at = DateTime::parse_from_rfc3339(&started_str)
            .map_err(|e| JobStoreError::Serialization(e.to_string()))?
            .with_timezone(&Utc);
        let completed_at = completed_str
            .map(|s| {
                DateTime::parse_from_rfc3339(&s)
                    .map(|dt| dt.with_timezone(&Utc))
                    .map_err(|e| JobStoreError::Serialization(e.to_string()))
            })
            .transpose()?;
        let status: JobRunStatus = status_str
            .parse()
            .map_err(|e: String| JobStoreError::Serialization(e))?;

        Ok(JobRunRecord {
            resolution: resolution_json
                .map(|s| serde_json::from_str(&s))
                .transpose()
                .map_err(|e| JobStoreError::Serialization(e.to_string()))?,
            admission_audit: admission_audit_json
                .map(|s| serde_json::from_str(&s))
                .transpose()
                .map_err(|e| JobStoreError::Serialization(e.to_string()))?,
            run_id,
            job_id,
            agent_id,
            started_at,
            completed_at,
            status,
            error,
            execution_time_ms: exec_time.map(|v| v as u64),
            execution: execution_json
                .map(|s| serde_json::from_str(&s))
                .transpose()
                .map_err(|e| JobStoreError::Serialization(e.to_string()))?,
        })
    })())
}

/// Extension trait so we can use `optional()` on rusqlite queries.
trait OptionalExt<T> {
    fn optional(self) -> Result<Option<T>, rusqlite::Error>;
}

impl<T> OptionalExt<T> for Result<T, rusqlite::Error> {
    fn optional(self) -> Result<Option<T>, rusqlite::Error> {
        match self {
            Ok(v) => Ok(Some(v)),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(e),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{
        AgentConfig, AgentId, ExecutionMode, Priority, ResourceLimits, SecurityTier,
    };
    use std::collections::HashMap;

    fn test_agent_config() -> AgentConfig {
        AgentConfig {
            id: AgentId::new(),
            name: "cron_test_agent".to_string(),
            dsl_source: "agent cron_test {}".to_string(),
            execution_mode: ExecutionMode::Ephemeral,
            security_tier: SecurityTier::Tier1,
            resource_limits: ResourceLimits::default(),
            capabilities: vec![],
            policies: vec![],
            metadata: HashMap::new(),
            priority: Priority::Normal,
        }
    }

    fn test_job() -> CronJobDefinition {
        CronJobDefinition::new(
            "hourly_check".to_string(),
            "0 * * * *".to_string(),
            "UTC".to_string(),
            test_agent_config(),
        )
    }

    #[tokio::test]
    async fn save_and_get_job() {
        let store = SqliteJobStore::open_in_memory().unwrap();
        let job = test_job();
        store.save_job(&job).await.unwrap();

        let loaded = store.get_job(job.job_id).await.unwrap().unwrap();
        assert_eq!(loaded.name, "hourly_check");
        assert_eq!(loaded.cron_expression, "0 * * * *");
        assert!(loaded.enabled);
    }

    #[tokio::test]
    async fn delete_job() {
        let store = SqliteJobStore::open_in_memory().unwrap();
        let job = test_job();
        store.save_job(&job).await.unwrap();

        assert!(store.delete_job(job.job_id).await.unwrap());
        assert!(store.get_job(job.job_id).await.unwrap().is_none());
        // Deleting again returns false.
        assert!(!store.delete_job(job.job_id).await.unwrap());
    }

    #[tokio::test]
    async fn list_jobs_with_filter() {
        let store = SqliteJobStore::open_in_memory().unwrap();

        let mut job1 = test_job();
        job1.name = "job1".to_string();
        store.save_job(&job1).await.unwrap();

        let mut job2 = test_job();
        job2.name = "job2".to_string();
        job2.status = CronJobStatus::Paused;
        store.save_job(&job2).await.unwrap();

        let all = store.list_jobs(None).await.unwrap();
        assert_eq!(all.len(), 2);

        let active = store.list_jobs(Some(CronJobStatus::Active)).await.unwrap();
        assert_eq!(active.len(), 1);
        assert_eq!(active[0].name, "job1");

        let paused = store.list_jobs(Some(CronJobStatus::Paused)).await.unwrap();
        assert_eq!(paused.len(), 1);
        assert_eq!(paused[0].name, "job2");
    }

    #[tokio::test]
    async fn get_due_jobs_filtering() {
        let store = SqliteJobStore::open_in_memory().unwrap();
        let now = Utc::now();

        // Job due in the past → should be returned.
        let mut due_job = test_job();
        due_job.name = "due".to_string();
        due_job.next_run = Some(now - chrono::Duration::minutes(5));
        store.save_job(&due_job).await.unwrap();

        // Job due in the future → should not be returned.
        let mut future_job = test_job();
        future_job.name = "future".to_string();
        future_job.next_run = Some(now + chrono::Duration::hours(1));
        store.save_job(&future_job).await.unwrap();

        // Disabled job → should not be returned.
        let mut disabled_job = test_job();
        disabled_job.name = "disabled".to_string();
        disabled_job.next_run = Some(now - chrono::Duration::minutes(1));
        disabled_job.enabled = false;
        store.save_job(&disabled_job).await.unwrap();

        let due = store.get_due_jobs(now).await.unwrap();
        assert_eq!(due.len(), 1);
        assert_eq!(due[0].name, "due");
    }

    #[tokio::test]
    async fn update_run_state() {
        let store = SqliteJobStore::open_in_memory().unwrap();
        let job = test_job();
        store.save_job(&job).await.unwrap();

        let now = Utc::now();
        let next = now + chrono::Duration::hours(1);
        store
            .update_run_state(job.job_id, now, Some(next), 1, CronJobStatus::Active, true)
            .await
            .unwrap();

        let loaded = store.get_job(job.job_id).await.unwrap().unwrap();
        assert_eq!(loaded.run_count, 1);
        assert!(loaded.last_run.is_some());
        assert!(loaded.next_run.is_some());
    }

    #[tokio::test]
    async fn record_failure_and_dead_letter() {
        let store = SqliteJobStore::open_in_memory().unwrap();
        let job = test_job();
        store.save_job(&job).await.unwrap();

        store
            .record_failure(job.job_id, 4, CronJobStatus::Failed)
            .await
            .unwrap();

        let loaded = store.get_job(job.job_id).await.unwrap().unwrap();
        assert_eq!(loaded.failure_count, 4);
        assert_eq!(loaded.status, CronJobStatus::Failed);
    }

    #[tokio::test]
    async fn run_history_crud() {
        let store = SqliteJobStore::open_in_memory().unwrap();
        let job = test_job();
        store.save_job(&job).await.unwrap();

        let record = JobRunRecord {
            resolution: None,
            admission_audit: None,
            run_id: Uuid::new_v4(),
            job_id: job.job_id,
            agent_id: job.agent_config.id,
            started_at: Utc::now(),
            completed_at: Some(Utc::now()),
            status: JobRunStatus::Succeeded,
            error: None,
            execution_time_ms: Some(1234),
            execution: None,
        };
        store.save_run_record(&record).await.unwrap();

        let history = store.get_run_history(job.job_id, 10).await.unwrap();
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].status, JobRunStatus::Succeeded);
        assert_eq!(history[0].execution_time_ms, Some(1234));
    }

    #[tokio::test]
    async fn concurrent_reads() {
        let store = std::sync::Arc::new(SqliteJobStore::open_in_memory().unwrap());
        let job = test_job();
        store.save_job(&job).await.unwrap();

        let mut handles = vec![];
        for _ in 0..10 {
            let s = store.clone();
            let jid = job.job_id;
            handles.push(tokio::spawn(async move {
                s.get_job(jid).await.unwrap().unwrap()
            }));
        }
        for h in handles {
            let loaded = h.await.unwrap();
            assert_eq!(loaded.name, "hourly_check");
        }
    }
    #[tokio::test]
    async fn old_schema_migration_preserves_history_and_adds_input_and_results() {
        let root = tempfile::tempdir().unwrap();
        let path = root.path().join("private/jobs.sqlite");
        let store = SqliteJobStore::open(&path).unwrap();
        let mut job = test_job();
        store.save_job(&job).await.unwrap();
        let record = JobRunRecord {
            resolution: None,
            admission_audit: None,
            run_id: Uuid::new_v4(),
            job_id: job.job_id,
            agent_id: job.agent_config.id,
            started_at: Utc::now(),
            completed_at: Some(Utc::now()),
            status: JobRunStatus::Succeeded,
            error: None,
            execution_time_ms: Some(10),
            execution: None,
        };
        store.save_run_record(&record).await.unwrap();
        drop(store);
        // These columns are absent from the prior persisted schema.
        let connection = rusqlite::Connection::open(&path).unwrap();
        connection.execute_batch("ALTER TABLE cron_jobs DROP COLUMN input_json; ALTER TABLE job_run_log DROP COLUMN execution_json;").unwrap();
        drop(connection);
        let store = SqliteJobStore::open(&path).unwrap();
        assert!(store
            .get_job(job.job_id)
            .await
            .unwrap()
            .unwrap()
            .input
            .is_null());
        assert_eq!(
            store.get_run_history(job.job_id, 10).await.unwrap()[0].run_id,
            record.run_id
        );
        assert!(
            store.save_run_record(&record).await.is_err(),
            "terminal runs are immutable"
        );
        job.input = serde_json::json!({"message":"payload"});
        store.save_job(&job).await.unwrap();
        let task = super::super::ScheduledTask::new(job.agent_config.clone());
        let mut completion = super::super::task_manager::TaskCompletion::new(
            &task,
            super::super::task_manager::TaskStatus::Completed,
            None,
        );
        completion.output = Some("actual result".into());
        let mut pending = record.clone();
        pending.run_id = Uuid::new_v4();
        pending.completed_at = None;
        pending.status = JobRunStatus::Running;
        store.save_run_record(&pending).await.unwrap();
        pending.completed_at = Some(Utc::now());
        pending.status = JobRunStatus::Succeeded;
        pending.execution = Some(completion);
        store.save_run_record(&pending).await.unwrap();
        drop(store);
        let store = SqliteJobStore::open(&path).unwrap();
        assert_eq!(
            store.get_job(job.job_id).await.unwrap().unwrap().input,
            job.input
        );
        let history = store.get_run_history(job.job_id, 10).await.unwrap();
        assert_eq!(history.len(), 2);
        let run = history.iter().find(|r| r.run_id == pending.run_id).unwrap();
        assert_eq!(
            run.execution.as_ref().unwrap().output.as_deref(),
            Some("actual result")
        );
    }
}