harn-vm 0.10.162

//! Machine-scoped monetary allowance. Every attempted provider call reserves
//! its catalog upper bound in one SQLite transaction before transport. A lost
//! process leaves its reservation in place: uncertainty is not a free call.
use std::future::{poll_fn, Future};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::task::Poll;
use std::time::Duration;

use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior};
use rust_decimal::prelude::ToPrimitive;
use rust_decimal::Decimal;
use time::{Date, OffsetDateTime};
use uuid::Uuid;

use super::{error, unavailable, DenialKind};
use crate::runtime_sqlite::{initialize_runtime_sqlite_migrating, RuntimeSqliteSchema};
use crate::value::{ErrorCategory, VmError};

const SCALE: i64 = 1_000_000;
const LIFETIME_PERIOD: &str = "L";
const SQLITE_SCHEMA: RuntimeSqliteSchema = RuntimeSqliteSchema::new(
    "machine_spend_quota",
    2,
    "CREATE TABLE IF NOT EXISTS spend_policy (scope TEXT PRIMARY KEY, daily_limit INTEGER, monthly_limit INTEGER, contract_broken INTEGER NOT NULL DEFAULT 0);
     CREATE TABLE IF NOT EXISTS spend_period (scope TEXT NOT NULL, period TEXT NOT NULL, reserved INTEGER NOT NULL CHECK(reserved >= 0), PRIMARY KEY(scope, period));
     CREATE TABLE IF NOT EXISTS spend_attempt (id TEXT PRIMARY KEY, scope TEXT NOT NULL, day TEXT NOT NULL, month TEXT NOT NULL, reserved INTEGER NOT NULL CHECK(reserved >= 0), actual INTEGER, settled INTEGER NOT NULL DEFAULT 0, created_ms INTEGER NOT NULL);
     CREATE TABLE IF NOT EXISTS spend_policy_audit (id INTEGER PRIMARY KEY AUTOINCREMENT, scope TEXT NOT NULL, at_ms INTEGER NOT NULL, approved_by TEXT NOT NULL, old_daily INTEGER, old_monthly INTEGER, new_daily INTEGER, new_monthly INTEGER);
     CREATE INDEX IF NOT EXISTS spend_attempt_scope ON spend_attempt(scope);",
);

#[derive(Clone, Debug)]
pub struct MachineSpendQuota {
    inner: Arc<QuotaConfig>,
}

#[derive(Debug)]
struct QuotaConfig {
    path: PathBuf,
    scope: String,
    clock: Arc<dyn harn_clock::Clock>,
}

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MachineSpendPolicy {
    pub daily_limit_microusd: Option<i64>,
    pub monthly_limit_microusd: Option<i64>,
    /// Ceiling for the stable billing scope across restarts and calendar periods.
    pub lifetime_limit_microusd: Option<i64>,
}

impl MachineSpendPolicy {
    fn valid(&self) -> bool {
        let limits = [
            self.daily_limit_microusd,
            self.monthly_limit_microusd,
            self.lifetime_limit_microusd,
        ];
        limits.iter().any(Option::is_some) && limits.iter().flatten().all(|limit| *limit >= 0)
    }
}

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct MachineSpendReceipt {
    pub reserved_microusd: i64,
    /// Known provider usage only. Missing usage is counted separately.
    pub actual_known_microusd: i64,
    pub usage_unknown_attempts: u64,
    pub contract_broken: bool,
    pub daily_remaining_microusd: Option<i64>,
    pub monthly_remaining_microusd: Option<i64>,
    pub lifetime_reserved_microusd: i64,
    pub lifetime_actual_known_microusd: i64,
    pub lifetime_usage_unknown_attempts: u64,
    pub lifetime_remaining_microusd: Option<i64>,
    pub daily_reset_unix_ms: i64,
    pub monthly_reset_unix_ms: i64,
}

#[derive(Clone, Copy, serde::Serialize)]
#[serde(rename_all = "snake_case")]
enum QuotaPeriod {
    Day,
    Month,
    Lifetime,
}

struct LifetimeSpend {
    reserved: i64,
    actual_known: i64,
    usage_unknown: u64,
}

fn lifetime_spend(connection: &Connection, scope: &str) -> Result<LifetimeSpend, VmError> {
    connection.query_row(
        "SELECT COALESCE(SUM(reserved), 0), COALESCE(SUM(actual), 0), COALESCE(SUM(CASE WHEN actual IS NULL THEN 1 ELSE 0 END), 0) FROM spend_attempt WHERE scope = ?1",
        [scope],
        |row| Ok(LifetimeSpend {
            reserved: row.get(0)?,
            actual_known: row.get(1)?,
            usage_unknown: row.get::<_, i64>(2)?.try_into().unwrap_or(u64::MAX),
        }),
    ).map_err(db_error)
}

#[derive(Debug)]
pub(super) struct DurableReservation {
    config: Arc<QuotaConfig>,
    id: String,
}

impl MachineSpendQuota {
    /// The host supplies one private, stable machine/user database path and
    /// billing scope shared by all its sessions. Limits are integer micro-USD;
    /// no floating-point rounding occurs in admission.
    pub fn open(
        path: impl AsRef<Path>,
        scope: impl Into<String>,
        policy: MachineSpendPolicy,
    ) -> Result<Self, VmError> {
        Self::open_with_clock(path, scope, policy, harn_clock::RealClock::arc())
    }

