frame-host 0.4.0

//! The real [`ManagementConversation`] over frame-conv, and the boot-line
//! handoff both sides share.
//!
//! Topology: the dev-started host OPENS the management conversation on
//! its own embedded bus and prints ONE handoff line on stdout naming the
//! bus endpoint and the conversation id — the same piped-stdout channel
//! the boot URL already rides. `frame dev` parses that line and JOINS
//! (enrollment mints the joiner's own credential; only the id travels).
//! S1 requests and S2 status events ride the one conversation on their
//! respective F-3a patterns.

use std::time::Duration;

use frame_conv::error::CallError;
use frame_conv::handle::ConversationHandle;
use frame_conv::id::{ConversationId, CorrelationId};
use frame_conv::outcome::InboundRequest;
use frame_conv::store::ResumeStore;
use frame_conv::{AttachError, BusAttachment};

use super::adapter::{InboundManagement, ManagementConversation};
use super::{DevReply, DevRequest, DevStatusEvent, StageRefusal};

/// The handoff line's stable prefix — parsed by `frame dev`, printed by
/// the dev-started host, tested from both directions below.
pub const MANAGEMENT_LINE_PREFIX: &str = "dev management at ";

/// Formats the one handoff line: `dev management at ENDPOINT conversation ID`.
#[must_use]
pub fn format_management_line(endpoint: &str, conversation: ConversationId) -> String {
    format!("{MANAGEMENT_LINE_PREFIX}{endpoint} conversation {conversation}")
}

/// Parses a host output line as the management handoff; `None` for every
/// other line (the pump forwards all lines verbatim, this picks the one).
#[must_use]
pub fn parse_management_line(line: &str) -> Option<(String, ConversationId)> {
    let after = line.split(MANAGEMENT_LINE_PREFIX).nth(1)?;
    let (endpoint, id) = after.split_once(" conversation ")?;
    let conversation = id.trim().parse().ok()?;
    Some((endpoint.to_owned(), conversation))
}

/// In-memory resume state for the dev management conversation. A dev
/// session deliberately does NOT resume across processes — the operator
/// reruns the one command (R2's recovery guidance) — so durable
/// persistence would advertise a durability this channel does not have.
#[derive(Default)]
pub struct DevResumeStore(Vec<u8>);

impl ResumeStore for DevResumeStore {
    fn persist(&mut self, canonical_state: &[u8]) -> Result<(), frame_conv::error::StoreError> {
        self.0 = canonical_state.to_vec();
        Ok(())
    }
}

/// The node-side conversation: opened by the dev host, served by the
/// adapter loop.
pub struct DevManagementConversation<S: ResumeStore> {
    handle: ConversationHandle<S>,
}

impl<S: ResumeStore> DevManagementConversation<S> {
    /// Opens the management conversation on the embedded bus, returning
    /// the id the handoff line advertises.
    ///
    /// # Errors
    ///
    /// The typed attach failure.
    pub fn open(
        attachment: &BusAttachment,
        store: S,
    ) -> Result<(Self, ConversationId), AttachError> {
        let (handle, grant) = ConversationHandle::open(attachment, store)?;
        Ok((Self { handle }, grant.conversation))
    }

    /// Joins an advertised management conversation (the `frame dev` side).
    ///
    /// # Errors
    ///
    /// The typed attach failure.
    pub fn join(
        attachment: &BusAttachment,
        conversation: ConversationId,
        store: S,
    ) -> Result<Self, AttachError> {
        let (handle, _grant) = ConversationHandle::join(attachment, conversation, store)?;
        Ok(Self { handle })
    }

    /// The underlying handle, for the client side's own request/subscribe
    /// calls (`frame dev` drives requests directly; only the NODE side
    /// runs the adapter serve loop).
    pub fn handle_mut(&mut self) -> &mut ConversationHandle<S> {
        &mut self.handle
    }
}

impl<S: ResumeStore> ManagementConversation for DevManagementConversation<S> {
    fn next_request(&mut self, wait: Duration) -> Result<Option<InboundManagement>, String> {
        match self.handle.next_request::<DevRequest>(wait) {
            Ok(None) => Ok(None),
            Ok(Some(InboundRequest::Valid(incoming))) => Ok(Some(InboundManagement {
                request: incoming.body,
                correlation: incoming.correlation,
            })),
            Ok(Some(InboundRequest::SchemaInvalid {
                correlation,
                detail,
                ..
            })) => {
                // Version skew answered LOUDLY at the requester (typed
                // refusal), then this pump arm yields quiet — the reply
                // here is the anomaly's handling, not opportunistic work.
                let refusal = DevReply::Refused(StageRefusal::SchemaInvalid {
                    detail: detail.clone(),
                });
                self.handle
                    .reply(correlation, &refusal)
                    .map_err(|error| format!("schema-invalid refusal reply: {error}"))?;
                tracing::error!(%detail, "management request failed schema validation; refused typed");
                Ok(None)
            }
            Err(CallError::ConnectionLost { detail }) => Err(detail),
            Err(other) => Err(other.to_string()),
        }
    }

    fn reply(&mut self, correlation: CorrelationId, reply: &DevReply) -> Result<(), String> {
        self.handle
            .reply(correlation, reply)
            .map(|_| ())
            .map_err(|error| error.to_string())
    }

    fn publish_status(&mut self, event: &DevStatusEvent) -> Result<(), String> {
        self.handle
            .publish_event(event)
            .map(|_| ())
            .map_err(|error| error.to_string())
    }
}

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

    use super::{format_management_line, parse_management_line};
    use frame_conv::id::ConversationId;

    /// The handoff line survives its own round trip, and ordinary host
    /// log lines (the URL line included) never parse as one.
    #[test]
    fn handoff_line_round_trips_and_rejects_noise() {
        let id: ConversationId = "12345".parse().expect("id parses");
        let line = format_management_line("127.0.0.1:9410", id);
        assert_eq!(
            parse_management_line(&line),
            Some(("127.0.0.1:9410".to_owned(), id))
        );
        assert_eq!(
            parse_management_line(&format!("prefix {line}")),
            Some(("127.0.0.1:9410".to_owned(), id)),
            "a logger prefix does not defeat the parse"
        );
        assert!(parse_management_line("demo serving at http://127.0.0.1:4191").is_none());
        assert!(parse_management_line("dev management at half a line").is_none());
        assert!(
            parse_management_line("dev management at 1.2.3.4:1 conversation notanumber").is_none()
        );
    }
}