tenant.openagents/omega

1//! What pressing send during a running turn does, per executor class (omega#79).
2//!
3//! ## The hole this closes at design time
4//!
5//! Every incumbent has the same concurrency defect: a second send arriving
6//! while a turn is running is quietly reinterpreted. Sometimes it steers the
7//! provider, sometimes it cancels and restarts, sometimes it is dropped — and
8//! the user cannot tell which happened, because the three look identical from
9//! the composer.
10//!
11//! Omega inherited a partial fix. `MessageQueue` already distinguishes a steer
12//! from an enqueue, but only the **native loop** ever learned about it:
13//! `sync_queue_flag_to_native_thread` sets the boundary flag on a
14//! `agent::Thread` and no-ops for anything else. An external ACP thread and an
15//! engine lane both fell through to cancel-then-send, which is the silent
16//! reinterpretation this packet exists to end. Two classes worked and one
17//! dropped.
18//!
19//! ## The shape
20//!
21//! [`disposition`] is a total function from (what the user asked for, which
22//! class is running it, what the peer said it can do) to a **declared** outcome.
23//! There is no `None`, no fallthrough, and no variant meaning "whatever the
24//! executor does". Every class has a stated answer, and every answer a user
25//! would experience as different is a different variant.
26//!
27//! Where a class cannot do what was asked, the answer is a typed refusal
28//! carrying its declared fallback ([`SendDisposition::Refused`]) rather than a
29//! quiet substitution. The user is told the steer was not available *and* what
30//! happened instead.
31//!
32//! ## Why an engine lane refuses a steer
33//!
34//! An engine lane **is** Full Auto authority (`OMEGA-DELTA-0029`), and its
35//! controls are bound to the run generation the Full Auto surface minted them
36//! for (`OMEGA-DELTA-0030`). A composer that could interrupt a run mid-flight
37//! would be a second place that believes it can command a run, reading a
38//! projection. So the engine lane's declared answer to a steer is a refusal
39//! with a durable-hold fallback, and its answer to an enqueue is the hold. That
40//! is a stated behavior, not an omission — which is the whole difference the
41//! exit asks about.
42//!
43//! This module is pure and clock-free, like [`super::router`]: same inputs,
44//! same disposition, every time. It starts nothing, and it holds no state.
45
46use crate::ExecutorClass;
47
48/// What the person asked for when they pressed send during a running turn.
49///
50/// These are two different intentions, and conflating them is the defect.
51/// "Interrupt what you are doing and take this into account" is not "run this
52/// next".
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub enum SendCommand {
55    /// Reach the running turn. The user wants the executor to change course.
56    Steer,
57    /// Do not reach the running turn. Run this after it finishes.
58    Enqueue,
59}
60
61impl SendCommand {
62    #[must_use]
63    pub const fn token(self) -> &'static str {
64        match self {
65            Self::Steer => "steer",
66            Self::Enqueue => "enqueue",
67        }
68    }
69
70    #[must_use]
71    pub const fn all() -> &'static [Self] {
72        &[Self::Steer, Self::Enqueue]
73    }
74
75    #[must_use]
76    pub fn parse_token(token: &str) -> Option<Self> {
77        match token {
78            "steer" => Some(Self::Steer),
79            "enqueue" => Some(Self::Enqueue),
80            _ => None,
81        }
82    }
83}
84
85/// What an executor said it can do with a send that arrives mid-turn.
86///
87/// Only [`ExecutorClass::ExternalAcp`] negotiates: the native loop's answer is
88/// fixed by its own turn loop, and an engine lane's is fixed by where run
89/// authority lives. An external peer is asked, and a peer that has not answered
90/// is [`Unknown`](Self::Unknown) — which is deliberately *not* the same as
91/// [`CannotSteer`](Self::CannotSteer). Treating silence as a capability is how
92/// a steer becomes a cancel.
93#[derive(Debug, Clone, Copy, PartialEq, Eq)]
94pub enum SteerCapability {
95    /// The peer declared it can take a mid-turn message.
96    CanSteer,
97    /// The peer declared it cannot.
98    CannotSteer,
99    /// The peer has not been asked, or did not answer.
100    Unknown,
101}
102
103impl SteerCapability {
104    #[must_use]
105    pub const fn token(self) -> &'static str {
106        match self {
107            Self::CanSteer => "can_steer",
108            Self::CannotSteer => "cannot_steer",
109            Self::Unknown => "unknown",
110        }
111    }
112
113    #[must_use]
114    pub const fn all() -> &'static [Self] {
115        &[Self::CanSteer, Self::CannotSteer, Self::Unknown]
116    }
117}
118
119/// Why a steer was not performed as asked.
120#[derive(Debug, Clone, Copy, PartialEq, Eq)]
121pub enum SteerRefusal {
122    /// The peer declared it cannot take a mid-turn message.
123    PeerCannotSteer,
124    /// The peer has not declared a steer capability. Not the same as declaring
125    /// that it cannot: Omega refuses rather than guessing, because guessing
126    /// wrong means cancelling somebody's turn.
127    PeerCapabilityUnknown,
128    /// An engine lane is Full Auto authority. Its run is commanded from the
129    /// surface that minted its controls, not from a thread composer.
130    EngineLaneIsRunAuthority,
131}
132
133impl SteerRefusal {
134    #[must_use]
135    pub const fn token(self) -> &'static str {
136        match self {
137            Self::PeerCannotSteer => "peer_cannot_steer",
138            Self::PeerCapabilityUnknown => "peer_capability_unknown",
139            Self::EngineLaneIsRunAuthority => "engine_lane_is_run_authority",
140        }
141    }
142
143    /// The sentence a user is shown. Derived, never stored — the same rule
144    /// [`crate::ExecutorDisclosure`] holds to.
145    #[must_use]
146    pub const fn phrase(self) -> &'static str {
147        match self {
148            Self::PeerCannotSteer => "this agent cannot take a message mid-turn",
149            Self::PeerCapabilityUnknown => {
150                "this agent has not said whether it can take a message mid-turn"
151            }
152            Self::EngineLaneIsRunAuthority => {
153                "a Full Auto run is steered from the run surface, not from here"
154            }
155        }
156    }
157
158    #[must_use]
159    pub const fn all() -> &'static [Self] {
160        &[
161            Self::PeerCannotSteer,
162            Self::PeerCapabilityUnknown,
163            Self::EngineLaneIsRunAuthority,
164        ]
165    }
166}
167
168/// What actually happens to a send that arrives during a running turn.
169///
170/// Every variant is something a user would experience as different. There is no
171/// variant meaning "it depends on the executor" — that is the state this type
172/// replaces.
173#[derive(Debug, Clone, Copy, PartialEq, Eq)]
174pub enum SendDisposition {
175    /// The running turn is ended at its next message boundary and the message
176    /// is delivered. The native loop's `end_turn_at_next_boundary`.
177    SteerAtMessageBoundary,
178    /// The message is handed to the running turn without ending it. What an
179    /// external peer that declared the capability does.
180    SteerInFlight,
181    /// The message is held until the prior turn is proven quiescent, then
182    /// promoted. Nothing reaches the running turn.
183    HeldUntilQuiescent,
184    /// The steer was not available. The fallback is stated, not implied.
185    Refused {
186        refusal: SteerRefusal,
187        fallback: SendFallback,
188    },
189}
190
191/// What happens instead, when a steer is refused.
192///
193/// A separate closed type so a refusal cannot be constructed without saying
194/// what it did with the message. "Refused" on its own would be the drop this
195/// packet exists to prevent.
196#[derive(Debug, Clone, Copy, PartialEq, Eq)]
197pub enum SendFallback {
198    /// Admitted to the durable queue and promoted after quiescence.
199    HeldUntilQuiescent,
200}
201
202impl SendFallback {
203    #[must_use]
204    pub const fn token(self) -> &'static str {
205        match self {
206            Self::HeldUntilQuiescent => "held_until_quiescent",
207        }
208    }
209}
210
211impl SendDisposition {
212    /// The stable token. Persisted and compared; the user sees
213    /// [`phrase`](Self::phrase).
214    #[must_use]
215    pub fn token(self) -> String {
216        match self {
217            Self::SteerAtMessageBoundary => "steer_at_message_boundary".to_owned(),
218            Self::SteerInFlight => "steer_in_flight".to_owned(),
219            Self::HeldUntilQuiescent => "held_until_quiescent".to_owned(),
220            Self::Refused { refusal, fallback } => {
221                format!("refused:{}:{}", refusal.token(), fallback.token())
222            }
223        }
224    }
225
226    /// Whether the running turn is reached at all.
227    ///
228    /// The single question the falsifier for this packet asks: "a second send
229    /// reaches a running provider turn without a guard".
230    #[must_use]
231    pub const fn reaches_running_turn(self) -> bool {
232        matches!(self, Self::SteerAtMessageBoundary | Self::SteerInFlight)
233    }
234
235    /// The line the composer shows. Derived from the parts on every call.
236    #[must_use]
237    pub fn phrase(self) -> String {
238        match self {
239            Self::SteerAtMessageBoundary => {
240                "Steering: the current turn ends at its next step.".to_owned()
241            }
242            Self::SteerInFlight => "Steering: sent to the running turn.".to_owned(),
243            Self::HeldUntilQuiescent => "Queued: sends after this turn finishes.".to_owned(),
244            Self::Refused { refusal, fallback } => match fallback {
245                SendFallback::HeldUntilQuiescent => {
246                    format!(
247                        "Not steered — {}. Queued: sends after this turn finishes.",
248                        refusal.phrase()
249                    )
250                }
251            },
252        }
253    }
254}
255
256/// The law. Total over every (command, class, capability) triple.
257///
258/// A caller cannot reach a case this does not answer, which is what makes
259/// "declared visible behavior on every executor class" checkable rather than
260/// asserted.
261#[must_use]
262pub const fn disposition(
263    command: SendCommand,
264    class: ExecutorClass,
265    capability: SteerCapability,
266) -> SendDisposition {
267    match command {
268        // An enqueue never reaches the running turn, on any class. This is the
269        // half that was already honest, and it stays uniform on purpose: a
270        // queued message that sometimes interrupts is the original defect.
271        SendCommand::Enqueue => SendDisposition::HeldUntilQuiescent,
272        SendCommand::Steer => match class {
273            // The native loop stops at a message boundary. It does not need to
274            // negotiate, because Omega owns both sides of that loop.
275            ExecutorClass::NativeLoop => SendDisposition::SteerAtMessageBoundary,
276            ExecutorClass::ExternalAcp => match capability {
277                SteerCapability::CanSteer => SendDisposition::SteerInFlight,
278                SteerCapability::CannotSteer => SendDisposition::Refused {
279                    refusal: SteerRefusal::PeerCannotSteer,
280                    fallback: SendFallback::HeldUntilQuiescent,
281                },
282                // Silence is not consent. A peer that never answered is not
283                // assumed able, because the cost of assuming wrong is the
284                // user's running turn.
285                SteerCapability::Unknown => SendDisposition::Refused {
286                    refusal: SteerRefusal::PeerCapabilityUnknown,
287                    fallback: SendFallback::HeldUntilQuiescent,
288                },
289            },
290            // An engine lane is Full Auto authority. Its capability is not
291            // consulted, because the answer does not depend on what the engine
292            // can do — it depends on where run authority lives.
293            ExecutorClass::EngineLane => SendDisposition::Refused {
294                refusal: SteerRefusal::EngineLaneIsRunAuthority,
295                fallback: SendFallback::HeldUntilQuiescent,
296            },
297        },
298    }
299}
300
301/// Where a queued message is in its life.
302///
303/// Distinct from a draft (not yet sent) and from a steer (reaches the turn).
304/// The exit asks for these to be visible and distinct, so they are a closed
305/// enum rather than a pair of booleans.
306#[derive(Debug, Clone, Copy, PartialEq, Eq)]
307pub enum QueueItemState {
308    /// Admitted durably. Acknowledged to the user only once this holds.
309    Queued,
310    /// Promoted to a turn after the prior turn was proven quiescent.
311    Promoted,
312    /// Withdrawn by the user before promotion.
313    Cancelled,
314    /// Promotion was attempted and did not start a turn.
315    Failed,
316}
317
318impl QueueItemState {
319    #[must_use]
320    pub const fn token(self) -> &'static str {
321        match self {
322            Self::Queued => "queued",
323            Self::Promoted => "promoted",
324            Self::Cancelled => "cancelled",
325            Self::Failed => "failed",
326        }
327    }
328
329    #[must_use]
330    pub fn parse_token(token: &str) -> Option<Self> {
331        match token {
332            "queued" => Some(Self::Queued),
333            "promoted" => Some(Self::Promoted),
334            "cancelled" => Some(Self::Cancelled),
335            "failed" => Some(Self::Failed),
336            _ => None,
337        }
338    }
339
340    #[must_use]
341    pub const fn all() -> &'static [Self] {
342        &[
343            Self::Queued,
344            Self::Promoted,
345            Self::Cancelled,
346            Self::Failed,
347        ]
348    }
349
350    /// Whether this item can still be promoted.
351    ///
352    /// A terminal item that promoted again is the duplicate the falsifier for
353    /// reconnect and restart names.
354    #[must_use]
355    pub const fn is_open(self) -> bool {
356        matches!(self, Self::Queued)
357    }
358}
359
360/// Whether the prior turn is proven finished, or merely believed to be.
361///
362/// A scheduler that promotes on "believed" is how a queued message races the
363/// turn it was supposed to follow. Only [`Proven`](Self::Proven) promotes.
364#[derive(Debug, Clone, Copy, PartialEq, Eq)]
365pub enum Quiescence {
366    /// The executor reported the turn stopped.
367    Proven,
368    /// A turn is running.
369    Running,
370    /// The connection dropped and no stop was observed. Not proof of anything.
371    Unknown,
372}
373
374impl Quiescence {
375    #[must_use]
376    pub const fn token(self) -> &'static str {
377        match self {
378            Self::Proven => "proven",
379            Self::Running => "running",
380            Self::Unknown => "unknown",
381        }
382    }
383
384    #[must_use]
385    pub const fn all() -> &'static [Self] {
386        &[Self::Proven, Self::Running, Self::Unknown]
387    }
388}
389
390/// Whether the head of the queue may be promoted right now.
391///
392/// One thread-owned scheduler asks this. It is deliberately conservative about
393/// [`Quiescence::Unknown`]: after a reconnect Omega has not seen the prior turn
394/// stop, and promoting there is exactly the duplicate the falsifier describes.
395#[must_use]
396pub const fn may_promote(state: QueueItemState, quiescence: Quiescence) -> bool {
397    matches!(
398        (state, quiescence),
399        (QueueItemState::Queued, Quiescence::Proven)
400    )
401}
402
403#[cfg(test)]
404mod tests {
405    use super::*;
406
407    /// The exit, as a check. Every class answers every command, and the answers
408    /// are not all the same — a uniform answer would mean the classes were not
409    /// really consulted.
410    #[test]
411    fn every_executor_class_has_a_declared_answer_to_a_send_during_a_turn() {
412        let mut dispositions = Vec::new();
413        for command in SendCommand::all() {
414            for class in ExecutorClass::all() {
415                for capability in SteerCapability::all() {
416                    let decided = disposition(*command, *class, *capability);
417                    assert!(
418                        !decided.phrase().is_empty(),
419                        "{}/{}/{} has no phrase to show",
420                        command.token(),
421                        class.token(),
422                        capability.token()
423                    );
424                    dispositions.push(decided);
425                }
426            }
427        }
428        assert_eq!(dispositions.len(), 2 * 3 * 3);
429        // Each class contributes at least one answer no other class gives, so a
430        // reader cannot conclude the class was ignored.
431        for class in ExecutorClass::all() {
432            let mine: Vec<_> = SteerCapability::all()
433                .iter()
434                .map(|capability| disposition(SendCommand::Steer, *class, *capability))
435                .collect();
436            let others: Vec<_> = ExecutorClass::all()
437                .iter()
438                .filter(|other| *other != class)
439                .flat_map(|other| {
440                    SteerCapability::all()
441                        .iter()
442                        .map(move |capability| disposition(SendCommand::Steer, *other, *capability))
443                })
444                .collect();
445            assert!(
446                mine.iter().any(|decided| !others.contains(decided)),
447                "{} gives no answer distinct from the other classes",
448                class.token()
449            );
450        }
451    }
452
453    /// The named falsifier: "a second send reaches a running provider turn
454    /// without a guard".
455    #[test]
456    fn nothing_reaches_a_running_turn_without_an_explicit_steer() {
457        for class in ExecutorClass::all() {
458            for capability in SteerCapability::all() {
459                assert!(
460                    !disposition(SendCommand::Enqueue, *class, *capability).reaches_running_turn(),
461                    "an enqueue reached the running turn on {}",
462                    class.token()
463                );
464            }
465        }
466    }
467
468    /// Silence is not a capability. This is the case that turns a steer into a
469    /// cancelled turn on a peer nobody asked.
470    #[test]
471    fn an_undeclared_peer_is_refused_rather_than_assumed_able() {
472        let decided = disposition(
473            SendCommand::Steer,
474            ExecutorClass::ExternalAcp,
475            SteerCapability::Unknown,
476        );
477        assert!(!decided.reaches_running_turn());
478        assert_eq!(
479            decided,
480            SendDisposition::Refused {
481                refusal: SteerRefusal::PeerCapabilityUnknown,
482                fallback: SendFallback::HeldUntilQuiescent,
483            }
484        );
485    }
486
487    /// An engine lane is Full Auto authority. `OMEGA-DELTA-0029` reaches it only
488    /// through an explicit pin, and `OMEGA-DELTA-0030` keeps run commands bound
489    /// to the generation the run surface minted. A composer steer would be a
490    /// second commanding surface.
491    #[test]
492    fn an_engine_lane_steer_is_refused_whatever_the_engine_can_do() {
493        for capability in SteerCapability::all() {
494            let decided = disposition(
495                SendCommand::Steer,
496                ExecutorClass::EngineLane,
497                *capability,
498            );
499            assert_eq!(
500                decided,
501                SendDisposition::Refused {
502                    refusal: SteerRefusal::EngineLaneIsRunAuthority,
503                    fallback: SendFallback::HeldUntilQuiescent,
504                },
505                "capability {} changed an engine lane's answer",
506                capability.token()
507            );
508            assert!(!decided.reaches_running_turn());
509        }
510    }
511
512    /// A refusal that did not say what happened to the message is a drop with
513    /// better manners. The type has no way to express one, and the rendered
514    /// line always carries both halves.
515    #[test]
516    fn every_refusal_states_what_happened_to_the_message() {
517        for class in ExecutorClass::all() {
518            for capability in SteerCapability::all() {
519                let decided = disposition(SendCommand::Steer, *class, *capability);
520                if let SendDisposition::Refused { refusal, fallback } = decided {
521                    let phrase = decided.phrase();
522                    assert!(phrase.contains(refusal.phrase()), "{phrase} hides the reason");
523                    assert!(
524                        phrase.contains("Queued"),
525                        "{phrase} does not say the message was queued"
526                    );
527                    assert_eq!(fallback, SendFallback::HeldUntilQuiescent);
528                }
529            }
530        }
531    }
532
533    /// Only a proven stop promotes. Unknown is what a reconnect leaves behind,
534    /// and promoting there is the duplicate the acceptance names.
535    #[test]
536    fn only_a_proven_stop_promotes_the_queue_head() {
537        assert!(may_promote(QueueItemState::Queued, Quiescence::Proven));
538        for quiescence in Quiescence::all() {
539            for state in QueueItemState::all() {
540                let promoted = may_promote(*state, *quiescence);
541                if promoted {
542                    assert_eq!(*state, QueueItemState::Queued);
543                    assert_eq!(*quiescence, Quiescence::Proven);
544                }
545                if !state.is_open() {
546                    assert!(
547                        !promoted,
548                        "{} promoted a second time",
549                        state.token()
550                    );
551                }
552            }
553        }
554    }
555
556    #[test]
557    fn every_token_round_trips_and_is_distinct() {
558        for command in SendCommand::all() {
559            assert_eq!(SendCommand::parse_token(command.token()), Some(*command));
560        }
561        for state in QueueItemState::all() {
562            assert_eq!(QueueItemState::parse_token(state.token()), Some(*state));
563        }
564        let mut tokens = Vec::new();
565        for command in SendCommand::all() {
566            for class in ExecutorClass::all() {
567                for capability in SteerCapability::all() {
568                    tokens.push(disposition(*command, *class, *capability).token());
569                }
570            }
571        }
572        tokens.sort();
573        let before = tokens.len();
574        tokens.dedup();
575        assert!(tokens.len() < before, "the triples should collapse to a smaller answer set");
576        assert!(
577            tokens.contains(&"steer_at_message_boundary".to_owned()),
578            "the native loop's boundary stop is not reachable"
579        );
580    }
581}
582