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    /// Override the auto-assigned request id (JOE-2221 host-stable ids).
98    pub fn with_request_id(mut self, id: u64) -> Self {
99        self.request_id = id;
100        self
101    }
102
103    pub fn deadline(&self) -> Option<Instant> {
104        self.deadline
105    }
106
107    pub fn remaining(&self) -> Option<Duration> {
108        self.deadline
109            .map(|d| d.saturating_duration_since(Instant::now()))
110    }
111
112    /// Check cancel then deadline. Prefer cancel when both apply.
113    pub fn check(&self) -> Result<()> {
114        if self.cancel.is_cancelled() {
115            return Err(ProviderError::Cancelled.into());
116        }
117        if let Some(d) = self.deadline {
118            if Instant::now() >= d {
119                return Err(ProviderError::DeadlineExceeded.into());
120            }
121        }
122        Ok(())
123    }
124
125    pub fn emit(&self, stage: &'static str, detail: impl Into<String>) {
126        if let Some(sink) = &self.progress {
127            sink(OpProgress {
128                stage,
129                detail: detail.into(),
130            });
131        }
132    }
133
134    /// Tokio-compatible timeout duration for the remaining budget (or fallback).
135    pub fn timeout_or(&self, fallback: Duration) -> Duration {
136        self.remaining().unwrap_or(fallback)
137    }
138}
139
140#[cfg(test)]
141mod tests {
142    use super::*;
143
144    #[test]
145    fn cancel_beats_deadline() {
146        let ctx = OpContext::new().with_deadline_from_now(Duration::from_secs(60));
147        ctx.cancel.cancel();
148        let err = ctx.check().unwrap_err();
149        assert!(matches!(
150            err,
151            crate::error::TranscriptionError::Provider(ProviderError::Cancelled)
152        ));
153    }
154
155    #[test]
156    fn deadline_fires() {
157        let ctx = OpContext::new().with_deadline_from_now(Duration::from_millis(1));
158        std::thread::sleep(Duration::from_millis(5));
159        let err = ctx.check().unwrap_err();
160        assert!(matches!(
161            err,
162            crate::error::TranscriptionError::Provider(ProviderError::DeadlineExceeded)
163        ));
164    }
165
166    #[test]
167    fn unique_request_ids() {
168        let a = OpContext::new();
169        let b = OpContext::new();
170        assert_ne!(a.request_id, b.request_id);
171    }
172}
aurum_core/runtime/op_context.rs

aurum_core/runtime/
op_context.rs
