frame_host/dev/
conversation.rs

1//! The real [`ManagementConversation`] over frame-conv, and the boot-line
2//! handoff both sides share.
3//!
4//! Topology: the dev-started host OPENS the management conversation on
5//! its own embedded bus and prints ONE handoff line on stdout naming the
6//! bus endpoint and the conversation id — the same piped-stdout channel
7//! the boot URL already rides. `frame dev` parses that line and JOINS
8//! (enrollment mints the joiner's own credential; only the id travels).
9//! S1 requests and S2 status events ride the one conversation on their
10//! respective F-3a patterns.
11
12use std::time::Duration;
13
14use frame_conv::error::CallError;
15use frame_conv::handle::ConversationHandle;
16use frame_conv::id::{ConversationId, CorrelationId};
17use frame_conv::outcome::InboundRequest;
18use frame_conv::store::ResumeStore;
19use frame_conv::{AttachError, BusAttachment};
20
21use super::adapter::{InboundManagement, ManagementConversation};
22use super::{DevReply, DevRequest, DevStatusEvent, StageRefusal};
23
24/// The handoff line's stable prefix — parsed by `frame dev`, printed by
25/// the dev-started host, tested from both directions below.
26pub const MANAGEMENT_LINE_PREFIX: &str = "dev management at ";
27
28/// Formats the one handoff line: `dev management at ENDPOINT conversation ID`.
29#[must_use]
30pub fn format_management_line(endpoint: &str, conversation: ConversationId) -> String {
31    format!("{MANAGEMENT_LINE_PREFIX}{endpoint} conversation {conversation}")
32}
33
34/// Parses a host output line as the management handoff; `None` for every
35/// other line (the pump forwards all lines verbatim, this picks the one).
36#[must_use]
37pub fn parse_management_line(line: &str) -> Option<(String, ConversationId)> {
38    let after = line.split(MANAGEMENT_LINE_PREFIX).nth(1)?;
39    let (endpoint, id) = after.split_once(" conversation ")?;
40    let conversation = id.trim().parse().ok()?;
41    Some((endpoint.to_owned(), conversation))
42}
43
44/// In-memory resume state for the dev management conversation. A dev
45/// session deliberately does NOT resume across processes — the operator
46/// reruns the one command (R2's recovery guidance) — so durable
47/// persistence would advertise a durability this channel does not have.
48#[derive(Default)]
49pub struct DevResumeStore(Vec<u8>);
50
51impl ResumeStore for DevResumeStore {
52    fn persist(&mut self, canonical_state: &[u8]) -> Result<(), frame_conv::error::StoreError> {
53        self.0 = canonical_state.to_vec();
54        Ok(())
55    }
56}
57
58/// The node-side conversation: opened by the dev host, served by the
59/// adapter loop.
60pub struct DevManagementConversation<S: ResumeStore> {
61    handle: ConversationHandle<S>,
62}
63
64impl<S: ResumeStore> DevManagementConversation<S> {
65    /// Opens the management conversation on the embedded bus, returning
66    /// the id the handoff line advertises.
67    ///
68    /// # Errors
69    ///
70    /// The typed attach failure.
71    pub fn open(
72        attachment: &BusAttachment,
73        store: S,
74    ) -> Result<(Self, ConversationId), AttachError> {
75        let (handle, grant) = ConversationHandle::open(attachment, store)?;
76        Ok((Self { handle }, grant.conversation))
77    }
78
79    /// Joins an advertised management conversation (the `frame dev` side).
80    ///
81    /// # Errors
82    ///
83    /// The typed attach failure.
84    pub fn join(
85        attachment: &BusAttachment,
86        conversation: ConversationId,
87        store: S,
88    ) -> Result<Self, AttachError> {
89        let (handle, _grant) = ConversationHandle::join(attachment, conversation, store)?;
90        Ok(Self { handle })
91    }
92
93    /// The underlying handle, for the client side's own request/subscribe
94    /// calls (`frame dev` drives requests directly; only the NODE side
95    /// runs the adapter serve loop).
96    pub fn handle_mut(&mut self) -> &mut ConversationHandle<S> {
97        &mut self.handle
98    }
99}
100
101impl<S: ResumeStore> ManagementConversation for DevManagementConversation<S> {
102    fn next_request(&mut self, wait: Duration) -> Result<Option<InboundManagement>, String> {
103        match self.handle.next_request::<DevRequest>(wait) {
104            Ok(None) => Ok(None),
105            Ok(Some(InboundRequest::Valid(incoming))) => Ok(Some(InboundManagement {
106                request: incoming.body,
107                correlation: incoming.correlation,
108            })),
109            Ok(Some(InboundRequest::SchemaInvalid {
110                correlation,
111                detail,
112                ..
113            })) => {
114                // Version skew answered LOUDLY at the requester (typed
115                // refusal), then this pump arm yields quiet — the reply
116                // here is the anomaly's handling, not opportunistic work.
117                let refusal = DevReply::Refused(StageRefusal::SchemaInvalid {
118                    detail: detail.clone(),
119                });
120                self.handle
121                    .reply(correlation, &refusal)
122                    .map_err(|error| format!("schema-invalid refusal reply: {error}"))?;
123                tracing::error!(%detail, "management request failed schema validation; refused typed");
124                Ok(None)
125            }
126            Err(CallError::ConnectionLost { detail }) => Err(detail),
127            Err(other) => Err(other.to_string()),
128        }
129    }
130
131    fn reply(&mut self, correlation: CorrelationId, reply: &DevReply) -> Result<(), String> {
132        self.handle
133            .reply(correlation, reply)
134            .map(|_| ())
135            .map_err(|error| error.to_string())
136    }
137
138    fn publish_status(&mut self, event: &DevStatusEvent) -> Result<(), String> {
139        self.handle
140            .publish_event(event)
141            .map(|_| ())
142            .map_err(|error| error.to_string())
143    }
144}
145
146#[cfg(test)]
147mod tests {
148    #![allow(clippy::expect_used)]
149
150    use super::{format_management_line, parse_management_line};
151    use frame_conv::id::ConversationId;
152
153    /// The handoff line survives its own round trip, and ordinary host
154    /// log lines (the URL line included) never parse as one.
155    #[test]
156    fn handoff_line_round_trips_and_rejects_noise() {
157        let id: ConversationId = "12345".parse().expect("id parses");
158        let line = format_management_line("127.0.0.1:9410", id);
159        assert_eq!(
160            parse_management_line(&line),
161            Some(("127.0.0.1:9410".to_owned(), id))
162        );
163        assert_eq!(
164            parse_management_line(&format!("prefix {line}")),
165            Some(("127.0.0.1:9410".to_owned(), id)),
166            "a logger prefix does not defeat the parse"
167        );
168        assert!(parse_management_line("demo serving at http://127.0.0.1:4191").is_none());
169        assert!(parse_management_line("dev management at half a line").is_none());
170        assert!(
171            parse_management_line("dev management at 1.2.3.4:1 conversation notanumber").is_none()
172        );
173    }
174}
frame_host/dev/conversation.rs

frame_host/dev/
conversation.rs
