laser_wire/
session.rs

1use crate::agent::{
2    AgentId, Budget, ConversationId, CorrelationId, LogPosition, SdkInfo, SessionStatus, TokenUsage,
3};
4use crate::error::InvalidError;
5use crate::graph::SourceRef;
6use crate::limits::{
7    MAX_FILTER_SOURCE_NAME_BYTES, MAX_HEARTBEAT_SESSIONS, MAX_SESSION_CURSOR_BYTES,
8    MAX_SESSION_LABEL_BYTES, MAX_SESSION_TEXT_QUERY_BYTES,
9};
10use crate::schema::Digest32;
11use crate::validate::Validate;
12use serde::{Deserialize, Serialize};
13use std::collections::BTreeMap;
14
15/// The resource surface used to narrow a session link read.
16#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(rename_all = "snake_case")]
18pub enum LinkSurface {
19    Memory,
20    Kv,
21    GraphNode,
22    GraphEdge,
23    Projection,
24    Child,
25}
26
27/// The relationship a session has with a linked resource.
28#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
29#[serde(rename_all = "snake_case")]
30pub enum LinkRelation {
31    Wrote,
32    Recalled,
33    Touched,
34}
35
36/// Read requests. Every request names the stream before any other field.
37pub mod request {
38    use super::*;
39
40    #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
41    pub struct SessionGet {
42        pub stream: String,
43        pub id: ConversationId,
44    }
45
46    #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
47    pub struct SessionList {
48        pub stream: String,
49        #[serde(default, skip_serializing_if = "Option::is_none")]
50        pub status: Option<SessionStatus>,
51        /// Only sessions in the tree rooted at this session.
52        #[serde(default, skip_serializing_if = "Option::is_none")]
53        pub root: Option<ConversationId>,
54        /// Only sessions whose label starts with this prefix.
55        #[serde(default, skip_serializing_if = "Option::is_none")]
56        pub label_prefix: Option<String>,
57        #[serde(default, skip_serializing_if = "Option::is_none")]
58        pub agent: Option<AgentId>,
59        #[serde(default, skip_serializing_if = "Option::is_none")]
60        pub text: Option<String>,
61        #[serde(default, skip_serializing_if = "Option::is_none")]
62        pub cursor: Option<String>,
63        /// The page size. Zero leaves it to the server.
64        pub limit: u32,
65        pub want_total: bool,
66    }
67
68    #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
69    pub struct SessionEvents {
70        pub stream: String,
71        pub id: ConversationId,
72        #[serde(default, skip_serializing_if = "Option::is_none")]
73        pub cursor: Option<String>,
74        /// The page size. Zero leaves it to the server.
75        pub limit: u32,
76        pub fixed_frontier: bool,
77    }
78
79    #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
80    pub struct SessionState {
81        pub stream: String,
82        pub id: ConversationId,
83        /// The most history rows. Zero leaves it to the server.
84        pub history_limit: u32,
85    }
86
87    #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
88    pub struct SessionLinks {
89        pub stream: String,
90        pub id: ConversationId,
91        #[serde(default, skip_serializing_if = "Option::is_none")]
92        pub surface: Option<LinkSurface>,
93    }
94
95    #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
96    pub struct SessionSources {
97        pub stream: String,
98        pub id: ConversationId,
99        /// Return only the registered lane identity for a native lane writer.
100        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
101        pub lane_only: bool,
102    }
103
104    #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
105    pub struct SessionChanges {
106        pub stream: String,
107        pub after: u64,
108        /// The page size. Zero leaves it to the server.
109        pub limit: u32,
110        /// The change series `after` was read from. A different series answers a
111        /// resync with the current series and the newest sequence as `floor`, so
112        /// `Some(0)` with `after` zero is the one-call start of a follower. `None`
113        /// skips the series check and returns the retained rows after `after`.
114        #[serde(default, skip_serializing_if = "Option::is_none")]
115        pub series: Option<u64>,
116    }
117}
118
119/// How far the session index has folded one source partition. Offsets are
120/// never compared across topics. `topic_generation` is the topic creation time
121/// in microseconds, so a recreated topic with a reused id is a different source.
122#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
123pub struct SourceFrontier {
124    pub topic_id: u32,
125    pub topic_generation: u64,
126    pub partition_id: u32,
127    /// The last offset folded into the index, absent before the first record.
128    #[serde(default, skip_serializing_if = "Option::is_none")]
129    pub folded: Option<u64>,
130    /// The live partition head when the read probed it.
131    #[serde(default, skip_serializing_if = "Option::is_none")]
132    pub head: Option<u64>,
133    /// The first offset still retained by the source.
134    #[serde(default, skip_serializing_if = "Option::is_none")]
135    pub retained_from: Option<u64>,
136}
137
138/// Why part of a session timeline is missing.
139#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
140#[serde(rename_all = "snake_case")]
141pub enum GapReason {
142    /// The source expired these offsets before the index folded them.
143    ExpiredBeforeFold,
144    /// The index pruned these offsets after the source expired them.
145    Pruned,
146    /// The session reached its event cap, so these records were counted only.
147    Truncated,
148    /// The source is being rebuilt after a reset.
149    Rebuilding,
150    #[serde(other)]
151    Unrecognized,
152}
153
154/// A missing inclusive offset range on one source partition.
155#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
156pub struct SourceGap {
157    pub topic_id: u32,
158    pub topic_generation: u64,
159    pub partition_id: u32,
160    pub from: u64,
161    pub to: u64,
162    pub reason: GapReason,
163}
164
165/// The inclusive offset range on one source partition that holds the payloads
166/// of a timeline page. A range is a hint for bounded polls, not a promise that
167/// every offset in it belongs to the session.
168#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
169pub struct PayloadRange {
170    pub topic_id: u32,
171    pub topic_generation: u64,
172    pub partition_id: u32,
173    pub first: u64,
174    pub last: u64,
175}
176
177/// What the fold did with one state record.
178#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
179#[serde(rename_all = "snake_case")]
180pub enum StateOutcome {
181    /// The patch or snapshot changed the document.
182    Applied,
183    /// A patch operation failed, so the whole patch was rejected.
184    Rejected,
185    /// The base revision did not match the applied revision.
186    Stale,
187    /// The operation id was already applied.
188    Duplicate,
189    #[serde(other)]
190    Unrecognized,
191}
192
193/// One entry in the state history of a session.
194#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
195pub struct StateChange {
196    /// The document revision after this record.
197    pub revision: u64,
198    #[serde(default, skip_serializing_if = "Option::is_none")]
199    pub op_id: Option<String>,
200    pub outcome: StateOutcome,
201    pub at: LogPosition,
202    pub broker_ts: u64,
203    #[serde(default, skip_serializing_if = "Option::is_none")]
204    pub old_digest: Option<Digest32>,
205    #[serde(default, skip_serializing_if = "Option::is_none")]
206    pub new_digest: Option<Digest32>,
207    #[serde(default, skip_serializing_if = "Option::is_none")]
208    pub reason: Option<String>,
209}
210
211/// Conditions that make a session summary partial.
212#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
213pub struct SessionFlags {
214    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
215    pub label_truncated: bool,
216    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
217    pub overflow: bool,
218    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
219    pub events_truncated: bool,
220    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
221    pub lane_conflict: bool,
222    /// A control record arrived on a partition other than the session's
223    /// control route and was not applied.
224    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
225    pub control_conflict: bool,
226    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
227    pub rebuilding: bool,
228    /// Liveness is not known yet because the heartbeat tail has not caught
229    /// up, so `idle` is not meaningful.
230    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
231    pub liveness_unknown: bool,
232}
233
234/// The summary of one session.
235#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
236pub struct SessionInfo {
237    pub stream: String,
238    pub id: ConversationId,
239    #[serde(default, skip_serializing_if = "Option::is_none")]
240    pub label: Option<String>,
241    #[serde(default, skip_serializing_if = "Option::is_none")]
242    pub namespace: Option<String>,
243    #[serde(default, skip_serializing_if = "Option::is_none")]
244    pub agent: Option<AgentId>,
245    #[serde(default, skip_serializing_if = "Option::is_none")]
246    pub parent: Option<ConversationId>,
247    #[serde(default, skip_serializing_if = "Option::is_none")]
248    pub root: Option<ConversationId>,
249    pub status: SessionStatus,
250    pub idle: bool,
251    pub over_budget: bool,
252    pub pause_requested: bool,
253    pub cancel_requested: bool,
254    #[serde(default, skip_serializing_if = "Option::is_none")]
255    pub started_at: Option<u64>,
256    #[serde(default, skip_serializing_if = "Option::is_none")]
257    pub ended_at: Option<u64>,
258    #[serde(default, skip_serializing_if = "Option::is_none")]
259    pub first_event_at: Option<u64>,
260    #[serde(default, skip_serializing_if = "Option::is_none")]
261    pub last_event_at: Option<u64>,
262    #[serde(default, skip_serializing_if = "Option::is_none")]
263    pub last_heartbeat_at: Option<u64>,
264    pub events: u64,
265    pub model_calls: u64,
266    pub tool_calls: u64,
267    pub tokens_in: u64,
268    pub tokens_out: u64,
269    pub cost_micros: u64,
270    pub errors: u64,
271    #[serde(default, skip_serializing_if = "Option::is_none")]
272    pub budget: Option<Budget>,
273    #[serde(default, skip_serializing_if = "Option::is_none")]
274    pub sdk: Option<SdkInfo>,
275    #[serde(default)]
276    pub flags: SessionFlags,
277    /// Work records held while the session was paused and not yet handled.
278    #[serde(default, skip_serializing_if = "is_zero")]
279    pub held: u64,
280    #[serde(default, skip_serializing_if = "Vec::is_empty")]
281    pub frontier: Vec<SourceFrontier>,
282}
283
284fn is_zero(value: &u64) -> bool {
285    *value == 0
286}
287
288/// One page of session summaries.
289#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
290pub struct SessionPage {
291    pub items: Vec<SessionInfo>,
292    #[serde(default, skip_serializing_if = "Option::is_none")]
293    pub cursor: Option<String>,
294    #[serde(default, skip_serializing_if = "Option::is_none")]
295    pub total: Option<u64>,
296    #[serde(default, skip_serializing_if = "Option::is_none")]
297    pub searched: Option<u64>,
298    pub truncated: bool,
299    /// The index is rebuilding on this node, so `items` and `total` can be incomplete.
300    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
301    pub rebuilding: bool,
302}
303
304/// One indexed record on a session timeline. The summary never holds prompt
305/// text or tool argument values.
306#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
307pub struct SessionEvent {
308    pub at: SourceRef,
309    pub session: ConversationId,
310    pub broker_ts: u64,
311    pub kind: String,
312    #[serde(default, skip_serializing_if = "Option::is_none")]
313    pub operation: Option<String>,
314    pub display: String,
315    #[serde(default, skip_serializing_if = "Option::is_none")]
316    pub agent: Option<AgentId>,
317    #[serde(default, skip_serializing_if = "Option::is_none")]
318    pub addressee: Option<String>,
319    #[serde(default, skip_serializing_if = "Option::is_none")]
320    pub correlation: Option<CorrelationId>,
321    #[serde(default, skip_serializing_if = "Option::is_none")]
322    pub cause: Option<LogPosition>,
323    #[serde(default, skip_serializing_if = "Option::is_none")]
324    pub tool: Option<String>,
325    #[serde(default, skip_serializing_if = "Option::is_none")]
326    pub usage: Option<TokenUsage>,
327    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
328    pub after_end: bool,
329    /// The principal whose enrolled key verified this record's signature.
330    #[serde(default, skip_serializing_if = "Option::is_none")]
331    pub verified_actor: Option<String>,
332    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
333    pub summary: BTreeMap<String, serde_json::Value>,
334}
335
336/// One page of a session timeline.
337#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
338pub struct SessionEventsPage {
339    pub items: Vec<SessionEvent>,
340    #[serde(default, skip_serializing_if = "Option::is_none")]
341    pub cursor: Option<String>,
342    #[serde(default, skip_serializing_if = "Vec::is_empty")]
343    pub ranges: Vec<PayloadRange>,
344    #[serde(default, skip_serializing_if = "Vec::is_empty")]
345    pub frontier: Vec<SourceFrontier>,
346    pub fixed_frontier: bool,
347    #[serde(default, skip_serializing_if = "Vec::is_empty")]
348    pub gaps: Vec<SourceGap>,
349}
350
351/// The folded state document of a session.
352#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
353pub struct SessionStateView {
354    pub revision: u64,
355    pub document: serde_json::Value,
356    #[serde(default, skip_serializing_if = "Vec::is_empty")]
357    pub history: Vec<StateChange>,
358    #[serde(default, skip_serializing_if = "Option::is_none")]
359    pub frontier: Option<SourceFrontier>,
360    pub complete: bool,
361}
362
363/// One resource a session wrote, recalled, or touched.
364#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
365pub struct SessionLink {
366    pub surface: LinkSurface,
367    pub resource: String,
368    pub item: String,
369    pub relation: LinkRelation,
370    pub first: LogPosition,
371    pub last: LogPosition,
372}
373
374/// The resources linked to one session.
375#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
376pub struct SessionLinksView {
377    pub links: Vec<SessionLink>,
378    #[serde(default, skip_serializing_if = "Vec::is_empty")]
379    pub frontier: Vec<SourceFrontier>,
380    pub truncated: bool,
381}
382
383/// The source partitions that hold records of one session.
384#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
385pub struct SessionSources {
386    pub sources: Vec<SourceFrontier>,
387    /// Registered lane topic ID, creation generation, and partition count.
388    #[serde(default, skip_serializing_if = "Option::is_none")]
389    pub lane: Option<(u32, u64, u32)>,
390}
391
392/// One committed change batch in a stream.
393#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
394pub struct SessionChangeRow {
395    pub seq: u64,
396    pub sessions: Vec<ConversationId>,
397    /// The records behind the change: the folded records for `records`, the
398    /// link source records for `links`, and for `expiry` the folded record
399    /// whose time expired the sessions. Empty for `maintenance` and for an
400    /// expiry the clock decided.
401    #[serde(default, skip_serializing_if = "Vec::is_empty")]
402    pub positions: Vec<LogPosition>,
403    pub truncated: bool,
404    #[serde(default, skip_serializing_if = "SessionChangeOrigin::is_records")]
405    pub origin: SessionChangeOrigin,
406}
407
408/// What made the sessions of one change row change.
409#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
410#[serde(rename_all = "snake_case")]
411pub enum SessionChangeOrigin {
412    /// Records folded from the session sources.
413    #[default]
414    Records,
415    /// Links written by managed mutations or projections.
416    Links,
417    /// A source partition was reset or left the registration, and its
418    /// sessions were rebuilt.
419    Maintenance,
420    /// Sessions idle past their retention were removed.
421    Expiry,
422}
423
424impl SessionChangeOrigin {
425    pub fn is_records(&self) -> bool {
426        *self == Self::Records
427    }
428}
429
430/// Change rows after a sequence. On a series mismatch `floor` is the newest
431/// sequence and `resync` is set. Otherwise `floor` is the oldest retained
432/// sequence, or the next one when none is retained, and `resync` is set when
433/// `after + 1 < floor`, even with rows present. On `resync` the reader adopts
434/// `series` and `floor` and lists sessions again.
435#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
436pub struct SessionChanges {
437    pub rows: Vec<SessionChangeRow>,
438    pub floor: u64,
439    pub resync: bool,
440    /// The change series these sequence numbers belong to. Zero from an index
441    /// that does not name one.
442    #[serde(default, skip_serializing_if = "is_zero_series")]
443    pub series: u64,
444}
445
446/// The reply to every session read: `Ok` with the outcome, or `Err` with a
447/// failure.
448#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
449#[non_exhaustive]
450pub enum SessionReply {
451    Ok(SessionOutcome),
452    Err(SessionError),
453}
454
455/// The successful outcome of a session read, shaped per request.
456#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
457#[non_exhaustive]
458pub enum SessionOutcome {
459    Info(Box<SessionInfo>),
460    Page(SessionPage),
461    Events(SessionEventsPage),
462    State(SessionStateView),
463    Links(SessionLinksView),
464    Sources(SessionSources),
465    Changes(SessionChanges),
466}
467
468/// Why a session read failed.
469#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
470#[non_exhaustive]
471pub enum SessionError {
472    #[error("sessions not supported: {0}")]
473    Unsupported(String),
474    #[error("session not found: {0}")]
475    NotFound(String),
476    #[error("stream is not registered for sessions: {0}")]
477    NotRegistered(String),
478    #[error("invalid session request: {0}")]
479    Invalid(String),
480    #[error("session read not authorized: {0}")]
481    Unauthorized(String),
482    /// The request named a stream or topic generation the index does not hold.
483    #[error("stale session generation: {0}")]
484    Stale(String),
485    #[error("session backend error: {0}")]
486    Backend(String),
487    /// The identical request may succeed later.
488    #[error("temporarily unavailable: {0}")]
489    Unavailable(String),
490}
491
492/// The body of one process heartbeat on `agent.heartbeats`: the sessions in
493/// one stream that the process holds leases on. A process with sessions in
494/// several streams writes one heartbeat per stream.
495#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
496pub struct SessionHeartbeat {
497    pub process: String,
498    pub stream: String,
499    pub sessions: Vec<ConversationId>,
500}
501
502impl Validate for request::SessionList {
503    fn validate(&self) -> Result<(), InvalidError> {
504        if self.stream.is_empty() || self.stream.len() > MAX_FILTER_SOURCE_NAME_BYTES {
505            return Err(InvalidError::new(format!(
506                "session list stream must be 1..={MAX_FILTER_SOURCE_NAME_BYTES} bytes"
507            )));
508        }
509        let bounded = [
510            ("label prefix", &self.label_prefix, MAX_SESSION_LABEL_BYTES),
511            ("text", &self.text, MAX_SESSION_TEXT_QUERY_BYTES),
512            ("cursor", &self.cursor, MAX_SESSION_CURSOR_BYTES),
513        ];
514        for (label, value, cap) in bounded {
515            if let Some(value) = value
516                && value.len() > cap
517            {
518                return Err(InvalidError::new(format!(
519                    "session list {label} is {}B, exceeds cap {cap}B",
520                    value.len()
521                )));
522            }
523        }
524        Ok(())
525    }
526}
527
528impl Validate for SessionHeartbeat {
529    fn validate(&self) -> Result<(), InvalidError> {
530        if self.sessions.len() > MAX_HEARTBEAT_SESSIONS {
531            return Err(InvalidError::new("heartbeat lists too many sessions"));
532        }
533        Ok(())
534    }
535}
536
537#[expect(clippy::trivially_copy_pass_by_ref, reason = "serde skip predicate")]
538fn is_zero_series(series: &u64) -> bool {
539    *series == 0
540}
541
542#[cfg(test)]
543mod tests {
544    use super::*;
545
546    #[test]
547    fn given_a_session_list_over_its_caps_when_validated_then_should_refuse_it() {
548        let list = request::SessionList {
549            stream: "agents".to_owned(),
550            status: None,
551            root: None,
552            label_prefix: None,
553            agent: None,
554            text: None,
555            cursor: None,
556            limit: 0,
557            want_total: false,
558        };
559        list.validate().expect("a plain session list is valid");
560        request::SessionList {
561            stream: "s".repeat(MAX_FILTER_SOURCE_NAME_BYTES),
562            label_prefix: Some("l".repeat(MAX_SESSION_LABEL_BYTES)),
563            text: Some("t".repeat(MAX_SESSION_TEXT_QUERY_BYTES)),
564            cursor: Some("c".repeat(MAX_SESSION_CURSOR_BYTES)),
565            ..list.clone()
566        }
567        .validate()
568        .expect("every filter at its cap is valid");
569        request::SessionList {
570            text: Some(String::new()),
571            cursor: Some(String::new()),
572            ..list.clone()
573        }
574        .validate()
575        .expect("an empty text and cursor are within their caps");
576        for invalid in [
577            request::SessionList {
578                stream: String::new(),
579                ..list.clone()
580            },
581            request::SessionList {
582                stream: "s".repeat(MAX_FILTER_SOURCE_NAME_BYTES + 1),
583                ..list.clone()
584            },
585            request::SessionList {
586                label_prefix: Some("l".repeat(MAX_SESSION_LABEL_BYTES + 1)),
587                ..list.clone()
588            },
589            request::SessionList {
590                text: Some("t".repeat(MAX_SESSION_TEXT_QUERY_BYTES + 1)),
591                ..list.clone()
592            },
593            request::SessionList {
594                cursor: Some("c".repeat(MAX_SESSION_CURSOR_BYTES + 1)),
595                ..list.clone()
596            },
597        ] {
598            assert!(invalid.validate().is_err(), "{invalid:?}");
599        }
600    }
601}
laser_wire/session.rs

laser_wire/
session.rs
