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    /// Authenticated identity, set by the server. The key-value store scopes its
70    /// rows by this `user_id`, and the SDK cannot set it.
71    pub user_id: u32,
72    /// The caller's consensus client id, for logging and audit only. Never zero
73    /// for a client, zero on a request the server issues itself.
74    pub client_id: u128,
75    /// `gen_ai.conversation.id` echoed for the audit log, never used to route.
76    pub correlation: Option<String>,
77    /// Stable identity minted by the originating SDK for a logical managed
78    /// mutation. Read-only commands leave it absent.
79    #[serde(default, skip_serializing_if = "Option::is_none")]
80    pub operation_id: Option<u128>,
81    /// Set by the server, like `user_id`: true when the caller holds a global
82    /// read permission. Widens the read-side ops (key-value scan/namespaces,
83    /// fork list) to every user's rows, while writes stay scoped to `user_id`
84    /// regardless.
85    #[serde(default)]
86    pub read_all: bool,
87    /// Server-stamped stream identity. A session command without it fails closed.
88    #[serde(default, skip_serializing_if = "Option::is_none")]
89    pub scope: Option<ForwardedScope>,
90    /// The managed command code (browse, key-value, or fork block). The backend
91    /// dispatches on it.
92    pub command_code: u32,
93    /// Opaque CBOR request the SDK sent, not decoded in transit.
94    #[serde(with = "crate::encoding::bin_bytes")]
95    pub payload: Vec<u8>,
96    /// The caller's effective capability grants, stamped by the server. Empty
97    /// (the default, skipped on the wire) when authorization is not in effect.
98    #[serde(default, skip_serializing_if = "Vec::is_empty")]
99    pub grants: Vec<crate::authz::Grant>,
100    /// Set by the server when the client wrapped this read in a
101    /// [`crate::mutation::ManagedReadEnvelope`]. The backend answers only once
102    /// its fold has reached the surface's mutation log head.
103    #[serde(default, skip_serializing_if = "is_false")]
104    pub read_barrier: bool,
105    /// Set by the server from [`crate::mutation::ManagedReadEnvelope`]: the
106    /// backend answers only once its fold for the read's surface has applied
107    /// this position.
108    #[serde(default, skip_serializing_if = "Option::is_none")]
109    pub min_position: Option<crate::mutation::MutationPosition>,
110    /// Set by the server for a managed request an SDK client sent over the
111    /// binary protocol. The backend answers with a
112    /// [`crate::mutation::ManagedReplyEnvelope`] carrying its applied
113    /// frontier.
114    #[serde(default, skip_serializing_if = "is_false")]
115    pub positioned_reply: bool,
116    /// Set by the server from [`crate::mutation::ManagedReadEnvelope`]: the
117    /// session positions a session read must find applied.
118    #[serde(default, skip_serializing_if = "Vec::is_empty")]
119    pub min_sessions: Vec<crate::mutation::SessionPosition>,
120    /// The tenancy of the server's replicated authorization state. The
121    /// backend refuses a frame whose tenancy differs from its own.
122    #[serde(default, skip_serializing_if = "crate::authz::Tenancy::is_deployment")]
123    pub tenancy: crate::authz::Tenancy,
124    /// Set by the server on a mutation a client sent over the binary protocol:
125    /// the agent the calling connection declared in its presence. A claim the
126    /// client makes, never verified. Absent over HTTP and on the server's own
127    /// requests.
128    #[serde(default, skip_serializing_if = "Option::is_none")]
129    pub agent: Option<crate::agent::AgentId>,
130    /// Set by the server from [`ON_BEHALF_OF_HEADER`](crate::http::ON_BEHALF_OF_HEADER)
131    /// on an HTTP write: the end user an intermediary wrote for. A claim, never
132    /// verified.
133    #[serde(default, skip_serializing_if = "Option::is_none")]
134    pub on_behalf_of: Option<String>,
135}
136
137#[expect(clippy::trivially_copy_pass_by_ref, reason = "serde skip predicate")]
138fn is_false(value: &bool) -> bool {
139    !*value
140}
141
142#[cfg(all(test, feature = "cbor"))]
143mod tests {
144    use super::*;
145    use crate::framing::{decode_named, encode_named};
146
147    // A contiguous byte-string run inside the encoded frame proves the opaque
148    // field rode as a CBOR byte string (major type 2), not an array of ints.
149    // The values are all >= 0x18, which an array would have to widen to two
150    // bytes each, so the byte string is also strictly smaller.
151    fn contains_byte_string(frame: &[u8], len: u8, fill: u8) -> bool {
152        let head = 0x40 | len; // byte string, length in the low 5 bits
153        let mut want = vec![head];
154        want.extend(std::iter::repeat_n(fill, len as usize));
155        frame.windows(want.len()).any(|w| w == want.as_slice())
156    }
157
158    #[test]
159    fn given_forwarded_query_when_encoded_then_envelope_rides_as_a_byte_string() {
160        let frame = encode_named(&ForwardedQuery {
161            user_id: 7,
162            client_id: 42,
163            correlation: Some("conv-1".to_owned()),
164            query_envelope: vec![0x90; 5],
165            grants: Vec::new(),
166            tenancy: Default::default(),
167            scope: None,
168        })
169        .expect("encodes");
170        assert!(
171            contains_byte_string(&frame, 5, 0x90),
172            "query_envelope must encode as a CBOR byte string, got {frame:02x?}"
173        );
174    }
175
176    #[test]
177    fn given_forwarded_command_when_encoded_then_payload_rides_as_a_byte_string() {
178        let frame = encode_named(&ForwardedCommand {
179            read_barrier: false,
180            min_position: None,
181            positioned_reply: false,
182            min_sessions: Vec::new(),
183            tenancy: Default::default(),
184            user_id: 7,
185            client_id: 42,
186            correlation: None,
187            operation_id: Some(17),
188            read_all: true,
189            scope: None,
190            command_code: 1_000_000,
191            payload: vec![0x90; 5],
192            grants: Vec::new(),
193            agent: None,
194            on_behalf_of: None,
195        })
196        .expect("encodes");
197        assert!(
198            contains_byte_string(&frame, 5, 0x90),
199            "payload must encode as a CBOR byte string, got {frame:02x?}"
200        );
201    }
202
203    #[test]
204    fn given_forwarded_frames_when_round_tripped_then_should_preserve_fields() {
205        let query = ForwardedQuery {
206            user_id: 1,
207            client_id: u128::MAX,
208            correlation: Some("c".to_owned()),
209            query_envelope: vec![0xff, 0x00, 0x18, 0x7f],
210            grants: Vec::new(),
211            tenancy: Default::default(),
212            scope: Some(ForwardedScope {
213                stream_id: 3,
214                stream: "acme".to_owned(),
215                stream_created_at_micros: 9,
216            }),
217        };
218        let back: ForwardedQuery =
219            decode_named(&encode_named(&query).expect("encodes")).expect("decodes");
220        assert_eq!(back, query);
221
222        let command = ForwardedCommand {
223            read_barrier: false,
224            min_position: None,
225            positioned_reply: false,
226            min_sessions: Vec::new(),
227            tenancy: Default::default(),
228            user_id: 2,
229            client_id: 9,
230            correlation: None,
231            operation_id: Some(9),
232            read_all: false,
233            scope: None,
234            command_code: 42,
235            payload: vec![0xde, 0xad, 0xbe, 0xef],
236            grants: Vec::new(),
237            agent: None,
238            on_behalf_of: None,
239        };
240        let back: ForwardedCommand =
241            decode_named(&encode_named(&command).expect("encodes")).expect("decodes");
242        assert_eq!(back, command);
243    }
244
245    #[test]
246    fn given_a_root_caller_with_an_agent_when_round_tripped_then_should_keep_both() {
247        let caller = Caller {
248            user_id: 0,
249            agent: Some("orbit-planner".parse().expect("agent id")),
250            on_behalf_of: None,
251        };
252
253        let back: Caller = decode_named(&encode_named(&caller).expect("encodes")).expect("decodes");
254
255        assert_eq!(back, caller);
256    }
257
258    #[test]
259    fn given_a_caller_written_for_an_end_user_when_round_tripped_then_should_keep_who_it_was_for() {
260        let caller = Caller {
261            user_id: 3,
262            agent: None,
263            on_behalf_of: Some("1793690355287073".to_owned()),
264        };
265
266        let back: Caller = decode_named(&encode_named(&caller).expect("encodes")).expect("decodes");
267
268        assert_eq!(back, caller);
269    }
270
271    #[test]
272    fn given_a_caller_with_an_empty_agent_when_decoded_then_should_be_refused() {
273        #[derive(serde::Serialize)]
274        struct Unchecked<'a> {
275            user_id: u32,
276            agent: &'a str,
277        }
278        let frame = encode_named(&Unchecked {
279            user_id: 3,
280            agent: "",
281        })
282        .expect("encodes");
283
284        assert!(decode_named::<Caller>(&frame).is_err());
285    }
286
287    #[test]
288    fn given_a_forwarded_command_with_an_agent_when_read_without_the_field_then_should_ignore_it() {
289        #[derive(Debug, PartialEq, serde::Deserialize)]
290        struct BeforeAgent {
291            user_id: u32,
292            client_id: u128,
293            command_code: u32,
294        }
295        let command = ForwardedCommand {
296            read_barrier: false,
297            min_position: None,
298            positioned_reply: false,
299            min_sessions: Vec::new(),
300            tenancy: Default::default(),
301            user_id: 0,
302            client_id: 9,
303            correlation: None,
304            operation_id: Some(9),
305            read_all: false,
306            scope: None,
307            command_code: 42,
308            payload: vec![1],
309            grants: Vec::new(),
310            agent: Some("orbit-planner".parse().expect("agent id")),
311            on_behalf_of: None,
312        };
313
314        let before: BeforeAgent =
315            decode_named(&encode_named(&command).expect("encodes")).expect("decodes");
316
317        assert_eq!(
318            before,
319            BeforeAgent {
320                user_id: 0,
321                client_id: 9,
322                command_code: 42,
323            }
324        );
325    }
326}
laser_wire/forward.rs

laser_wire/
forward.rs