    /// Supply the clock used for billing periods and audit timestamps.
    pub fn open_with_clock(
        path: impl AsRef<Path>,
        scope: impl Into<String>,
        policy: MachineSpendPolicy,
        clock: Arc<dyn harn_clock::Clock>,
    ) -> Result<Self, VmError> {
        let scope = scope.into();
        if !path.as_ref().is_absolute() || scope.trim().is_empty() || !policy.valid() {
            return Err(error(
                DenialKind::InvalidBudget,
                "machine spend quota requires an absolute path, a scope, and valid limits",
            ));
        }
        let config = Arc::new(QuotaConfig {
            path: path.as_ref().to_path_buf(),
            scope,
            clock,
        });
        let connection = connect(&config.path)?;
        connection
            .execute(
                "INSERT OR IGNORE INTO spend_policy(scope, daily_limit, monthly_limit, lifetime_limit) VALUES (?1, ?2, ?3, ?4)",
                params![config.scope, policy.daily_limit_microusd, policy.monthly_limit_microusd, policy.lifetime_limit_microusd],
            )
            .map_err(db_error)?;
        let stored = read_policy(&connection, &config.scope)?;
        if stored != policy {
            return Err(error(
                DenialKind::InvalidBudget,
                "machine spend policy differs from the durable policy; use an authorized policy update",
            ));
        }
        Ok(Self { inner: config })
    }

    pub fn policy(&self) -> Result<MachineSpendPolicy, VmError> {
        read_policy(&connect(&self.inner.path)?, &self.inner.scope)
    }

    /// Install this machine allowance for a provider execution tree. A nested
    /// call inherits its parent's machine ledger and cannot replace it with a
    /// different path or billing scope.
    pub async fn scope<F: Future>(&self, inner: F) -> Result<F::Output, VmError> {
        self.validate_parent(true)?;
        let mut admission = super::SCOPE.with(|slot| slot.borrow().clone());
        let parent_ledger = admission.ledger.clone();
        admission.machine = Some(self.clone());
        admission.host_owned = true;
        let mut ambient = crate::orchestration::AmbientExecutionScope::capture_for_inline_subtask();
        ambient.set_llm_admission(admission);
        let mut inner = std::pin::pin!(crate::orchestration::scope_ambient(ambient, inner));
        poll_fn(|context| {
            if let Err(error) = self.validate_parent(false) {
                return Poll::Ready(Err(error));
            }
            let active = super::SCOPE.with(|slot| slot.borrow().clone());
            if (active.host_owned || crate::current_execution_scope().is_some())
                && !Arc::ptr_eq(&active.ledger, &parent_ledger)
            {
                return Poll::Ready(Err(error(
                    DenialKind::ScopeUnavailable,
                    "machine spend scope moved under an unrelated execution budget",
                )));
            }
            inner.as_mut().poll(context).map(Ok)
        })
        .await
    }

    fn validate_parent(&self, check_late_activation: bool) -> Result<(), VmError> {
        let active = super::SCOPE.with(|slot| slot.borrow().clone());
        if active
            .machine
            .as_ref()
            .is_some_and(|parent| !parent.same_scope(self))
        {
            return Err(error(
                DenialKind::ScopeUnavailable,
                "an unrelated machine allowance cannot replace the active budget",
            ));
        }
        if check_late_activation && active.machine.is_none() {
            let ledger = active
                .ledger
                .lock()
                .map_err(|_| error(DenialKind::ScopeUnavailable, "admission ledger poisoned"))?;
            if ledger.prior_unreserved_attempt || ledger.attempts_started > 0 {
                return Err(error(
                    DenialKind::LateActivation,
                    "machine spend quota must start before the first provider attempt",
                ));
            }
        }
        Ok(())
    }

    pub(super) fn same_scope(&self, other: &Self) -> bool {
        self.inner.path == other.inner.path && self.inner.scope == other.inner.scope
    }

    /// An approved host-side budget action changes the durable ceiling and
    /// leaves an audit row. The host must authenticate `approved_by` before
    /// calling; a program in the VM cannot invoke this method.
    pub fn update_policy(
        &self,
        policy: MachineSpendPolicy,
        approved_by: &str,
    ) -> Result<(), VmError> {
        if approved_by.trim().is_empty() || !policy.valid() {
            return Err(error(
                DenialKind::InvalidBudget,
                "invalid approved spend policy update",
            ));
        }
        let mut connection = connect(&self.inner.path)?;
        let tx = connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(db_error)?;
        let old = read_policy(&tx, &self.inner.scope)?;
        tx.execute(
            "UPDATE spend_policy SET daily_limit = ?2, monthly_limit = ?3, lifetime_limit = ?4 WHERE scope = ?1",
            params![
                self.inner.scope,
                policy.daily_limit_microusd,
                policy.monthly_limit_microusd,
                policy.lifetime_limit_microusd
            ],
        )
        .map_err(db_error)?;
        tx.execute(
            "INSERT INTO spend_policy_audit(scope, at_ms, approved_by, old_daily, old_monthly, new_daily, new_monthly, old_lifetime, new_lifetime) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![self.inner.scope, harn_clock::now_wall_ms(self.inner.clock.as_ref()), approved_by, old.daily_limit_microusd, old.monthly_limit_microusd, policy.daily_limit_microusd, policy.monthly_limit_microusd, old.lifetime_limit_microusd, policy.lifetime_limit_microusd],
        )
        .map_err(db_error)?;
        tx.commit().map_err(db_error)
    }

