laser_wire/
forward.rs

1// Frames the managed backend consumes on behalf of a client. The SDK never
2// sends these directly. They live here so the encoding has one definition and
3// the golden corpus pins them like every other frame.
4
5use serde::{Deserialize, Serialize};
6
7/// Who asked for a request the managed backend records, present only when a
8/// client asked. A record the backend or the streaming server issues on its own
9/// carries none.
10#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
11pub struct Caller {
12    /// The authenticated user. The root user is id zero.
13    pub user_id: u32,
14    /// The agent the caller's connection declared in its presence. A claim the
15    /// client makes, never verified.
16    #[serde(default, skip_serializing_if = "Option::is_none")]
17    pub agent: Option<crate::agent::AgentId>,
18    /// The end user an intermediary wrote for, as it named that user in
19    /// [`ON_BEHALF_OF_HEADER`](crate::http::ON_BEHALF_OF_HEADER). A claim the
20    /// backend records, never verified.
21    #[serde(default, skip_serializing_if = "Option::is_none")]
22    pub on_behalf_of: Option<String>,
23}
24
25/// Trusted stream identity stamped by the server before forwarding a session request.
26#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
27pub struct ForwardedScope {
28    pub stream_id: u32,
29    pub stream: String,
30    pub stream_created_at_micros: u64,
31}
32
33/// A forwarded managed query. Carries the authenticated identity the SDK cannot
34/// set itself, plus the opaque request the SDK sent. CBOR-encoded, named fields.
35#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
36pub struct ForwardedQuery {
37    /// Authenticated identity. Audit today, per-user stream scoping later.
38    /// Trusted: set by the server, not the client.
39    pub user_id: u32,
40    /// Originating client id, for logging and audit only. Never zero for a
41    /// client, zero on a request the server issues itself.
42    pub client_id: u128,
43    /// `gen_ai.conversation.id` echoed for the audit log, never used to route.
44    pub correlation: Option<String>,
45    /// Opaque CBOR `QueryEnvelope` from the SDK, not decoded in transit.
46    #[serde(with = "crate::encoding::bin_bytes")]
47    pub query_envelope: Vec<u8>,
48    /// The caller's effective capability grants, stamped by the server. Empty
49    /// (the default, skipped on the wire) when authorization is not in effect.
50    #[serde(default, skip_serializing_if = "Vec::is_empty")]
51    pub grants: Vec<crate::authz::Grant>,
52    /// The tenancy of the server's replicated authorization state. The
53    /// backend refuses a frame whose tenancy differs from its own.
54    #[serde(default, skip_serializing_if = "crate::authz::Tenancy::is_deployment")]
55    pub tenancy: crate::authz::Tenancy,
56    /// The stream of the queried index, stamped once the caller's native read
57    /// authority over it was checked. The backend refuses a query of a
58    /// stream-scoped index that arrives without this scope or with another.
59    #[serde(default, skip_serializing_if = "Option::is_none")]
60    pub scope: Option<ForwardedScope>,
61}
62
63/// A forwarded keyed managed command (registry browse, key-value, forks). Unlike
64/// `ForwardedQuery`, several op kinds share one path, so the frame carries
65/// `command_code` and the backend dispatches on it. `payload` is the opaque
66/// CBOR request the SDK sent.
67#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
68pub struct ForwardedCommand {
69    /// The client that asked, stamped by the server, so the SDK cannot set it.
70    /// The backend records it and authorizes with `grants`. Absent on a request
71    /// the server issues on its own.
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub caller: Option<Caller>,
74    /// The caller's consensus client id, for logs and correlation only, never
75    /// identity.
76    pub client_id: u128,
77    /// `gen_ai.conversation.id` echoed for the audit log, never used to route.
78    pub correlation: Option<String>,
79    /// Stable identity minted by the originating SDK for a logical managed
80    /// mutation. Read-only commands leave it absent.
81    #[serde(default, skip_serializing_if = "Option::is_none")]
82    pub operation_id: Option<u128>,
83    /// Vestigial. The server stamps `false`, the plane ignores it, and the
84    /// caller's grants govern access. It stays on the wire as the contract's
85    /// current structure, never as a widening flag.
86    #[serde(default)]
87    pub read_all: bool,
88    /// Server-stamped stream identity. A session command without it fails closed.
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub scope: Option<ForwardedScope>,
91    /// The managed command code (browse, key-value, or fork block). The backend
92    /// dispatches on it.
93    pub command_code: u32,
94    /// Opaque CBOR request the SDK sent, not decoded in transit.
95    #[serde(with = "crate::encoding::bin_bytes")]
96    pub payload: Vec<u8>,
97    /// The caller's effective capability grants, stamped by the server. Empty
98    /// (the default, skipped on the wire) when authorization is not in effect.
99    #[serde(default, skip_serializing_if = "Vec::is_empty")]
100    pub grants: Vec<crate::authz::Grant>,
101    /// Set by the server when the client wrapped this read in a
102    /// [`crate::mutation::ManagedReadEnvelope`]. The backend answers only once
103    /// its fold has reached the surface's mutation log head.
104    #[serde(default, skip_serializing_if = "is_false")]
105    pub read_barrier: bool,
106    /// Set by the server from [`crate::mutation::ManagedReadEnvelope`]: the
107    /// backend answers only once its fold for the read's surface has applied
108    /// this position.
109    #[serde(default, skip_serializing_if = "Option::is_none")]
110    pub min_position: Option<crate::mutation::MutationPosition>,
111    /// Set by the server for a managed request an SDK client sent over the
112    /// binary protocol. The backend answers with a
113    /// [`crate::mutation::ManagedReplyEnvelope`] carrying its applied
114    /// frontier.
115    #[serde(default, skip_serializing_if = "is_false")]
116    pub positioned_reply: bool,
117    /// Set by the server from [`crate::mutation::ManagedReadEnvelope`]: the
118    /// session positions a session read must find applied.
119    #[serde(default, skip_serializing_if = "Vec::is_empty")]
120    pub min_sessions: Vec<crate::mutation::SessionPosition>,
121    /// The tenancy of the server's replicated authorization state. The
122    /// backend refuses a frame whose tenancy differs from its own.
123    #[serde(default, skip_serializing_if = "crate::authz::Tenancy::is_deployment")]
124    pub tenancy: crate::authz::Tenancy,
125}
126
127#[expect(clippy::trivially_copy_pass_by_ref, reason = "serde skip predicate")]
128fn is_false(value: &bool) -> bool {
129    !*value
130}
131
132#[cfg(all(test, feature = "cbor"))]
133mod tests {
134    use super::*;
135    use crate::framing::{decode_named, encode_named};
136
137    // A contiguous byte-string run inside the encoded frame proves the opaque
138    // field rode as a CBOR byte string (major type 2), not an array of ints.
139    // The values are all >= 0x18, which an array would have to widen to two
140    // bytes each, so the byte string is also strictly smaller.
141    fn contains_byte_string(frame: &[u8], len: u8, fill: u8) -> bool {
142        let head = 0x40 | len; // byte string, length in the low 5 bits
143        let mut want = vec![head];
144        want.extend(std::iter::repeat_n(fill, len as usize));
145        frame.windows(want.len()).any(|w| w == want.as_slice())
146    }
147
148    #[test]
149    fn given_forwarded_query_when_encoded_then_envelope_rides_as_a_byte_string() {
150        let frame = encode_named(&ForwardedQuery {
151            user_id: 7,
152            client_id: 42,
153            correlation: Some("conv-1".to_owned()),
154            query_envelope: vec![0x90; 5],
155            grants: Vec::new(),
156            tenancy: Default::default(),
157            scope: None,
158        })
159        .expect("encodes");
160        assert!(
161            contains_byte_string(&frame, 5, 0x90),
162            "query_envelope must encode as a CBOR byte string, got {frame:02x?}"
163        );
164    }
165
166    #[test]
167    fn given_forwarded_command_when_encoded_then_payload_rides_as_a_byte_string() {
168        let frame = encode_named(&ForwardedCommand {
169            read_barrier: false,
170            min_position: None,
171            positioned_reply: false,
172            min_sessions: Vec::new(),
173            tenancy: Default::default(),
174            caller: Some(Caller {
175                user_id: 7,
176                agent: None,
177                on_behalf_of: None,
178            }),
179            client_id: 42,
180            correlation: None,
181            operation_id: Some(17),
182            read_all: true,
183            scope: None,
184            command_code: 1_000_000,
185            payload: vec![0x90; 5],
186            grants: Vec::new(),
187        })
188        .expect("encodes");
189        assert!(
190            contains_byte_string(&frame, 5, 0x90),
191            "payload must encode as a CBOR byte string, got {frame:02x?}"
192        );
193    }
194
195    #[test]
196    fn given_forwarded_frames_when_round_tripped_then_should_preserve_fields() {
197        let query = ForwardedQuery {
198            user_id: 1,
199            client_id: u128::MAX,
200            correlation: Some("c".to_owned()),
201            query_envelope: vec![0xff, 0x00, 0x18, 0x7f],
202            grants: Vec::new(),
203            tenancy: Default::default(),
204            scope: Some(ForwardedScope {
205                stream_id: 3,
206                stream: "acme".to_owned(),
207                stream_created_at_micros: 9,
208            }),
209        };
210        let back: ForwardedQuery =
211            decode_named(&encode_named(&query).expect("encodes")).expect("decodes");
212        assert_eq!(back, query);
213
214        let command = ForwardedCommand {
215            read_barrier: false,
216            min_position: None,
217            positioned_reply: false,
218            min_sessions: Vec::new(),
219            tenancy: Default::default(),
220            caller: Some(Caller {
221                user_id: 2,
222                agent: Some("orbit-planner".parse().expect("agent id")),
223                on_behalf_of: Some("1793690355287073".to_owned()),
224            }),
225            client_id: 9,
226            correlation: None,
227            operation_id: Some(9),
228            read_all: false,
229            scope: None,
230            command_code: 42,
231            payload: vec![0xde, 0xad, 0xbe, 0xef],
232            grants: Vec::new(),
233        };
234        let back: ForwardedCommand =
235            decode_named(&encode_named(&command).expect("encodes")).expect("decodes");
236        assert_eq!(back, command);
237    }
238
239    #[test]
240    fn given_a_root_caller_with_an_agent_when_round_tripped_then_should_keep_both() {
241        let caller = Caller {
242            user_id: 0,
243            agent: Some("orbit-planner".parse().expect("agent id")),
244            on_behalf_of: None,
245        };
246
247        let back: Caller = decode_named(&encode_named(&caller).expect("encodes")).expect("decodes");
248
249        assert_eq!(back, caller);
250    }
251
252    #[test]
253    fn given_a_caller_written_for_an_end_user_when_round_tripped_then_should_keep_who_it_was_for() {
254        let caller = Caller {
255            user_id: 3,
256            agent: None,
257            on_behalf_of: Some("1793690355287073".to_owned()),
258        };
259
260        let back: Caller = decode_named(&encode_named(&caller).expect("encodes")).expect("decodes");
261
262        assert_eq!(back, caller);
263    }
264
265    #[test]
266    fn given_a_caller_with_an_empty_agent_when_decoded_then_should_be_refused() {
267        #[derive(serde::Serialize)]
268        struct Unchecked<'a> {
269            user_id: u32,
270            agent: &'a str,
271        }
272        let frame = encode_named(&Unchecked {
273            user_id: 3,
274            agent: "",
275        })
276        .expect("encodes");
277
278        assert!(decode_named::<Caller>(&frame).is_err());
279    }
280
281    #[test]
282    fn given_a_request_the_server_issues_on_its_own_when_encoded_then_should_carry_no_caller() {
283        let command = ForwardedCommand {
284            read_barrier: false,
285            min_position: None,
286            positioned_reply: false,
287            min_sessions: Vec::new(),
288            tenancy: Default::default(),
289            caller: None,
290            client_id: 0,
291            correlation: None,
292            operation_id: Some(9),
293            read_all: false,
294            scope: None,
295            command_code: 42,
296            payload: vec![1],
297            grants: Vec::new(),
298        };
299
300        let frame = encode_named(&command).expect("encodes");
301        let back: ForwardedCommand = decode_named(&frame).expect("decodes");
302
303        assert_eq!(back, command);
304        assert!(
305            !frame
306                .windows(b"caller".len())
307                .any(|window| window == b"caller"),
308            "an absent caller is omitted from the frame"
309        );
310    }
311}
laser_wire/forward.rs

laser_wire/
forward.rs
