tenant.openagents/omega

1//! Durable admission for messages sent during a running turn (omega#79).
2//!
3//! `OMEGA-DELTA-0032`.
4//!
5//! ## What was wrong
6//!
7//! Omega inherited a real queue. `MessageQueue` distinguishes a steer from an
8//! enqueue, and its state machine is careful about the cancel it has to absorb.
9//! Two things were missing, and the issue's own falsifier names both.
10//!
11//! **"Queue state lives only in renderer memory."** It did. `MessageQueue` is a
12//! field on `ThreadView`, holds live GPUI editor handles, and dies with
13//! the view. Close the panel, reconnect, or restart, and a message the composer
14//! had already acknowledged as queued was gone with no trace that it ever
15//! existed. This module is the durable half: an item is written down *before*
16//! the UI says it is queued, so what the user was told and what survives a
17//! restart are the same fact.
18//!
19//! **"A second send reaches a running provider turn without a guard."**
20//! `sync_queue_flag_to_native_thread` set the boundary flag on
21//! `agent::Thread` and did nothing for anything else, so an external ACP thread
22//! and an engine lane both fell through to `dispatch_queued_entry`'s
23//! cancel-then-send. The guard is
24//! [`omega_front_door::disposition`](omega_front_door::disposition), a total
25//! law over all three executor classes; this module records what it decided
26//! beside the item, so the reason a message was not steered outlives the
27//! session that decided it.
28//!
29//! ## Shape
30//!
31//! Same shape as [`crate::omega_router::RouteJournal`], for the same reasons:
32//! a `BTreeMap` so the file does not change when nothing decided differently,
33//! an atomic temporary-file rewrite so a crash leaves the previous journal
34//! rather than a truncated one, and typed round-tripping so a hand-edited file
35//! is refused rather than believed.
36//!
37//! One rule this journal has that the route journal does not: **a terminal item
38//! is never reopened.** `promoted`, `cancelled` and `failed` are final. A
39//! restart that could move an item back to `queued` would promote it a second
40//! time, which is exactly the duplication the acceptance criterion names.
41
42use std::cell::RefCell;
43use std::collections::BTreeMap;
44use std::path::{Path, PathBuf};
45
46use omega_front_door::{
47    ExecutorClass, QueueItemState, Quiescence, SendCommand, SendDisposition, SteerCapability,
48    disposition, may_promote,
49};
50use serde_json::Value;
51
52/// Schema of the durable queue document.
53pub const SEND_QUEUE_JOURNAL_SCHEMA: &str = "openagents.omega.agent_send_queue.v1";
54
55/// File the document lives in, under the Omega data directory.
56pub const SEND_QUEUE_JOURNAL_FILE: &str = "agent-send-queue.json";
57
58/// One admitted message, as it survives a restart.
59///
60/// The editor handle and the tracked buffers are deliberately absent: they are
61/// live GPUI values and cannot be a durable fact. What has to survive is the
62/// text the person wrote, where it was going, what they asked for, what Omega
63/// decided, and where the item is in its life.
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub struct QueuedSend {
66    /// Stable across restarts and unique within a thread. Ordering identity as
67    /// well as name: promotion is by `(sequence, item_id)`, never by map order.
68    pub item_id: String,
69    pub thread_id: String,
70    pub sequence: u64,
71    /// The message body. Plain text only — a durable record of a message the
72    /// user typed, not a serialised content-block tree.
73    pub text: String,
74    pub command: SendCommand,
75    pub class: ExecutorClass,
76    pub capability: SteerCapability,
77    pub state: QueueItemState,
78}
79
80impl QueuedSend {
81    /// What Omega will do with this item, derived from the parts.
82    ///
83    /// Derived on every call and never stored, the same rule
84    /// `ExecutorDisclosure` holds to: a stored disposition could disagree with
85    /// the law that produced it, and then the record would be the lie.
86    #[must_use]
87    pub const fn disposition(&self) -> SendDisposition {
88        disposition(self.command, self.class, self.capability)
89    }
90
91    /// Whether this item may be promoted now.
92    #[must_use]
93    pub const fn may_promote(&self, quiescence: Quiescence) -> bool {
94        may_promote(self.state, quiescence)
95    }
96}
97
98/// Why an admission or a transition did not happen.
99#[derive(Debug, Clone, Copy, PartialEq, Eq)]
100pub enum SendQueueRefusal {
101    /// The item is already terminal. Promoting it again would duplicate a turn.
102    ItemIsTerminal,
103    /// No item with this id.
104    UnknownItem,
105    /// The prior turn is not proven finished.
106    NotQuiescent,
107    /// The record could not be written, so the user was not told it was queued.
108    NotPersisted,
109}
110
111impl SendQueueRefusal {
112    #[must_use]
113    pub const fn token(self) -> &'static str {
114        match self {
115            Self::ItemIsTerminal => "item_is_terminal",
116            Self::UnknownItem => "unknown_item",
117            Self::NotQuiescent => "not_quiescent",
118            Self::NotPersisted => "not_persisted",
119        }
120    }
121}
122
123/// The durable queue.
124///
125/// One journal for the whole app, partitioned by thread id. Promotion is the
126/// job of a single thread-owned scheduler, which asks
127/// [`head_for`](Self::head_for) and then [`promote`](Self::promote).
128pub struct SendQueueJournal {
129    path: PathBuf,
130    /// `BTreeMap` so the serialised file is a function of the items, not of
131    /// hash order — a journal that changes when nothing decided differently is
132    /// a journal nobody can diff.
133    items: RefCell<BTreeMap<String, QueuedSend>>,
134    next_sequence: RefCell<u64>,
135}
136
137impl SendQueueJournal {
138    /// The journal at the Omega data directory's usual place.
139    #[must_use]
140    pub fn at_data_dir() -> Self {
141        Self::at(
142            paths::data_dir()
143                .join("openagents")
144                .join(SEND_QUEUE_JOURNAL_FILE),
145        )
146    }
147
148    /// The journal at an explicit path. Loads what is already there.
149    ///
150    /// An unreadable journal starts empty **and says so loudly**. It does not
151    /// silently continue, because a queue that quietly forgot an admitted
152    /// message is the defect this module exists to prevent, and a warning is
153    /// the only honest thing available at this layer.
154    #[must_use]
155    pub fn at(path: PathBuf) -> Self {
156        let items = load(&path).unwrap_or_else(|error| {
157            log::warn!(
158                "OMEGA-DELTA-0032: send queue at {} could not be read ({error:#}); \
159                 starting from empty. Any admitted message it held is lost.",
160                path.display()
161            );
162            BTreeMap::new()
163        });
164        let next = items.values().map(|item| item.sequence).max().unwrap_or(0) + 1;
165        Self {
166            path,
167            items: RefCell::new(items),
168            next_sequence: RefCell::new(next),
169        }
170    }
171
172    /// Admit a message to the queue, durably, before acknowledging it.
173    ///
174    /// The `Ok` here is the acknowledgement the UI is allowed to show. A caller
175    /// that renders "queued" on the `Err` path has reintroduced the defect.
176    ///
177    /// # Errors
178    ///
179    /// [`SendQueueRefusal::NotPersisted`] when the record could not be written.
180    pub fn admit(
181        &self,
182        thread_id: &str,
183        item_id: &str,
184        text: &str,
185        command: SendCommand,
186        class: ExecutorClass,
187        capability: SteerCapability,
188    ) -> Result<QueuedSend, SendQueueRefusal> {
189        let sequence = {
190            let mut next = self.next_sequence.borrow_mut();
191            let sequence = *next;
192            *next += 1;
193            sequence
194        };
195        let item = QueuedSend {
196            item_id: item_id.to_owned(),
197            thread_id: thread_id.to_owned(),
198            sequence,
199            text: text.to_owned(),
200            command,
201            class,
202            capability,
203            state: QueueItemState::Queued,
204        };
205        self.items
206            .borrow_mut()
207            .insert(Self::key(thread_id, item_id), item.clone());
208        match self.persist() {
209            Ok(()) => Ok(item),
210            Err(error) => {
211                log::error!(
212                    "OMEGA-DELTA-0032: {item_id} could not be admitted to {}: {error:#}",
213                    self.path.display()
214                );
215                self.items
216                    .borrow_mut()
217                    .remove(&Self::key(thread_id, item_id));
218                Err(SendQueueRefusal::NotPersisted)
219            }
220        }
221    }
222
223    /// Every open item for a thread, in promotion order.
224    #[must_use]
225    pub fn open_items(&self, thread_id: &str) -> Vec<QueuedSend> {
226        let mut items: Vec<QueuedSend> = self
227            .items
228            .borrow()
229            .values()
230            .filter(|item| item.thread_id == thread_id && item.state.is_open())
231            .cloned()
232            .collect();
233        items.sort_by(|left, right| {
234            left.sequence
235                .cmp(&right.sequence)
236                .then_with(|| left.item_id.cmp(&right.item_id))
237        });
238        items
239    }
240
241    /// Every item for a thread, including terminal ones, in the same order.
242    #[must_use]
243    pub fn items(&self, thread_id: &str) -> Vec<QueuedSend> {
244        let mut items: Vec<QueuedSend> = self
245            .items
246            .borrow()
247            .values()
248            .filter(|item| item.thread_id == thread_id)
249            .cloned()
250            .collect();
251        items.sort_by(|left, right| {
252            left.sequence
253                .cmp(&right.sequence)
254                .then_with(|| left.item_id.cmp(&right.item_id))
255        });
256        items
257    }
258
259    /// The item the scheduler would promote next, if any.
260    #[must_use]
261    pub fn head_for(&self, thread_id: &str) -> Option<QueuedSend> {
262        self.open_items(thread_id).into_iter().next()
263    }
264
265    /// Promote the queue head, but only after proven quiescence.
266    ///
267    /// # Errors
268    ///
269    /// [`SendQueueRefusal::NotQuiescent`] when the prior turn is running or its
270    /// stop was never observed — the reconnect case, where promoting is how a
271    /// message gets sent twice.
272    pub fn promote(
273        &self,
274        thread_id: &str,
275        item_id: &str,
276        quiescence: Quiescence,
277    ) -> Result<QueuedSend, SendQueueRefusal> {
278        let key = Self::key(thread_id, item_id);
279        let current = self
280            .items
281            .borrow()
282            .get(&key)
283            .cloned()
284            .ok_or(SendQueueRefusal::UnknownItem)?;
285        if !current.state.is_open() {
286            return Err(SendQueueRefusal::ItemIsTerminal);
287        }
288        if !may_promote(current.state, quiescence) {
289            return Err(SendQueueRefusal::NotQuiescent);
290        }
291        self.transition(&key, QueueItemState::Promoted)
292    }
293
294    /// Withdraw an item before it is promoted.
295    ///
296    /// # Errors
297    ///
298    /// [`SendQueueRefusal::ItemIsTerminal`] for an item that already ended.
299    pub fn cancel(&self, thread_id: &str, item_id: &str) -> Result<QueuedSend, SendQueueRefusal> {
300        let key = Self::key(thread_id, item_id);
301        self.require_open(&key)?;
302        self.transition(&key, QueueItemState::Cancelled)
303    }
304
305    /// Record that promotion was attempted and did not start a turn.
306    ///
307    /// # Errors
308    ///
309    /// [`SendQueueRefusal::ItemIsTerminal`] for an item that already ended.
310    pub fn fail(&self, thread_id: &str, item_id: &str) -> Result<QueuedSend, SendQueueRefusal> {
311        let key = Self::key(thread_id, item_id);
312        self.require_open(&key)?;
313        self.transition(&key, QueueItemState::Failed)
314    }
315
316    fn require_open(&self, key: &str) -> Result<(), SendQueueRefusal> {
317        let items = self.items.borrow();
318        let item = items.get(key).ok_or(SendQueueRefusal::UnknownItem)?;
319        if item.state.is_open() {
320            Ok(())
321        } else {
322            Err(SendQueueRefusal::ItemIsTerminal)
323        }
324    }
325
326    fn transition(
327        &self,
328        key: &str,
329        state: QueueItemState,
330    ) -> Result<QueuedSend, SendQueueRefusal> {
331        let updated = {
332            let mut items = self.items.borrow_mut();
333            let item = items.get_mut(key).ok_or(SendQueueRefusal::UnknownItem)?;
334            item.state = state;
335            item.clone()
336        };
337        match self.persist() {
338            Ok(()) => Ok(updated),
339            Err(error) => {
340                log::error!(
341                    "OMEGA-DELTA-0032: {key} could not record {}: {error:#}",
342                    state.token()
343                );
344                Err(SendQueueRefusal::NotPersisted)
345            }
346        }
347    }
348
349    fn key(thread_id: &str, item_id: &str) -> String {
350        format!("{thread_id}\u{1f}{item_id}")
351    }
352
353    fn persist(&self) -> anyhow::Result<()> {
354        let items = self.items.borrow();
355        let document = serde_json::json!({
356            "schema": SEND_QUEUE_JOURNAL_SCHEMA,
357            "items": items
358                .values()
359                .map(|item| serde_json::json!({
360                    "itemId": item.item_id,
361                    "threadId": item.thread_id,
362                    "sequence": item.sequence,
363                    "text": item.text,
364                    "command": item.command.token(),
365                    "class": item.class.token(),
366                    "capability": item.capability.token(),
367                    "state": item.state.token(),
368                }))
369                .collect::<Vec<_>>(),
370        });
371        if let Some(parent) = self.path.parent() {
372            std::fs::create_dir_all(parent)?;
373        }
374        let temporary = self.path.with_extension("json.tmp");
375        std::fs::write(&temporary, serde_json::to_vec_pretty(&document)?)?;
376        std::fs::rename(&temporary, &self.path)?;
377        Ok(())
378    }
379}
380
381fn parse_class(token: &str) -> Option<ExecutorClass> {
382    ExecutorClass::all()
383        .iter()
384        .copied()
385        .find(|class| class.token() == token)
386}
387
388fn parse_capability(token: &str) -> Option<SteerCapability> {
389    SteerCapability::all()
390        .iter()
391        .copied()
392        .find(|capability| capability.token() == token)
393}
394
395fn load(path: &Path) -> anyhow::Result<BTreeMap<String, QueuedSend>> {
396    if !path.exists() {
397        return Ok(BTreeMap::new());
398    }
399    let document: Value = serde_json::from_slice(&std::fs::read(path)?)?;
400    let schema = document.get("schema").and_then(Value::as_str);
401    anyhow::ensure!(
402        schema == Some(SEND_QUEUE_JOURNAL_SCHEMA),
403        "unsupported send queue schema {schema:?}"
404    );
405    let mut items = BTreeMap::new();
406    for entry in document
407        .get("items")
408        .and_then(Value::as_array)
409        .unwrap_or(&Vec::new())
410    {
411        let (
412            Some(item_id),
413            Some(thread_id),
414            Some(sequence),
415            Some(text),
416            Some(command),
417            Some(class),
418            Some(capability),
419            Some(state),
420        ) = (
421            entry.get("itemId").and_then(Value::as_str),
422            entry.get("threadId").and_then(Value::as_str),
423            entry.get("sequence").and_then(Value::as_u64),
424            entry.get("text").and_then(Value::as_str),
425            entry
426                .get("command")
427                .and_then(Value::as_str)
428                .and_then(SendCommand::parse_token),
429            entry
430                .get("class")
431                .and_then(Value::as_str)
432                .and_then(parse_class),
433            entry
434                .get("capability")
435                .and_then(Value::as_str)
436                .and_then(parse_capability),
437            entry
438                .get("state")
439                .and_then(Value::as_str)
440                .and_then(QueueItemState::parse_token),
441        )
442        else {
443            anyhow::bail!("send queue entry is not a complete queued send");
444        };
445        items.insert(
446            SendQueueJournal::key(thread_id, item_id),
447            QueuedSend {
448                item_id: item_id.to_owned(),
449                thread_id: thread_id.to_owned(),
450                sequence,
451                text: text.to_owned(),
452                command,
453                class,
454                capability,
455                state,
456            },
457        );
458    }
459    Ok(items)
460}
461
462/// The line the composer shows for one admitted item.
463///
464/// Derived from the item, so the queue cannot render a promise its disposition
465/// does not support. Nothing stores this.
466#[must_use]
467pub fn queue_item_phrase(item: &QueuedSend) -> String {
468    match item.state {
469        QueueItemState::Queued => item.disposition().phrase(),
470        QueueItemState::Promoted => "Sent.".to_owned(),
471        QueueItemState::Cancelled => "Removed from the queue.".to_owned(),
472        QueueItemState::Failed => "Could not be sent. Still in the queue history.".to_owned(),
473    }
474}
475
476#[cfg(test)]
477mod tests {
478    use super::*;
479    use omega_front_door::{SendFallback, SteerRefusal};
480
481    fn journal() -> (SendQueueJournal, tempfile::TempDir) {
482        let dir = tempfile::tempdir().expect("temp dir");
483        let path = dir.path().join(SEND_QUEUE_JOURNAL_FILE);
484        (SendQueueJournal::at(path), dir)
485    }
486
487    fn admit(
488        journal: &SendQueueJournal,
489        item_id: &str,
490        command: SendCommand,
491        class: ExecutorClass,
492        capability: SteerCapability,
493    ) -> QueuedSend {
494        journal
495            .admit("thread-1", item_id, "look at the other file too", command, class, capability)
496            .expect("admitted")
497    }
498
499    /// The falsifier's second half: "queue state lives only in renderer
500    /// memory". A journal reopened from the same path is the restart.
501    #[test]
502    fn an_admitted_message_survives_a_restart() {
503        let dir = tempfile::tempdir().expect("temp dir");
504        let path = dir.path().join(SEND_QUEUE_JOURNAL_FILE);
505        {
506            let journal = SendQueueJournal::at(path.clone());
507            journal
508                .admit(
509                    "thread-1",
510                    "item-1",
511                    "and check the tests",
512                    SendCommand::Enqueue,
513                    ExecutorClass::NativeLoop,
514                    SteerCapability::Unknown,
515                )
516                .expect("admitted");
517        }
518        let reopened = SendQueueJournal::at(path);
519        let items = reopened.open_items("thread-1");
520        assert_eq!(items.len(), 1);
521        assert_eq!(items[0].text, "and check the tests");
522        assert_eq!(items[0].state, QueueItemState::Queued);
523    }
524
525    /// A promoted item that a restart reopened would be sent twice. The
526    /// acceptance criterion names this exact case.
527    #[test]
528    fn a_promoted_item_is_not_reopened_by_a_restart_and_cannot_promote_twice() {
529        let dir = tempfile::tempdir().expect("temp dir");
530        let path = dir.path().join(SEND_QUEUE_JOURNAL_FILE);
531        {
532            let journal = SendQueueJournal::at(path.clone());
533            admit(
534                &journal,
535                "item-1",
536                SendCommand::Enqueue,
537                ExecutorClass::NativeLoop,
538                SteerCapability::Unknown,
539            );
540            journal
541                .promote("thread-1", "item-1", Quiescence::Proven)
542                .expect("promoted once");
543        }
544        let reopened = SendQueueJournal::at(path);
545        assert!(reopened.open_items("thread-1").is_empty());
546        assert_eq!(reopened.items("thread-1").len(), 1);
547        assert_eq!(
548            reopened.promote("thread-1", "item-1", Quiescence::Proven),
549            Err(SendQueueRefusal::ItemIsTerminal)
550        );
551    }
552
553    /// After a reconnect Omega never saw the prior turn stop. Promoting there
554    /// is how a queued message races the turn it was meant to follow.
555    #[test]
556    fn a_reconnect_does_not_promote_on_an_unobserved_stop() {
557        let (journal, _dir) = journal();
558        admit(
559            &journal,
560            "item-1",
561            SendCommand::Enqueue,
562            ExecutorClass::ExternalAcp,
563            SteerCapability::Unknown,
564        );
565        assert_eq!(
566            journal.promote("thread-1", "item-1", Quiescence::Unknown),
567            Err(SendQueueRefusal::NotQuiescent)
568        );
569        assert_eq!(
570            journal.promote("thread-1", "item-1", Quiescence::Running),
571            Err(SendQueueRefusal::NotQuiescent)
572        );
573        assert!(journal.promote("thread-1", "item-1", Quiescence::Proven).is_ok());
574    }
575
576    /// Ordering is the item's own, not the map's. Two items admitted in order
577    /// promote in that order however their ids sort.
578    #[test]
579    fn promotion_order_is_admission_order_not_identifier_order() {
580        let (journal, _dir) = journal();
581        admit(
582            &journal,
583            "zzz-first",
584            SendCommand::Enqueue,
585            ExecutorClass::NativeLoop,
586            SteerCapability::Unknown,
587        );
588        admit(
589            &journal,
590            "aaa-second",
591            SendCommand::Enqueue,
592            ExecutorClass::NativeLoop,
593            SteerCapability::Unknown,
594        );
595        assert_eq!(journal.head_for("thread-1").expect("head").item_id, "zzz-first");
596    }
597
598    /// Two threads, one file, no crossing. A queue that leaked between threads
599    /// would deliver somebody's message to the wrong agent.
600    #[test]
601    fn items_do_not_cross_between_threads() {
602        let (journal, _dir) = journal();
603        admit(
604            &journal,
605            "item-1",
606            SendCommand::Enqueue,
607            ExecutorClass::NativeLoop,
608            SteerCapability::Unknown,
609        );
610        journal
611            .admit(
612                "thread-2",
613                "item-1",
614                "different thread, same item id",
615                SendCommand::Enqueue,
616                ExecutorClass::ExternalAcp,
617                SteerCapability::CanSteer,
618            )
619            .expect("admitted");
620        assert_eq!(journal.open_items("thread-1").len(), 1);
621        assert_eq!(journal.open_items("thread-2").len(), 1);
622        assert_eq!(
623            journal.open_items("thread-2")[0].text,
624            "different thread, same item id"
625        );
626    }
627
628    /// The disposition is derived from the stored parts on every read, so a
629    /// restored item cannot claim an outcome the law does not give it.
630    #[test]
631    fn a_restored_item_re_derives_its_disposition_rather_than_replaying_one() {
632        let dir = tempfile::tempdir().expect("temp dir");
633        let path = dir.path().join(SEND_QUEUE_JOURNAL_FILE);
634        {
635            let journal = SendQueueJournal::at(path.clone());
636            admit(
637                &journal,
638                "item-1",
639                SendCommand::Steer,
640                ExecutorClass::EngineLane,
641                SteerCapability::CanSteer,
642            );
643        }
644        let reopened = SendQueueJournal::at(path.clone());
645        let item = reopened.head_for("thread-1").expect("head");
646        assert_eq!(
647            item.disposition(),
648            SendDisposition::Refused {
649                refusal: SteerRefusal::EngineLaneIsRunAuthority,
650                fallback: SendFallback::HeldUntilQuiescent,
651            }
652        );
653        assert!(!item.disposition().reaches_running_turn());
654        // And the file itself holds no rendered disposition to disagree with.
655        let raw = std::fs::read_to_string(&path).expect("readable");
656        assert!(!raw.contains("held_until_quiescent"));
657        assert!(!raw.contains("Queued:"));
658    }
659
660    /// A journal it cannot read starts empty rather than pretending. The
661    /// schema check is what stops it adopting somebody else's file.
662    #[test]
663    fn a_foreign_document_is_refused_rather_than_adopted() {
664        let dir = tempfile::tempdir().expect("temp dir");
665        let path = dir.path().join(SEND_QUEUE_JOURNAL_FILE);
666        std::fs::write(
667            &path,
668            serde_json::to_vec_pretty(&serde_json::json!({
669                "schema": "openagents.omega.agent_route_journal.v1",
670                "items": [],
671            }))
672            .expect("json"),
673        )
674        .expect("written");
675        assert!(load(&path).is_err());
676        assert!(SendQueueJournal::at(path).open_items("thread-1").is_empty());
677    }
678
679    #[test]
680    fn a_cancelled_item_leaves_the_queue_and_stays_gone() {
681        let (journal, _dir) = journal();
682        admit(
683            &journal,
684            "item-1",
685            SendCommand::Enqueue,
686            ExecutorClass::NativeLoop,
687            SteerCapability::Unknown,
688        );
689        assert!(journal.cancel("thread-1", "item-1").is_ok());
690        assert!(journal.open_items("thread-1").is_empty());
691        assert_eq!(
692            journal.promote("thread-1", "item-1", Quiescence::Proven),
693            Err(SendQueueRefusal::ItemIsTerminal)
694        );
695        assert_eq!(
696            journal.cancel("thread-1", "item-1"),
697            Err(SendQueueRefusal::ItemIsTerminal)
698        );
699    }
700
701    /// Every state a queued item can be in says something different to the
702    /// person who queued it. A shared phrase would make two of them
703    /// indistinguishable in the composer.
704    #[test]
705    fn every_queue_state_says_something_a_reader_can_tell_apart() {
706        let (journal, _dir) = journal();
707        let mut phrases = Vec::new();
708        for (index, state) in QueueItemState::all().iter().enumerate() {
709            let item_id = format!("item-{index}");
710            let mut item = admit(
711                &journal,
712                &item_id,
713                SendCommand::Enqueue,
714                ExecutorClass::NativeLoop,
715                SteerCapability::Unknown,
716            );
717            item.state = *state;
718            phrases.push(queue_item_phrase(&item));
719        }
720        let unique: std::collections::BTreeSet<_> = phrases.iter().collect();
721        assert_eq!(unique.len(), phrases.len(), "{phrases:?} are not distinct");
722    }
723
724    /// The two commands are not the same command with a flag. An enqueue never
725    /// reaches the running turn on any class, and the record says which was
726    /// asked for.
727    #[test]
728    fn steer_and_enqueue_are_recorded_as_different_commands() {
729        let (journal, _dir) = journal();
730        let steered = admit(
731            &journal,
732            "item-steer",
733            SendCommand::Steer,
734            ExecutorClass::NativeLoop,
735            SteerCapability::Unknown,
736        );
737        let queued = admit(
738            &journal,
739            "item-queue",
740            SendCommand::Enqueue,
741            ExecutorClass::NativeLoop,
742            SteerCapability::Unknown,
743        );
744        assert_ne!(steered.command, queued.command);
745        assert!(steered.disposition().reaches_running_turn());
746        assert!(!queued.disposition().reaches_running_turn());
747    }
748}
749