agent_queue/
lib.rs

1//! # Job Queue
2//!
3//! Production-grade background agent job queue system extracted from tauri-queue.
4//! Framework-agnostic — no Tauri dependency.
5//!
6//! ## Features
7//!
8//! - Priority-based scheduling (High, Normal, Low)
9//! - SQLite persistence with crash recovery
10//! - Hardware throttling (cooldown, max consecutive runs)
11//! - Real-time cancellation during job execution
12//! - Progress tracking via pluggable event emitter
13//! - Pause/resume capability
14//!
15//! ## Quick Start
16//!
17//! 1. Define a job type implementing [`JobHandler`]
18//! 2. Create a [`QueueManager`] with a [`QueueConfig`]
19//! 3. Add jobs with [`QueueManager::add()`]
20//! 4. Spawn the executor with [`QueueManager::spawn()`]
21
22pub mod config;
23pub mod db;
24pub mod error;
25pub mod events;
26pub mod executor;
27pub mod queue;
28pub mod types;
29
30pub use config::{QueueConfig, QueueConfigBuilder};
31pub use error::QueueError;
32pub use executor::ProcessedJob;
33pub use queue::QueueManager;
34pub use types::{
35    FailureClass, JobResult, QueueJob, QueueJobDetails, QueueJobStatus, QueuePriority, QueueStats,
36};
37
38use rusqlite::Connection;
39use std::sync::{Arc, Mutex};
40
41/// Trait for emitting queue lifecycle events.
42///
43/// Implement this trait to receive notifications about job state changes.
44/// For example, a Tauri app would implement this to emit events to the frontend,
45/// while a CLI app might log to stdout or update a progress bar.
46pub trait QueueEventEmitter: Send + Sync + 'static {
47    /// Called when a job starts executing.
48    fn emit_job_started(&self, event: events::JobStartedEvent);
49    /// Called when a job completes successfully.
50    fn emit_job_completed(&self, event: events::JobCompletedEvent);
51    /// Called when a job fails.
52    fn emit_job_failed(&self, event: events::JobFailedEvent);
53    /// Called when a job reports progress.
54    fn emit_job_progress(&self, event: events::JobProgressEvent);
55    /// Called when a job is cancelled.
56    fn emit_job_cancelled(&self, event: events::JobCancelledEvent);
57}
58
59/// A no-op event emitter that discards all events.
60///
61/// Useful for testing or when you don't need event notifications.
62pub struct NoopEventEmitter;
63
64impl QueueEventEmitter for NoopEventEmitter {
65    fn emit_job_started(&self, _event: events::JobStartedEvent) {}
66    fn emit_job_completed(&self, _event: events::JobCompletedEvent) {}
67    fn emit_job_failed(&self, _event: events::JobFailedEvent) {}
68    fn emit_job_progress(&self, _event: events::JobProgressEvent) {}
69    fn emit_job_cancelled(&self, _event: events::JobCancelledEvent) {}
70}
71
72/// A logging event emitter that logs events via `tracing`.
73///
74/// Useful for daemon or CLI usage where events should appear in logs.
75pub struct LoggingEventEmitter;
76
77#[allow(deprecated)]
78impl QueueEventEmitter for LoggingEventEmitter {
79    fn emit_job_started(&self, event: events::JobStartedEvent) {
80        // Prefer canonical trace_ctx over legacy trace_id
81        let trace = event
82            .trace_ctx
83            .as_ref()
84            .map(|ctx| ctx.trace_id.as_str())
85            .or(event.trace_id.as_deref())
86            .unwrap_or("");
87        let attempt = event
88            .attempt_id
89            .as_ref()
90            .map(|a| a.to_string())
91            .unwrap_or_else(|| {
92                event
93                    .attempt_count
94                    .map(|c| c.to_string())
95                    .unwrap_or_default()
96            });
97        tracing::info!(
98            job_id = %event.job_id,
99            worker_id = event.worker_id.as_deref().unwrap_or(""),
100            trace_id = %trace,
101            attempt = %attempt,
102            trial_id = event.trial_id.as_ref().map(|t| t.to_string()).unwrap_or_default().as_str(),
103            "Job started"
104        );
105    }
106    fn emit_job_completed(&self, event: events::JobCompletedEvent) {
107        let trace = event
108            .trace_ctx
109            .as_ref()
110            .map(|ctx| ctx.trace_id.as_str())
111            .or(event.trace_id.as_deref())
112            .unwrap_or("");
113        let attempt = event
114            .attempt_id
115            .as_ref()
116            .map(|a| a.to_string())
117            .unwrap_or_else(|| {
118                event
119                    .attempt_count
120                    .map(|c| c.to_string())
121                    .unwrap_or_default()
122            });
123        tracing::info!(
124            job_id = %event.job_id,
125            worker_id = event.worker_id.as_deref().unwrap_or(""),
126            trace_id = %trace,
127            attempt = %attempt,
128            trial_id = event.trial_id.as_ref().map(|t| t.to_string()).unwrap_or_default().as_str(),
129            "Job completed"
130        );
131    }
132    fn emit_job_failed(&self, event: events::JobFailedEvent) {
133        let trace = event
134            .trace_ctx
135            .as_ref()
136            .map(|ctx| ctx.trace_id.as_str())
137            .or(event.trace_id.as_deref())
138            .unwrap_or("");
139        let attempt = event
140            .attempt_id
141            .as_ref()
142            .map(|a| a.to_string())
143            .unwrap_or_else(|| {
144                event
145                    .attempt_count
146                    .map(|c| c.to_string())
147                    .unwrap_or_default()
148            });
149        tracing::warn!(
150            job_id = %event.job_id,
151            worker_id = event.worker_id.as_deref().unwrap_or(""),
152            trace_id = %trace,
153            attempt = %attempt,
154            trial_id = event.trial_id.as_ref().map(|t| t.to_string()).unwrap_or_default().as_str(),
155            failure_class = event.failure_class.as_deref().unwrap_or(""),
156            error = %event.error,
157            "Job failed"
158        );
159    }
160    fn emit_job_progress(&self, event: events::JobProgressEvent) {
161        let trace = event
162            .trace_ctx
163            .as_ref()
164            .map(|ctx| ctx.trace_id.as_str())
165            .or(event.trace_id.as_deref())
166            .unwrap_or("");
167        tracing::debug!(
168            job_id = %event.job_id,
169            worker_id = event.worker_id.as_deref().unwrap_or(""),
170            trace_id = %trace,
171            progress = %event.progress,
172            step = %event.current_step,
173            total = %event.total_steps,
174            "Job progress"
175        );
176    }
177    fn emit_job_cancelled(&self, event: events::JobCancelledEvent) {
178        let trace = event
179            .trace_ctx
180            .as_ref()
181            .map(|ctx| ctx.trace_id.as_str())
182            .or(event.trace_id.as_deref())
183            .unwrap_or("");
184        tracing::info!(
185            job_id = %event.job_id,
186            worker_id = event.worker_id.as_deref().unwrap_or(""),
187            trace_id = %trace,
188            "Job cancelled"
189        );
190    }
191}
192
193/// Context provided to job handlers during execution.
194///
195/// Gives access to an event emitter for reporting progress and
196/// methods for checking cancellation.
197#[allow(deprecated)]
198pub struct JobContext {
199    /// The ID of the currently executing job.
200    pub job_id: String,
201    /// Phase status: compatibility / migration-only.
202    ///
203    /// Legacy trace ID for correlating queue work with upstream orchestration.
204    /// Prefer [`trace_ctx`](Self::trace_ctx) for new code.
205    #[deprecated(note = "Use trace_ctx instead. Will be removed when all consumers migrate.")]
206    pub trace_id: Option<String>,
207    /// Canonical trace context for end-to-end correlation.
208    pub trace_ctx: Option<stack_ids::TraceCtx>,
209    /// Canonical attempt identity — one per re-enqueue.
210    pub attempt_id: Option<stack_ids::AttemptId>,
211    /// Canonical trial identity — one per concrete execution.
212    pub trial_id: Option<stack_ids::TrialId>,
213    /// Worker identity currently holding the job lease.
214    pub worker_id: Option<String>,
215    /// Phase status: compatibility / migration-only.
216    ///
217    /// Current attempt count (legacy counter).
218    #[deprecated(
219        note = "Use attempt_id/trial_id instead. Will be removed when all consumers migrate."
220    )]
221    pub attempt_count: u32,
222    /// Event emitter for reporting progress.
223    pub(crate) event_emitter: Arc<dyn QueueEventEmitter>,
224    /// Shared database connection for cancellation checks.
225    pub(crate) db: Arc<Mutex<Connection>>,
226}
227
228#[allow(deprecated)]
229impl JobContext {
230    /// Create a context for direct (non-queued) job execution.
231    ///
232    /// This creates a lightweight context backed by an in-memory SQLite database
233    /// and a no-op event emitter. Useful for CLI tools that execute jobs directly
234    /// without going through the queue infrastructure.
235    pub fn new_direct(job_id: &str) -> Self {
236        let conn = Connection::open_in_memory().expect("in-memory DB");
237        let _ = conn.execute_batch(
238            "CREATE TABLE IF NOT EXISTS queue_jobs (
239                id TEXT PRIMARY KEY,
240                priority INTEGER DEFAULT 2,
241                status TEXT DEFAULT 'processing',
242                data_json TEXT NOT NULL DEFAULT '{}',
243                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
244                started_at DATETIME,
245                completed_at DATETIME,
246                error_message TEXT
247            )",
248        );
249        Self {
250            job_id: job_id.to_string(),
251            trace_id: None,
252            trace_ctx: None,
253            attempt_id: Some(stack_ids::AttemptId::generate()),
254            trial_id: Some(stack_ids::TrialId::generate()),
255            worker_id: None,
256            attempt_count: 1,
257            event_emitter: Arc::new(NoopEventEmitter),
258            db: Arc::new(Mutex::new(conn)),
259        }
260    }
261
262    /// Emit a progress event.
263    ///
264    /// # Arguments
265    /// * `current` - Current step number
266    /// * `total` - Total number of steps
267    pub fn emit_progress(&self, current: u32, total: u32) {
268        self.event_emitter
269            .emit_job_progress(events::JobProgressEvent {
270                job_id: self.job_id.clone(),
271                current_step: current,
272                total_steps: total,
273                progress: if total > 0 {
274                    current as f64 / total as f64
275                } else {
276                    0.0
277                },
278                trace_id: self.trace_id.clone(),
279                worker_id: self.worker_id.clone(),
280                attempt_count: Some(self.attempt_count),
281                status: Some("processing".to_string()),
282                trace_ctx: self.trace_ctx.clone(),
283                attempt_id: self.attempt_id.clone(),
284                trial_id: self.trial_id.clone(),
285            });
286    }
287
288    /// Check if this job has been cancelled.
289    ///
290    /// Call this periodically during long-running jobs to support
291    /// cooperative cancellation. If it returns `true`, your handler
292    /// should return `Err(QueueError::Cancelled)`.
293    pub fn is_cancelled(&self) -> bool {
294        match self.db.lock() {
295            Ok(conn) => db::is_cancelled(&conn, &self.job_id).unwrap_or(false),
296            Err(_) => false,
297        }
298    }
299}
300
301/// Trait that job types must implement to be processed by the queue.
302///
303/// Your job type must be serializable (stored as JSON in SQLite),
304/// cloneable, and thread-safe.
305///
306/// # Example
307///
308/// ```ignore
309/// use serde::{Serialize, Deserialize};
310/// use agent_queue::*;
311///
312/// #[derive(Debug, Clone, Serialize, Deserialize)]
313/// struct EmailJob {
314///     to: String,
315///     subject: String,
316/// }
317///
318/// impl JobHandler for EmailJob {
319///     async fn execute(&self, ctx: &JobContext) -> Result<JobResult, QueueError> {
320///         // Send email...
321///         ctx.emit_progress(1, 1);
322///         Ok(JobResult::success())
323///     }
324/// }
325/// ```
326pub trait JobHandler: Send + Sync + serde::Serialize + serde::de::DeserializeOwned + Clone {
327    /// Execute the job. This is called by the executor when the job is picked up.
328    ///
329    /// Use `ctx.emit_progress()` to report progress and `ctx.is_cancelled()`
330    /// to check for cancellation during long-running operations.
331    fn execute(
332        &self,
333        ctx: &JobContext,
334    ) -> impl std::future::Future<Output = Result<JobResult, QueueError>> + Send;
335
336    /// Optional: a human-readable name for this job type, used in logging.
337    fn job_type(&self) -> &str {
338        std::any::type_name::<Self>()
339    }
340}
agent_queue/lib.rs

agent_queue/
lib.rs
