diff --git a/Cargo.lock b/Cargo.lock index f87f82b0..33348b5e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -318,6 +318,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "tracing", "uuid", ] @@ -455,6 +456,7 @@ dependencies = [ "ardur-cost-gate", "ardur-cron", "ardur-cron-ui", + "ardur-durability", "ardur-embeddings", "ardur-fused-runtime", "ardur-memory", @@ -473,6 +475,7 @@ dependencies = [ "clap", "comrak", "crossterm", + "flate2", "futures", "hex", "ignore", @@ -488,6 +491,7 @@ dependencies = [ "serde_json", "sha2 0.11.0", "syntect", + "tar", "tempfile", "thiserror 2.0.18", "tokio", @@ -567,6 +571,36 @@ dependencies = [ "uuid", ] +[[package]] +name = "ardur-delegate-tool" +version = "0.0.1" +dependencies = [ + "anyhow", + "ardur-cap-token", + "ardur-cost-gate", + "ardur-multi-agent", + "ardur-receipt", + "ardur-runtime", + "ardur-tool-registry", + "async-trait", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "uuid", +] + +[[package]] +name = "ardur-durability" +version = "0.0.1" +dependencies = [ + "rustix", + "tempfile", + "thiserror 2.0.18", + "tracing", + "uuid", +] + [[package]] name = "ardur-e2e-tests" version = "0.0.0" @@ -581,6 +615,7 @@ dependencies = [ "ardur-channel-telegram", "ardur-cli", "ardur-cost-gate", + "ardur-cron", "ardur-embeddings", "ardur-fused-runtime", "ardur-hooks-openclaw-compat", @@ -697,23 +732,6 @@ dependencies = [ "serde", ] -[[package]] -name = "ardur-governance" -version = "0.0.1" -dependencies = [ - "ardur-cap-token", - "ardur-receipt", - "base64 0.22.1", - "chrono", - "p256 0.13.2", - "rand_core 0.6.4", - "serde", - "serde_json", - "sha2 0.11.0", - "thiserror 2.0.18", - "uuid", -] - [[package]] name = "ardur-health" version = "0.0.1" @@ -793,19 +811,6 @@ dependencies = [ "uuid", ] -[[package]] -name = "ardur-loop-detector" -version = "0.0.1" -dependencies = [ - "ardur-receipt", - "proptest", - "serde", - "serde_json", - "sha2 0.11.0", - "thiserror 2.0.18", - "uuid", -] - [[package]] name = "ardur-media-audio" version = "0.0.1" @@ -1123,6 +1128,8 @@ dependencies = [ "ardur-channel-matrix", "ardur-channel-telegram", "ardur-cost-gate", + "ardur-delegate-tool", + "ardur-durability", "ardur-fused-runtime", "ardur-hooks-openclaw-compat", "ardur-media-audio", @@ -3809,6 +3816,16 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24" +[[package]] +name = "filetime" +version = "0.2.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" +dependencies = [ + "cfg-if", + "libc", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -9222,6 +9239,17 @@ dependencies = [ "serde", ] +[[package]] +name = "tar" +version = "0.4.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f6221d9a6003c78398e3b239969f352578258df48c8eb051caadae0015bc840" +dependencies = [ + "filetime", + "libc", + "xattr", +] + [[package]] name = "teloxide" version = "0.17.0" @@ -10820,6 +10848,16 @@ dependencies = [ "zeroize", ] +[[package]] +name = "xattr" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" +dependencies = [ + "libc", + "rustix", +] + [[package]] name = "xxhash-rust" version = "0.8.15" diff --git a/crates/automation/Cargo.toml b/crates/automation/Cargo.toml index cadf4f8b..198105d4 100644 --- a/crates/automation/Cargo.toml +++ b/crates/automation/Cargo.toml @@ -29,6 +29,7 @@ serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } +tracing = { workspace = true } uuid = { workspace = true } [dev-dependencies] diff --git a/crates/automation/src/driver.rs b/crates/automation/src/driver.rs new file mode 100644 index 00000000..f0f78861 --- /dev/null +++ b/crates/automation/src/driver.rs @@ -0,0 +1,310 @@ +//! Timer → executor bridge for proactive automation schedules (issue #347). +//! +//! [`ProactiveAutomationLoop::fire_due`](crate::proactive::ProactiveAutomationLoop::fire_due) +//! already executes due schedules end-to-end — validate → fused-runtime submit → +//! channel delivery → fire-count persistence — but nothing in the substrate ever +//! drove it on a clock. Persisted schedules were therefore inert: created, +//! stored, and never run. [`ScheduleDriver`] is the missing bridge. It owns a +//! loop and ticks it on a fixed interval, firing every due schedule each tick. +//! +//! The driver deliberately mirrors the shape of +//! [`ardur_cron::CronScheduler`](https://docs.rs) — `new` / `start` / `stop` / +//! `is_running` — but where that type only marks job lifecycle, this one calls +//! the real executor. Callers that want a bounded, inline run (a CLI +//! `schedule run --max-ticks`, or a test) use [`ScheduleDriver::run_bounded`]; +//! callers that want an always-on background driver (a server boot) use +//! [`ScheduleDriver::start`]. + +use std::sync::Arc; +use std::time::Duration; + +use chrono::Utc; +use tokio::sync::RwLock; +use tokio::task::JoinHandle; +use tracing::{info, warn}; + +use crate::proactive::{ + AutomationChannel, AutomationRuntime, FireReport, ProactiveAutomationLoop, ScheduleStore, +}; + +/// Drives a [`ProactiveAutomationLoop`] on a fixed tick interval, firing every +/// due schedule each tick. +pub struct ScheduleDriver { + loop_: Arc>, + tick_interval: Duration, + handle: Arc>>>, +} + +impl ScheduleDriver +where + R: AutomationRuntime, + S: ScheduleStore, + C: AutomationChannel, +{ + /// Wrap an automation loop and tick it every `tick_interval`. + #[must_use] + pub fn new(loop_: Arc>, tick_interval: Duration) -> Self { + Self { + loop_, + tick_interval, + handle: Arc::new(RwLock::new(None)), + } + } + + /// Fire every schedule due at `now` exactly once. A store-read failure is + /// logged and swallowed so one bad tick never tears the driver down; the + /// returned reports cover the schedules that were actually considered. + pub async fn tick_once(&self, now: chrono::DateTime) -> Vec { + match self.loop_.fire_due(now).await { + Ok(reports) => { + for report in &reports { + if report.delivered { + info!(schedule = %report.schedule_id.0, "fired due schedule"); + } else if let Some(err) = &report.error { + warn!(schedule = %report.schedule_id.0, error = %err, "schedule fire failed"); + } + } + reports + } + Err(err) => { + warn!(error = %err, "failed to load due schedules for this tick"); + Vec::new() + } + } + } + + /// Run the driver inline until `max_ticks` ticks have elapsed, or forever + /// when `max_ticks` is `None`. + /// + /// The first tick fires immediately (matching [`tokio::time::interval`] + /// semantics), so `run_bounded(Some(1))` is "fire everything due right now, + /// once, then return" — the shape a one-shot CLI drive and the E2E tests + /// rely on. + pub async fn run_bounded(&self, max_ticks: Option) { + let mut ticker = tokio::time::interval(self.tick_interval); + let mut fired = 0usize; + loop { + ticker.tick().await; + self.tick_once(Utc::now()).await; + fired = fired.saturating_add(1); + if max_ticks.is_some_and(|max| fired >= max) { + break; + } + } + } + + /// Spawn a background task that drives the loop forever on the tick + /// interval. Idempotent-guarded: a second `start` while already running is a + /// no-op that returns `false`. + pub async fn start(&self) -> bool { + let mut handle = self.handle.write().await; + if handle.is_some() { + return false; + } + let loop_ = Arc::clone(&self.loop_); + let interval = self.tick_interval; + let h = tokio::spawn(async move { + let mut ticker = tokio::time::interval(interval); + loop { + ticker.tick().await; + let now = Utc::now(); + match loop_.fire_due(now).await { + Ok(reports) => { + for report in &reports { + if report.delivered { + info!(schedule = %report.schedule_id.0, "fired due schedule"); + } else if let Some(err) = &report.error { + warn!(schedule = %report.schedule_id.0, error = %err, "schedule fire failed"); + } + } + } + Err(err) => { + warn!(error = %err, "failed to load due schedules for this tick"); + } + } + } + }); + *handle = Some(h); + info!("automation schedule driver started"); + true + } + + /// Abort the background driver. Returns `false` when it was not running. + pub async fn stop(&self) -> bool { + let mut handle = self.handle.write().await; + if let Some(h) = handle.take() { + h.abort(); + info!("automation schedule driver stopped"); + true + } else { + false + } + } + + /// Whether a background driver spawned by [`start`](Self::start) is live. + pub async fn is_running(&self) -> bool { + self.handle.read().await.is_some() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::proactive::{ + AutomationAttenuation, AutomationDeliveryEvent, AutomationSchedule, InMemoryScheduleStore, + ProactiveAutomationError, ScheduledCapToken, + }; + + use ardur_cost_gate::CostTuple as GateCostTuple; + use ardur_cron::CronExpression; + use ardur_fused_runtime::PerRequestProvisioning; + use ardur_runtime::{ + CapTokenRef, ChatMessage, CostTuple as RuntimeCostTuple, ReceiptId, RuntimeError, + SessionId, SubmitRequest, SubmitResult, + }; + use async_trait::async_trait; + use tokio::sync::Mutex; + + #[derive(Debug, Default)] + struct RecordingRuntime { + submits: Mutex, + } + + #[async_trait] + impl AutomationRuntime for RecordingRuntime { + async fn submit( + &self, + _req: SubmitRequest, + _provisioning: PerRequestProvisioning, + ) -> Result { + *self.submits.lock().await += 1; + Ok(SubmitResult { + receipt_id: ReceiptId::new(), + response: ChatMessage::assistant("driven"), + cost: RuntimeCostTuple::default(), + }) + } + } + + #[derive(Debug, Default)] + struct RecordingChannel { + delivered: Mutex, + } + + #[async_trait] + impl AutomationChannel for RecordingChannel { + async fn deliver( + &self, + _event: AutomationDeliveryEvent, + ) -> Result<(), ProactiveAutomationError> { + *self.delivered.lock().await += 1; + Ok(()) + } + } + + fn token() -> ScheduledCapToken { + ScheduledCapToken::attenuated( + CapTokenRef("attenuated".to_string()), + vec![AutomationAttenuation { + rule: "restrict_tools:chat.submit".to_string(), + evidence: None, + }], + ) + } + + fn budget() -> GateCostTuple { + GateCostTuple { + tokens_in: 100, + tokens_out: 100, + cents: 10, + wall_ms: 1_000, + attention_score: 1, + } + } + + /// The timer bridge actually executes a persisted due schedule: one bounded + /// tick submits through the runtime, delivers to the channel, and bumps the + /// store's fire count. This is the regression guard for #347 — before the + /// driver existed, no timer ever reached `fire_due`. + #[tokio::test] + async fn bounded_tick_fires_due_schedule() { + let runtime = Arc::new(RecordingRuntime::default()); + let store = Arc::new(InMemoryScheduleStore::new()); + let channel = Arc::new(RecordingChannel::default()); + let loop_ = Arc::new(ProactiveAutomationLoop::new( + runtime.clone(), + store.clone(), + channel.clone(), + )); + + let schedule = AutomationSchedule::new( + "every-minute", + CronExpression::every_minute(), + SessionId::new(), + token(), + budget(), + "status update", + ); + let id = schedule.id.clone(); + loop_.upsert_schedule(schedule).await.expect("upsert"); + + let driver = ScheduleDriver::new(loop_, Duration::from_millis(10)); + driver.run_bounded(Some(1)).await; + + assert_eq!( + *runtime.submits.lock().await, + 1, + "the timer drove one submit" + ); + assert_eq!(*channel.delivered.lock().await, 1, "the fire was delivered"); + let stored = store.load_all().await.expect("load"); + let fired = stored + .iter() + .find(|s| s.id == id) + .expect("schedule still present"); + assert_eq!(fired.fire_count, 1, "the successful fire was persisted"); + } + + /// A background driver started and then stopped runs at least one real tick + /// against the executor — proving `start`/`stop` drive execution, not just + /// lifecycle bookkeeping. + #[tokio::test] + async fn background_driver_start_stop_fires() { + let runtime = Arc::new(RecordingRuntime::default()); + let store = Arc::new(InMemoryScheduleStore::new()); + let channel = Arc::new(RecordingChannel::default()); + let loop_ = Arc::new(ProactiveAutomationLoop::new( + runtime.clone(), + store.clone(), + channel.clone(), + )); + let schedule = AutomationSchedule::new( + "every-minute", + CronExpression::every_minute(), + SessionId::new(), + token(), + budget(), + "status update", + ); + loop_.upsert_schedule(schedule).await.expect("upsert"); + + let driver = ScheduleDriver::new(loop_, Duration::from_millis(10)); + assert!(driver.start().await, "first start spawns the driver"); + assert!( + !driver.start().await, + "second start is a no-op while running" + ); + assert!(driver.is_running().await); + + // Give the background ticker time for at least one tick (first tick is + // immediate, but yield generously to avoid a slow-CI race). + tokio::time::sleep(Duration::from_millis(60)).await; + assert!(driver.stop().await, "stop aborts the running driver"); + assert!(!driver.is_running().await); + + assert!( + *runtime.submits.lock().await >= 1, + "the background driver fired the due schedule at least once" + ); + } +} diff --git a/crates/automation/src/lib.rs b/crates/automation/src/lib.rs index c936c7f2..9f0dcd3d 100644 --- a/crates/automation/src/lib.rs +++ b/crates/automation/src/lib.rs @@ -22,12 +22,14 @@ #![forbid(unsafe_code)] #![warn(missing_docs)] +pub mod driver; mod error; pub mod learning; pub mod proactive; mod task_record; pub mod tasks; +pub use driver::ScheduleDriver; pub use error::TaskFlowError; pub use proactive::{ AutomationAttenuation, AutomationChannel, AutomationDeliveryEvent, AutomationRuntime, diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index 878bacc9..128af151 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -71,6 +71,7 @@ syntect = { workspace = true } crossterm = { workspace = true } anyhow = { workspace = true } +async-trait = { workspace = true } thiserror = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/cli/src/lib.rs b/crates/cli/src/lib.rs index d8bce625..97e21102 100644 --- a/crates/cli/src/lib.rs +++ b/crates/cli/src/lib.rs @@ -40,6 +40,7 @@ mod error; mod fused; mod links; mod markdown; +mod schedule_exec; mod secure_io; mod slash; mod state; @@ -70,6 +71,10 @@ pub use error::CliError; pub use fused::FusedEngine; pub use links::{osc8_from_env, terminal_supports_osc8}; pub use markdown::{render_markdown, render_markdown_with}; +pub use schedule_exec::{ + DEFAULT_DRIVER_INTERVAL_SECS, ScheduleRecord, read_schedule_records, run_schedule_fire, + run_schedule_run, +}; pub use secure_io::{ create_private_file_no_follow, directory_modified_no_follow, list_directory_names_no_follow, read_file_no_follow, read_string_no_follow, remove_directory_tree_no_follow, diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 71b0ad0c..52d45ea0 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -23,9 +23,10 @@ use std::path::{Path, PathBuf}; use std::process::ExitCode; use ardur_cli::{ - ChatArgs, CliError, Config, SessionMetadata, StateDirs, directory_modified_no_follow, - list_directory_names_no_follow, read_string_no_follow, remove_directory_tree_no_follow, - run_chat, write_private_file_atomic_no_follow, write_private_file_no_follow, + ChatArgs, CliError, Config, DEFAULT_DRIVER_INTERVAL_SECS, ScheduleRecord, SessionMetadata, + StateDirs, directory_modified_no_follow, list_directory_names_no_follow, read_schedule_records, + read_string_no_follow, remove_directory_tree_no_follow, run_chat, run_schedule_fire, + run_schedule_run, write_private_file_atomic_no_follow, write_private_file_no_follow, }; use ardur_session_journals::{ JournalEntry, default_secret_patterns, redact_entries_default, redact_text, @@ -2193,22 +2194,20 @@ enum ScheduleAction { /// Schedule ID. id: String, }, - /// Test fire a schedule now (dry-run). + /// Fire a schedule now, executing it end-to-end through the pipeline. Fire { /// Schedule ID. id: String, }, -} - -/// Parsed schedule record stored in the state directory. -#[derive(serde::Serialize, serde::Deserialize)] -struct ScheduleRecord { - schedule_id: String, - label: String, - pattern: String, - prompt: String, - created_at: u64, - enabled: bool, + /// Drive every due schedule on an interval, executing each end-to-end. + Run { + /// Seconds between ticks. + #[arg(long, default_value_t = DEFAULT_DRIVER_INTERVAL_SECS)] + interval_secs: u64, + /// Stop after this many ticks (omit to run until interrupted). + #[arg(long)] + max_ticks: Option, + }, } /// Simple cron-like parser for the most common NL patterns. @@ -2269,24 +2268,6 @@ fn parse_time_to_cron(time_str: &str) -> Option { Some(format!("{minute} {hour} * * *")) } -/// Read schedule records from the state directory. -fn read_schedules(root: &Path) -> Result, CliError> { - let dir = root.join("schedules"); - let mut records = Vec::new(); - if let Ok(entries) = std::fs::read_dir(&dir) { - for entry in entries.flatten() { - if entry.path().extension().is_some_and(|e| e == "json") { - if let Ok(content) = read_string_no_follow(&entry.path()) { - if let Ok(v) = serde_json::from_str::(&content) { - records.push(v); - } - } - } - } - } - Ok(records) -} - /// Parse a 5-field cron string into a CronExpression. /// Supported: `*` (any), `n` (exact), `n-m` (range), `a,b` (list), `*/n` (step), /// `a-b/n` (range step), and `JAN`/`MON` names — validated up front so an @@ -2332,7 +2313,8 @@ fn next_fire_times( /// Run `ardur schedule` subcommands. fn run_schedule(args: ScheduleArgs) -> Result<(), CliError> { - let root = StateDirs::resolve()?.root; + let dirs = StateDirs::resolve()?; + let root = dirs.root.clone(); let schedules_dir = root.join("schedules"); std::fs::create_dir_all(&schedules_dir)?; @@ -2358,6 +2340,8 @@ fn run_schedule(args: ScheduleArgs) -> Result<(), CliError> { .map(|d| d.as_secs()) .unwrap_or(0), enabled: true, + last_fire_at: None, + fire_count: 0, }; write_private_file_atomic_no_follow( &schedules_dir.join(format!("{id}.json")), @@ -2372,7 +2356,7 @@ fn run_schedule(args: ScheduleArgs) -> Result<(), CliError> { } } ScheduleAction::List => { - let records = read_schedules(&root)?; + let records = read_schedule_records(&schedules_dir); if records.is_empty() { println!("no schedules"); } else { @@ -2395,7 +2379,7 @@ fn run_schedule(args: ScheduleArgs) -> Result<(), CliError> { } } ScheduleAction::Next { id, count } => { - let records = read_schedules(&root)?; + let records = read_schedule_records(&schedules_dir); let found = records.iter().find(|r| r.schedule_id == id); match found { Some(r) => { @@ -2420,19 +2404,15 @@ fn run_schedule(args: ScheduleArgs) -> Result<(), CliError> { println!("deleted schedule {id}"); } ScheduleAction::Fire { id } => { - let records = read_schedules(&root)?; - let found = records.iter().find(|r| r.schedule_id == id); - match found { - Some(r) => { - println!("dry-run fire schedule {id}"); - println!(" prompt: {}", r.prompt); - println!(" pattern: {}", r.pattern); - println!(" note: execution engine not yet wired"); - } - None => { - return Err(CliError::State(format!("schedule `{id}` not found"))); - } - } + let config = Config::load(None)?; + run_schedule_fire(&dirs, &config, &id)?; + } + ScheduleAction::Run { + interval_secs, + max_ticks, + } => { + let config = Config::load(None)?; + run_schedule_run(&dirs, &config, interval_secs, max_ticks)?; } } Ok(()) diff --git a/crates/cli/src/schedule_exec.rs b/crates/cli/src/schedule_exec.rs new file mode 100644 index 00000000..319bef94 --- /dev/null +++ b/crates/cli/src/schedule_exec.rs @@ -0,0 +1,661 @@ +//! Executes persisted `ardur schedule` jobs through the real automation +//! executor (issue #347). +//! +//! Before this module, `ardur schedule create` persisted a [`ScheduleRecord`] +//! to `/schedules/.json` and nothing ever ran it: `ardur schedule +//! fire` printed `execution engine not yet wired`, and no timer drove the +//! `ardur-automation` executor (which was fully built but had zero callers). +//! Three disjoint schedule stores existed and none was connected to execution. +//! +//! This module reconciles that. The `/schedules` directory is the single +//! durable store; [`CliScheduleStore`] adapts it to the automation crate's +//! [`ScheduleStore`] contract by *materializing* each persisted record into an +//! [`AutomationSchedule`] at fire time — minting a fresh, short-lived, +//! **attenuated** cap-token and a per-fire budget top-up for every unattended +//! fire (a token minted at *create* time would have expired by the time a daily +//! job fires days later, so per-fire minting is the only correct design for a +//! durable scheduler). Fires then run through the ordinary +//! [`FusedRuntime`](ardur_fused_runtime::FusedRuntime) ten-stage pipeline — +//! cap-token verify → Cedar → cost admission → provider → signed receipt → +//! journal — exactly like an `ardur chat` turn, and successful responses are +//! delivered to a caller-supplied [`AutomationChannel`]. +//! +//! Two entry points drive it: +//! - [`run_schedule_fire`] — fire one persisted schedule now, end-to-end +//! (backs `ardur schedule fire `). +//! - [`run_schedule_run`] — drive every *due* schedule on an interval via +//! [`ScheduleDriver`] (backs `ardur schedule run`), bounded by `--max-ticks` +//! or unbounded. +//! +//! The provider/runtime wiring here mirrors [`crate::FusedEngine`]'s builder: +//! the same provider selection (with an offline stub fallback when credentials +//! are absent), the same persistent issuer/receipt keys and Cedar policies, and +//! the same file-backed receipt log — so a fired schedule appends to the very +//! same signed receipt chain a chat turn would. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use ardur_automation::{ + AutomationAttenuation, AutomationChannel, AutomationDeliveryEvent, AutomationSchedule, + AutomationScheduleId, AutomationScheduleStatus, FireReport, FusedAutomationRuntime, + ProactiveAutomationError, ProactiveAutomationLoop, ScheduleDriver, ScheduleStore, + ScheduledCapToken, +}; +use ardur_cap_token::{BiscuitCapTokenIssuer, CapScope, CapTokenIssuer, HolderId as CapHolderId}; +use ardur_cost_gate::{CostEnvelope, CostTuple as GateCostTuple, HolderId as GateHolderId}; +use ardur_cron::CronExpression; +use ardur_fused_runtime::{FusedRuntime, FusedRuntimeBuilder}; +use ardur_memory::InMemoryMemoryRuntime; +use ardur_provider_runtime::{ + AnthropicProvider, InstrumentedProvider, ModelId, Provider, ProviderError, +}; +use ardur_provider_selector as provider_selector; +use ardur_runtime::{CapTokenRef, SessionId}; +use ardur_session_journals::FileSessionJournal; +use async_trait::async_trait; +use chrono::{DateTime, TimeZone, Utc}; +use serde::{Deserialize, Serialize}; + +use crate::config::Config; +use crate::error::CliError; +use crate::secure_io::{read_string_no_follow, write_private_file_atomic_no_follow}; +use crate::state::StateDirs; + +/// The audience unattended-fire cap-tokens are scoped to — the same audience +/// [`crate::FusedEngine`] verifies chat turns against, so the scheduler runtime +/// accepts them. +const SCHEDULE_AUDIENCE: &str = "cli"; +/// The tool/capability every scheduled turn exercises. +const SCHEDULE_TOOL: &str = "chat.submit"; +/// A fixed scheduler-process journal session id (a stable v5-ish constant), so +/// all fires from one process share one journal file. Individual fires still +/// carry their own per-schedule `session_id` in the receipt/request. +const SCHEDULER_JOURNAL_SESSION_UUID: u128 = 0x5c4ed000_0000_4000_8000_000000000001; +/// A per-fire attenuated cap-token lives only long enough to run one turn. +const FIRE_CAP_TTL_SECS: u64 = 300; +/// Per-fire cents ceiling admitted onto the schedule subject before a fire. The +/// projected envelope gates only the cents axis, so this bounds a single fire's +/// spend. +const DEFAULT_PER_FIRE_CENTS: u64 = 100; +/// Default driver tick interval (seconds) for `ardur schedule run`. +pub const DEFAULT_DRIVER_INTERVAL_SECS: u64 = 60; + +/// A persisted schedule record under `/schedules/.json`. +/// +/// This is the single durable schedule shape the CLI reads and writes; +/// `ardur schedule create` writes it, and [`CliScheduleStore`] materializes it +/// into an executable [`AutomationSchedule`]. The `last_fire_at` / `fire_count` +/// fields are additive (serde-defaulted) so records written before #347 load +/// unchanged. +#[derive(Clone, Serialize, Deserialize)] +pub struct ScheduleRecord { + /// Stable schedule id (a UUID string; also the file stem). + pub schedule_id: String, + /// Human-readable label. + pub label: String, + /// Five-field cron pattern the schedule fires on. + pub pattern: String, + /// Prompt submitted as the user turn on each fire. + pub prompt: String, + /// Creation time (unix seconds). + pub created_at: u64, + /// Whether the schedule is eligible to fire. + pub enabled: bool, + /// Last successful fire time, persisted so a driver does not double-fire + /// within one cron minute across ticks/restarts. + #[serde(default)] + pub last_fire_at: Option>, + /// Number of successful fires. + #[serde(default)] + pub fire_count: u64, +} + +impl ScheduleRecord { + /// The on-disk path for this record under `dir`. + fn path_in(dir: &Path, id: &str) -> PathBuf { + dir.join(format!("{id}.json")) + } +} + +/// Whether `id` is a safe on-disk file stem (no path traversal). Schedule ids +/// are UUIDs, so this rejects anything that is not a bare +/// `[A-Za-z0-9._-]` token — the same guarantee `sanitize_state_id` gives the +/// other state commands. +fn is_safe_id(id: &str) -> bool { + !id.is_empty() + && id.len() <= 128 + && id + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-')) + && id != "." + && id != ".." +} + +/// Read every persisted schedule record from `dir`, skipping unreadable or +/// malformed files. +pub fn read_schedule_records(dir: &Path) -> Vec { + let mut records = Vec::new(); + if let Ok(entries) = std::fs::read_dir(dir) { + for entry in entries.flatten() { + let path = entry.path(); + if path.extension().is_some_and(|e| e == "json") { + if let Ok(content) = read_string_no_follow(&path) { + if let Ok(record) = serde_json::from_str::(&content) { + records.push(record); + } + } + } + } + } + records +} + +fn write_schedule_record(dir: &Path, record: &ScheduleRecord) -> Result<(), std::io::Error> { + std::fs::create_dir_all(dir)?; + let path = ScheduleRecord::path_in(dir, &record.schedule_id); + let bytes = serde_json::to_vec_pretty(record) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + write_private_file_atomic_no_follow(&path, &bytes) +} + +/// Parse a five-field cron pattern string into a validated [`CronExpression`]. +fn pattern_to_expression(pattern: &str) -> Result { + let fields: Vec<&str> = pattern.split_whitespace().collect(); + if fields.len() != 5 { + return Err(ProactiveAutomationError::InvalidSchedule(format!( + "cron pattern must have 5 fields, got {}", + fields.len() + ))); + } + let expr = CronExpression::new(fields[0], fields[1], fields[2], fields[3], fields[4]); + expr.validate().map_err(|e| { + ProactiveAutomationError::InvalidSchedule(format!("invalid cron `{pattern}`: {e}")) + })?; + Ok(expr) +} + +/// Serialize a [`CronExpression`] back to its five-field pattern string. +fn expression_to_pattern(expr: &CronExpression) -> String { + format!( + "{} {} {} {} {}", + expr.minute, expr.hour, expr.day_of_month, expr.month, expr.day_of_week + ) +} + +/// A [`ScheduleStore`] backed by the CLI's `/schedules/*.json` records. +/// +/// `load_all` materializes each record into an executable [`AutomationSchedule`] +/// — minting a fresh attenuated cap-token and per-fire budget for every fire — +/// so the automation executor can run persisted jobs. `record_successful_fire` +/// persists the updated `last_fire_at`/`fire_count` back to the record file. +pub struct CliScheduleStore { + dir: PathBuf, + issuer: Arc, + subject: String, + per_fire_cents: u64, +} + +impl CliScheduleStore { + /// Build a store over `dir`, minting per-fire cap-tokens with `issuer` for + /// `subject`. + #[must_use] + pub fn new( + dir: PathBuf, + issuer: Arc, + subject: String, + per_fire_cents: u64, + ) -> Self { + Self { + dir, + issuer, + subject, + per_fire_cents, + } + } + + /// Mint a fresh, short-lived, attenuated cap-token for one unattended fire. + fn mint_fire_token(&self) -> Result { + let now_unix = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let cap = self + .issuer + .issue( + CapHolderId(self.subject.clone()), + CapScope { + audience: SCHEDULE_AUDIENCE.to_string(), + expires_unix: now_unix + FIRE_CAP_TTL_SECS, + budget_remaining: self.per_fire_cents.max(1), + tool_allowlist: vec![ + SCHEDULE_TOOL.to_string(), + ardur_memory::MEMORY_READ_CAPABILITY.to_string(), + ardur_memory::MEMORY_WRITE_CAPABILITY.to_string(), + ], + }, + ) + .map_err(|e| { + ProactiveAutomationError::InvalidSchedule(format!( + "minting unattended-fire cap-token: {e}" + )) + })?; + let token = CapTokenRef(cap.to_base64().map_err(|e| { + ProactiveAutomationError::InvalidSchedule(format!( + "serializing unattended-fire cap-token: {e}" + )) + })?); + Ok(ScheduledCapToken::attenuated( + token, + vec![AutomationAttenuation { + rule: format!("restrict_tools:{SCHEDULE_TOOL}"), + evidence: Some("ardur schedule unattended fire".to_string()), + }], + )) + } + + fn per_fire_budget(&self) -> GateCostTuple { + GateCostTuple { + tokens_in: 1_000_000, + tokens_out: 1_000_000, + cents: self.per_fire_cents, + wall_ms: 1_000_000, + attention_score: 1_000_000, + } + } + + /// Materialize one persisted record into an executable schedule. + fn materialize( + &self, + record: &ScheduleRecord, + ) -> Result { + let expression = pattern_to_expression(&record.pattern)?; + // A record id is a UUID; reuse it as the fire's session id so receipts + // and journals attribute the fire to a stable session. Fall back to a + // fresh session id if a legacy record carried a non-UUID id. + let session_id = uuid::Uuid::parse_str(&record.schedule_id) + .map(SessionId) + .unwrap_or_default(); + let mut schedule = AutomationSchedule::new( + record.label.clone(), + expression, + session_id, + self.mint_fire_token()?, + self.per_fire_budget(), + record.prompt.clone(), + ); + schedule.id = AutomationScheduleId(record.schedule_id.clone()); + schedule.status = if record.enabled { + AutomationScheduleStatus::Enabled + } else { + AutomationScheduleStatus::Paused + }; + schedule.created_at = Utc + .timestamp_opt(record.created_at as i64, 0) + .single() + .unwrap_or_else(Utc::now); + schedule.last_fire_at = record.last_fire_at; + schedule.fire_count = record.fire_count; + Ok(schedule) + } + + /// The record shape for an [`AutomationSchedule`] (the lossy inverse of + /// [`materialize`](Self::materialize): the cap-token/budget are re-minted on + /// load, so only the durable fields round-trip). + fn to_record(schedule: &AutomationSchedule) -> ScheduleRecord { + ScheduleRecord { + schedule_id: schedule.id.0.clone(), + label: schedule.name.clone(), + pattern: expression_to_pattern(&schedule.expression), + prompt: schedule.prompt.clone(), + created_at: u64::try_from(schedule.created_at.timestamp().max(0)).unwrap_or(0), + enabled: schedule.status == AutomationScheduleStatus::Enabled, + last_fire_at: schedule.last_fire_at, + fire_count: schedule.fire_count, + } + } +} + +#[async_trait] +impl ScheduleStore for CliScheduleStore { + async fn load_all(&self) -> Result, ProactiveAutomationError> { + let mut schedules = Vec::new(); + for record in read_schedule_records(&self.dir) { + match self.materialize(&record) { + Ok(schedule) => schedules.push(schedule), + Err(err) => { + tracing::warn!( + schedule = %record.schedule_id, + error = %err, + "skipping unmaterializable schedule record" + ); + } + } + } + Ok(schedules) + } + + async fn upsert(&self, schedule: AutomationSchedule) -> Result<(), ProactiveAutomationError> { + write_schedule_record(&self.dir, &Self::to_record(&schedule))?; + Ok(()) + } + + async fn remove(&self, id: &AutomationScheduleId) -> Result<(), ProactiveAutomationError> { + if !is_safe_id(&id.0) { + return Err(ProactiveAutomationError::ScheduleNotFound(id.0.clone())); + } + let path = ScheduleRecord::path_in(&self.dir, &id.0); + match std::fs::remove_file(&path) { + Ok(()) => Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + Err(ProactiveAutomationError::ScheduleNotFound(id.0.clone())) + } + Err(err) => Err(err.into()), + } + } + + async fn record_successful_fire( + &self, + id: &AutomationScheduleId, + fired_at: DateTime, + ) -> Result { + if !is_safe_id(&id.0) { + return Err(ProactiveAutomationError::ScheduleNotFound(id.0.clone())); + } + let path = ScheduleRecord::path_in(&self.dir, &id.0); + let raw = match read_string_no_follow(&path) { + Ok(raw) => raw, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + return Err(ProactiveAutomationError::ScheduleNotFound(id.0.clone())); + } + Err(err) => return Err(err.into()), + }; + let mut record: ScheduleRecord = serde_json::from_str(&raw)?; + record.last_fire_at = Some(fired_at); + record.fire_count = record.fire_count.saturating_add(1); + write_schedule_record(&self.dir, &record)?; + self.materialize(&record) + } +} + +/// An [`AutomationChannel`] that prints a delivered fire's response to stdout — +/// the CLI's "delivery" for an unattended fire the operator is watching. +struct StdoutAutomationChannel; + +#[async_trait] +impl AutomationChannel for StdoutAutomationChannel { + async fn deliver( + &self, + event: AutomationDeliveryEvent, + ) -> Result<(), ProactiveAutomationError> { + println!( + "fired schedule {} ({}) at {}", + event.schedule_id.0, event.schedule_name, event.fired_at + ); + println!(" receipt: {}", event.result.receipt_id.0); + println!(" response: {}", event.result.response.content); + Ok(()) + } +} + +/// The fused runtime plus the issuer/subject a fired schedule needs. +struct SchedulerRuntime { + runtime: Arc, + issuer: Arc, + subject: String, + /// Whether the selected provider fell back to the offline stub (no creds). + offline: bool, +} + +/// Build the fused runtime the scheduler fires through. Mirrors +/// [`crate::FusedEngine`]'s builder: provider selection with an offline stub +/// fallback, persistent issuer/receipt keys and Cedar policies, a file-backed +/// receipt log, and a generously provisioned subject budget (the per-fire +/// top-up is layered on at submit time by each schedule's provisioning). +async fn build_scheduler_runtime( + config: &Config, + dirs: &StateDirs, + budget_cents: u64, +) -> Result { + let model = ModelId::new(&config.model); + + let (provider, offline): (Arc, bool) = + match provider_selector::from_env(model.clone()) { + Ok(live) => (live, false), + Err(e @ ProviderError::InvalidSelection(_)) => return Err(CliError::Provider(e)), + Err(_) => { + let stub: Arc = Arc::new(AnthropicProvider::stub(model.clone())); + tracing::info!( + offline = true, + "selected provider unavailable; scheduler using offline stub" + ); + (stub, true) + } + }; + let provider = InstrumentedProvider::wrap(provider); + + let issuer = Arc::new(dirs.load_or_create_issuer()?); + let cap_root = issuer.public_key(); + let receipt_key = dirs.load_or_create_receipt_key()?; + let policies = dirs.load_cedar_policies()?; + let subject = dirs.local_subject(); + let holder = GateHolderId(subject.clone()); + + // Gate only the cents axis per fire, like the chat path, so a cents-scoped + // per-fire top-up covers one turn. + let envelope = CostEnvelope { + tokens_in_max: 0, + tokens_out_max: 0, + cents_max: u32::try_from(DEFAULT_PER_FIRE_CENTS).unwrap_or(u32::MAX), + wall_ms_max: 0, + attention_score_max: 0, + }; + + let journal_session = SessionId(uuid::Uuid::from_u128(SCHEDULER_JOURNAL_SESSION_UUID)); + let journal = FileSessionJournal::new(&dirs.journals, journal_session) + .map_err(|e| CliError::State(format!("opening the scheduler journal: {e}")))?; + let memory = Arc::new(InMemoryMemoryRuntime::new()); + + let (runtime, _reconciliation) = + FusedRuntimeBuilder::new(cap_root, policies, provider, receipt_key, model) + .audience(SCHEDULE_AUDIENCE) + .tool(SCHEDULE_TOOL) + .provision_budget( + holder, + GateCostTuple { + tokens_in: 1_000_000_000, + tokens_out: 1_000_000_000, + cents: budget_cents.max(DEFAULT_PER_FIRE_CENTS), + wall_ms: 1_000_000_000, + attention_score: 1_000_000_000, + }, + ) + .projected_envelope(envelope) + .with_memory(memory) + .with_journal(Arc::new(journal)) + .with_default_injection_filters() + .receipt_log(dirs.receipt_log()) + .build_reconciled() + .await + .map_err(|e| CliError::State(format!("building the scheduler runtime: {e}")))?; + + Ok(SchedulerRuntime { + runtime: Arc::new(runtime), + issuer, + subject, + offline, + }) +} + +/// Assemble the automation loop over the CLI schedule store, the scheduler +/// runtime, and the stdout delivery channel. +type CliLoop = + ProactiveAutomationLoop; + +fn build_loop(scheduler: &SchedulerRuntime, dirs: &StateDirs) -> Arc { + let store = CliScheduleStore::new( + dirs.root.join("schedules"), + Arc::clone(&scheduler.issuer), + scheduler.subject.clone(), + DEFAULT_PER_FIRE_CENTS, + ); + let runtime = FusedAutomationRuntime::new(Arc::clone(&scheduler.runtime)); + Arc::new(ProactiveAutomationLoop::new( + Arc::new(runtime), + Arc::new(store), + Arc::new(StdoutAutomationChannel), + )) +} + +/// Ensure the state directories a fire touches (keys, journals, receipts, +/// schedules) exist. Deliberately does *not* run the state-tree schema +/// migration [`StateDirs::create`] performs: `ardur schedule create` never +/// stamps the schema either, and the fire path must not fail when it is the +/// first `ardur` command to touch the state tree (e.g. an unattended driver +/// booting before any interactive session). +fn prepare_dirs(dirs: &StateDirs) -> Result<(), CliError> { + let schedules = dirs.root.join("schedules"); + for dir in [&dirs.keys, &dirs.journals, &dirs.receipts, &schedules] { + std::fs::create_dir_all(dir)?; + } + Ok(()) +} + +/// Fire one persisted schedule now, end-to-end. Backs `ardur schedule fire`. +/// +/// # Errors +/// If the id is unknown, the state tree cannot be prepared, the runtime cannot +/// be built, or the fire itself fails (a delivered-but-errored fire is reported +/// as an error so the operator sees non-zero exit and the reason). +pub fn run_schedule_fire(dirs: &StateDirs, config: &Config, id: &str) -> Result<(), CliError> { + install_stderr_tracing(); + if !is_safe_id(id) { + return Err(CliError::State(format!("invalid schedule id `{id}`"))); + } + prepare_dirs(dirs)?; + let schedules_dir = dirs.root.join("schedules"); + if !ScheduleRecord::path_in(&schedules_dir, id).is_file() { + return Err(CliError::State(format!("schedule `{id}` not found"))); + } + + let tokio_rt = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()?; + tokio_rt.block_on(async move { + let scheduler = build_scheduler_runtime(config, dirs, config.budget_cents).await?; + if scheduler.offline { + eprintln!( + "note: no live provider credentials found; firing against the offline stub provider" + ); + } + let loop_ = build_loop(&scheduler, dirs); + let report = loop_ + .fire_now(&AutomationScheduleId(id.to_string())) + .await + .map_err(|e| CliError::State(format!("firing schedule `{id}`: {e}")))?; + report_or_err(id, &report) + }) +} + +/// Drive every *due* schedule on an interval. Backs `ardur schedule run`. +/// +/// `max_ticks` bounds the number of ticks (`Some(1)` fires everything due right +/// now once and returns); `None` runs until the process is interrupted. The +/// first tick fires immediately. +/// +/// # Errors +/// If the state tree cannot be prepared or the runtime cannot be built. A fire +/// that fails mid-run is logged (per schedule) but does not abort the driver. +pub fn run_schedule_run( + dirs: &StateDirs, + config: &Config, + interval_secs: u64, + max_ticks: Option, +) -> Result<(), CliError> { + install_stderr_tracing(); + prepare_dirs(dirs)?; + + let tokio_rt = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()?; + tokio_rt.block_on(async move { + let scheduler = build_scheduler_runtime(config, dirs, config.budget_cents).await?; + if scheduler.offline { + eprintln!( + "note: no live provider credentials found; driving schedules against the offline stub provider" + ); + } + let loop_ = build_loop(&scheduler, dirs); + let interval = Duration::from_secs(interval_secs.max(1)); + let driver = ScheduleDriver::new(loop_, interval); + match max_ticks { + Some(ticks) => println!( + "driving due schedules every {interval_secs}s for {ticks} tick(s)" + ), + None => println!( + "driving due schedules every {interval_secs}s (Ctrl-C to stop)" + ), + } + driver.run_bounded(max_ticks).await; + Ok::<(), CliError>(()) + }) +} + +/// Turn a [`FireReport`] into a printed success or a [`CliError`]. +fn report_or_err(id: &str, report: &FireReport) -> Result<(), CliError> { + if report.delivered { + Ok(()) + } else { + Err(CliError::State(format!( + "schedule `{id}` did not fire: {}", + report.error.as_deref().unwrap_or("unknown error") + ))) + } +} + +/// Install a stderr tracing subscriber if none is set (a no-op otherwise), so +/// scheduler warnings surface without clobbering an already-installed one. +fn install_stderr_tracing() { + let _ = tracing_subscriber::fmt() + .with_writer(std::io::stderr) + .try_init(); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn is_safe_id_rejects_traversal() { + assert!(is_safe_id("59d683b9-0000-4000-8000-000000000000")); + assert!(!is_safe_id("../etc/passwd")); + assert!(!is_safe_id("a/b")); + assert!(!is_safe_id("..")); + assert!(!is_safe_id("")); + } + + #[test] + fn pattern_round_trips_through_expression() { + let expr = pattern_to_expression("0,5 * * * *").expect("valid pattern"); + assert_eq!(expression_to_pattern(&expr), "0,5 * * * *"); + assert!(pattern_to_expression("bad").is_err()); + assert!(pattern_to_expression("99 * * * *").is_err()); + } + + #[test] + fn record_fire_fields_default_on_legacy_json() { + // A pre-#347 record (no last_fire_at / fire_count) still deserializes. + let legacy = r#"{ + "schedule_id": "id-1", + "label": "daily", + "pattern": "0 9 * * *", + "prompt": "post standup", + "created_at": 1750000000, + "enabled": true + }"#; + let record: ScheduleRecord = serde_json::from_str(legacy).expect("legacy record loads"); + assert_eq!(record.fire_count, 0); + assert!(record.last_fire_at.is_none()); + } +} diff --git a/crates/cli/tests/schedule_commands.rs b/crates/cli/tests/schedule_commands.rs index 259a69ba..31848087 100644 --- a/crates/cli/tests/schedule_commands.rs +++ b/crates/cli/tests/schedule_commands.rs @@ -62,13 +62,40 @@ fn schedule_create_list_next_delete_lifecycle() { let next_stdout = String::from_utf8(next).expect("stdout utf8"); assert!(next_stdout.contains("next fire times"), "{next_stdout}"); - // Fire dry-run. - Command::cargo_bin("ardur") + // Fire the schedule end-to-end (issue #347). With no provider credentials + // the fire runs against the offline stub; `ARDUR_DEV_PERMISSIVE_POLICY` + // supplies the permit the fused pipeline's Cedar stage requires. A signed + // receipt must land on the chain — proof the job actually executed rather + // than printing the old "execution engine not yet wired" stub. + let fire = Command::cargo_bin("ardur") .expect("the `ardur` binary builds") .env("HOME", dir.path()) + .env("ARDUR_DEV_PERMISSIVE_POLICY", "true") + .env_remove("ANTHROPIC_API_KEY") + .env_remove("OPENROUTER_API_KEY") + .env_remove("OPENAI_API_KEY") .args(["schedule", "fire", &id]) .assert() - .success(); + .success() + .get_output() + .stdout + .clone(); + let fire_stdout = String::from_utf8(fire).expect("stdout utf8"); + assert!( + fire_stdout.contains("fired schedule") && fire_stdout.contains("receipt:"), + "{fire_stdout}" + ); + let chain = dir + .path() + .join(".ardur") + .join("receipts") + .join("chain.jsonl"); + let chain_lines = std::fs::read_to_string(&chain) + .expect("the receipt chain exists after a fire") + .lines() + .filter(|l| !l.is_empty()) + .count(); + assert_eq!(chain_lines, 1, "the fire minted exactly one receipt"); // Delete. Command::cargo_bin("ardur") diff --git a/crates/e2e-tests/Cargo.toml b/crates/e2e-tests/Cargo.toml index bf83dadc..ea1287bb 100644 --- a/crates/e2e-tests/Cargo.toml +++ b/crates/e2e-tests/Cargo.toml @@ -31,6 +31,7 @@ ardur-cedar-policy = { workspace = true } # (it is no other crate's library dep), so it is pulled in by an explicit path. ardur-cli = { path = "../cli" } ardur-cost-gate = { workspace = true } +ardur-cron = { workspace = true } ardur-fused-runtime = { workspace = true } ardur-lifecycle-hooks = { workspace = true } ardur-memory = { workspace = true } diff --git a/crates/e2e-tests/tests/scenario_scheduled_job_execution.rs b/crates/e2e-tests/tests/scenario_scheduled_job_execution.rs new file mode 100644 index 00000000..df6dd694 --- /dev/null +++ b/crates/e2e-tests/tests/scenario_scheduled_job_execution.rs @@ -0,0 +1,174 @@ +//! Scenario §2.E — `scheduled_job_execution` (regression guard for issue #347). +//! +//! The unattended-execution path used to be inert: a schedule could be +//! persisted, but no timer ever drove the `ardur-automation` executor, so a due +//! job never ran and never minted a receipt. This scenario proves the bridge is +//! now real end-to-end: +//! +//! 1. A persisted [`AutomationSchedule`] sits in a store, due to fire. +//! 2. A [`ScheduleDriver`] — the timer→executor bridge added for #347 — ticks +//! once and fires it through the **real** [`FusedRuntime`] pipeline +//! (cap-token verify → Cedar → cost admission → provider → signed receipt). +//! 3. The fire is delivered to the channel, the store's fire-count is bumped, +//! and a signed receipt lands on the on-disk chain and verifies under the +//! publishing JWKS. +//! +//! If any of the three disjoint halves regresses back to "persist but never +//! run", the receipt assertion at the end fails. + +use ardur_e2e_tests::fixtures; + +use std::sync::Arc; +use std::time::Duration; + +use ardur_automation::{ + AutomationAttenuation, AutomationChannel, AutomationDeliveryEvent, AutomationSchedule, + FusedAutomationRuntime, InMemoryScheduleStore, ProactiveAutomationError, + ProactiveAutomationLoop, ScheduleDriver, ScheduleStore, ScheduledCapToken, +}; +use ardur_cap_token::{CapScope, CapTokenIssuer, HolderId as CapHolderId}; +use ardur_cost_gate::{CostEnvelope, CostTuple as GateCostTuple}; +use ardur_cron::CronExpression; +use ardur_fused_runtime::{load_persisted_chain, verify_persisted_chain_with_jwks}; +use ardur_receipt::Jwks; +use ardur_runtime::{CapTokenRef, SessionId}; +use async_trait::async_trait; +use tokio::sync::Mutex; + +/// A channel that captures delivered fire events so the test can assert the +/// runtime response reached delivery. +#[derive(Default)] +struct CapturingChannel { + events: Mutex>, +} + +#[async_trait] +impl AutomationChannel for CapturingChannel { + async fn deliver( + &self, + event: AutomationDeliveryEvent, + ) -> Result<(), ProactiveAutomationError> { + self.events.lock().await.push(event); + Ok(()) + } +} + +/// Mint an attenuated cap-token the automation loop will accept, scoped to the +/// fixtures' audience/tool/holder so the fused runtime verifies it. +fn attenuated_token() -> ScheduledCapToken { + let raw = fixtures::dev_cap_issuer() + .issue( + CapHolderId(fixtures::TEST_HOLDER.to_string()), + CapScope { + audience: fixtures::AUDIENCE.to_string(), + expires_unix: fixtures::NOW_UNIX + 3_600, + budget_remaining: 1_000_000, + tool_allowlist: vec![fixtures::TOOL.to_string()], + }, + ) + .expect("the fire cap-token issues") + .to_base64() + .expect("the fire cap-token serializes"); + ScheduledCapToken::attenuated( + CapTokenRef(raw), + vec![AutomationAttenuation { + rule: format!("restrict_tools:{}", fixtures::TOOL), + evidence: Some("scheduled unattended fire".to_string()), + }], + ) +} + +#[tokio::test] +async fn due_schedule_fires_through_pipeline_and_mints_receipt() { + // A file-backed receipt log so we can prove a signed receipt was actually + // persisted by the fire — not merely returned in memory. + let session_root = fixtures::temp_session_root(); + let receipt_log = session_root.path().join("chain.jsonl"); + + // The real fused runtime over the deterministic stub provider + a generous + // budget, writing receipts to `receipt_log`. + let runtime = fixtures::fused_builder(Arc::new(fixtures::stub_provider())) + .projected_envelope(CostEnvelope { + cents_max: 1_000, + ..Default::default() + }) + .receipt_log(&receipt_log) + .build() + .expect("the fused runtime wires"); + + // Wire the automation executor over the real runtime. + let auto_runtime = Arc::new(FusedAutomationRuntime::new(Arc::new(runtime))); + let store = Arc::new(InMemoryScheduleStore::new()); + let channel = Arc::new(CapturingChannel::default()); + let loop_ = Arc::new(ProactiveAutomationLoop::new( + auto_runtime, + store.clone(), + channel.clone(), + )); + + // A persisted, always-due schedule. + let mut schedule = AutomationSchedule::new( + "nightly-digest", + CronExpression::every_minute(), + SessionId::new(), + attenuated_token(), + GateCostTuple { + tokens_in: 1_000_000, + tokens_out: 1_000_000, + cents: 100, + wall_ms: 1_000_000, + attention_score: 1_000_000, + }, + "summarize today's activity", + ); + // Spend against the same holder the runtime provisioned. + schedule.budget_subject = Some(fixtures::TEST_HOLDER.to_string()); + let id = schedule.id.clone(); + loop_ + .upsert_schedule(schedule) + .await + .expect("schedule persists"); + + // Nothing has run yet: no receipt on the chain. + assert!( + load_persisted_chain(&receipt_log) + .map(|c| c.is_empty()) + .unwrap_or(true), + "no receipt exists before the driver ticks" + ); + + // The timer→executor bridge: one bounded tick fires everything due now. + let driver = ScheduleDriver::new(loop_, Duration::from_millis(10)); + driver.run_bounded(Some(1)).await; + + // 1. The fire was delivered with the runtime's response. + let events = channel.events.lock().await; + assert_eq!( + events.len(), + 1, + "exactly one due schedule fired and delivered" + ); + assert_eq!(events[0].schedule_id, id); + assert_eq!( + events[0].result.response.content, "[anthropic stub]", + "the delivered response is the runtime's real completion" + ); + + // 2. The store recorded the successful fire. + let stored = store.load_all().await.expect("store reloads"); + let fired = stored + .iter() + .find(|s| s.id == id) + .expect("schedule present"); + assert_eq!(fired.fire_count, 1, "the successful fire was persisted"); + assert!(fired.last_fire_at.is_some(), "the fire time was recorded"); + + // 3. A signed receipt was minted and chained on disk, and it verifies. + let chain = load_persisted_chain(&receipt_log).expect("the receipt chain loads"); + assert_eq!(chain.len(), 1, "the fire minted exactly one receipt"); + let jwks = Jwks::from_public_key(&fixtures::dev_receipt_key().public_key()); + verify_persisted_chain_with_jwks(&chain, &jwks) + .expect("the scheduled fire's receipt chain verifies under the publishing JWKS"); + + drop(session_root); +}