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

laser_wire/
session.rs