    pub(super) fn reserve(&self, amount: Decimal) -> Result<DurableReservation, VmError> {
        let amount = micros_ceil(amount)?;
        let now = self.inner.clock.now_utc();
        let (day, month, daily_reset, monthly_reset) = periods(now);
        let mut connection = connect(&self.inner.path)?;
        let tx = connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(db_error)?;
        let policy = read_policy(&tx, &self.inner.scope)?;
        if contract_broken(&tx, &self.inner.scope)? {
            return Err(error(
                DenialKind::ProviderContractViolation,
                "machine spend scope has an unresolved provider contract violation",
            ));
        }
        for (period, kind, limit, reset) in [
            (
                day.as_str(),
                QuotaPeriod::Day,
                policy.daily_limit_microusd,
                Some(daily_reset),
            ),
            (
                month.as_str(),
                QuotaPeriod::Month,
                policy.monthly_limit_microusd,
                Some(monthly_reset),
            ),
            (
                LIFETIME_PERIOD,
                QuotaPeriod::Lifetime,
                policy.lifetime_limit_microusd,
                None,
            ),
        ] {
            if let Some(limit) = limit {
                let used = period_reserved(&tx, &self.inner.scope, period)?;
                if amount > limit.saturating_sub(used) {
                    return Err(exhausted(kind, limit, used, reset));
                }
            }
        }
        let id = Uuid::new_v4().to_string();
        tx.execute(
            "INSERT INTO spend_attempt(id, scope, day, month, reserved, created_ms) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
            params![id, self.inner.scope, day, month, amount, harn_clock::now_wall_ms(self.inner.clock.as_ref())],
        )
        .map_err(db_error)?;
        for period in [day.as_str(), month.as_str(), LIFETIME_PERIOD] {
            tx.execute(
                "INSERT INTO spend_period(scope, period, reserved) VALUES (?1, ?2, ?3) ON CONFLICT(scope, period) DO UPDATE SET reserved = reserved + excluded.reserved",
                params![self.inner.scope, period, amount],
            )
            .map_err(db_error)?;
        }
        tx.commit().map_err(db_error)?;
        Ok(DurableReservation {
            config: self.inner.clone(),
            id,
        })
    }

    /// Budget preflight reads indexed counters, without scanning usage history.
    pub(super) fn remaining_microusd(&self) -> Result<i64, VmError> {
        let (day, month, _, _) = periods(self.inner.clock.now_utc());
        let mut connection = connect(&self.inner.path)?;
        let snapshot = connection
            .transaction_with_behavior(TransactionBehavior::Deferred)
            .map_err(db_error)?;
        let policy = read_policy(&snapshot, &self.inner.scope)?;
        if contract_broken(&snapshot, &self.inner.scope)? {
            return Ok(0);
        }
        let mut remaining = None;
        for (period, limit) in [
            (day.as_str(), policy.daily_limit_microusd),
            (month.as_str(), policy.monthly_limit_microusd),
            (LIFETIME_PERIOD, policy.lifetime_limit_microusd),
        ] {
            if let Some(limit) = limit {
                let used = period_reserved(&snapshot, &self.inner.scope, period)?;
                let available = limit.saturating_sub(used).max(0);
                remaining = Some(remaining.map_or(available, |prior: i64| prior.min(available)));
            }
        }
        snapshot.commit().map_err(db_error)?;
        remaining.ok_or_else(|| error(DenialKind::InvalidBudget, "machine spend limit is absent"))
    }

    pub fn receipt(&self) -> Result<MachineSpendReceipt, VmError> {
        let now = self.inner.clock.now_utc();
        let (day, month, daily_reset_unix_ms, monthly_reset_unix_ms) = periods(now);
        let mut connection = connect(&self.inner.path)?;
        let snapshot = connection
            .transaction_with_behavior(TransactionBehavior::Deferred)
            .map_err(db_error)?;
        let policy = read_policy(&snapshot, &self.inner.scope)?;
        let contract_broken = contract_broken(&snapshot, &self.inner.scope)?;
        let day_used = period_reserved(&snapshot, &self.inner.scope, &day)?;
        let month_used = period_reserved(&snapshot, &self.inner.scope, &month)?;
        let (actual_known, usage_unknown): (i64, i64) = snapshot
            .query_row(
                "SELECT COALESCE(SUM(actual), 0), SUM(CASE WHEN actual IS NULL THEN 1 ELSE 0 END) FROM spend_attempt WHERE scope = ?1 AND month = ?2",
                params![self.inner.scope, month],
                |row| Ok((row.get(0)?, row.get::<_, Option<i64>>(1)?.unwrap_or(0))),
            )
            .map_err(db_error)?;
        let lifetime = lifetime_spend(&snapshot, &self.inner.scope)?;
        snapshot.commit().map_err(db_error)?;
        Ok(MachineSpendReceipt {
            reserved_microusd: month_used,
            actual_known_microusd: actual_known,
            usage_unknown_attempts: usage_unknown.try_into().unwrap_or(u64::MAX),
            contract_broken,
            lifetime_reserved_microusd: lifetime.reserved,
            lifetime_actual_known_microusd: lifetime.actual_known,
            lifetime_usage_unknown_attempts: lifetime.usage_unknown,
            lifetime_remaining_microusd: policy
                .lifetime_limit_microusd
                .map(|limit| limit.saturating_sub(lifetime.reserved).max(0)),
            daily_remaining_microusd: policy
                .daily_limit_microusd
                .map(|limit| limit.saturating_sub(day_used).max(0)),
            monthly_remaining_microusd: policy
                .monthly_limit_microusd
                .map(|limit| limit.saturating_sub(month_used).max(0)),
            daily_reset_unix_ms,
            monthly_reset_unix_ms,
        })
    }
}

