running_process/broker/
session_codec.rs

1//! Phase 3 SESSION-lane codec (soldr#2365, slice 3a): bridge the proxy pump's
2//! `SessionFrame`s onto the v1 `Frame` envelope and back.
3//!
4//! The pump ([`crate::broker::session_pump`]) speaks in-memory `SessionFrame`
5//! channels; this module is the byte-boundary twin that lets those frames cross
6//! a real transport. Each `SessionFrame` rides exactly one `Frame` on the
7//! [`SESSION_PAYLOAD_PROTOCOL`] lane, encoded with the same
8//! `[u8 framing_version=1][u32 LE len][prost Frame]` wire shape every other
9//! lane uses (via [`encode_framed`] / [`try_decode_framed`]).
10//!
11//! **Direction.** The frame kind is derived from the `SessionFrame` variant:
12//! client→daemon frames (`Stdin`, `StdinEof`) are `REQUEST`, daemon→client
13//! frames (`Stdout`, `Stderr`, `Exit`) are `RESPONSE`. This is a provisional
14//! hint, not request/response correlation — a compile session is one long-lived
15//! bidirectional exchange, so `request_id` carries a session-local **sequence
16//! number** for observability, not a request/response pairing. A later slice
17//! (3b) that multiplexes many sessions over one endpoint will introduce a real
18//! session id; until then one transport carries one session and direction is
19//! implied by which half of the duplex a frame arrives on.
20//!
21//! This module deliberately does **not** touch the broker socket, `FrameClient`,
22//! or any OS handle: it is a pure `SessionFrame <-> bytes` codec so the fidelity
23//! it guarantees (byte-transparency across partial-frame boundaries) can be
24//! proven in isolation and reused by whatever transport slice 3b lands.
25
26use prost::Message;
27
28use crate::broker::protocol::{
29    encode_framed, try_decode_framed, Frame, FrameKind, FramingError, SESSION_PAYLOAD_PROTOCOL,
30};
31use crate::broker::protocol_v2::{session_frame, SessionFrame};
32
33/// The `Frame` kind a `SessionFrame` rides under, derived from its direction.
34///
35/// Inbound (client→daemon) stdin frames are `REQUEST`; outbound
36/// (daemon→client) stdout/stderr/exit frames are `RESPONSE`. See the module
37/// docs for why this is a hint rather than request/response correlation.
38fn frame_kind_for(kind: &session_frame::Kind) -> FrameKind {
39    match kind {
40        // Client → daemon: stdin stream and the opening command carriage.
41        session_frame::Kind::Stdin(_)
42        | session_frame::Kind::StdinEof(_)
43        | session_frame::Kind::Start(_) => FrameKind::Request,
44        // Daemon → client: stdout/stderr stream and the terminal exit.
45        session_frame::Kind::Stdout(_)
46        | session_frame::Kind::Stderr(_)
47        | session_frame::Kind::Exit(_) => FrameKind::Response,
48    }
49}
50
51/// Wrap one `SessionFrame` in a SESSION-lane `Frame`.
52///
53/// The envelope carries the v1 defaults (`envelope_version`,
54/// `payload_encoding = NONE`, no deadline, empty trace context) from
55/// [`Frame::request`], with `payload_protocol` pinned to
56/// [`SESSION_PAYLOAD_PROTOCOL`], `request_id` set to the session-local `seq`,
57/// and `kind` derived from the frame's direction. A `SessionFrame` with no
58/// `kind` set (an empty oneof — never produced by the pump) defaults to
59/// `REQUEST`; it still round-trips, carrying an empty payload.
60pub fn session_frame_to_frame(frame: &SessionFrame, seq: u64) -> Frame {
61    let kind = frame
62        .kind
63        .as_ref()
64        .map_or(FrameKind::Request, frame_kind_for);
65    let mut wrapped =
66        Frame::request(SESSION_PAYLOAD_PROTOCOL, frame.encode_to_vec()).with_request_id(seq);
67    wrapped.kind = kind as i32;
68    wrapped
69}
70
71/// Encode one `SessionFrame` to complete wire bytes
72/// (`[1][u32 len][prost Frame]`), ready to write to a transport.
73///
74/// # Errors
75///
76/// [`FramingError::FrameTooLarge`] when the encoded envelope exceeds the frame
77/// cap — propagated from [`encode_framed`].
78pub fn encode_session_frame(frame: &SessionFrame, seq: u64) -> Result<Vec<u8>, FramingError> {
79    encode_framed(&session_frame_to_frame(frame, seq))
80}
81
82/// One `SessionFrame` decoded from the front of a byte buffer, plus how many
83/// wire bytes it occupied.
84///
85/// `PartialEq` only, not `Eq`: prost stops auto-deriving `Eq` for `SessionFrame`
86/// once its oneof includes `SessionStart` (which carries a repeated message
87/// field), and equality here is only ever used through `assert_eq!` (which needs
88/// `PartialEq`).
89#[derive(Debug, Clone, PartialEq)]
90pub struct DecodedSessionFrame {
91    /// The decoded session frame.
92    pub frame: SessionFrame,
93    /// Total wire bytes consumed (outer header + envelope body). The caller
94    /// advances its read buffer by exactly this many bytes.
95    pub consumed: usize,
96}
97
98/// Incrementally decode one SESSION-lane `SessionFrame` from the front of `buf`.
99///
100/// Returns `Ok(None)` when `buf` does not yet hold a complete frame — the
101/// caller reads more bytes and retries. On `Ok(Some(decoded))` the caller
102/// consumes `decoded.consumed` bytes. This mirrors [`try_decode_framed`] and
103/// adds the SESSION-lane payload decode on top.
104///
105/// # Errors
106///
107/// - [`SessionCodecError::Framing`] for a malformed outer frame (bad framing
108///   version, oversize, or undecodable envelope).
109/// - [`SessionCodecError::WrongProtocol`] when the envelope is well-formed but
110///   is not on the SESSION lane — a caller multiplexing lanes must route by
111///   `payload_protocol` before calling this.
112/// - [`SessionCodecError::Decode`] when the envelope payload is not a valid
113///   `SessionFrame`.
114pub fn try_decode_session_frame(
115    buf: &[u8],
116) -> Result<Option<DecodedSessionFrame>, SessionCodecError> {
117    let Some(decoded) = try_decode_framed(buf).map_err(SessionCodecError::Framing)? else {
118        return Ok(None);
119    };
120    if decoded.frame.payload_protocol != SESSION_PAYLOAD_PROTOCOL {
121        return Err(SessionCodecError::WrongProtocol {
122            got: decoded.frame.payload_protocol,
123        });
124    }
125    let frame = SessionFrame::decode(decoded.frame.payload.as_slice())
126        .map_err(SessionCodecError::Decode)?;
127    Ok(Some(DecodedSessionFrame {
128        frame,
129        consumed: decoded.consumed,
130    }))
131}
132
133/// Errors from [`try_decode_session_frame`].
134#[derive(Debug, thiserror::Error)]
135pub enum SessionCodecError {
136    /// The outer v1 frame was malformed (see [`FramingError`]).
137    #[error(transparent)]
138    Framing(FramingError),
139    /// The frame decoded but was not on the SESSION lane. The value is the
140    /// `payload_protocol` that was seen; the expected lane is `0x5350`.
141    #[error("frame carried payload_protocol {got:#06X}, expected SESSION lane 0x5350")]
142    WrongProtocol {
143        /// The payload protocol the frame actually carried.
144        got: u32,
145    },
146    /// The envelope payload was not a valid `SessionFrame`.
147    #[error("failed to decode SessionFrame payload: {0}")]
148    Decode(prost::DecodeError),
149}
150
151#[cfg(test)]
152mod tests;
running_process/broker/session_codec.rs

running_process/broker/
session_codec.rs
