meerkat_runtime/meerkat_machine/
comms_drain.rs

1use super::*;
2
3use std::fmt;
4
5/// Phase of the comms drain slot.
6///
7/// Shell-side mechanics tracking for the runtime-owned drain task. The DSL's
8/// `drain_phase` is the canonical lifecycle authority; this slot phase is the
9/// mechanical companion that tracks whether a tokio `JoinHandle` is in flight.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum CommsDrainPhase {
12    Inactive,
13    Starting,
14    Running,
15    ExitedRespawnable,
16    Stopped,
17}
18
19impl fmt::Display for CommsDrainPhase {
20    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
21        match self {
22            Self::Inactive => write!(f, "Inactive"),
23            Self::Starting => write!(f, "Starting"),
24            Self::Running => write!(f, "Running"),
25            Self::ExitedRespawnable => write!(f, "ExitedRespawnable"),
26            Self::Stopped => write!(f, "Stopped"),
27        }
28    }
29}
30
31/// Mode for the comms drain task.
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub enum CommsDrainMode {
34    /// Legacy timed drain with idle timeout.
35    Timed,
36    /// Live session ingress while a runtime-backed session is attached.
37    AttachedSession,
38    /// Long-lived host drain (no idle timeout, respawnable on failure).
39    PersistentHost,
40}
41
42/// Reason the drain task exited.
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub enum DrainExitReason {
45    IdleTimeout,
46    Dismissed,
47    Failed,
48    Aborted,
49    SessionShutdown,
50}
51
52impl From<DrainExitReason> for crate::meerkat_machine::dsl::DrainExitReason {
53    fn from(reason: DrainExitReason) -> Self {
54        match reason {
55            DrainExitReason::IdleTimeout => Self::IdleTimeout,
56            DrainExitReason::Dismissed => Self::Dismissed,
57            DrainExitReason::Failed => Self::Failed,
58            DrainExitReason::Aborted => Self::Aborted,
59            DrainExitReason::SessionShutdown => Self::SessionShutdown,
60        }
61    }
62}
63
64impl From<DrainExitReason> for meerkat_core::handles::DrainExitReason {
65    fn from(reason: DrainExitReason) -> Self {
66        match reason {
67            DrainExitReason::IdleTimeout => Self::IdleTimeout,
68            DrainExitReason::Dismissed => Self::Dismissed,
69            DrainExitReason::Failed => Self::Failed,
70            DrainExitReason::Aborted => Self::Aborted,
71            DrainExitReason::SessionShutdown => Self::SessionShutdown,
72        }
73    }
74}
75
76impl From<crate::meerkat_machine::dsl::DrainPhase> for CommsDrainPhase {
77    fn from(phase: crate::meerkat_machine::dsl::DrainPhase) -> Self {
78        match phase {
79            crate::meerkat_machine::dsl::DrainPhase::Inactive => Self::Inactive,
80            crate::meerkat_machine::dsl::DrainPhase::Running => Self::Running,
81            crate::meerkat_machine::dsl::DrainPhase::Stopped => Self::Stopped,
82            crate::meerkat_machine::dsl::DrainPhase::ExitedRespawnable => Self::ExitedRespawnable,
83        }
84    }
85}
86
87impl From<crate::meerkat_machine::dsl::DrainMode> for CommsDrainMode {
88    fn from(mode: crate::meerkat_machine::dsl::DrainMode) -> Self {
89        match mode {
90            crate::meerkat_machine::dsl::DrainMode::Timed => Self::Timed,
91            crate::meerkat_machine::dsl::DrainMode::AttachedSession => Self::AttachedSession,
92            crate::meerkat_machine::dsl::DrainMode::PersistentHost => Self::PersistentHost,
93        }
94    }
95}
96
97/// Typed view of the peer-ingress transport capability owner (W2-G).
98///
99/// Projected from the DSL's tagged-union state
100/// (`peer_ingress_owner_kind` + companion fields). The
101/// `peer_ingress_owner_consistency` invariant guarantees the companion
102/// fields are populated exactly for variants that name them.
103#[derive(Debug, Clone, PartialEq, Eq)]
104#[non_exhaustive]
105pub enum PeerIngressOwner {
106    Unattached,
107    SessionOwned {
108        comms_runtime_id: crate::meerkat_machine::dsl::CommsRuntimeId,
109    },
110    MobOwned {
111        comms_runtime_id: crate::meerkat_machine::dsl::CommsRuntimeId,
112        mob_id: crate::meerkat_machine::dsl::MobId,
113    },
114}
115
116impl PeerIngressOwner {
117    /// Returns `true` iff the owner is `MobOwned`.
118    pub fn is_mob_owned(&self) -> bool {
119        matches!(self, PeerIngressOwner::MobOwned { .. })
120    }
121}
122
123/// Typed view of the per-session supervisor-bridge binding (Wave 3 D Row 21).
124///
125/// Projected from the DSL's tagged-union state
126/// (`supervisor_binding_kind` +
127/// `supervisor_bound_{name, peer_id, address, signing_public_key, epoch}`).
128/// The `supervisor_binding_consistency` invariant guarantees the companion
129/// fields are populated exactly when the kind is `Bound`.
130#[derive(Debug, Clone, PartialEq, Eq)]
131#[non_exhaustive]
132pub enum SupervisorBinding {
133    /// No supervisor bound. The initial state and the state after a
134    /// successful `RevokeSupervisor`.
135    Unbound,
136    /// Supervisor authorized. The companion fields travel together:
137    /// `name` + `peer_id` + `address` + `signing_public_key` derive from the
138    /// initial bind or the latest `AuthorizeSupervisor` rotation; `epoch`
139    /// monotonically increases across rotations.
140    Bound {
141        name: String,
142        peer_id: String,
143        address: String,
144        signing_public_key: String,
145        epoch: u64,
146    },
147}
148
149pub struct CommsDrainSlot {
150    handle: Option<tokio::task::JoinHandle<()>>,
151    task_runtime: Option<Arc<dyn meerkat_core::agent::CommsRuntime>>,
152}
153
154impl CommsDrainSlot {
155    pub fn new() -> Self {
156        Self {
157            handle: None,
158            task_runtime: None,
159        }
160    }
161
162    pub(crate) fn task_runtime_matches(
163        &self,
164        runtime: &Arc<dyn meerkat_core::agent::CommsRuntime>,
165    ) -> bool {
166        self.task_runtime
167            .as_ref()
168            .is_some_and(|current| Arc::ptr_eq(current, runtime))
169    }
170
171    pub(crate) fn task_runtime(&self) -> Option<Arc<dyn meerkat_core::agent::CommsRuntime>> {
172        self.task_runtime.clone()
173    }
174
175    pub(crate) fn handle_present(&self) -> bool {
176        self.handle.is_some()
177    }
178
179    pub(crate) fn install_task(
180        &mut self,
181        runtime: Arc<dyn meerkat_core::agent::CommsRuntime>,
182        handle: tokio::task::JoinHandle<()>,
183    ) {
184        if let Some(existing) = self.handle.take() {
185            existing.abort();
186        }
187        self.task_runtime = Some(runtime);
188        self.handle = Some(handle);
189    }
190
191    pub(crate) fn take_handle(&mut self) -> Option<tokio::task::JoinHandle<()>> {
192        self.handle.take()
193    }
194
195    pub(crate) fn clear_after_exit(&mut self, keep_runtime: bool) {
196        self.handle.take();
197        if !keep_runtime {
198            self.task_runtime = None;
199        }
200    }
201
202    pub(crate) fn abort(&mut self) {
203        self.task_runtime = None;
204        if let Some(handle) = self.handle.take() {
205            handle.abort();
206        }
207    }
208
209    /// Signal cancellation on the drain task and return its `JoinHandle` so the
210    /// caller can await quiescence. The two-phase unregister drain uses this to
211    /// abort the drain task *and* join it before committing teardown; awaiting
212    /// the returned handle yields `Err(JoinError::is_cancelled())`, which is
213    /// benign. The slot is left empty (no runtime, no handle).
214    pub(crate) fn abort_keeping_handle(&mut self) -> Option<tokio::task::JoinHandle<()>> {
215        self.task_runtime = None;
216        let handle = self.handle.take()?;
217        handle.abort();
218        Some(handle)
219    }
220}
221
222pub fn abort_slot(slot: &mut CommsDrainSlot) {
223    slot.abort();
224}
225
226#[derive(Debug, Clone)]
227pub(super) struct DrainAuthorityState {
228    pub phase: crate::meerkat_machine::dsl::DrainPhase,
229    pub mode: Option<crate::meerkat_machine::dsl::DrainMode>,
230    pub peer_owner_kind: crate::meerkat_machine::dsl::PeerIngressOwnerKind,
231    pub peer_runtime_id: Option<crate::meerkat_machine::dsl::CommsRuntimeId>,
232}
233
234impl DrainAuthorityState {
235    pub(super) fn can_spawn(&self) -> bool {
236        matches!(
237            self.phase,
238            crate::meerkat_machine::dsl::DrainPhase::Inactive
239                | crate::meerkat_machine::dsl::DrainPhase::Stopped
240                | crate::meerkat_machine::dsl::DrainPhase::ExitedRespawnable
241        )
242    }
243
244    pub(super) fn has_peer_runtime(
245        &self,
246        runtime_id: &crate::meerkat_machine::dsl::CommsRuntimeId,
247    ) -> bool {
248        self.peer_owner_kind != crate::meerkat_machine::dsl::PeerIngressOwnerKind::Unattached
249            && self.peer_runtime_id.as_ref() == Some(runtime_id)
250    }
251}
252
253impl MeerkatMachine {
254    async fn apply_supervisor_binding_input(
255        &self,
256        session_id: &SessionId,
257        input: crate::meerkat_machine::dsl::MeerkatMachineInput,
258    ) -> Result<crate::meerkat_machine::dsl::MeerkatMachineTransition, SupervisorBindingStageError>
259    {
260        #[cfg(target_arch = "wasm32")]
261        let mut sessions = self
262            .sessions
263            .try_write()
264            .map_err(|_| SupervisorBindingStageError::SessionRegistryBusy)?;
265        #[cfg(not(target_arch = "wasm32"))]
266        let mut sessions = self.sessions.write().await;
267
268        let entry = sessions
269            .get_mut(session_id)
270            .ok_or(SupervisorBindingStageError::SessionNotRegistered)?;
271
272        #[cfg(target_arch = "wasm32")]
273        let mut authority = entry
274            .dsl_authority
275            .try_lock()
276            .map_err(|_| SupervisorBindingStageError::SessionAuthorityBusy)?;
277        #[cfg(not(target_arch = "wasm32"))]
278        let mut authority = entry
279            .dsl_authority
280            .lock()
281            .unwrap_or_else(std::sync::PoisonError::into_inner);
282
283        crate::meerkat_machine::dsl::MeerkatMachineMutator::apply(&mut *authority, input)
284            .map_err(SupervisorBindingStageError::Dsl)
285    }
286
287    pub async fn update_peer_ingress_context(
288        self: &Arc<Self>,
289        session_id: &SessionId,
290        keep_alive: bool,
291        comms_runtime: Option<Arc<dyn meerkat_core::agent::CommsRuntime>>,
292    ) -> Result<bool, RuntimeDriverError> {
293        match self
294            .execute_meerkat_machine_drain_command(MeerkatMachineCommand::SetPeerIngressContext {
295                session_id: session_id.clone(),
296                keep_alive,
297                comms_runtime,
298                mob_id: None,
299            })
300            .await?
301        {
302            MeerkatMachineCommandResult::Spawned(spawned) => Ok(spawned),
303            other => Err(RuntimeDriverError::Internal(format!(
304                "update_peer_ingress_context: unexpected command result variant: {other:?}"
305            ))),
306        }
307    }
308
309    /// Manage the comms drain lifecycle for a session based on keep_alive intent.
310    ///
311    /// When `keep_alive` is true, spawns a drain if one is not already running.
312    /// When `keep_alive` is false, aborts any running drain for the session.
313    /// Returns `true` if a new drain was spawned.
314    pub async fn maybe_spawn_comms_drain(
315        self: &Arc<Self>,
316        session_id: &SessionId,
317        keep_alive: bool,
318        comms_runtime: Option<Arc<dyn meerkat_core::agent::CommsRuntime>>,
319    ) -> Result<bool, RuntimeDriverError> {
320        match self
321            .execute_meerkat_machine_drain_command(MeerkatMachineCommand::SetPeerIngressContext {
322                session_id: session_id.clone(),
323                keep_alive,
324                comms_runtime,
325                mob_id: None,
326            })
327            .await?
328        {
329            MeerkatMachineCommandResult::Spawned(spawned) => Ok(spawned),
330            other => Err(RuntimeDriverError::Internal(format!(
331                "maybe_spawn_comms_drain: unexpected command result variant: {other:?}"
332            ))),
333        }
334    }
335
336    /// Refresh a session-owned peer ingress drain without re-attaching
337    /// ownership.
338    ///
339    /// This is intentionally narrower than
340    /// [`MeerkatMachine::update_peer_ingress_context`]: callers that only need
341    /// the existing authorized session-owned transport to be healthy can
342    /// respawn a missing/respawnable drain task without staging
343    /// `AttachSessionIngress` with a possibly different runtime handle.
344    /// Mob-owned ingress and missing cached session runtimes are left
345    /// untouched.
346    pub async fn refresh_session_owned_peer_ingress(
347        self: &Arc<Self>,
348        session_id: &SessionId,
349    ) -> Result<bool, RuntimeDriverError> {
350        if !self.sessions.read().await.contains_key(session_id) {
351            return Err(RuntimeDriverError::NotReady {
352                state: RuntimeState::Destroyed,
353            });
354        }
355        if matches!(
356            self.existing_session_runtime_state(session_id).await,
357            Some(RuntimeState::Destroyed)
358        ) {
359            return Err(RuntimeDriverError::Destroyed);
360        }
361
362        let gate = self.session_mutation_gate(session_id).await;
363        let _gate_guard = match gate {
364            Some(ref g) => Some(g.lock().await),
365            None => None,
366        };
367
368        let Some(comms_runtime) = self.session_owned_drain_runtime(session_id).await else {
369            return Ok(false);
370        };
371
372        self.update_peer_ingress_context_inner(session_id, true, Some(comms_runtime))
373            .await
374    }
375
376    /// Mob-owned variant of [`MeerkatMachine::maybe_spawn_comms_drain`]
377    /// (W2-G / issue #264).
378    ///
379    /// Shell calls this from the mob provisioning path to claim peer-ingress
380    /// ownership as `MobOwned { comms_runtime_id, mob_id }`. The DSL
381    /// transition permits promotion from `Unattached` or `SessionOwned`, so
382    /// a mob can take over a session-owned drain at spawn; silent downgrades
383    /// back to `SessionOwned` are impossible by construction.
384    pub async fn maybe_spawn_mob_comms_drain(
385        self: &Arc<Self>,
386        session_id: &SessionId,
387        comms_runtime: Arc<dyn meerkat_core::agent::CommsRuntime>,
388        mob_id: crate::meerkat_machine::dsl::MobId,
389    ) -> Result<bool, RuntimeDriverError> {
390        match self
391            .execute_meerkat_machine_drain_command(MeerkatMachineCommand::SetPeerIngressContext {
392                session_id: session_id.clone(),
393                keep_alive: true,
394                comms_runtime: Some(comms_runtime),
395                mob_id: Some(mob_id),
396            })
397            .await?
398        {
399            MeerkatMachineCommandResult::Spawned(spawned) => Ok(spawned),
400            other => Err(RuntimeDriverError::Internal(format!(
401                "maybe_spawn_mob_comms_drain: unexpected command result variant: {other:?}"
402            ))),
403        }
404    }
405
406    /// Read the current peer-ingress owner from DSL state.
407    ///
408    /// Returns `PeerIngressOwner::Unattached` for sessions that have no
409    /// registered DSL state (unknown / destroyed sessions). Used by the
410    /// session-runtime to refuse reconfiguration of mob-owned drains at
411    /// turn-start.
412    ///
413    /// The `peer_ingress_owner_consistency` invariant guarantees that
414    /// companion fields are populated for non-`Unattached` kinds, but if
415    /// the invariant were ever violated at runtime, we gracefully degrade
416    /// to `Unattached` rather than panic.
417    pub async fn peer_ingress_owner(&self, session_id: &SessionId) -> PeerIngressOwner {
418        let sessions = self.sessions.read().await;
419        let Some(entry) = sessions.get(session_id) else {
420            return PeerIngressOwner::Unattached;
421        };
422        let authority = entry
423            .dsl_authority
424            .lock()
425            .unwrap_or_else(std::sync::PoisonError::into_inner);
426        match authority.state().peer_ingress_owner_kind {
427            crate::meerkat_machine::dsl::PeerIngressOwnerKind::Unattached => {
428                PeerIngressOwner::Unattached
429            }
430            crate::meerkat_machine::dsl::PeerIngressOwnerKind::SessionOwned => {
431                match authority.state().peer_ingress_comms_runtime_id.clone() {
432                    Some(comms_runtime_id) => PeerIngressOwner::SessionOwned { comms_runtime_id },
433                    None => {
434                        tracing::error!(
435                            %session_id,
436                            "peer_ingress_owner_consistency invariant violation: SessionOwned without comms_runtime_id"
437                        );
438                        PeerIngressOwner::Unattached
439                    }
440                }
441            }
442            crate::meerkat_machine::dsl::PeerIngressOwnerKind::MobOwned => {
443                match (
444                    authority.state().peer_ingress_comms_runtime_id.clone(),
445                    authority.state().peer_ingress_mob_id.clone(),
446                ) {
447                    (Some(comms_runtime_id), Some(mob_id)) => PeerIngressOwner::MobOwned {
448                        comms_runtime_id,
449                        mob_id,
450                    },
451                    _ => {
452                        tracing::error!(
453                            %session_id,
454                            "peer_ingress_owner_consistency invariant violation: MobOwned without companion fields"
455                        );
456                        PeerIngressOwner::Unattached
457                    }
458                }
459            }
460        }
461    }
462
463    async fn session_owned_drain_runtime(
464        &self,
465        session_id: &SessionId,
466    ) -> Option<Arc<dyn meerkat_core::agent::CommsRuntime>> {
467        let sessions = self.sessions.read().await;
468        let entry = sessions.get(session_id)?;
469        let authority = entry
470            .dsl_authority
471            .lock()
472            .unwrap_or_else(std::sync::PoisonError::into_inner);
473        let state = authority.state();
474        if state.peer_ingress_owner_kind
475            != crate::meerkat_machine::dsl::PeerIngressOwnerKind::SessionOwned
476        {
477            return None;
478        }
479        let expected_runtime_id = state.peer_ingress_comms_runtime_id.as_ref()?;
480        let runtime = entry.drain_slot.task_runtime()?;
481        let actual_runtime_id = crate::meerkat_machine::dsl::CommsRuntimeId::from_runtime(&runtime);
482        (expected_runtime_id == &actual_runtime_id).then_some(runtime)
483    }
484
485    pub(super) async fn drain_authority_state(
486        &self,
487        session_id: &SessionId,
488    ) -> Option<DrainAuthorityState> {
489        let sessions = self.sessions.read().await;
490        let entry = sessions.get(session_id)?;
491        let authority = entry
492            .dsl_authority
493            .lock()
494            .unwrap_or_else(std::sync::PoisonError::into_inner);
495        let state = authority.state();
496        Some(DrainAuthorityState {
497            phase: state.drain_phase,
498            mode: state.drain_mode,
499            peer_owner_kind: state.peer_ingress_owner_kind,
500            peer_runtime_id: state.peer_ingress_comms_runtime_id.clone(),
501        })
502    }
503
504    pub(super) async fn update_peer_ingress_context_inner(
505        self: &Arc<Self>,
506        session_id: &SessionId,
507        keep_alive: bool,
508        comms_runtime: Option<Arc<dyn meerkat_core::agent::CommsRuntime>>,
509    ) -> Result<bool, RuntimeDriverError> {
510        if !keep_alive {
511            // Explicit disable: stop any running drain for this session. A
512            // failed abort is a typed control fault — the drain would keep
513            // running while the caller believes keep-alive was disabled — so
514            // it propagates instead of being silently discarded.
515            self.execute_meerkat_machine_drain_local_command(MeerkatMachineCommand::Abort {
516                session_id: session_id.clone(),
517            })
518            .await?;
519            return Ok(false);
520        }
521
522        let mode = CommsDrainMode::PersistentHost;
523
524        let comms = match comms_runtime {
525            Some(c) => c,
526            None => return Ok(false),
527        };
528
529        let runtime_id = crate::meerkat_machine::dsl::CommsRuntimeId::from_runtime(&comms);
530        let Some(authority_state) = self.drain_authority_state(session_id).await else {
531            tracing::warn!(
532                %session_id,
533                "refusing to spawn comms drain without generated drain authority"
534            );
535            return Ok(false);
536        };
537        if !authority_state.has_peer_runtime(&runtime_id) {
538            tracing::warn!(
539                %session_id,
540                "refusing to spawn comms drain without matching generated peer-ingress authority"
541            );
542            return Ok(false);
543        }
544
545        let dsl_mode = crate::meerkat_machine::dsl::DrainMode::from(mode);
546        let needs_spawn = authority_state.can_spawn();
547        let needs_task_refresh = if needs_spawn {
548            false
549        } else if authority_state.phase == crate::meerkat_machine::dsl::DrainPhase::Running
550            && authority_state.mode == Some(dsl_mode)
551        {
552            let sessions = self.sessions.read().await;
553            let Some(entry) = sessions.get(session_id) else {
554                tracing::warn!(
555                    %session_id,
556                    "refusing to spawn comms drain for unregistered session"
557                );
558                return Ok(false);
559            };
560            !entry.drain_slot.handle_present() || !entry.drain_slot.task_runtime_matches(&comms)
561        } else {
562            false
563        };
564
565        if !needs_spawn && !needs_task_refresh {
566            return Ok(false);
567        }
568
569        if needs_spawn {
570            // Stage DSL SpawnDrain only when the machine is transitioning from
571            // not-running into running. A runtime-instance refresh keeps the
572            // conceptual drain alive and only swaps the mechanical task after
573            // peer-ingress authority has accepted the runtime identity.
574            if let Err(err) = self
575                .stage_session_dsl_input(
576                    session_id,
577                    crate::meerkat_machine::dsl::MeerkatMachineInput::SpawnDrain { mode: dsl_mode },
578                    "SpawnDrain",
579                )
580                .await
581            {
582                tracing::warn!(
583                    %session_id,
584                    error = %err,
585                    "DSL rejected SpawnDrain; skipping drain spawn"
586                );
587                return Ok(false);
588            }
589        } else if needs_task_refresh {
590            tracing::warn!(
591                %session_id,
592                "refreshing persistent comms drain task from generated peer-ingress authority"
593            );
594        }
595
596        let idle_timeout = match mode {
597            CommsDrainMode::PersistentHost => Some(std::time::Duration::MAX),
598            CommsDrainMode::Timed | CommsDrainMode::AttachedSession => None,
599        };
600        let handle = crate::comms_drain::spawn_comms_drain(
601            Arc::clone(self),
602            session_id.clone(),
603            comms.clone(),
604            idle_timeout,
605        );
606        let mut sessions = self.sessions.write().await;
607        if let Some(entry) = sessions.get_mut(session_id) {
608            entry.drain_slot.install_task(comms.clone(), handle);
609        } else {
610            handle.abort();
611            return Ok(false);
612        }
613
614        Ok(true)
615    }
616
617    /// Notify the authority that a drain task has exited with the given reason.
618    ///
619    /// Called from drain task exit paths (or by wrappers that detect task
620    /// completion). The generated `NotifyDrainExited` input owns whether the
621    /// exit enters `ExitedRespawnable` or `Stopped`; this method only projects
622    /// the accepted authority state into task-handle mechanics.
623    pub async fn notify_comms_drain_exited(
624        self: &Arc<Self>,
625        session_id: &SessionId,
626        reason: DrainExitReason,
627    ) -> Result<(), RuntimeDriverError> {
628        self.execute_meerkat_machine_command(
629            Some(Arc::clone(self)),
630            MeerkatMachineCommand::NotifyDrainExited {
631                session_id: session_id.clone(),
632                reason,
633            },
634        )
635        .await
636        .map_err(MeerkatMachine::driver_error_from_command_error)?;
637        Ok(())
638    }
639
640    pub(super) async fn notify_comms_drain_exited_inner(
641        &self,
642        session_id: &SessionId,
643        reason: DrainExitReason,
644    ) {
645        let keep_runtime = self
646            .drain_authority_state(session_id)
647            .await
648            .is_some_and(|state| {
649                state.phase == crate::meerkat_machine::dsl::DrainPhase::ExitedRespawnable
650            });
651        let mut sessions = self.sessions.write().await;
652        if let Some(entry) = sessions.get_mut(session_id) {
653            entry.drain_slot.clear_after_exit(keep_runtime);
654        }
655        if std::env::var_os("RKAT_TRACE_COMMS_DRAIN_BIND").is_some() {
656            tracing::info!(
657                %session_id,
658                ?reason,
659                respawnable = keep_runtime,
660                "comms drain exited"
661            );
662        }
663    }
664
665    pub(crate) async fn project_comms_drain_failed_safety_net(&self, session_id: &SessionId) {
666        let keep_runtime = match self.drain_authority_state(session_id).await {
667            Some(state) => {
668                state.phase == crate::meerkat_machine::dsl::DrainPhase::ExitedRespawnable
669            }
670            None => false,
671        };
672        let mut sessions = self.sessions.write().await;
673        if let Some(entry) = sessions.get_mut(session_id) {
674            entry.drain_slot.clear_after_exit(keep_runtime);
675        }
676    }
677
678    /// Abort all active comms drain tasks.
679    pub async fn abort_comms_drains(&self) -> Result<(), RuntimeDriverError> {
680        self.execute_meerkat_machine_command(None, MeerkatMachineCommand::AbortAll)
681            .await
682            .map_err(MeerkatMachine::driver_error_from_command_error)?;
683        Ok(())
684    }
685
686    /// Abort the comms drain task for a specific session.
687    pub async fn abort_comms_drain(
688        &self,
689        session_id: &SessionId,
690    ) -> Result<(), RuntimeDriverError> {
691        self.execute_meerkat_machine_command(
692            None,
693            MeerkatMachineCommand::Abort {
694                session_id: session_id.clone(),
695            },
696        )
697        .await
698        .map_err(MeerkatMachine::driver_error_from_command_error)?;
699        Ok(())
700    }
701
702    /// Wait for a session's comms drain task to finish.
703    ///
704    /// Returns immediately if no drain is active for the session.
705    /// If the task already notified the authority (normal exit), this is a no-op
706    /// for authority state. If the task panicked without notifying, this submits
707    /// `TaskExited { Failed }` as a safety net.
708    pub async fn wait_comms_drain(&self, session_id: &SessionId) -> Result<(), RuntimeDriverError> {
709        self.execute_meerkat_machine_command(
710            None,
711            MeerkatMachineCommand::Wait {
712                session_id: session_id.clone(),
713            },
714        )
715        .await
716        .map_err(MeerkatMachine::driver_error_from_command_error)?;
717        Ok(())
718    }
719
720    /// Read the current supervisor binding from DSL state (Wave 3 D Row 21).
721    ///
722    /// Returns `SupervisorBinding::Unbound` for sessions that have no
723    /// registered DSL state (unknown / destroyed sessions). The
724    /// `supervisor_binding_consistency` invariant guarantees the four
725    /// companion fields are populated exactly when the kind is `Bound`; if
726    /// that invariant were ever violated at runtime, we gracefully degrade
727    /// to `Unbound` rather than panic.
728    pub async fn supervisor_binding(&self, session_id: &SessionId) -> SupervisorBinding {
729        let sessions = self.sessions.read().await;
730        let Some(entry) = sessions.get(session_id) else {
731            return SupervisorBinding::Unbound;
732        };
733        let authority = entry
734            .dsl_authority
735            .lock()
736            .unwrap_or_else(std::sync::PoisonError::into_inner);
737        match authority.state().supervisor_binding_kind {
738            crate::meerkat_machine::dsl::SupervisorBindingKind::Unbound => {
739                SupervisorBinding::Unbound
740            }
741            crate::meerkat_machine::dsl::SupervisorBindingKind::Bound => {
742                match (
743                    authority.state().supervisor_bound_name.clone(),
744                    authority.state().supervisor_bound_peer_id.clone(),
745                    authority.state().supervisor_bound_address.clone(),
746                    authority
747                        .state()
748                        .supervisor_bound_signing_public_key
749                        .clone(),
750                    authority.state().supervisor_bound_epoch,
751                ) {
752                    (
753                        Some(name),
754                        Some(peer_id),
755                        Some(address),
756                        Some(signing_public_key),
757                        Some(epoch),
758                    ) => SupervisorBinding::Bound {
759                        name,
760                        peer_id,
761                        address,
762                        signing_public_key,
763                        epoch,
764                    },
765                    _ => {
766                        tracing::error!(
767                            %session_id,
768                            "supervisor_binding_consistency invariant violation: Bound without all companion fields"
769                        );
770                        SupervisorBinding::Unbound
771                    }
772                }
773            }
774        }
775    }
776
777    fn local_endpoint_for_comms_runtime(
778        comms_runtime: &dyn meerkat_core::agent::CommsRuntime,
779    ) -> Result<crate::meerkat_machine::dsl::PeerEndpoint, String> {
780        let peer_id = comms_runtime
781            .peer_id()
782            .ok_or_else(|| "runtime peer_id unavailable".to_string())?;
783        let name = comms_runtime
784            .comms_name()
785            .ok_or_else(|| "runtime comms_name unavailable".to_string())?;
786        let address = comms_runtime
787            .advertised_address()
788            .ok_or_else(|| "runtime advertised_address unavailable".to_string())?;
789        let pubkey = comms_runtime
790            .public_key_bytes()
791            .ok_or_else(|| "runtime public_key_bytes unavailable".to_string())?;
792        Ok(crate::meerkat_machine::dsl::PeerEndpoint::new(
793            name,
794            peer_id.to_string(),
795            address,
796            pubkey,
797        ))
798    }
799
800    /// Publish the target runtime's own endpoint into MeerkatMachine before
801    /// generated trust handoffs mint authority scoped to that trust store.
802    pub async fn stage_local_endpoint_for_comms_runtime(
803        &self,
804        session_id: &SessionId,
805        comms_runtime: &dyn meerkat_core::agent::CommsRuntime,
806    ) -> Result<(), SupervisorBindingStageError> {
807        tracing::debug!(
808            %session_id,
809            "MeerkatMachine::stage_local_endpoint_for_comms_runtime building endpoint"
810        );
811        let endpoint = Self::local_endpoint_for_comms_runtime(comms_runtime)
812            .map_err(SupervisorBindingStageError::LocalEndpoint)?;
813        tracing::debug!(
814            %session_id,
815            "MeerkatMachine::stage_local_endpoint_for_comms_runtime built endpoint"
816        );
817        #[cfg(target_arch = "wasm32")]
818        let mut sessions = self
819            .sessions
820            .try_write()
821            .map_err(|_| SupervisorBindingStageError::SessionRegistryBusy)?;
822        #[cfg(not(target_arch = "wasm32"))]
823        let mut sessions = self.sessions.write().await;
824        let entry = sessions
825            .get_mut(session_id)
826            .ok_or(SupervisorBindingStageError::SessionNotRegistered)?;
827        #[cfg(target_arch = "wasm32")]
828        let mut authority = entry
829            .dsl_authority
830            .try_lock()
831            .map_err(|_| SupervisorBindingStageError::SessionAuthorityBusy)?;
832        #[cfg(not(target_arch = "wasm32"))]
833        let mut authority = entry
834            .dsl_authority
835            .lock()
836            .unwrap_or_else(std::sync::PoisonError::into_inner);
837        tracing::debug!(
838            %session_id,
839            "MeerkatMachine::stage_local_endpoint_for_comms_runtime applying endpoint"
840        );
841        if authority.state().local_endpoint.as_ref() == Some(&endpoint) {
842            tracing::debug!(
843                %session_id,
844                "MeerkatMachine::stage_local_endpoint_for_comms_runtime endpoint already applied"
845            );
846            return Ok(());
847        }
848        crate::meerkat_machine::dsl::MeerkatMachineMutator::apply(
849            &mut *authority,
850            crate::meerkat_machine::dsl::MeerkatMachineInput::PublishLocalEndpoint { endpoint },
851        )
852        .map_err(SupervisorBindingStageError::Dsl)?;
853        tracing::debug!(
854            %session_id,
855            "MeerkatMachine::stage_local_endpoint_for_comms_runtime applied endpoint"
856        );
857        Ok(())
858    }
859
860    /// Stage a DSL `BindSupervisor` input (Wave 3 D Row 21).
861    ///
862    /// Returns the classified result from the DSL mutator so callers can
863    /// surface typed rejections (e.g. "already bound"). The shell uses
864    /// this after validating the incoming bridge request's bootstrap
865    /// token; the DSL is the authority that flips `Unbound → Bound`.
866    pub async fn stage_supervisor_bind(
867        &self,
868        session_id: &SessionId,
869        name: String,
870        peer_id: String,
871        address: String,
872        signing_public_key: String,
873        epoch: u64,
874    ) -> Result<crate::meerkat_machine::dsl::MeerkatMachineTransition, SupervisorBindingStageError>
875    {
876        self.apply_supervisor_binding_input(
877            session_id,
878            crate::meerkat_machine::dsl::MeerkatMachineInput::BindSupervisor {
879                name,
880                peer_id,
881                address,
882                signing_public_key,
883                epoch,
884            },
885        )
886        .await
887    }
888
889    pub async fn supervisor_trust_publish_freshness_authority(
890        &self,
891        session_id: &SessionId,
892    ) -> Result<
893        crate::protocol_supervisor_trust_publish::SupervisorTrustFreshnessAuthority,
894        SupervisorBindingStageError,
895    > {
896        let sessions = self.sessions.read().await;
897        let entry = sessions
898            .get(session_id)
899            .ok_or(SupervisorBindingStageError::SessionNotRegistered)?;
900        Ok(
901            crate::protocol_supervisor_trust_publish::SupervisorTrustFreshnessAuthority::from_authority(
902                Arc::clone(&entry.dsl_authority),
903            ),
904        )
905    }
906
907    pub async fn supervisor_trust_revoke_freshness_authority(
908        &self,
909        session_id: &SessionId,
910    ) -> Result<
911        crate::protocol_supervisor_trust_revoke::SupervisorTrustFreshnessAuthority,
912        SupervisorBindingStageError,
913    > {
914        let sessions = self.sessions.read().await;
915        let entry = sessions
916            .get(session_id)
917            .ok_or(SupervisorBindingStageError::SessionNotRegistered)?;
918        Ok(
919            crate::protocol_supervisor_trust_revoke::SupervisorTrustFreshnessAuthority::from_authority(
920                Arc::clone(&entry.dsl_authority),
921            ),
922        )
923    }
924
925    /// Stage a DSL `AuthorizeSupervisor` input (Wave 3 D Row 21).
926    ///
927    /// Rotates the current binding to a new supervisor + epoch. The shell
928    /// must have already verified the rotation is authorized by the
929    /// *current* supervisor before calling this method.
930    pub async fn stage_supervisor_authorize(
931        &self,
932        session_id: &SessionId,
933        name: String,
934        peer_id: String,
935        address: String,
936        signing_public_key: String,
937        epoch: u64,
938    ) -> Result<crate::meerkat_machine::dsl::MeerkatMachineTransition, SupervisorBindingStageError>
939    {
940        self.apply_supervisor_binding_input(
941            session_id,
942            crate::meerkat_machine::dsl::MeerkatMachineInput::AuthorizeSupervisor {
943                name,
944                peer_id,
945                address,
946                signing_public_key,
947                epoch,
948            },
949        )
950        .await
951    }
952
953    /// Stage a DSL `RequestSupervisorTrustPublish` input.
954    ///
955    /// Used when the current supervisor binding is already correct but
956    /// the shell still needs a fresh generated publish obligation before
957    /// repairing or reasserting the concrete trust edge.
958    pub async fn stage_supervisor_trust_publish_request(
959        &self,
960        session_id: &SessionId,
961        name: String,
962        peer_id: String,
963        address: String,
964        signing_public_key: String,
965        epoch: u64,
966    ) -> Result<crate::meerkat_machine::dsl::MeerkatMachineTransition, SupervisorBindingStageError>
967    {
968        self.apply_supervisor_binding_input(
969            session_id,
970            crate::meerkat_machine::dsl::MeerkatMachineInput::RequestSupervisorTrustPublish {
971                name,
972                peer_id,
973                address,
974                signing_public_key,
975                epoch,
976            },
977        )
978        .await
979    }
980
981    /// Stage a DSL `RevokeSupervisor` input (Wave 3 D Row 21).
982    ///
983    /// Returns to `Unbound`. The DSL guard enforces that the supplied
984    /// `peer_id` and `epoch` match the current binding exactly; a stale
985    /// revoke cannot tear down a freshly rotated binding.
986    pub async fn stage_supervisor_revoke(
987        &self,
988        session_id: &SessionId,
989        peer_id: String,
990        epoch: u64,
991    ) -> Result<crate::meerkat_machine::dsl::MeerkatMachineTransition, SupervisorBindingStageError>
992    {
993        self.apply_supervisor_binding_input(
994            session_id,
995            crate::meerkat_machine::dsl::MeerkatMachineInput::RevokeSupervisor { peer_id, epoch },
996        )
997        .await
998    }
999
1000    /// Stage a DSL `SupervisorTrustEdgePublished` feedback input (C-F2 /
1001    /// wave-d D-d).
1002    ///
1003    /// Invoked by `try_handle_supervisor_bridge_command` after a
1004    /// successful `Router::add_trusted_peer` call. The `epoch` passed
1005    /// through is the one observed on the originating
1006    /// `PublishSupervisorTrustEdge` effect (i.e. the epoch of the
1007    /// `BindSupervisor` / `AuthorizeSupervisor` commit that triggered
1008    /// the publication). The DSL guard rejects the ack if the binding
1009    /// has since rotated forward — a stale ack cannot close the
1010    /// outstanding obligation for the newer epoch.
1011    pub async fn stage_supervisor_trust_published(
1012        &self,
1013        session_id: &SessionId,
1014        peer_id: String,
1015        epoch: u64,
1016    ) -> Result<(), SupervisorBindingStageError> {
1017        self.apply_supervisor_binding_input(
1018            session_id,
1019            crate::meerkat_machine::dsl::MeerkatMachineInput::SupervisorTrustEdgePublished {
1020                peer_id,
1021                epoch,
1022            },
1023        )
1024        .await?;
1025        Ok(())
1026    }
1027
1028    /// Stage a DSL `SupervisorTrustEdgePublishFailed` feedback input
1029    /// (C-F2 / wave-d D-d).
1030    ///
1031    /// Invoked when `Router::add_trusted_peer` returns an error. The
1032    /// `epoch` comes from the originating producer effect; the DSL
1033    /// guard rejects a stale-epoch ack arriving after the binding has
1034    /// rotated forward.
1035    pub async fn stage_supervisor_trust_publish_failed(
1036        &self,
1037        session_id: &SessionId,
1038        peer_id: String,
1039        epoch: u64,
1040        reason: String,
1041    ) -> Result<(), SupervisorBindingStageError> {
1042        self.apply_supervisor_binding_input(
1043            session_id,
1044            crate::meerkat_machine::dsl::MeerkatMachineInput::SupervisorTrustEdgePublishFailed {
1045                peer_id,
1046                epoch,
1047                reason,
1048            },
1049        )
1050        .await?;
1051        Ok(())
1052    }
1053
1054    /// Stage a DSL `SupervisorTrustEdgeRevoked` feedback input (C-F2 /
1055    /// wave-d D-d).
1056    ///
1057    /// Invoked after a successful `Router::remove_trusted_peer` call.
1058    /// Epoch guard semantics mirror `stage_supervisor_trust_published`.
1059    pub async fn stage_supervisor_trust_revoked(
1060        &self,
1061        session_id: &SessionId,
1062        peer_id: String,
1063        epoch: u64,
1064    ) -> Result<(), SupervisorBindingStageError> {
1065        self.apply_supervisor_binding_input(
1066            session_id,
1067            crate::meerkat_machine::dsl::MeerkatMachineInput::SupervisorTrustEdgeRevoked {
1068                peer_id,
1069                epoch,
1070            },
1071        )
1072        .await?;
1073        Ok(())
1074    }
1075
1076    /// Stage a DSL `SupervisorTrustEdgeRevokeFailed` feedback input
1077    /// (C-F2 / wave-d D-d).
1078    ///
1079    /// Invoked when `Router::remove_trusted_peer` returns an error.
1080    /// Epoch guard semantics mirror `stage_supervisor_trust_published`.
1081    pub async fn stage_supervisor_trust_revoke_failed(
1082        &self,
1083        session_id: &SessionId,
1084        peer_id: String,
1085        epoch: u64,
1086        reason: String,
1087    ) -> Result<(), SupervisorBindingStageError> {
1088        self.apply_supervisor_binding_input(
1089            session_id,
1090            crate::meerkat_machine::dsl::MeerkatMachineInput::SupervisorTrustEdgeRevokeFailed {
1091                peer_id,
1092                epoch,
1093                reason,
1094            },
1095        )
1096        .await?;
1097        Ok(())
1098    }
1099}
1100
1101/// Errors raised when staging a supervisor-binding input against the DSL
1102/// (Wave 3 D Row 21).
1103#[derive(Debug)]
1104pub enum SupervisorBindingStageError {
1105    /// The session is not registered with the runtime.
1106    SessionNotRegistered,
1107    /// The runtime session registry was already borrowed in a non-reentrant
1108    /// WASM turn while staging supervisor binding authority.
1109    SessionRegistryBusy,
1110    /// The per-session DSL authority was already borrowed in a non-reentrant
1111    /// WASM turn while staging supervisor binding authority.
1112    SessionAuthorityBusy,
1113    /// The DSL mutator rejected the transition (e.g. guard failure). The
1114    /// boxed inner is the typed DSL transition error; callers that need to
1115    /// distinguish guard rejections from missing-transition failures can
1116    /// match on it.
1117    Dsl(crate::meerkat_machine::dsl::MeerkatMachineTransitionError),
1118    /// The target runtime did not expose a complete typed local endpoint for
1119    /// generated trust-store ownership.
1120    LocalEndpoint(String),
1121}
1122
1123impl std::fmt::Display for SupervisorBindingStageError {
1124    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1125        match self {
1126            Self::SessionNotRegistered => write!(f, "session not registered with runtime"),
1127            Self::SessionRegistryBusy => {
1128                write!(f, "runtime session registry busy during supervisor binding")
1129            }
1130            Self::SessionAuthorityBusy => {
1131                write!(f, "session authority busy during supervisor binding")
1132            }
1133            Self::Dsl(err) => write!(f, "DSL rejected supervisor binding input: {err}"),
1134            Self::LocalEndpoint(err) => {
1135                write!(f, "local endpoint unavailable for supervisor trust: {err}")
1136            }
1137        }
1138    }
1139}
1140
1141impl std::error::Error for SupervisorBindingStageError {}
1142
1143/// Previous supervisor binding carried by generated authorize admission
1144/// feedback. The shell uses it mechanically for revoke/rollback after
1145/// MeerkatMachine has accepted the rotation.
1146#[derive(Debug, Clone, PartialEq, Eq)]
1147pub(crate) struct GeneratedSupervisorBinding {
1148    pub name: String,
1149    pub peer_id: String,
1150    pub address: String,
1151    pub signing_public_key: String,
1152    pub epoch: u64,
1153}
1154
1155#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1156pub(crate) enum SupervisorBindAdmission {
1157    Bootstrap,
1158    IdempotentAck,
1159    Rejected(crate::meerkat_machine::dsl::SupervisorBindRejectionKind),
1160}
1161
1162#[derive(Debug, Clone, PartialEq, Eq)]
1163pub(crate) enum SupervisorAuthorizeAdmission {
1164    Proceed(GeneratedSupervisorBinding),
1165    IdempotentAck,
1166    Rejected(crate::meerkat_machine::dsl::SupervisorAuthorizeRejectionKind),
1167}
1168
1169#[derive(Debug)]
1170pub(crate) enum SupervisorAdmissionStageError {
1171    SessionNotRegistered,
1172    Dsl(crate::meerkat_machine::dsl::MeerkatMachineTransitionError),
1173    MissingAdmissionEffect(&'static str),
1174    MalformedAdmissionEffect(&'static str),
1175}
1176
1177impl std::fmt::Display for SupervisorAdmissionStageError {
1178    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1179        match self {
1180            Self::SessionNotRegistered => write!(f, "session not registered with runtime"),
1181            Self::Dsl(err) => write!(f, "DSL rejected supervisor admission input: {err}"),
1182            Self::MissingAdmissionEffect(context) => write!(
1183                f,
1184                "{context} admission transition committed without admission feedback"
1185            ),
1186            Self::MalformedAdmissionEffect(context) => write!(
1187                f,
1188                "{context} admission feedback carried inconsistent result fields"
1189            ),
1190        }
1191    }
1192}
1193
1194impl std::error::Error for SupervisorAdmissionStageError {}
1195
1196impl MeerkatMachine {
1197    pub(crate) async fn resolve_supervisor_bind_admission(
1198        &self,
1199        session_id: &SessionId,
1200        supervisor_peer_id: String,
1201        supervisor_epoch: u64,
1202        sender_peer_id: Option<String>,
1203    ) -> Result<SupervisorBindAdmission, SupervisorAdmissionStageError> {
1204        let mut sessions = self.sessions.write().await;
1205        let entry = sessions
1206            .get_mut(session_id)
1207            .ok_or(SupervisorAdmissionStageError::SessionNotRegistered)?;
1208        let effects = {
1209            let mut authority = entry
1210                .dsl_authority
1211                .lock()
1212                .unwrap_or_else(std::sync::PoisonError::into_inner);
1213            crate::meerkat_machine::dsl::MeerkatMachineMutator::apply(
1214                &mut *authority,
1215                crate::meerkat_machine::dsl::MeerkatMachineInput::ResolveSupervisorBindAdmission {
1216                    supervisor_peer_id,
1217                    supervisor_epoch,
1218                    sender_peer_id,
1219                },
1220            )
1221            .map_err(SupervisorAdmissionStageError::Dsl)?
1222            .into_effects()
1223        };
1224        effects
1225            .iter()
1226            .find_map(|effect| {
1227                match effect {
1228                crate::meerkat_machine::dsl::MeerkatMachineEffect::SupervisorBindAdmissionResolved {
1229                    result,
1230                    rejection,
1231                } => Some((*result, *rejection)),
1232                _ => None,
1233            }
1234            })
1235            .ok_or(SupervisorAdmissionStageError::MissingAdmissionEffect(
1236                "bind supervisor",
1237            ))
1238            .and_then(|(result, rejection)| match (result, rejection) {
1239                (
1240                    crate::meerkat_machine::dsl::SupervisorBindAdmissionResultKind::Bootstrap,
1241                    None,
1242                ) => Ok(SupervisorBindAdmission::Bootstrap),
1243                (
1244                    crate::meerkat_machine::dsl::SupervisorBindAdmissionResultKind::IdempotentAck,
1245                    None,
1246                ) => Ok(SupervisorBindAdmission::IdempotentAck),
1247                (
1248                    crate::meerkat_machine::dsl::SupervisorBindAdmissionResultKind::Reject,
1249                    Some(rejection),
1250                ) => Ok(SupervisorBindAdmission::Rejected(rejection)),
1251                _ => Err(SupervisorAdmissionStageError::MalformedAdmissionEffect(
1252                    "bind supervisor",
1253                )),
1254            })
1255    }
1256
1257    /// Resolve the material `BindMember` admission verdict (advertised-address
1258    /// match, raw supervisor-peer sender match, expected runtime peer-id match,
1259    /// bootstrap-token match) through MeerkatMachine authority. The shell
1260    /// supplies the four pure boolean observations it already computes; the
1261    /// machine emits the verdict in the precedence order address → sender →
1262    /// peer-id → token, else accept. The shell mirrors the returned verdict.
1263    pub(crate) async fn resolve_supervisor_bind_material_admission(
1264        &self,
1265        session_id: &SessionId,
1266        address_matches: bool,
1267        sender_matches_supervisor: bool,
1268        expected_peer_id_matches: bool,
1269        bootstrap_token_matches: bool,
1270    ) -> Result<
1271        crate::meerkat_machine::dsl::SupervisorBindMaterialAdmissionKind,
1272        SupervisorAdmissionStageError,
1273    > {
1274        let mut sessions = self.sessions.write().await;
1275        let entry = sessions
1276            .get_mut(session_id)
1277            .ok_or(SupervisorAdmissionStageError::SessionNotRegistered)?;
1278        let effects = {
1279            let mut authority = entry
1280                .dsl_authority
1281                .lock()
1282                .unwrap_or_else(std::sync::PoisonError::into_inner);
1283            crate::meerkat_machine::dsl::MeerkatMachineMutator::apply(
1284                &mut *authority,
1285                crate::meerkat_machine::dsl::MeerkatMachineInput::ResolveSupervisorBindMaterialAdmission {
1286                    address_matches,
1287                    sender_matches_supervisor,
1288                    expected_peer_id_matches,
1289                    bootstrap_token_matches,
1290                },
1291            )
1292            .map_err(SupervisorAdmissionStageError::Dsl)?
1293            .into_effects()
1294        };
1295        effects
1296            .iter()
1297            .find_map(|effect| match effect {
1298                crate::meerkat_machine::dsl::MeerkatMachineEffect::SupervisorBindMaterialAdmissionResolved {
1299                    verdict,
1300                } => Some(*verdict),
1301                _ => None,
1302            })
1303            .ok_or(SupervisorAdmissionStageError::MissingAdmissionEffect(
1304                "bind supervisor material",
1305            ))
1306    }
1307
1308    pub(crate) async fn resolve_supervisor_authorize_admission(
1309        &self,
1310        session_id: &SessionId,
1311        supervisor_peer_id: String,
1312        supervisor_epoch: u64,
1313        sender_peer_id: Option<String>,
1314    ) -> Result<SupervisorAuthorizeAdmission, SupervisorAdmissionStageError> {
1315        let mut sessions = self.sessions.write().await;
1316        let entry = sessions
1317            .get_mut(session_id)
1318            .ok_or(SupervisorAdmissionStageError::SessionNotRegistered)?;
1319        let effects = {
1320            let mut authority = entry
1321                .dsl_authority
1322                .lock()
1323                .unwrap_or_else(std::sync::PoisonError::into_inner);
1324            crate::meerkat_machine::dsl::MeerkatMachineMutator::apply(
1325                &mut *authority,
1326                crate::meerkat_machine::dsl::MeerkatMachineInput::ResolveSupervisorAuthorizeAdmission {
1327                    supervisor_peer_id,
1328                    supervisor_epoch,
1329                    sender_peer_id,
1330                },
1331            )
1332            .map_err(SupervisorAdmissionStageError::Dsl)?
1333            .into_effects()
1334        };
1335        effects
1336            .iter()
1337            .find_map(|effect| match effect {
1338                crate::meerkat_machine::dsl::MeerkatMachineEffect::SupervisorAuthorizeAdmissionResolved {
1339                    result,
1340                    rejection,
1341                    previous_name,
1342                    previous_peer_id,
1343                    previous_address,
1344                    previous_signing_public_key,
1345                    previous_epoch,
1346                } => Some((
1347                    *result,
1348                    *rejection,
1349                    previous_name.clone(),
1350                    previous_peer_id.clone(),
1351                    previous_address.clone(),
1352                    previous_signing_public_key.clone(),
1353                    *previous_epoch,
1354                )),
1355                _ => None,
1356            })
1357            .ok_or(SupervisorAdmissionStageError::MissingAdmissionEffect(
1358                "authorize supervisor",
1359            ))
1360            .and_then(
1361                |(
1362                    result,
1363                    rejection,
1364                    previous_name,
1365                    previous_peer_id,
1366                    previous_address,
1367                    previous_signing_public_key,
1368                    previous_epoch,
1369                )| {
1370                    match (
1371                        result,
1372                        rejection,
1373                        previous_name,
1374                        previous_peer_id,
1375                        previous_address,
1376                        previous_signing_public_key,
1377                        previous_epoch,
1378                    ) {
1379                        (
1380                            crate::meerkat_machine::dsl::SupervisorAuthorizeAdmissionResultKind::Proceed,
1381                            None,
1382                            Some(name),
1383                            Some(peer_id),
1384                            Some(address),
1385                            Some(signing_public_key),
1386                            Some(epoch),
1387                        ) => Ok(SupervisorAuthorizeAdmission::Proceed(
1388                            GeneratedSupervisorBinding {
1389                                name,
1390                                peer_id,
1391                                address,
1392                                signing_public_key,
1393                                epoch,
1394                            },
1395                        )),
1396                        (
1397                            crate::meerkat_machine::dsl::SupervisorAuthorizeAdmissionResultKind::IdempotentAck,
1398                            None,
1399                            None,
1400                            None,
1401                            None,
1402                            None,
1403                            None,
1404                        ) => Ok(SupervisorAuthorizeAdmission::IdempotentAck),
1405                        (
1406                            crate::meerkat_machine::dsl::SupervisorAuthorizeAdmissionResultKind::Reject,
1407                            Some(rejection),
1408                            None,
1409                            None,
1410                            None,
1411                            None,
1412                            None,
1413                        ) => Ok(SupervisorAuthorizeAdmission::Rejected(rejection)),
1414                        _ => Err(SupervisorAdmissionStageError::MalformedAdmissionEffect(
1415                            "authorize supervisor",
1416                        )),
1417                    }
1418                },
1419            )
1420    }
1421}
1422
1423/// Generated admission result for a supervisor bridge command that requires
1424/// the currently bound supervisor.
1425#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1426pub(crate) enum SupervisorBridgeCommandAdmission {
1427    Accepted,
1428    Rejected(crate::meerkat_machine::dsl::SupervisorBridgeCommandRejectionKind),
1429}
1430
1431#[derive(Debug)]
1432pub(crate) enum SupervisorBridgeCommandAdmissionStageError {
1433    SessionNotRegistered,
1434    Dsl(crate::meerkat_machine::dsl::MeerkatMachineTransitionError),
1435    MissingAdmissionEffect,
1436    MalformedAdmissionEffect,
1437}
1438
1439impl std::fmt::Display for SupervisorBridgeCommandAdmissionStageError {
1440    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1441        match self {
1442            Self::SessionNotRegistered => write!(f, "session not registered with runtime"),
1443            Self::Dsl(err) => write!(f, "DSL rejected supervisor bridge admission input: {err}"),
1444            Self::MissingAdmissionEffect => write!(
1445                f,
1446                "supervisor bridge admission transition committed without admission feedback"
1447            ),
1448            Self::MalformedAdmissionEffect => write!(
1449                f,
1450                "supervisor bridge admission feedback carried inconsistent result fields"
1451            ),
1452        }
1453    }
1454}
1455
1456impl std::error::Error for SupervisorBridgeCommandAdmissionStageError {}
1457
1458impl MeerkatMachine {
1459    /// Return the generated MeerkatMachine-owned direct peer endpoint set for
1460    /// callers that must target an exact `RemoveDirectPeerEndpoint` input.
1461    pub async fn direct_peer_endpoints(
1462        &self,
1463        session_id: &SessionId,
1464    ) -> Result<BTreeSet<crate::meerkat_machine::dsl::PeerEndpoint>, PeerEndpointStageError> {
1465        let sessions = self.sessions.read().await;
1466        let entry = sessions
1467            .get(session_id)
1468            .ok_or(PeerEndpointStageError::SessionNotRegistered)?;
1469        let authority = entry
1470            .dsl_authority
1471            .lock()
1472            .unwrap_or_else(std::sync::PoisonError::into_inner);
1473        Ok(authority.state().direct_peer_endpoints.clone())
1474    }
1475
1476    pub(crate) async fn resolve_supervisor_bridge_command_admission(
1477        &self,
1478        session_id: &SessionId,
1479        supervisor_peer_id: String,
1480        supervisor_epoch: u64,
1481        sender_peer_id: Option<String>,
1482    ) -> Result<SupervisorBridgeCommandAdmission, SupervisorBridgeCommandAdmissionStageError> {
1483        let mut sessions = self.sessions.write().await;
1484        let entry = sessions
1485            .get_mut(session_id)
1486            .ok_or(SupervisorBridgeCommandAdmissionStageError::SessionNotRegistered)?;
1487        let effects = {
1488            let mut authority = entry
1489                .dsl_authority
1490                .lock()
1491                .unwrap_or_else(std::sync::PoisonError::into_inner);
1492            crate::meerkat_machine::dsl::MeerkatMachineMutator::apply(
1493                &mut *authority,
1494                crate::meerkat_machine::dsl::MeerkatMachineInput::ResolveSupervisorBridgeCommandAdmission {
1495                    supervisor_peer_id,
1496                    supervisor_epoch,
1497                    sender_peer_id,
1498                },
1499            )
1500            .map_err(SupervisorBridgeCommandAdmissionStageError::Dsl)?
1501            .into_effects()
1502        };
1503        effects
1504            .iter()
1505            .find_map(|effect| match effect {
1506                crate::meerkat_machine::dsl::MeerkatMachineEffect::SupervisorBridgeCommandAdmissionResolved {
1507                    result,
1508                    rejection,
1509                } => Some((*result, *rejection)),
1510                _ => None,
1511            })
1512            .ok_or(SupervisorBridgeCommandAdmissionStageError::MissingAdmissionEffect)
1513            .and_then(|(result, rejection)| match (result, rejection) {
1514                (
1515                    crate::meerkat_machine::dsl::SupervisorBridgeCommandAdmissionResultKind::Accept,
1516                    None,
1517                ) => Ok(SupervisorBridgeCommandAdmission::Accepted),
1518                (
1519                    crate::meerkat_machine::dsl::SupervisorBridgeCommandAdmissionResultKind::Reject,
1520                    Some(rejection),
1521                ) => Ok(SupervisorBridgeCommandAdmission::Rejected(rejection)),
1522                _ => Err(SupervisorBridgeCommandAdmissionStageError::MalformedAdmissionEffect),
1523            })
1524    }
1525
1526    /// D-track-b: stage an `AddDirectPeerEndpoint` DSL input and drive
1527    /// trust reconciliation against the caller-supplied runtime.
1528    ///
1529    /// Closes the emitter→consumer gap documented in
1530    /// `docs/wave-d-prep/track-b-producer-wiring.md`: the DSL owns the
1531    /// declarative peer set (`direct_peer_endpoints` +
1532    /// `mob_overlay_peer_endpoints`) and emits
1533    /// `CommsTrustReconcileRequested`; the reconciler consumes that
1534    /// effect and mechanically reconciles the underlying
1535    /// [`meerkat_core::agent::CommsRuntime`] trust store.
1536    ///
1537    /// The caller supplies the session's current `CommsRuntime`.
1538    /// Reconciliation reads that runtime's canonical trust-store
1539    /// snapshot every pass, so rebinds do not pin peer projection to
1540    /// an older transport instance.
1541    pub async fn stage_add_direct_peer_endpoint(
1542        &self,
1543        session_id: &SessionId,
1544        endpoint: crate::meerkat_machine::dsl::PeerEndpoint,
1545        comms_runtime: Arc<dyn meerkat_core::agent::CommsRuntime>,
1546    ) -> Result<(), PeerEndpointStageError> {
1547        // Parse-at-boundary: reject a malformed endpoint BEFORE it mutates the
1548        // machine peer set or emits CommsTrustReconcileRequested.
1549        validate_peer_endpoint_for_stage(&endpoint)?;
1550        let (reconciler, reconcile_obligation) = self
1551            .stage_peer_projection_input(
1552                session_id,
1553                crate::meerkat_machine::dsl::MeerkatMachineInput::AddDirectPeerEndpoint {
1554                    endpoint,
1555                },
1556                comms_runtime,
1557            )
1558            .await?;
1559        drive_reconciler(&reconciler, reconcile_obligation).await
1560    }
1561
1562    /// D-track-b: stage a `RemoveDirectPeerEndpoint` DSL input and
1563    /// drive trust reconciliation. See
1564    /// [`Self::stage_add_direct_peer_endpoint`] for the architectural
1565    /// contract.
1566    pub async fn stage_remove_direct_peer_endpoint(
1567        &self,
1568        session_id: &SessionId,
1569        endpoint: crate::meerkat_machine::dsl::PeerEndpoint,
1570        comms_runtime: Arc<dyn meerkat_core::agent::CommsRuntime>,
1571    ) -> Result<(), PeerEndpointStageError> {
1572        let (reconciler, reconcile_obligation) = self
1573            .stage_peer_projection_input(
1574                session_id,
1575                crate::meerkat_machine::dsl::MeerkatMachineInput::RemoveDirectPeerEndpoint {
1576                    endpoint,
1577                },
1578                comms_runtime,
1579            )
1580            .await?;
1581        drive_reconciler(&reconciler, reconcile_obligation).await
1582    }
1583
1584    /// Stage the generated absent-endpoint repair path for a direct peer id.
1585    ///
1586    /// This is used when machine state already says the direct endpoint is
1587    /// absent, but the caller needs to re-emit the generated reconciliation
1588    /// effect so stale trust-store projection rows cannot stay behaviorally
1589    /// active.
1590    pub async fn stage_repair_remove_direct_peer_id(
1591        &self,
1592        session_id: &SessionId,
1593        peer_id: String,
1594        comms_runtime: Arc<dyn meerkat_core::agent::CommsRuntime>,
1595    ) -> Result<(), PeerEndpointStageError> {
1596        let endpoint = crate::meerkat_machine::dsl::PeerEndpoint::new(
1597            "generated-remove-repair",
1598            peer_id,
1599            "generated-repair://absent-direct-peer",
1600            [0; 32],
1601        );
1602        self.stage_remove_direct_peer_endpoint(session_id, endpoint, comms_runtime)
1603            .await
1604    }
1605
1606    /// Stage a supervisor-observed mob peer overlay through generated
1607    /// MeerkatMachine authority before driving trust reconciliation.
1608    #[allow(clippy::too_many_arguments)]
1609    pub async fn stage_authorized_supervisor_mob_peer_overlay(
1610        &self,
1611        session_id: &SessionId,
1612        supervisor_peer_id: String,
1613        supervisor_epoch: u64,
1614        recipient_peer_id: String,
1615        overlay_epoch: u64,
1616        endpoints: BTreeSet<crate::meerkat_machine::dsl::PeerEndpoint>,
1617        endpoint_count: u64,
1618        command_peer_id: String,
1619        command_endpoint: crate::meerkat_machine::dsl::PeerEndpoint,
1620        command_kind: crate::meerkat_machine::dsl::MobPeerOverlayCommandKind,
1621        comms_runtime: Arc<dyn meerkat_core::agent::CommsRuntime>,
1622    ) -> Result<(), PeerEndpointStageError> {
1623        // Parse-at-boundary: reject any malformed overlay endpoint (the overlay
1624        // set and the command endpoint) BEFORE mutating the machine peer set.
1625        for endpoint in &endpoints {
1626            validate_peer_endpoint_for_stage(endpoint)?;
1627        }
1628        validate_peer_endpoint_for_stage(&command_endpoint)?;
1629        let (reconciler, reconcile_obligation) = self
1630            .stage_peer_projection_input(
1631                session_id,
1632                crate::meerkat_machine::dsl::MeerkatMachineInput::AuthorizeSupervisorMobPeerOverlay {
1633                    supervisor_peer_id,
1634                    supervisor_epoch,
1635                    recipient_peer_id,
1636                    overlay_epoch,
1637                    endpoints,
1638                    endpoint_count,
1639                    command_peer_id,
1640                    command_endpoint,
1641                    command_kind,
1642                },
1643                comms_runtime,
1644            )
1645            .await?;
1646        drive_reconciler(&reconciler, reconcile_obligation).await
1647    }
1648
1649    /// Apply a peer-projection DSL input, sample the emitted
1650    /// `CommsTrustReconcileRequested` effect under the same DSL lock,
1651    /// and return a reconciler for the current runtime with the generated
1652    /// obligation carrying the post-transition effective peer facts.
1653    ///
1654    /// The reconciler is driven OUTSIDE the `sessions` RwLock to avoid
1655    /// blocking other adapter operations behind trust-store I/O. There
1656    /// is no helper-local applied truth: each reconcile pass diffs the
1657    /// supplied runtime's canonical trust-store snapshot against the
1658    /// DSL-owned effective peer set carried by the generated obligation.
1659    async fn stage_peer_projection_input(
1660        &self,
1661        session_id: &SessionId,
1662        input: crate::meerkat_machine::dsl::MeerkatMachineInput,
1663        comms_runtime: Arc<dyn meerkat_core::agent::CommsRuntime>,
1664    ) -> Result<
1665        (
1666            Arc<crate::comms_trust_reconcile::CommsTrustReconciler>,
1667            crate::protocol_comms_trust_reconcile::CommsTrustReconcileObligation,
1668        ),
1669        PeerEndpointStageError,
1670    > {
1671        let mut sessions = self.sessions.write().await;
1672        let entry = sessions
1673            .get_mut(session_id)
1674            .ok_or(PeerEndpointStageError::SessionNotRegistered)?;
1675        let local_endpoint = Self::local_endpoint_for_comms_runtime(comms_runtime.as_ref())
1676            .map_err(PeerEndpointStageError::LocalEndpoint)?;
1677
1678        let reconcile_obligation = {
1679            let freshness_authority =
1680                crate::protocol_comms_trust_reconcile::PeerProjectionFreshnessAuthority::from_authority(
1681                    Arc::clone(&entry.dsl_authority),
1682                );
1683            let mut authority = entry
1684                .dsl_authority
1685                .lock()
1686                .unwrap_or_else(std::sync::PoisonError::into_inner);
1687            crate::meerkat_machine::dsl::MeerkatMachineMutator::apply(
1688                &mut *authority,
1689                crate::meerkat_machine::dsl::MeerkatMachineInput::PublishLocalEndpoint {
1690                    endpoint: local_endpoint,
1691                },
1692            )
1693            .map_err(PeerEndpointStageError::Dsl)?;
1694            let transition =
1695                crate::meerkat_machine::dsl::MeerkatMachineMutator::apply(&mut *authority, input)
1696                    .map_err(PeerEndpointStageError::Dsl)?;
1697            crate::protocol_comms_trust_reconcile::extract_obligations_with_freshness(
1698                &transition,
1699                freshness_authority,
1700            )
1701            .into_iter()
1702            .next()
1703            .ok_or(PeerEndpointStageError::MissingReconcileEffect)?
1704        };
1705
1706        let reconciler = Arc::new(crate::comms_trust_reconcile::CommsTrustReconciler::new(
1707            comms_runtime,
1708        ));
1709
1710        Ok((reconciler, reconcile_obligation))
1711    }
1712}
1713
1714async fn drive_reconciler(
1715    reconciler: &crate::comms_trust_reconcile::CommsTrustReconciler,
1716    reconcile_obligation: crate::protocol_comms_trust_reconcile::CommsTrustReconcileObligation,
1717) -> Result<(), PeerEndpointStageError> {
1718    reconciler
1719        .reconcile(&reconcile_obligation)
1720        .await
1721        .map(|_report| ())
1722        .map_err(PeerEndpointStageError::Reconcile)
1723}
1724
1725/// Errors raised when staging a peer-projection input against the DSL
1726/// and driving the session-scoped trust reconciler (D-track-b).
1727#[derive(Debug)]
1728pub enum PeerEndpointStageError {
1729    /// The session is not registered with the runtime.
1730    SessionNotRegistered,
1731    /// The DSL mutator rejected the transition (e.g. duplicate endpoint,
1732    /// stale overlay epoch, or per-phase guard failure).
1733    Dsl(crate::meerkat_machine::dsl::MeerkatMachineTransitionError),
1734    /// The DSL transition committed but did not emit
1735    /// `CommsTrustReconcileRequested`. This indicates a contract
1736    /// violation between the schema and the runtime — the three
1737    /// peer-projection transitions are specified to emit the effect
1738    /// unconditionally.
1739    MissingReconcileEffect,
1740    /// The target runtime did not expose a complete typed local endpoint for
1741    /// generated trust-store ownership.
1742    LocalEndpoint(String),
1743    /// The reconciler failed to mechanically reconcile the trust
1744    /// store.
1745    Reconcile(crate::comms_trust_reconcile::CommsTrustReconcileError),
1746    /// A staged `PeerEndpoint` carried a malformed `peer_id`/`address`/`name`.
1747    /// Rejected at the ingress boundary (parse-at-boundary) BEFORE any machine
1748    /// peer-set mutation or effect emission, so invalid identity atoms never
1749    /// reach `direct_peer_endpoints`/`mob_overlay_peer_endpoints`.
1750    InvalidEndpoint(crate::comms_trust_reconcile::CommsTrustReconcileError),
1751}
1752
1753impl std::fmt::Display for PeerEndpointStageError {
1754    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1755        match self {
1756            Self::SessionNotRegistered => write!(f, "session not registered with runtime"),
1757            Self::Dsl(err) => write!(f, "DSL rejected peer-projection input: {err}"),
1758            Self::MissingReconcileEffect => write!(
1759                f,
1760                "peer-projection DSL transition committed without emitting CommsTrustReconcileRequested"
1761            ),
1762            Self::LocalEndpoint(err) => {
1763                write!(
1764                    f,
1765                    "local endpoint unavailable for trust reconciliation: {err}"
1766                )
1767            }
1768            Self::Reconcile(err) => write!(f, "trust reconciliation failed: {err}"),
1769            Self::InvalidEndpoint(err) => {
1770                write!(f, "peer endpoint rejected at ingress boundary: {err}")
1771            }
1772        }
1773    }
1774}
1775
1776impl std::error::Error for PeerEndpointStageError {}
1777
1778/// Parse-at-boundary validation for a peer endpoint about to be staged into the
1779/// MeerkatMachine peer set. Reuses the canonical
1780/// [`endpoint_to_descriptor`](crate::comms_trust_reconcile::endpoint_to_descriptor)
1781/// parse so the machine never admits a malformed `peer_id`/`address`/`name`.
1782fn validate_peer_endpoint_for_stage(
1783    endpoint: &crate::meerkat_machine::dsl::PeerEndpoint,
1784) -> Result<(), PeerEndpointStageError> {
1785    crate::comms_trust_reconcile::endpoint_to_descriptor(endpoint)
1786        .map(|_| ())
1787        .map_err(PeerEndpointStageError::InvalidEndpoint)
1788}
meerkat_runtime/meerkat_machine/comms_drain.rs

meerkat_runtime/meerkat_machine/
comms_drain.rs