impl DurableReservation {
    pub(super) fn invalidate(&self) -> Result<(), VmError> {
        let connection = connect(&self.config.path)?;
        connection
            .execute(
                "UPDATE spend_policy SET contract_broken = 1 WHERE scope = ?1",
                [&self.config.scope],
            )
            .map_err(db_error)?;
        Ok(())
    }

    /// Release only the part the provider's complete usage proves unused.
    /// A failed settlement leaves the full durable hold in place.
    pub(super) fn settle(&self, upper: Decimal, actual: Option<Decimal>) -> Result<(), VmError> {
        let upper = micros_ceil(upper)?;
        let actual = actual.map(micros_ceil).transpose()?;
        let mut connection = connect(&self.config.path)?;
        let tx = connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(db_error)?;
        let (day, month, reserved): (String, String, i64) = tx
            .query_row(
                "SELECT day, month, reserved FROM spend_attempt WHERE id = ?1 AND scope = ?2 AND settled = 0",
                params![self.id, self.config.scope],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .map_err(db_error)?;
        if upper > reserved || actual.is_some_and(|actual| actual > upper) {
            tx.execute(
                "UPDATE spend_policy SET contract_broken = 1 WHERE scope = ?1",
                [&self.config.scope],
            )
            .map_err(db_error)?;
            tx.commit().map_err(db_error)?;
            return Err(error(
                DenialKind::ProviderContractViolation,
                "provider usage exceeded the machine spend reservation",
            ));
        }
        tx.execute(
            "UPDATE spend_attempt SET reserved = ?2, actual = ?3, settled = 1 WHERE id = ?1",
            params![self.id, upper, actual],
        )
        .map_err(db_error)?;
        for period in [day.as_str(), month.as_str(), LIFETIME_PERIOD] {
            tx.execute(
                "UPDATE spend_period SET reserved = reserved - ?3 WHERE scope = ?1 AND period = ?2",
                params![self.config.scope, period, reserved - upper],
            )
            .map_err(db_error)?;
        }
        tx.commit().map_err(db_error)
    }
}

fn connect(path: &Path) -> Result<Connection, VmError> {
    let parent = path.parent().ok_or_else(|| {
        error(
            DenialKind::InvalidBudget,
            "spend ledger needs a parent directory",
        )
    })?;
    std::fs::create_dir_all(parent).map_err(db_error)?;
    let connection = Connection::open(path).map_err(db_error)?;
    connection
        .busy_timeout(Duration::from_secs(5))
        .map_err(db_error)?;
    // WAL promotion belongs to `initialize_runtime_sqlite`, which does it under
    // a cross-process lock. Switching a fresh file here first raced every other
    // process opening it and failed about one round in twenty with
    // "database is locked" (harn#8818).
    initialize_runtime_sqlite_migrating(
        &connection,
        Duration::from_secs(5),
        &SQLITE_SCHEMA,
        |tx| {
            for (table, column) in [
                ("spend_policy", "lifetime_limit"),
                ("spend_policy_audit", "old_lifetime"),
                ("spend_policy_audit", "new_lifetime"),
            ] {
                let mut statement = tx.prepare(&format!("PRAGMA table_info({table})"))?;
                let columns = statement
                    .query_map([], |row| row.get::<_, String>(1))?
                    .collect::<rusqlite::Result<Vec<_>>>()?;
                if !columns.iter().any(|name| name == column) {
                    tx.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} INTEGER"))?;
                }
            }
            // Upgrade the existing period projection once, in the schema
            // transaction. Admission then reads one indexed row per ceiling.
            tx.execute(
                "INSERT INTO spend_period(scope, period, reserved) SELECT scope, ?1, SUM(reserved) FROM spend_attempt GROUP BY scope",
                [LIFETIME_PERIOD],
            )?;
            Ok(())
        },
    )
    .map_err(db_error)?;
    Ok(connection)
}

fn period_reserved(connection: &Connection, scope: &str, period: &str) -> Result<i64, VmError> {
    Ok(connection
        .query_row(
            "SELECT reserved FROM spend_period WHERE scope = ?1 AND period = ?2",
            params![scope, period],
            |row| row.get(0),
        )
        .optional()
        .map_err(db_error)?
        .unwrap_or(0))
}

fn read_policy(connection: &Connection, scope: &str) -> Result<MachineSpendPolicy, VmError> {
    connection
        .query_row(
            "SELECT daily_limit, monthly_limit, lifetime_limit FROM spend_policy WHERE scope = ?1",
            [scope],
            |row| {
                Ok(MachineSpendPolicy {
                    daily_limit_microusd: row.get(0)?,
                    monthly_limit_microusd: row.get(1)?,
                    lifetime_limit_microusd: row.get(2)?,
                })
            },
        )
        .map_err(db_error)
}

