frame_host/dev/

adapter.rs

//! The node-side management adapter: serves S1 requests and publishes S2
//! status over the shared `frame:dev-management@v1` vocabulary, driving
//! the [`BarrierEngine`] behind the [`require_dev_node`] gate.
//!
//! ## The pump, sanctioned (brief §Adapter pump sanction, ruled 2026-07-24)
//!
//! Requests arrive via frame-conv's `next_request(wait)` — a
//! quantum-bounded pump whose `Ok(None)` is a benign quiet wait (the
//! substrate's hardcoded 5 s IO quantum is upstream ASK-2, pinned at
//! F-3a's tear: "elapsed IO quanta while waiting are benign re-arms").
//! Ruling B targets polling a consumer INVENTS, not a substrate pump the
//! consumer has no alternative to. The binding condition is structural:
//! **quiet iterations are EMPTY** — the `Ok(None)` arm does nothing but
//! re-arm the same pump (see [`serve`]'s single match; the witness test
//! `quiet_ticks_do_no_work` fails if the quiet arm ever grows a body).
//! When upstream ships a close-wakeable wait, this adapter inherits it
//! for free and the disclosure self-retires.
//!
//! [`require_dev_node`]: super::require_dev_node

use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use frame_conv::id::CorrelationId;

use super::{BarrierEngine, DevReply, DevRequest, DevStatusEvent, NodeControl};

/// One decoded inbound management request awaiting its reply.
pub struct InboundManagement {
    /// The typed request.
    pub request: DevRequest,
    /// Correlation to answer with.
    pub correlation: CorrelationId,
}

/// The conversation surface the adapter serves over — the real
/// implementation wraps `frame_conv::ConversationHandle`; tests script
/// it. Errors are rendered transport fates: any `Err` means the dev
/// client's connection is gone and the serve loop exits with it.
pub trait ManagementConversation {
    /// Waits up to `wait` for the next inbound request; `Ok(None)` is the
    /// benign quiet tick.
    ///
    /// # Errors
    ///
    /// A typed connection fate, rendered.
    fn next_request(&mut self, wait: Duration) -> Result<Option<InboundManagement>, String>;
    /// Publishes the exactly-one typed reply for `correlation`.
    ///
    /// # Errors
    ///
    /// A typed publish failure, rendered.
    fn reply(&mut self, correlation: CorrelationId, reply: &DevReply) -> Result<(), String>;
    /// Publishes one S2 status event.
    ///
    /// # Errors
    ///
    /// A typed publish failure, rendered.
    fn publish_status(&mut self, event: &DevStatusEvent) -> Result<(), String>;
}

/// Why the serve loop returned.
#[derive(Debug, PartialEq, Eq)]
pub enum ServeExit {
    /// The close flag was raised: ordered teardown.
    Closed,
    /// The conversation's connection died (S3): the dev client is gone.
    ConnectionLost {
        /// Exact rendered transport fate.
        detail: String,
    },
}

/// Serves management requests until closed or the connection dies.
///
/// Status events raised by the engine mid-activation are published as
/// they happen; a status-publish failure is recorded and surfaces as the
/// connection loss it is AFTER the in-flight activation completes — the
/// node-side barrier never aborts half-way because the observer left.
pub fn serve<C: ManagementConversation>(
    conversation: &mut C,
    engine: &mut BarrierEngine,
    control: &mut dyn NodeControl,
    close: &AtomicBool,
    quantum: Duration,
) -> ServeExit {
    loop {
        if close.load(Ordering::Acquire) {
            return ServeExit::Closed;
        }
        match conversation.next_request(quantum) {
            // The sanctioned quiet tick: EMPTY by ruling — nothing but
            // the loop's own re-arm happens here.
            Ok(None) => {}
            Ok(Some(inbound)) => {
                let mut publish_failure: Option<String> = None;
                let reply = {
                    let failure = &mut publish_failure;
                    engine.handle(inbound.request, control, &mut |event| {
                        if failure.is_none()
                            && let Err(detail) = conversation_publish(conversation, &event)
                        {
                            *failure = Some(detail);
                        }
                    })
                };
                if let Err(detail) = conversation.reply(inbound.correlation, &reply) {
                    return ServeExit::ConnectionLost { detail };
                }
                if let Some(detail) = publish_failure {
                    return ServeExit::ConnectionLost { detail };
                }
            }
            Err(detail) => return ServeExit::ConnectionLost { detail },
        }
    }
}

/// Free helper so the closure above can publish without capturing
/// `conversation` mutably twice.
fn conversation_publish<C: ManagementConversation>(
    conversation: &mut C,
    event: &DevStatusEvent,
) -> Result<(), String> {
    conversation.publish_status(event)
}

#[cfg(test)]
mod tests {
    #![allow(clippy::expect_used, clippy::panic)]

    use std::sync::atomic::AtomicBool;
    use std::time::Duration;

    use frame_conv::id::CorrelationId;

    use super::{InboundManagement, ManagementConversation, ServeExit, serve};
    use crate::dev::{
        BarrierEngine, CandidateBytes, DevReply, DevRequest, DevStatusEvent, GenerationReport,
        NodeControl, NodeMode, StageRefusal, StartFailure,
    };

    const QUANTUM: Duration = Duration::from_millis(1);

    /// Scripted conversation: a queue of pump outcomes, recorded replies
    /// and publishes.
    struct FakeConversation {
        script: Vec<Option<InboundManagement>>,
        replies: Vec<DevReply>,
        published: Vec<DevStatusEvent>,
        pump_ticks: usize,
    }

