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

laser_wire/
session.rs