fn contract_broken(connection: &Connection, scope: &str) -> Result<bool, VmError> {
    connection
        .query_row(
            "SELECT contract_broken FROM spend_policy WHERE scope = ?1",
            [scope],
            |row| row.get::<_, bool>(0),
        )
        .map_err(db_error)
}

fn micros_ceil(amount: Decimal) -> Result<i64, VmError> {
    if amount.is_sign_negative() {
        return Err(error(DenialKind::InvalidBudget, "negative spend amount"));
    }
    (amount * Decimal::from(SCALE))
        .ceil()
        .to_i64()
        .ok_or_else(|| {
            error(
                DenialKind::InvalidBudget,
                "spend amount exceeds ledger range",
            )
        })
}

fn periods(now: OffsetDateTime) -> (String, String, i64, i64) {
    let date = now.date();
    let next_day = date.next_day().expect("UTC date within supported range");
    let first_next_month = if date.month() == time::Month::December {
        Date::from_calendar_date(date.year() + 1, time::Month::January, 1)
    } else {
        Date::from_calendar_date(date.year(), date.month().next(), 1)
    }
    .expect("UTC month within supported range");
    (
        format!("D:{date}"),
        format!("M:{:04}-{:02}", date.year(), date.month() as u8),
        next_day.midnight().assume_utc().unix_timestamp() * 1000,
        first_next_month.midnight().assume_utc().unix_timestamp() * 1000,
    )
}

/// A ledger failure that can say whether another connection held the lock.
trait LedgerFailure: std::fmt::Display {
    fn contended(&self) -> bool {
        false
    }
}

impl LedgerFailure for rusqlite::Error {
    fn contended(&self) -> bool {
        harn_sqlite::sqlite_contention(self).is_some()
    }
}

impl LedgerFailure for crate::runtime_sqlite::RuntimeSqliteError {
    fn contended(&self) -> bool {
        self.is_busy_or_locked()
    }
}

impl LedgerFailure for std::io::Error {}

/// An unreadable ledger is not an exhausted budget: a contended one is
/// `resource_busy`, anything else an `environment` failure (harn#8818).
fn db_error(error_value: impl LedgerFailure) -> VmError {
    let category = if error_value.contended() {
        ErrorCategory::ResourceBusy
    } else {
        ErrorCategory::Environment
    };
    unavailable(
        category,
        &format!("machine spend ledger unavailable: {error_value}"),
    )
}

fn exhausted(period: QuotaPeriod, limit: i64, used: i64, reset_unix_ms: Option<i64>) -> VmError {
    VmError::Thrown(crate::schema::json_to_vm_value(&serde_json::json!({
        "category": "budget_exceeded",
        "origin": "local",
        "kind": "terminal",
        "reason": "budget_exceeded",
        "admission_reason": "insufficient_allowance",
        "quota_scope": "machine",
        "period": period,
        "limit_microusd": limit,
        "reserved_microusd": used,
        "remaining_microusd": limit.saturating_sub(used).max(0),
        "reset_at_unix_ms": reset_unix_ms,
        "message": "machine spend quota exhausted before provider transport"
    })))
}

#[cfg(test)]
mod tests {
    use super::*;

    fn policy(limit: i64) -> MachineSpendPolicy {
        MachineSpendPolicy {
            daily_limit_microusd: Some(limit),
            monthly_limit_microusd: Some(limit),
            lifetime_limit_microusd: None,
        }
    }

    #[test]
    fn utc_periods_reset_at_day_month_and_year_boundaries() {
        let leap_day = Date::from_calendar_date(2028, time::Month::February, 29)
            .unwrap()
            .midnight()
            .assume_utc();
        let (day, month, daily_reset, monthly_reset) = periods(leap_day);
        assert_eq!(day, "D:2028-02-29");
        assert_eq!(month, "M:2028-02");
        assert_eq!(daily_reset, monthly_reset);
        assert_eq!(
            monthly_reset,
            Date::from_calendar_date(2028, time::Month::March, 1)
                .unwrap()
                .midnight()
                .assume_utc()
                .unix_timestamp()
                * 1000
        );
        let year_end = Date::from_calendar_date(2028, time::Month::December, 31)
            .unwrap()
            .midnight()
            .assume_utc();
        let (_, month, daily_reset, monthly_reset) = periods(year_end);
        assert_eq!(month, "M:2028-12");
        assert_eq!(daily_reset, monthly_reset);
        assert_eq!(
            monthly_reset,
            Date::from_calendar_date(2029, time::Month::January, 1)
                .unwrap()
                .midnight()
                .assume_utc()
                .unix_timestamp()
                * 1000
        );
    }

    #[test]
    fn fractional_microdollars_round_up_before_admission() {
        assert_eq!(micros_ceil(Decimal::new(1, 7)).unwrap(), 1);
        assert_eq!(micros_ceil(Decimal::new(1, 6)).unwrap(), 1);
        assert_eq!(micros_ceil(Decimal::new(11, 7)).unwrap(), 2);
    }

    #[test]
    fn crashed_attempt_keeps_both_period_reservations_after_reopen() {
        let temp = tempfile::tempdir().unwrap();
        let path = temp.path().join("spend.sqlite");
        let first = MachineSpendQuota::open(&path, "person", policy(1_000_000)).unwrap();
        drop(first.reserve(Decimal::new(6, 1)).unwrap());
        drop(first);
        let restarted = MachineSpendQuota::open(&path, "person", policy(1_000_000)).unwrap();
        assert!(restarted.reserve(Decimal::new(5, 1)).is_err());
        let receipt = restarted.receipt().unwrap();
        assert_eq!(receipt.reserved_microusd, 600_000);
        assert_eq!(receipt.actual_known_microusd, 0);
        assert_eq!(receipt.usage_unknown_attempts, 1);
        assert_eq!(receipt.daily_remaining_microusd, Some(400_000));
        assert_eq!(receipt.monthly_remaining_microusd, Some(400_000));
    }

