laser_wire/
session.rs

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

laser_wire/
session.rs
