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_LABEL_BYTES,
8 MAX_SESSION_TEXT_QUERY_BYTES,
9};
10use crate::schema::Digest32;
11use crate::validate::Validate;
12use serde::{Deserialize, Serialize};
13use std::collections::BTreeMap;
14
15#[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#[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
36pub 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 #[serde(default, skip_serializing_if = "Option::is_none")]
53 pub root: Option<ConversationId>,
54 #[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 pub page: u32,
63 pub results: u32,
65 #[serde(default, skip_serializing_if = "Option::is_none")]
68 pub as_of: Option<u64>,
69 #[serde(default, skip_serializing_if = "Option::is_none")]
71 pub series: Option<u64>,
72 }
73
74 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
75 pub struct SessionEvents {
76 pub stream: String,
77 pub id: ConversationId,
78 pub page: u32,
80 pub results: u32,
82 #[serde(default, skip_serializing_if = "Option::is_none")]
85 pub as_of: Option<u64>,
86 #[serde(default, skip_serializing_if = "Option::is_none")]
88 pub series: Option<u64>,
89 }
90
91 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
92 pub struct SessionState {
93 pub stream: String,
94 pub id: ConversationId,
95 pub history_limit: u32,
97 }
98
99 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
100 pub struct SessionLinks {
101 pub stream: String,
102 pub id: ConversationId,
103 #[serde(default, skip_serializing_if = "Option::is_none")]
104 pub surface: Option<LinkSurface>,
105 }
106
107 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
108 pub struct SessionSources {
109 pub stream: String,
110 pub id: ConversationId,
111 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
113 pub lane_only: bool,
114 }
115
116 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
120 pub struct SessionChanges {
121 pub stream: String,
122 pub page: u32,
124 pub results: u32,
126 }
127}
128
129#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
133pub struct SourceFrontier {
134 pub topic_id: u32,
135 pub topic_generation: u64,
136 pub partition_id: u32,
137 #[serde(default, skip_serializing_if = "Option::is_none")]
139 pub folded: Option<u64>,
140 #[serde(default, skip_serializing_if = "Option::is_none")]
142 pub head: Option<u64>,
143 #[serde(default, skip_serializing_if = "Option::is_none")]
145 pub retained_from: Option<u64>,
146}
147
148#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
150#[serde(rename_all = "snake_case")]
151pub enum GapReason {
152 ExpiredBeforeFold,
154 Pruned,
156 Truncated,
158 Rebuilding,
160 #[serde(other)]
161 Unrecognized,
162}
163
164#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
166pub struct SourceGap {
167 pub topic_id: u32,
168 pub topic_generation: u64,
169 pub partition_id: u32,
170 pub from: u64,
171 pub to: u64,
172 pub reason: GapReason,
173}
174
175#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
179pub struct PayloadRange {
180 pub topic_id: u32,
181 pub topic_generation: u64,
182 pub partition_id: u32,
183 pub first: u64,
184 pub last: u64,
185}
186
187#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
189#[serde(rename_all = "snake_case")]
190pub enum StateOutcome {
191 Applied,
193 Rejected,
195 Stale,
197 Duplicate,
199 #[serde(other)]
200 Unrecognized,
201}
202
203#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
205pub struct StateChange {
206 pub revision: u64,
208 #[serde(default, skip_serializing_if = "Option::is_none")]
209 pub op_id: Option<String>,
210 pub outcome: StateOutcome,
211 pub at: LogPosition,
212 pub broker_ts: u64,
213 #[serde(default, skip_serializing_if = "Option::is_none")]
214 pub old_digest: Option<Digest32>,
215 #[serde(default, skip_serializing_if = "Option::is_none")]
216 pub new_digest: Option<Digest32>,
217 #[serde(default, skip_serializing_if = "Option::is_none")]
218 pub reason: Option<String>,
219}
220
221#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
223pub struct SessionFlags {
224 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
225 pub label_truncated: bool,
226 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
227 pub overflow: bool,
228 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
229 pub events_truncated: bool,
230 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
231 pub lane_conflict: bool,
232 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
235 pub control_conflict: bool,
236 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
237 pub rebuilding: bool,
238 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
241 pub liveness_unknown: bool,
242}
243
244#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
246pub struct SessionInfo {
247 pub stream: String,
248 pub id: ConversationId,
249 #[serde(default, skip_serializing_if = "Option::is_none")]
250 pub label: Option<String>,
251 #[serde(default, skip_serializing_if = "Option::is_none")]
252 pub namespace: Option<String>,
253 #[serde(default, skip_serializing_if = "Option::is_none")]
254 pub agent: Option<AgentId>,
255 #[serde(default, skip_serializing_if = "Option::is_none")]
256 pub parent: Option<ConversationId>,
257 #[serde(default, skip_serializing_if = "Option::is_none")]
258 pub root: Option<ConversationId>,
259 pub status: SessionStatus,
260 pub idle: bool,
261 pub over_budget: bool,
262 pub pause_requested: bool,
263 #[serde(default, skip_serializing_if = "Vec::is_empty")]
265 pub awaiting: Vec<AgentId>,
266 pub cancel_requested: bool,
267 #[serde(default, skip_serializing_if = "Option::is_none")]
268 pub started_at: Option<u64>,
269 #[serde(default, skip_serializing_if = "Option::is_none")]
270 pub ended_at: Option<u64>,
271 #[serde(default, skip_serializing_if = "Option::is_none")]
272 pub first_event_at: Option<u64>,
273 #[serde(default, skip_serializing_if = "Option::is_none")]
274 pub last_event_at: Option<u64>,
275 #[serde(default, skip_serializing_if = "Option::is_none")]
276 pub last_heartbeat_at: Option<u64>,
277 pub events: u64,
278 pub model_calls: u64,
279 pub tool_calls: u64,
280 pub tokens_in: u64,
281 pub tokens_out: u64,
282 pub cost_micros: u64,
283 pub errors: u64,
284 #[serde(default, skip_serializing_if = "Option::is_none")]
285 pub budget: Option<Budget>,
286 #[serde(default, skip_serializing_if = "Option::is_none")]
287 pub sdk: Option<SdkInfo>,
288 #[serde(default)]
289 pub flags: SessionFlags,
290 #[serde(default, skip_serializing_if = "is_zero")]
292 pub held: u64,
293 #[serde(default, skip_serializing_if = "Vec::is_empty")]
294 pub frontier: Vec<SourceFrontier>,
295}
296
297fn is_zero(value: &u64) -> bool {
298 *value == 0
299}
300
301#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
303pub struct SessionPage {
304 pub items: Vec<SessionInfo>,
305 pub page: u32,
306 pub total_pages: u32,
307 pub total_results: u32,
308 pub as_of: u64,
311 pub series: u64,
313 #[serde(default, skip_serializing_if = "Option::is_none")]
315 pub searched: Option<u64>,
316 pub truncated: bool,
319 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
321 pub rebuilding: bool,
322}
323
324#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
327pub struct SessionEvent {
328 pub at: SourceRef,
329 pub session: ConversationId,
330 pub broker_ts: u64,
331 pub kind: String,
332 #[serde(default, skip_serializing_if = "Option::is_none")]
333 pub operation: Option<String>,
334 pub display: String,
335 #[serde(default, skip_serializing_if = "Option::is_none")]
336 pub agent: Option<AgentId>,
337 #[serde(default, skip_serializing_if = "Option::is_none")]
338 pub addressee: Option<String>,
339 #[serde(default, skip_serializing_if = "Option::is_none")]
340 pub correlation: Option<CorrelationId>,
341 #[serde(default, skip_serializing_if = "Option::is_none")]
342 pub cause: Option<LogPosition>,
343 #[serde(default, skip_serializing_if = "Option::is_none")]
344 pub tool: Option<String>,
345 #[serde(default, skip_serializing_if = "Option::is_none")]
346 pub usage: Option<TokenUsage>,
347 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
348 pub after_end: bool,
349 #[serde(default, skip_serializing_if = "Option::is_none")]
351 pub verified_actor: Option<String>,
352 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
353 pub summary: BTreeMap<String, serde_json::Value>,
354}
355
356#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
359pub struct SessionEventsPage {
360 pub items: Vec<SessionEvent>,
361 pub page: u32,
362 pub total_pages: u32,
363 pub total_results: u32,
364 pub as_of: u64,
367 pub series: u64,
369 #[serde(default, skip_serializing_if = "Vec::is_empty")]
370 pub ranges: Vec<PayloadRange>,
371 #[serde(default, skip_serializing_if = "Vec::is_empty")]
372 pub frontier: Vec<SourceFrontier>,
373 #[serde(default, skip_serializing_if = "Vec::is_empty")]
374 pub gaps: Vec<SourceGap>,
375}
376
377#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
379pub struct SessionStateView {
380 pub revision: u64,
381 pub document: serde_json::Value,
382 #[serde(default, skip_serializing_if = "Vec::is_empty")]
383 pub history: Vec<StateChange>,
384 #[serde(default, skip_serializing_if = "Option::is_none")]
385 pub frontier: Option<SourceFrontier>,
386 pub complete: bool,
387}
388
389#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
391pub struct SessionLink {
392 pub surface: LinkSurface,
393 pub resource: String,
394 pub item: String,
395 pub relation: LinkRelation,
396 pub first: LogPosition,
397 pub last: LogPosition,
398}
399
400#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
402pub struct SessionLinksView {
403 pub links: Vec<SessionLink>,
404 #[serde(default, skip_serializing_if = "Vec::is_empty")]
405 pub frontier: Vec<SourceFrontier>,
406 pub truncated: bool,
407}
408
409#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
411pub struct SessionSources {
412 pub sources: Vec<SourceFrontier>,
413 #[serde(default, skip_serializing_if = "Option::is_none")]
415 pub lane: Option<(u32, u64, u32)>,
416}
417
418#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
420pub struct SessionChangeRow {
421 pub seq: u64,
422 pub sessions: Vec<ConversationId>,
423 #[serde(default, skip_serializing_if = "Vec::is_empty")]
428 pub positions: Vec<LogPosition>,
429 pub truncated: bool,
430 #[serde(default, skip_serializing_if = "SessionChangeOrigin::is_records")]
431 pub origin: SessionChangeOrigin,
432}
433
434#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
436#[serde(rename_all = "snake_case")]
437pub enum SessionChangeOrigin {
438 #[default]
440 Records,
441 Links,
443 Maintenance,
446 Expiry,
448}
449
450impl SessionChangeOrigin {
451 pub fn is_records(&self) -> bool {
452 *self == Self::Records
453 }
454}
455
456#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
461pub struct SessionChanges {
462 pub items: Vec<SessionChangeRow>,
463 pub page: u32,
464 pub total_pages: u32,
465 pub total_results: u64,
466 pub results: u32,
468 pub floor: u64,
469 pub series: u64,
471}
472
473#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
476#[non_exhaustive]
477pub enum SessionReply {
478 Ok(SessionOutcome),
479 Err(SessionError),
480}
481
482#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
484#[non_exhaustive]
485pub enum SessionOutcome {
486 Info(Box<SessionInfo>),
487 Page(SessionPage),
488 Events(SessionEventsPage),
489 State(SessionStateView),
490 Links(SessionLinksView),
491 Sources(SessionSources),
492 Changes(SessionChanges),
493}
494
495#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
497#[non_exhaustive]
498pub enum SessionError {
499 #[error("sessions not supported: {0}")]
500 Unsupported(String),
501 #[error("session not found: {0}")]
502 NotFound(String),
503 #[error("stream is not registered for sessions: {0}")]
504 NotRegistered(String),
505 #[error("invalid session request: {0}")]
506 Invalid(String),
507 #[error("session read not authorized: {0}")]
508 Unauthorized(String),
509 #[error("stale session generation: {0}")]
511 Stale(String),
512 #[error("session backend error: {0}")]
513 Backend(String),
514 #[error("temporarily unavailable: {0}")]
516 Unavailable(String),
517}
518
519#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
523pub struct SessionHeartbeat {
524 pub process: String,
525 pub stream: String,
526 pub sessions: Vec<ConversationId>,
527}
528
529impl Validate for request::SessionList {
530 fn validate(&self) -> Result<(), InvalidError> {
531 if self.stream.is_empty() || self.stream.len() > MAX_FILTER_SOURCE_NAME_BYTES {
532 return Err(InvalidError::new(format!(
533 "session list stream must be 1..={MAX_FILTER_SOURCE_NAME_BYTES} bytes"
534 )));
535 }
536 let bounded = [
537 ("label prefix", &self.label_prefix, MAX_SESSION_LABEL_BYTES),
538 ("text", &self.text, MAX_SESSION_TEXT_QUERY_BYTES),
539 ];
540 for (label, value, cap) in bounded {
541 if let Some(value) = value
542 && value.len() > cap
543 {
544 return Err(InvalidError::new(format!(
545 "session list {label} is {}B, exceeds cap {cap}B",
546 value.len()
547 )));
548 }
549 }
550 if self.as_of.is_some() != self.series.is_some() {
551 return Err(InvalidError::new(
552 "session list as_of and series name one bound, set both or neither",
553 ));
554 }
555 Ok(())
556 }
557}
558
559impl Validate for SessionHeartbeat {
560 fn validate(&self) -> Result<(), InvalidError> {
561 if self.sessions.len() > MAX_HEARTBEAT_SESSIONS {
562 return Err(InvalidError::new("heartbeat lists too many sessions"));
563 }
564 Ok(())
565 }
566}
567
568#[cfg(test)]
569mod tests {
570 use super::*;
571
572 #[test]
573 fn given_a_session_list_over_its_caps_when_validated_then_should_refuse_it() {
574 let list = request::SessionList {
575 stream: "agents".to_owned(),
576 status: None,
577 root: None,
578 label_prefix: None,
579 agent: None,
580 text: None,
581 page: 0,
582 results: 0,
583 as_of: None,
584 series: None,
585 };
586 list.validate().expect("a plain session list is valid");
587 request::SessionList {
588 as_of: Some(4405),
589 series: Some(77),
590 ..list.clone()
591 }
592 .validate()
593 .expect("a bound names its sequence and series");
594 request::SessionList {
595 stream: "s".repeat(MAX_FILTER_SOURCE_NAME_BYTES),
596 label_prefix: Some("l".repeat(MAX_SESSION_LABEL_BYTES)),
597 text: Some("t".repeat(MAX_SESSION_TEXT_QUERY_BYTES)),
598 ..list.clone()
599 }
600 .validate()
601 .expect("every filter at its cap is valid");
602 request::SessionList {
603 text: Some(String::new()),
604 ..list.clone()
605 }
606 .validate()
607 .expect("an empty text is within its cap");
608 for invalid in [
609 request::SessionList {
610 stream: String::new(),
611 ..list.clone()
612 },
613 request::SessionList {
614 stream: "s".repeat(MAX_FILTER_SOURCE_NAME_BYTES + 1),
615 ..list.clone()
616 },
617 request::SessionList {
618 label_prefix: Some("l".repeat(MAX_SESSION_LABEL_BYTES + 1)),
619 ..list.clone()
620 },
621 request::SessionList {
622 text: Some("t".repeat(MAX_SESSION_TEXT_QUERY_BYTES + 1)),
623 ..list.clone()
624 },
625 request::SessionList {
626 as_of: Some(4405),
627 ..list.clone()
628 },
629 request::SessionList {
630 series: Some(77),
631 ..list.clone()
632 },
633 ] {
634 assert!(invalid.validate().is_err(), "{invalid:?}");
635 }
636 }
637}