    #[test]
    fn lifetime_allowance_keeps_interrupted_reservations_across_calendar_rollover() {
        let temp = tempfile::tempdir().unwrap();
        let path = temp.path().join("spend.sqlite");
        let clock = harn_clock::PausedClock::new(
            Date::from_calendar_date(2028, time::Month::December, 31)
                .unwrap()
                .midnight()
                .assume_utc(),
        );
        let policy = MachineSpendPolicy {
            daily_limit_microusd: None,
            monthly_limit_microusd: None,
            lifetime_limit_microusd: Some(1_000_000),
        };
        let first =
            MachineSpendQuota::open_with_clock(&path, "campaign", policy.clone(), clock.clone())
                .unwrap();
        drop(first.reserve(Decimal::new(6, 1)).unwrap());
        drop(first);
        clock.advance_time(time::Duration::days(2));
        let restarted =
            MachineSpendQuota::open_with_clock(&path, "campaign", policy, clock).unwrap();
        let VmError::Thrown(receipt) = restarted
            .reserve(Decimal::new(5, 1))
            .expect_err("calendar rollover replenished campaign allowance")
        else {
            panic!("rollover refusal did not produce a structured budget receipt");
        };
        let receipt = crate::value::vm_to_storage_json(&receipt).unwrap();
        assert_eq!(receipt["category"], "budget_exceeded");
        assert_eq!(receipt["period"], "lifetime");
        assert_eq!(receipt["reserved_microusd"], 600_000);
        assert_eq!(receipt["remaining_microusd"], 400_000);
        assert!(receipt["reset_at_unix_ms"].is_null());
        let ledger = restarted.receipt().unwrap();
        assert_eq!(ledger.lifetime_reserved_microusd, 600_000);
        assert_eq!(ledger.lifetime_actual_known_microusd, 0);
        assert_eq!(ledger.lifetime_usage_unknown_attempts, 1);
        assert_eq!(ledger.lifetime_remaining_microusd, Some(400_000));
        // A genuinely different billing scope remains independent.
        let other = MachineSpendQuota::open(
            &path,
            "other-campaign",
            MachineSpendPolicy {
                daily_limit_microusd: None,
                monthly_limit_microusd: None,
                lifetime_limit_microusd: Some(1_000_000),
            },
        )
        .unwrap();
        assert!(other.reserve(Decimal::new(5, 1)).is_ok());
    }

    #[test]
    fn lifetime_settlement_and_authorized_policy_updates_preserve_prior_spend() {
        let temp = tempfile::tempdir().unwrap();
        let path = temp.path().join("spend.sqlite");
        let policy = MachineSpendPolicy {
            daily_limit_microusd: None,
            monthly_limit_microusd: None,
            lifetime_limit_microusd: Some(1_000_000),
        };
        let quota = MachineSpendQuota::open(&path, "campaign", policy.clone()).unwrap();
        quota
            .reserve(Decimal::new(6, 1))
            .unwrap()
            .settle(Decimal::new(2, 1), Some(Decimal::new(2, 1)))
            .unwrap();
        drop(quota.reserve(Decimal::new(5, 1)).unwrap());
        let receipt = quota.receipt().unwrap();
        assert_eq!(receipt.lifetime_reserved_microusd, 700_000);
        assert_eq!(receipt.lifetime_actual_known_microusd, 200_000);
        assert_eq!(receipt.lifetime_usage_unknown_attempts, 1);
        let increased = MachineSpendPolicy {
            lifetime_limit_microusd: Some(2_000_000),
            ..policy
        };
        assert!(MachineSpendQuota::open(&path, "campaign", increased.clone()).is_err());
        quota.update_policy(increased.clone(), "").unwrap_err();
        quota
            .update_policy(increased.clone(), "approved-host-action")
            .unwrap();
        let reopened = MachineSpendQuota::open(&path, "campaign", increased).unwrap();
        assert_eq!(
            reopened.receipt().unwrap().lifetime_remaining_microusd,
            Some(1_300_000)
        );
        let audit: (i64, i64) = connect(&path).unwrap().query_row(
            "SELECT old_lifetime, new_lifetime FROM spend_policy_audit WHERE scope = 'campaign'",
            [], |row| Ok((row.get(0)?, row.get(1)?)),
        ).unwrap();
        assert_eq!(audit, (1_000_000, 2_000_000));
    }

