aurum_core/runtime/
op_context.rs

1//! Per-operation control context (JOE-1595).
2
3use crate::cancel::CancelFlag;
4use crate::error::{ProviderError, Result};
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::sync::Arc;
7use std::time::{Duration, Instant};
8
9static NEXT_REQUEST_ID: AtomicU64 = AtomicU64::new(1);
10
11/// Progress event for hosts/telemetry (best-effort).
12#[derive(Debug, Clone)]
13pub struct OpProgress {
14    pub stage: &'static str,
15    pub detail: String,
16}
17
18/// Optional sink for progress events.
19pub type ProgressSink = Arc<dyn Fn(OpProgress) + Send + Sync>;
20
21/// Immutable-ish control context for one operation.
22///
23/// Fresh cancel token per operation — never reset a shared engine flag across ops.
24///
25/// Deadline uses a single absolute monotonic [`Instant`]; nested APIs should call
26/// [`Self::remaining`] rather than stacking independent timeouts.
27#[derive(Clone)]
28pub struct OpContext {
29    pub request_id: u64,
30    pub cancel: CancelFlag,
31    /// Absolute deadline (monotonic). None = no deadline.
32    deadline: Option<Instant>,
33    pub progress: Option<ProgressSink>,
34}
35
36impl std::fmt::Debug for OpContext {
37    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        f.debug_struct("OpContext")
39            .field("request_id", &self.request_id)
40            .field("cancelled", &self.cancel.is_cancelled())
41            .field("has_deadline", &self.deadline.is_some())
42            .finish()
43    }
44}
45
46impl Default for OpContext {
47    fn default() -> Self {
48        Self::new()
49    }
50}
51
52impl OpContext {
53    pub fn new() -> Self {
54        Self {
55            request_id: NEXT_REQUEST_ID.fetch_add(1, Ordering::Relaxed),
56            cancel: CancelFlag::new(),
57            deadline: None,
58            progress: None,
59        }
60    }
61
62    /// Create from an optional host cancel flag (cloned ownership of the same token).
63    pub fn with_cancel(cancel: CancelFlag) -> Self {
64        Self {
65            request_id: NEXT_REQUEST_ID.fetch_add(1, Ordering::Relaxed),
66            cancel,
67            deadline: None,
68            progress: None,
69        }
70    }
71
72    /// Build from an optional cancel flag, creating a fresh token when absent.
73    pub fn from_optional_cancel(cancel: Option<CancelFlag>) -> Self {
74        match cancel {
75            Some(c) => Self::with_cancel(c),
76            None => Self::new(),
77        }
78    }
79
80    pub fn with_deadline_from_now(mut self, timeout: Duration) -> Self {
81        if !timeout.is_zero() {
82            self.deadline = Some(Instant::now() + timeout);
83        }
84        self
85    }
86
87    pub fn with_absolute_deadline(mut self, at: Instant) -> Self {
88        self.deadline = Some(at);
89        self
90    }
91
92    pub fn with_progress(mut self, sink: ProgressSink) -> Self {
93        self.progress = Some(sink);
94        self
95    }
96
97    pub fn deadline(&self) -> Option<Instant> {
98        self.deadline
99    }
100
101    pub fn remaining(&self) -> Option<Duration> {
102        self.deadline
103            .map(|d| d.saturating_duration_since(Instant::now()))
104    }
105
106    /// Check cancel then deadline. Prefer cancel when both apply.
107    pub fn check(&self) -> Result<()> {
108        if self.cancel.is_cancelled() {
109            return Err(ProviderError::Cancelled.into());
110        }
111        if let Some(d) = self.deadline {
112            if Instant::now() >= d {
113                return Err(ProviderError::DeadlineExceeded.into());
114            }
115        }
116        Ok(())
117    }
118
119    pub fn emit(&self, stage: &'static str, detail: impl Into<String>) {
120        if let Some(sink) = &self.progress {
121            sink(OpProgress {
122                stage,
123                detail: detail.into(),
124            });
125        }
126    }
127
128    /// Tokio-compatible timeout duration for the remaining budget (or fallback).
129    pub fn timeout_or(&self, fallback: Duration) -> Duration {
130        self.remaining().unwrap_or(fallback)
131    }
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137
138    #[test]
139    fn cancel_beats_deadline() {
140        let ctx = OpContext::new().with_deadline_from_now(Duration::from_secs(60));
141        ctx.cancel.cancel();
142        let err = ctx.check().unwrap_err();
143        assert!(matches!(
144            err,
145            crate::error::TranscriptionError::Provider(ProviderError::Cancelled)
146        ));
147    }
148
149    #[test]
150    fn deadline_fires() {
151        let ctx = OpContext::new().with_deadline_from_now(Duration::from_millis(1));
152        std::thread::sleep(Duration::from_millis(5));
153        let err = ctx.check().unwrap_err();
154        assert!(matches!(
155            err,
156            crate::error::TranscriptionError::Provider(ProviderError::DeadlineExceeded)
157        ));
158    }
159
160    #[test]
161    fn unique_request_ids() {
162        let a = OpContext::new();
163        let b = OpContext::new();
164        assert_ne!(a.request_id, b.request_id);
165    }
166}
aurum_core/runtime/op_context.rs

aurum_core/runtime/
op_context.rs