    impl FakeConversation {
        fn new(mut script: Vec<Option<InboundManagement>>) -> Self {
            script.reverse();
            Self {
                script,
                replies: Vec::new(),
                published: Vec::new(),
                pump_ticks: 0,
            }
        }
    }

    impl ManagementConversation for FakeConversation {
        fn next_request(&mut self, _wait: Duration) -> Result<Option<InboundManagement>, String> {
            self.pump_ticks += 1;
            match self.script.pop() {
                Some(entry) => Ok(entry),
                None => Err("script exhausted: connection closed".to_owned()),
            }
        }
        fn reply(&mut self, _correlation: CorrelationId, reply: &DevReply) -> Result<(), String> {
            self.replies.push(reply.clone());
            Ok(())
        }
        fn publish_status(&mut self, event: &DevStatusEvent) -> Result<(), String> {
            self.published.push(event.clone());
            Ok(())
        }
    }

    /// A node that counts every operation — the quiet-tick witness's
    /// authority.
    #[derive(Default)]
    struct CountingNode {
        operations: usize,
    }

    impl NodeControl for CountingNode {
        fn stop_component(&mut self) -> Result<(), String> {
            self.operations += 1;
            Ok(())
        }
        fn stage(&mut self, _candidate: &CandidateBytes) -> Result<(), String> {
            self.operations += 1;
            Ok(())
        }
        fn start_component(&mut self) -> Result<(), StartFailure> {
            self.operations += 1;
            Ok(())
        }
        fn witness_mailbox(&mut self) -> Result<(), String> {
            self.operations += 1;
            Ok(())
        }
        fn witness_content(&mut self) -> Result<(), String> {
            self.operations += 1;
            Ok(())
        }
        fn report(
            &mut self,
            build_generation: u64,
            content_digest: [u8; 32],
        ) -> Result<GenerationReport, String> {
            self.operations += 1;
            Ok(GenerationReport {
                build_generation,
                content_digest,
                component_incarnation: 1,
                module_generation: 1,
            })
        }
    }

    fn candidate(generation: u64) -> CandidateBytes {
        CandidateBytes {
            build_generation: generation,
            content_digest: [0; 32],
            component_beam: vec![1],
            ffi_beam: vec![2],
        }
    }

    fn inbound(generation: u64) -> InboundManagement {
        InboundManagement {
            request: DevRequest::StageAndActivate(candidate(generation)),
            correlation: CorrelationId::mint(),
        }
    }

    /// THE quiet-tick witness (the ruling's structural condition): quiet
    /// pump iterations perform ZERO node operations, publish nothing, and
    /// reply to nothing — the pump is the loop's only re-arm source.
    #[test]
    fn quiet_ticks_do_no_work() {
        let mut conversation = FakeConversation::new(vec![None, None, None, None, None]);
        let mut engine = BarrierEngine::new(NodeMode::Dev, candidate(0));
        let mut node = CountingNode::default();
        let close = AtomicBool::new(false);

        let exit = serve(&mut conversation, &mut engine, &mut node, &close, QUANTUM);
        // The script ends in a connection fate after five quiet ticks.
        assert!(matches!(exit, ServeExit::ConnectionLost { .. }));
        assert_eq!(conversation.pump_ticks, 6, "five quiet ticks + the fate");
        assert_eq!(node.operations, 0, "quiet ticks touch the node ZERO times");
        assert!(
            conversation.published.is_empty(),
            "quiet ticks publish nothing"
        );
        assert!(
            conversation.replies.is_empty(),
            "quiet ticks reply to nothing"
        );
    }

    /// A request between quiet ticks is served exactly once: one reply,
    /// status events published as the barrier moves.
    #[test]
    fn requests_are_served_between_quiet_ticks() {
        let mut conversation = FakeConversation::new(vec![None, Some(inbound(1)), None]);
        let mut engine = BarrierEngine::new(NodeMode::Dev, candidate(0));
        let mut node = CountingNode::default();
        let close = AtomicBool::new(false);

        let exit = serve(&mut conversation, &mut engine, &mut node, &close, QUANTUM);
        assert!(matches!(exit, ServeExit::ConnectionLost { .. }));
        assert_eq!(conversation.replies.len(), 1, "exactly one reply");
        assert!(matches!(conversation.replies[0], DevReply::Activated(_)));
        assert!(
            matches!(
                conversation.published.last(),
                Some(DevStatusEvent::RunningCurrent(_))
            ),
            "the barrier's status transitions were published"
        );
    }

    /// The close flag wins before the next pump arm — ordered teardown,
    /// not an error.
    #[test]
    fn close_flag_exits_ordered() {
        let mut conversation = FakeConversation::new(vec![None]);
        let mut engine = BarrierEngine::new(NodeMode::Dev, candidate(0));
        let mut node = CountingNode::default();
        let close = AtomicBool::new(true);

        let exit = serve(&mut conversation, &mut engine, &mut node, &close, QUANTUM);
        assert_eq!(exit, ServeExit::Closed);
        assert_eq!(conversation.pump_ticks, 0, "no pump arm after close");
    }

    /// A production engine behind the adapter still refuses with zero
    /// node calls — the C3 pin holds through the full serve path.
    #[test]
    fn production_refusal_holds_through_the_adapter() {
        let mut conversation = FakeConversation::new(vec![Some(inbound(1))]);
        let mut engine = BarrierEngine::new(NodeMode::Production, candidate(0));
        let mut node = CountingNode::default();
        let close = AtomicBool::new(false);

        let _exit = serve(&mut conversation, &mut engine, &mut node, &close, QUANTUM);
        assert_eq!(
            conversation.replies[0],
            DevReply::Refused(StageRefusal::NotADevNode)
        );
        assert_eq!(node.operations, 0);
        assert!(conversation.published.is_empty());
    }
}