    #[test]
    fn version_one_policy_migrates_without_changing_its_authority() {
        let temp = tempfile::tempdir().unwrap();
        let path = temp.path().join("spend.sqlite");
        let connection = Connection::open(&path).unwrap();
        harn_sqlite::initialize_file(&connection, Duration::from_secs(5),
            harn_sqlite::SchemaVersion::new("machine_spend_quota", 1), |tx| {
                tx.execute_batch("CREATE TABLE spend_policy (scope TEXT PRIMARY KEY, daily_limit INTEGER, monthly_limit INTEGER, contract_broken INTEGER NOT NULL DEFAULT 0);
                    CREATE TABLE spend_policy_audit (id INTEGER PRIMARY KEY AUTOINCREMENT, scope TEXT NOT NULL, at_ms INTEGER NOT NULL, approved_by TEXT NOT NULL, old_daily INTEGER, old_monthly INTEGER, new_daily INTEGER, new_monthly INTEGER);
                    CREATE TABLE spend_period (scope TEXT NOT NULL, period TEXT NOT NULL, reserved INTEGER NOT NULL CHECK(reserved >= 0), PRIMARY KEY(scope, period));
                    CREATE TABLE spend_attempt (id TEXT PRIMARY KEY, scope TEXT NOT NULL, day TEXT NOT NULL, month TEXT NOT NULL, reserved INTEGER NOT NULL CHECK(reserved >= 0), actual INTEGER, settled INTEGER NOT NULL DEFAULT 0, created_ms INTEGER NOT NULL);
                    INSERT INTO spend_policy VALUES ('person', 1000000, 1000000, 0);
                    INSERT INTO spend_period VALUES ('person', 'D:2029-01-02', 600000), ('person', 'M:2029-01', 600000);
                    INSERT INTO spend_attempt VALUES ('old-call', 'person', 'D:2029-01-02', 'M:2029-01', 600000, NULL, 0, 0);")
            }).unwrap();
        drop(connection);
        let now = Date::from_calendar_date(2029, time::Month::January, 2)
            .unwrap()
            .midnight()
            .assume_utc();
        let clock = harn_clock::PausedClock::new(now);
        let reopened =
            MachineSpendQuota::open_with_clock(&path, "person", policy(1_000_000), clock).unwrap();
        assert!(reopened.reserve(Decimal::new(5, 1)).is_err());
        assert_eq!(
            reopened.receipt().unwrap().lifetime_reserved_microusd,
            600_000
        );
        assert_eq!(reopened.policy().unwrap(), policy(1_000_000));
        assert_eq!(
            reopened.receipt().unwrap().lifetime_usage_unknown_attempts,
            1
        );
        let incompatible = MachineSpendPolicy {
            lifetime_limit_microusd: Some(2_000_000),
            ..policy(1_000_000)
        };
        assert!(MachineSpendQuota::open(&path, "person", incompatible).is_err());
        // An old reader must refuse the upgraded schema, never discard its limits.
        let old = harn_sqlite::require_file_initialized::<rusqlite::Error>(
            &Connection::open(&path).unwrap(),
            Duration::from_secs(5),
            harn_sqlite::SchemaVersion::new("machine_spend_quota", 1),
        );
        assert!(matches!(
            old,
            Err(harn_sqlite::InitializationError::NewerSchemaVersion { .. })
        ));
    }

    #[test]
    fn concurrent_connections_cannot_spend_the_same_remainder() {
        for policy in [
            policy(1_000_000),
            MachineSpendPolicy {
                daily_limit_microusd: None,
                monthly_limit_microusd: None,
                lifetime_limit_microusd: Some(1_000_000),
            },
        ] {
            let temp = tempfile::tempdir().unwrap();
            let path = temp.path().join("spend.sqlite");
            let quota = MachineSpendQuota::open(&path, "person", policy.clone()).unwrap();
            let handles: Vec<_> = (0..8)
                .map(|_| {
                    let path = path.clone();
                    let policy = policy.clone();
                    crate::runtime_stack::spawn(move || {
                        let quota = MachineSpendQuota::open(path, "person", policy).unwrap();
                        quota.reserve(Decimal::new(3, 1)).is_ok()
                    })
                })
                .collect();
            let admitted = handles
                .into_iter()
                .map(|handle| handle.join().unwrap())
                .filter(|admitted| *admitted)
                .count();
            assert_eq!(admitted, 3);
            let receipt = quota.receipt().unwrap();
            assert_eq!(receipt.reserved_microusd, 900_000);
            assert_eq!(receipt.lifetime_reserved_microusd, 900_000);
        }
    }

    #[test]
    fn separate_processes_share_one_machine_allowance() {
        if let Some(path) = std::env::var_os("HARN_MACHINE_SPEND_TEST_CHILD_PATH") {
            let quota = MachineSpendQuota::open(path, "person", policy(1_000_000)).unwrap();
            let admitted = quota.reserve(Decimal::new(6, 1)).is_ok();
            println!("MACHINE_SPEND_ADMITTED={admitted}");
            return;
        }
        let temp = tempfile::tempdir().unwrap();
        let path = temp.path().join("spend.sqlite");
        let binary = std::env::current_exe().unwrap();
        let children: Vec<_> = (0..2)
            .map(|_| {
                std::process::Command::new(&binary)
                    .arg("--exact")
                    .arg("llm::admission::durable::tests::separate_processes_share_one_machine_allowance")
                    .arg("--nocapture")
                    .env("HARN_MACHINE_SPEND_TEST_CHILD_PATH", &path)
                    .stdout(std::process::Stdio::piped())
                    .stderr(std::process::Stdio::piped())
                    .spawn()
                    .unwrap()
            })
            .collect();
        let admitted = children
            .into_iter()
            .map(|child| child.wait_with_output().unwrap())
            .filter(|child| {
                assert!(
                    child.status.success(),
                    "{}",
                    String::from_utf8_lossy(&child.stderr)
                );
                String::from_utf8_lossy(&child.stdout).contains("MACHINE_SPEND_ADMITTED=true")
            })
            .count();
        assert_eq!(admitted, 1);
        assert_eq!(
            MachineSpendQuota::open(path, "person", policy(1_000_000))
                .unwrap()
                .receipt()
                .unwrap()
                .reserved_microusd,
            600_000
        );
    }

    /// Before harn#8818 about one round in twenty of four processes opening a
    /// fresh ledger failed with "database is locked" at the WAL switch.
    #[test]
    fn concurrent_first_opens_of_a_fresh_ledger_all_succeed() {
        const TEST: &str =
            "llm::admission::durable::tests::concurrent_first_opens_of_a_fresh_ledger_all_succeed";
        if let Some(path) = std::env::var_os("HARN_MACHINE_SPEND_TEST_CHILD_PATH") {
            match MachineSpendQuota::open(path, "person", policy(1_000_000)) {
                Ok(_) => println!("MACHINE_SPEND_OPENED=true"),
                Err(error) => println!("MACHINE_SPEND_OPENED=false {error}"),
            }
            return;
        }
        let binary = std::env::current_exe().unwrap();
        for round in 0..100 {
            let temp = tempfile::tempdir().unwrap();
            let path = temp.path().join("spend.sqlite");
            let children: Vec<_> = (0..4)
                .map(|_| {
                    std::process::Command::new(&binary)
                        .args(["--exact", TEST, "--nocapture"])
                        .env("HARN_MACHINE_SPEND_TEST_CHILD_PATH", &path)
                        .stdout(std::process::Stdio::piped())
                        .stderr(std::process::Stdio::null())
                        .spawn()
                        .unwrap()
                })
                .collect();
            for child in children {
                let stdout =
                    String::from_utf8_lossy(&child.wait_with_output().unwrap().stdout).to_string();
                assert!(
                    stdout.contains("MACHINE_SPEND_OPENED=true"),
                    "round {round}: {stdout}"
                );
            }
        }
    }

    #[test]
    fn an_unreadable_ledger_is_not_reported_as_a_spent_budget() {
        let busy = rusqlite::Error::SqliteFailure(
            rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
            Some("database is locked".to_string()),
        );
        assert_eq!(
            crate::value::error_to_category(&db_error(busy)),
            ErrorCategory::ResourceBusy
        );
        let missing = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
        assert_eq!(
            crate::value::error_to_category(&db_error(missing)),
            ErrorCategory::Environment
        );
    }

    #[test]
    fn complete_usage_releases_only_proven_unused_amount() {
        let temp = tempfile::tempdir().unwrap();
        let quota = MachineSpendQuota::open(
            temp.path().join("spend.sqlite"),
            "person",
            policy(1_000_000),
        )
        .unwrap();
        let first = quota.reserve(Decimal::new(6, 1)).unwrap();
        first
            .settle(Decimal::new(2, 1), Some(Decimal::new(15, 2)))
            .unwrap();
        assert!(quota.reserve(Decimal::new(7, 1)).is_ok());
        let receipt = quota.receipt().unwrap();
        assert_eq!(receipt.reserved_microusd, 900_000);
        assert_eq!(receipt.actual_known_microusd, 150_000);
        assert_eq!(receipt.usage_unknown_attempts, 1);
    }

    #[test]
    fn provider_contract_violation_denies_future_processes() {
        let temp = tempfile::tempdir().unwrap();
        let path = temp.path().join("spend.sqlite");
        let quota = MachineSpendQuota::open(&path, "person", policy(1_000_000)).unwrap();
        let reservation = quota.reserve(Decimal::new(6, 1)).unwrap();
        assert!(reservation.settle(Decimal::new(7, 1), None).is_err());
        assert!(quota.receipt().unwrap().contract_broken);
        drop(quota);
        let restarted = MachineSpendQuota::open(path, "person", policy(1_000_000)).unwrap();
        assert!(restarted.receipt().unwrap().contract_broken);
        assert!(restarted.reserve(Decimal::new(1, 1)).is_err());
    }

    #[test]
    fn policy_raise_requires_named_approval_and_is_durably_audited() {
        let temp = tempfile::tempdir().unwrap();
        let path = temp.path().join("spend.sqlite");
        let quota = MachineSpendQuota::open(&path, "person", policy(1_000_000)).unwrap();
        assert!(quota.update_policy(policy(2_000_000), "").is_err());
        quota
            .update_policy(policy(2_000_000), "host-approval-1")
            .unwrap();
        assert_eq!(quota.policy().unwrap(), policy(2_000_000));
        let connection = connect(&path).unwrap();
        let audit_count: i64 = connection
            .query_row("SELECT COUNT(*) FROM spend_policy_audit", [], |row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(audit_count, 1);
    }

    #[test]
    fn utc_day_and_month_reset_are_explicit() {
        let now = Date::from_calendar_date(2026, time::Month::September, 30)
            .unwrap()
            .with_hms(23, 59, 0)
            .unwrap()
            .assume_utc();
        let (day, month, day_reset, month_reset) = periods(now);
        assert_eq!(day, "D:2026-09-30");
        assert_eq!(month, "M:2026-09");
        assert_eq!(day_reset, month_reset);
        assert_eq!(
            day_reset,
            Date::from_calendar_date(2026, time::Month::October, 1)
                .unwrap()
                .midnight()
                .assume_utc()
                .unix_timestamp()
                * 1000
        );
    }
}