rch_common/
queue_contract.rs

1//! Queue wait/stream/no-start CLI contract (bd-session-history-remediation-ocv9i.10.2).
2//!
3//! When a command hits a busy fleet it may be queued. The cardinal rule of the
4//! CLI boundary is that an agent must NEVER be left with "maybe running
5//! somewhere" uncertainty: every invocation resolves to a **definite**
6//! [`StartState`] — it either started (and we wait/stream), returns a job id in
7//! an explicit not-started state (reattachable later), or fails before
8//! admission. [`resolve_queue_contract`] is the pure decision function that
9//! guarantees this, honoring the CLI options (wait, follow/stream, wait
10//! timeout, no-start, strict-proof). [`QueueContractResponse`] is the
11//! JSON/human envelope.
12
13use serde::{Deserialize, Serialize};
14
15/// CLI options governing queue behavior.
16#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
17pub struct QueueOptions {
18    /// Wait for a queued job to start (and stream), rather than returning early.
19    pub wait: bool,
20    /// Stream the job's output once it starts (implies `wait`).
21    pub follow: bool,
22    /// Maximum seconds to wait before returning a reattachable not-started job.
23    #[serde(default, skip_serializing_if = "Option::is_none")]
24    pub wait_timeout_secs: Option<u64>,
25    /// Return a job id without waiting/starting (explicit not-started).
26    pub no_start: bool,
27    /// Strict (proof) mode: require immediate admission — never queue.
28    pub strict: bool,
29}
30
31/// What admission decided for this command.
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub enum AdmissionState {
34    /// Admitted immediately and running.
35    AdmittedRunning,
36    /// Capacity was full — the command was queued.
37    Queued,
38    /// Rejected before any job started (with reason).
39    RejectedBeforeAdmission(String),
40}
41
42/// What happened while waiting on a queued job (a runtime input).
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub enum WaitResult {
45    /// The job started during the wait.
46    StartedAfterWait,
47    /// The wait timed out before the job started.
48    TimedOut,
49    /// The job was cancelled before it started.
50    CancelledBeforeStart,
51}
52
53/// The definite, unambiguous start state of a command. There is deliberately no
54/// `Unknown` — eliminating "maybe running somewhere" is the whole point.
55#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
56#[serde(rename_all = "snake_case")]
57pub enum StartState {
58    Started,
59    NotStarted,
60}
61
62/// The resolved queue-contract outcome.
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
64#[serde(rename_all = "snake_case", tag = "outcome")]
65pub enum QueueContractOutcome {
66    /// Admitted immediately and running.
67    StartedRunning,
68    /// Was queued, then started while we waited (streaming).
69    WaitingStreamed,
70    /// Queued and returned with an explicit not-started state (reattachable).
71    QueuedNotStarted,
72    /// Queued, but the wait timed out — returned reattachable not-started.
73    TimedOutQueued,
74    /// Cancelled before it started.
75    CancelledBeforeStart,
76    /// Rejected before admission — no job ever started.
77    FailedBeforeAdmission { reason: String },
78}
79
80impl QueueContractOutcome {
81    /// The definite start state for this outcome.
82    #[must_use]
83    pub const fn start_state(&self) -> StartState {
84        match self {
85            Self::StartedRunning | Self::WaitingStreamed => StartState::Started,
86            Self::QueuedNotStarted
87            | Self::TimedOutQueued
88            | Self::CancelledBeforeStart
89            | Self::FailedBeforeAdmission { .. } => StartState::NotStarted,
90        }
91    }
92
93    /// Did the job start?
94    #[must_use]
95    pub const fn started(&self) -> bool {
96        matches!(self.start_state(), StartState::Started)
97    }
98
99    /// Stable snake_case id.
100    #[must_use]
101    pub const fn as_str(&self) -> &'static str {
102        match self {
103            Self::StartedRunning => "started_running",
104            Self::WaitingStreamed => "waiting_streamed",
105            Self::QueuedNotStarted => "queued_not_started",
106            Self::TimedOutQueued => "timed_out_queued",
107            Self::CancelledBeforeStart => "cancelled_before_start",
108            Self::FailedBeforeAdmission { .. } => "failed_before_admission",
109        }
110    }
111}
112
113/// Resolve the queue contract — pure and total. Given admission, the CLI
114/// options, and (if waited) the wait result, returns exactly one definite
115/// outcome. `wait` is ignored unless the command was queued and waiting.
116#[must_use]
117pub fn resolve_queue_contract(
118    admission: &AdmissionState,
119    opts: &QueueOptions,
120    wait: Option<WaitResult>,
121) -> QueueContractOutcome {
122    match admission {
123        AdmissionState::RejectedBeforeAdmission(reason) => {
124            QueueContractOutcome::FailedBeforeAdmission {
125                reason: reason.clone(),
126            }
127        }
128        AdmissionState::AdmittedRunning => QueueContractOutcome::StartedRunning,
129        AdmissionState::Queued => {
130            // Strict (proof) mode refuses to queue — fail fast, nothing started.
131            if opts.strict {
132                return QueueContractOutcome::FailedBeforeAdmission {
133                    reason: "strict mode requires immediate admission; refusing to queue"
134                        .to_string(),
135                };
136            }
137            // No-start or no-wait: return a reattachable not-started job.
138            if opts.no_start || !(opts.wait || opts.follow) {
139                return QueueContractOutcome::QueuedNotStarted;
140            }
141            match wait {
142                Some(WaitResult::StartedAfterWait) => QueueContractOutcome::WaitingStreamed,
143                Some(WaitResult::TimedOut) => QueueContractOutcome::TimedOutQueued,
144                Some(WaitResult::CancelledBeforeStart) => {
145                    QueueContractOutcome::CancelledBeforeStart
146                }
147                None => QueueContractOutcome::QueuedNotStarted,
148            }
149        }
150    }
151}
152
153/// The agent-facing response envelope (JSON/human).
154#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
155pub struct QueueContractResponse {
156    /// The local/remote job id, when a job exists (queued or started).
157    #[serde(default, skip_serializing_if = "Option::is_none")]
158    pub job_id: Option<String>,
159    /// Definite: did the job start? Never ambiguous.
160    pub started: bool,
161    /// The outcome id.
162    pub outcome: String,
163    /// Whether the agent can reattach later (a job id exists but didn't start).
164    pub reattachable: bool,
165    /// Operator-facing detail.
166    pub detail: String,
167}
168
169impl QueueContractResponse {
170    /// Build the response for an outcome and (optional) job id.
171    #[must_use]
172    pub fn build(outcome: &QueueContractOutcome, job_id: Option<String>) -> Self {
173        let started = outcome.started();
174        // A job id that didn't start is reattachable; a failed-before-admission
175        // never produced a job, so it is not.
176        let has_job = job_id.is_some();
177        let reattachable = has_job && !started;
178        let detail = match outcome {
179            QueueContractOutcome::StartedRunning => "admitted and running".to_string(),
180            QueueContractOutcome::WaitingStreamed => "queued, then started; streaming".to_string(),
181            QueueContractOutcome::QueuedNotStarted => {
182                "queued; not started — reattach with the job id".to_string()
183            }
184            QueueContractOutcome::TimedOutQueued => {
185                "wait timed out while queued; not started — reattach with the job id".to_string()
186            }
187            QueueContractOutcome::CancelledBeforeStart => "cancelled before it started".to_string(),
188            QueueContractOutcome::FailedBeforeAdmission { reason } => {
189                format!("failed before admission: {reason}")
190            }
191        };
192        Self {
193            job_id,
194            started,
195            outcome: outcome.as_str().to_string(),
196            reattachable,
197            detail,
198        }
199    }
200
201    /// Human-readable line.
202    #[must_use]
203    pub fn render(&self) -> String {
204        format!(
205            "queue: {} (started={}, reattachable={}{}) — {}",
206            self.outcome,
207            self.started,
208            self.reattachable,
209            self.job_id
210                .as_deref()
211                .map_or_else(String::new, |j| format!(", job={j}")),
212            self.detail,
213        )
214    }
215}
216
217#[cfg(test)]
218mod tests {
219    use super::*;
220
221    #[test]
222    fn admitted_running_starts() {
223        let o = resolve_queue_contract(
224            &AdmissionState::AdmittedRunning,
225            &QueueOptions::default(),
226            None,
227        );
228        assert_eq!(o, QueueContractOutcome::StartedRunning);
229        assert!(o.started());
230        assert_eq!(o.start_state(), StartState::Started);
231    }
232
233    #[test]
234    fn busy_workers_no_wait_returns_reattachable_not_started() {
235        // Busy fleet, default options (no wait): explicit not-started + job id.
236        let o = resolve_queue_contract(&AdmissionState::Queued, &QueueOptions::default(), None);
237        assert_eq!(o, QueueContractOutcome::QueuedNotStarted);
238        assert!(!o.started());
239        let resp = QueueContractResponse::build(&o, Some("job-1".to_string()));
240        assert!(!resp.started);
241        assert!(resp.reattachable);
242        assert_eq!(resp.job_id.as_deref(), Some("job-1"));
243    }
244
245    #[test]
246    fn no_start_option_returns_job_id_not_started() {
247        let opts = QueueOptions {
248            no_start: true,
249            ..QueueOptions::default()
250        };
251        let o = resolve_queue_contract(&AdmissionState::Queued, &opts, None);
252        assert_eq!(o, QueueContractOutcome::QueuedNotStarted);
253    }
254
255    #[test]
256    fn wait_then_started_streams() {
257        let opts = QueueOptions {
258            wait: true,
259            follow: true,
260            ..QueueOptions::default()
261        };
262        let o = resolve_queue_contract(
263            &AdmissionState::Queued,
264            &opts,
265            Some(WaitResult::StartedAfterWait),
266        );
267        assert_eq!(o, QueueContractOutcome::WaitingStreamed);
268        assert!(o.started());
269    }
270
271    #[test]
272    fn queue_timeout_returns_reattachable_not_started() {
273        let opts = QueueOptions {
274            wait: true,
275            wait_timeout_secs: Some(30),
276            ..QueueOptions::default()
277        };
278        let o = resolve_queue_contract(&AdmissionState::Queued, &opts, Some(WaitResult::TimedOut));
279        assert_eq!(o, QueueContractOutcome::TimedOutQueued);
280        assert!(!o.started());
281        let resp = QueueContractResponse::build(&o, Some("job-7".to_string()));
282        assert!(resp.reattachable);
283        assert!(resp.detail.contains("reattach"));
284    }
285
286    #[test]
287    fn cancellation_before_start() {
288        let opts = QueueOptions {
289            wait: true,
290            ..QueueOptions::default()
291        };
292        let o = resolve_queue_contract(
293            &AdmissionState::Queued,
294            &opts,
295            Some(WaitResult::CancelledBeforeStart),
296        );
297        assert_eq!(o, QueueContractOutcome::CancelledBeforeStart);
298        assert!(!o.started());
299    }
300
301    #[test]
302    fn strict_mode_refuses_to_queue() {
303        let opts = QueueOptions {
304            strict: true,
305            wait: true,
306            ..QueueOptions::default()
307        };
308        let o = resolve_queue_contract(&AdmissionState::Queued, &opts, None);
309        assert!(matches!(
310            o,
311            QueueContractOutcome::FailedBeforeAdmission { .. }
312        ));
313        assert!(!o.started());
314    }
315
316    #[test]
317    fn rejected_before_admission_never_starts_and_has_no_job() {
318        let o = resolve_queue_contract(
319            &AdmissionState::RejectedBeforeAdmission("no admissible workers".to_string()),
320            &QueueOptions::default(),
321            None,
322        );
323        assert_eq!(o.as_str(), "failed_before_admission");
324        let resp = QueueContractResponse::build(&o, None);
325        assert!(!resp.started);
326        assert!(!resp.reattachable, "no job => not reattachable");
327        assert!(resp.job_id.is_none());
328        assert!(resp.detail.contains("no admissible workers"));
329    }
330
331    #[test]
332    fn every_outcome_has_a_definite_start_state() {
333        // The contract's core guarantee: no outcome is ambiguous.
334        let outcomes = [
335            QueueContractOutcome::StartedRunning,
336            QueueContractOutcome::WaitingStreamed,
337            QueueContractOutcome::QueuedNotStarted,
338            QueueContractOutcome::TimedOutQueued,
339            QueueContractOutcome::CancelledBeforeStart,
340            QueueContractOutcome::FailedBeforeAdmission {
341                reason: "x".to_string(),
342            },
343        ];
344        for o in &outcomes {
345            // start_state() is total and returns exactly Started xor NotStarted.
346            let s = o.start_state();
347            assert!(s == StartState::Started || s == StartState::NotStarted);
348            assert_eq!(o.started(), s == StartState::Started);
349        }
350    }
351
352    #[test]
353    fn response_serde_and_render() {
354        let o = resolve_queue_contract(&AdmissionState::Queued, &QueueOptions::default(), None);
355        let resp = QueueContractResponse::build(&o, Some("job-9".to_string()));
356        let v = serde_json::to_value(&resp).unwrap();
357        assert_eq!(v["outcome"], "queued_not_started");
358        assert_eq!(v["started"], false);
359        assert_eq!(v["reattachable"], true);
360        let back: QueueContractResponse = serde_json::from_value(v).unwrap();
361        assert_eq!(resp, back);
362        assert!(resp.render().contains("job=job-9"));
363    }
364}
rch_common/queue_contract.rs

rch_common/
queue_contract.rs
