agent_queue/
types.rs

1use serde::{de::DeserializeOwned, Deserialize, Serialize};
2
3/// Priority levels for queue jobs.
4///
5/// Jobs are processed in priority order: High (1), Normal (2), Low (3).
6/// Within the same priority, jobs are processed in FIFO order.
7#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
8pub enum QueuePriority {
9    Low,
10    Normal,
11    High,
12}
13
14impl QueuePriority {
15    pub fn as_i32(&self) -> i32 {
16        match self {
17            QueuePriority::Low => 3,
18            QueuePriority::Normal => 2,
19            QueuePriority::High => 1,
20        }
21    }
22
23    pub fn from_i32(val: i32) -> Self {
24        match val {
25            1 => QueuePriority::High,
26            2 => QueuePriority::Normal,
27            _ => QueuePriority::Low,
28        }
29    }
30}
31
32/// Job status lifecycle: Pending -> Processing -> Completed/Failed/Cancelled
33#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
34pub enum QueueJobStatus {
35    Pending,
36    Processing,
37    Completed,
38    Failed,
39    Cancelled,
40}
41
42impl QueueJobStatus {
43    pub fn as_str(&self) -> &str {
44        match self {
45            QueueJobStatus::Pending => "pending",
46            QueueJobStatus::Processing => "processing",
47            QueueJobStatus::Completed => "completed",
48            QueueJobStatus::Failed => "failed",
49            QueueJobStatus::Cancelled => "cancelled",
50        }
51    }
52
53    pub fn parse(s: &str) -> Option<Self> {
54        match s {
55            "pending" => Some(QueueJobStatus::Pending),
56            "processing" => Some(QueueJobStatus::Processing),
57            "completed" => Some(QueueJobStatus::Completed),
58            "failed" => Some(QueueJobStatus::Failed),
59            "cancelled" => Some(QueueJobStatus::Cancelled),
60            _ => None,
61        }
62    }
63}
64
65/// A generic queue job carrying a custom data payload.
66///
67/// The data field is stored as JSON in SQLite and deserialized back when the
68/// executor picks up the job. Your data type must implement `Serialize`,
69/// `DeserializeOwned`, `Clone`, `Send`, and `Sync`.
70///
71/// When creating jobs via [`QueueJob::new`], only `id`, `priority`, and `data`
72/// are meaningful — the remaining fields (`status`, `created_at`, `started_at`,
73/// `completed_at`, `error_message`) are set to defaults and populated by the
74/// database when reading job records back.
75///
76/// ## Trace and retry primitives
77///
78/// The canonical trace/retry fields are:
79/// - `trace_ctx`: canonical `stack_ids::TraceCtx` for end-to-end correlation
80/// - `attempt_id`: one per re-enqueue (this crate is the retry owner)
81/// - `trial_id`: one per concrete execution attempt
82///
83/// The legacy `trace_id: Option<String>` field is preserved for backward
84/// compatibility and is kept in sync with `trace_ctx` automatically.
85#[derive(Debug, Clone, Serialize, Deserialize)]
86#[serde(bound(deserialize = "T: DeserializeOwned"))]
87pub struct QueueJob<T>
88where
89    T: Serialize + DeserializeOwned + Clone + Send + Sync,
90{
91    pub id: String,
92    /// Phase status: compatibility / migration-only.
93    ///
94    /// Legacy trace identifier for end-to-end correlation.
95    /// The canonical replacement is [`trace_ctx`](Self::trace_ctx).
96    /// Use `stack_ids::TraceCtx::from_legacy_trace_id()` to convert.
97    /// Removal condition: replaced when all callers migrate.
98    #[serde(default, skip_serializing_if = "Option::is_none")]
99    pub trace_id: Option<String>,
100    /// Canonical trace context for end-to-end correlation.
101    ///
102    /// Replaces the legacy `trace_id` field. The trace ID string is
103    /// extractable via `trace_ctx.trace_id` for DB storage (the DB schema
104    /// stores `trace_id TEXT` and is NOT changed).
105    #[serde(default, skip_serializing_if = "Option::is_none")]
106    pub trace_ctx: Option<stack_ids::TraceCtx>,
107    /// Canonical attempt identity — one per re-enqueue.
108    ///
109    /// This crate is the retry owner: each re-enqueue creates a new
110    /// `AttemptId`. Retries within the same attempt use different `TrialId`s.
111    #[serde(default, skip_serializing_if = "Option::is_none")]
112    pub attempt_id: Option<stack_ids::AttemptId>,
113    /// Canonical trial identity — one per concrete execution.
114    ///
115    /// Each time the executor picks up a job for execution, a new `TrialId`
116    /// is generated. Multiple trials may share the same `AttemptId`.
117    #[serde(default, skip_serializing_if = "Option::is_none")]
118    pub trial_id: Option<stack_ids::TrialId>,
119    pub priority: QueuePriority,
120    pub status: QueueJobStatus,
121    pub data: T,
122    /// Only populated when reading from DB.
123    pub created_at: Option<String>,
124    /// Only populated when reading from DB.
125    pub started_at: Option<String>,
126    /// Only populated when reading from DB.
127    pub completed_at: Option<String>,
128    /// Only populated when reading from DB.
129    pub error_message: Option<String>,
130}
131
132impl<T> QueueJob<T>
133where
134    T: Serialize + DeserializeOwned + Clone + Send + Sync,
135{
136    /// Create a new job with a generated UUID and Normal priority.
137    ///
138    /// A fresh `AttemptId` is generated automatically (this crate is the
139    /// retry owner — each enqueue is a new attempt).
140    pub fn new(data: T) -> Self {
141        Self {
142            id: uuid::Uuid::new_v4().to_string(),
143            trace_id: None,
144            trace_ctx: None,
145            attempt_id: Some(stack_ids::AttemptId::generate()),
146            trial_id: None,
147            priority: QueuePriority::Normal,
148            status: QueueJobStatus::Pending,
149            data,
150            created_at: None,
151            started_at: None,
152            completed_at: None,
153            error_message: None,
154        }
155    }
156
157    /// Set the priority for this job (builder pattern).
158    pub fn with_priority(mut self, priority: QueuePriority) -> Self {
159        self.priority = priority;
160        self
161    }
162
163    /// Set a custom ID for this job (builder pattern).
164    pub fn with_id(mut self, id: String) -> Self {
165        self.id = id;
166        self
167    }
168
169    /// Phase status: compatibility / migration-only.
170    ///
171    /// Attach an optional trace ID string for end-to-end correlation.
172    /// Prefer [`with_trace_ctx()`](Self::with_trace_ctx) for new code.
173    /// Removal condition: replaced when all callers migrate.
174    pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
175        let id = trace_id.into();
176        // Keep canonical field in sync when set via legacy path.
177        self.trace_ctx = Some(stack_ids::TraceCtx::from_legacy_trace_id(&id));
178        self.trace_id = Some(id);
179        self
180    }
181
182    /// Attach a canonical `stack_ids::TraceCtx` for end-to-end correlation.
183    ///
184    /// Also sets the legacy `trace_id` string for DB storage compatibility
185    /// (the DB schema stores `trace_id TEXT` and is NOT changed).
186    pub fn with_trace_ctx(mut self, ctx: stack_ids::TraceCtx) -> Self {
187        self.trace_id = Some(ctx.trace_id.clone());
188        self.trace_ctx = Some(ctx);
189        self
190    }
191
192    /// Set a pre-existing `AttemptId` (e.g., when reconstructing from DB).
193    ///
194    /// Normally callers do not need this — `new()` generates a fresh
195    /// `AttemptId` automatically.
196    pub fn with_attempt_id(mut self, attempt_id: stack_ids::AttemptId) -> Self {
197        self.attempt_id = Some(attempt_id);
198        self
199    }
200
201    /// Set a `TrialId` (normally assigned by the executor, not callers).
202    pub fn with_trial_id(mut self, trial_id: stack_ids::TrialId) -> Self {
203        self.trial_id = Some(trial_id);
204        self
205    }
206
207    /// Get the canonical `TraceCtx`, preferring the canonical field,
208    /// falling back to reconstruction from the legacy `trace_id` string.
209    ///
210    /// Returns `None` if neither field is set.
211    pub fn resolve_trace_ctx(&self) -> Option<stack_ids::TraceCtx> {
212        self.trace_ctx.clone().or_else(|| {
213            self.trace_id
214                .as_ref()
215                .map(stack_ids::TraceCtx::from_legacy_trace_id)
216        })
217    }
218
219    /// Phase status: compatibility / migration-only.
220    ///
221    /// Convert the stored trace ID to a canonical `stack_ids::TraceCtx`.
222    /// Returns `None` if no trace ID is set.
223    /// Prefer [`resolve_trace_ctx()`](Self::resolve_trace_ctx) for new code.
224    pub fn trace_ctx_compat(&self) -> Option<stack_ids::TraceCtx> {
225        self.resolve_trace_ctx()
226    }
227}
228
229/// Aggregate count of jobs by status.
230#[derive(Debug, Clone, Default, Serialize, Deserialize)]
231pub struct QueueStats {
232    pub pending: u32,
233    pub processing: u32,
234    pub completed: u32,
235    pub failed: u32,
236    pub cancelled: u32,
237}
238
239/// Classification of a job failure, used for retry strategy.
240#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
241pub enum FailureClass {
242    /// Failure is temporary (network blip, timeout). Will be retried.
243    Transient,
244    /// Failure is permanent (bad input, unrecoverable). Will not be retried.
245    Permanent,
246    /// Rate-limited. Retry after the specified delay.
247    RateLimited { retry_after_secs: u64 },
248}
249
250impl FailureClass {
251    pub fn as_str(&self) -> &str {
252        match self {
253            Self::Transient => "transient",
254            Self::Permanent => "permanent",
255            Self::RateLimited { .. } => "rate_limited",
256        }
257    }
258}
259
260/// Result returned by a job handler after execution.
261#[derive(Debug, Clone, Serialize, Deserialize)]
262pub struct JobResult {
263    pub success: bool,
264    pub output: Option<String>,
265    pub error: Option<String>,
266    #[serde(default, skip_serializing_if = "Option::is_none")]
267    pub failure_class: Option<FailureClass>,
268}
269
270impl JobResult {
271    /// Create a successful result with no output.
272    pub fn success() -> Self {
273        Self {
274            success: true,
275            output: None,
276            error: None,
277            failure_class: None,
278        }
279    }
280
281    /// Create a successful result with output data.
282    pub fn success_with_output(output: String) -> Self {
283        Self {
284            success: true,
285            output: Some(output),
286            error: None,
287            failure_class: None,
288        }
289    }
290
291    /// Create a failure result with an error message.
292    pub fn failure(error: String) -> Self {
293        Self {
294            success: false,
295            output: None,
296            error: Some(error),
297            failure_class: Some(FailureClass::Permanent),
298        }
299    }
300
301    /// Mark a failure result as retryable with structured classification.
302    pub fn with_failure_class(mut self, failure_class: FailureClass) -> Self {
303        self.failure_class = Some(failure_class);
304        self
305    }
306
307    /// Create a transient failure result that should be retried.
308    pub fn transient_failure(error: String) -> Self {
309        Self::failure(error).with_failure_class(FailureClass::Transient)
310    }
311
312    /// Create a rate-limited failure result that should be retried later.
313    pub fn rate_limited(error: String, retry_after_secs: u64) -> Self {
314        Self::failure(error).with_failure_class(FailureClass::RateLimited { retry_after_secs })
315    }
316}
317
318/// Queryable runtime details for a specific job.
319#[derive(Debug, Clone, Serialize, Deserialize)]
320#[serde(rename_all = "camelCase")]
321pub struct QueueJobDetails {
322    pub id: String,
323    pub priority: QueuePriority,
324    pub status: QueueJobStatus,
325    pub data_json: String,
326    pub trace_id: Option<String>,
327    pub created_at: Option<String>,
328    pub started_at: Option<String>,
329    pub completed_at: Option<String>,
330    pub error_message: Option<String>,
331    pub worker_id: Option<String>,
332    pub heartbeat_at: Option<String>,
333    pub visibility_timeout_secs: u64,
334    pub failure_class: Option<String>,
335    pub next_run_at: Option<String>,
336    pub attempt_count: u32,
337    /// Canonical attempt identity — one per re-enqueue. `None` for pre-v4 rows.
338    pub attempt_id: Option<String>,
339    /// Canonical trial identity — one per concrete execution. `None` for pre-v4 rows.
340    pub trial_id: Option<String>,
341}
agent_queue/types.rs

agent_queue/
types.rs
