talos_core/
session.rs

1//! Session protocol types for the AppServerSession seam (ADR-005).
2//!
3//! SQ (Submission Queue): bounded `mpsc::Sender<SessionOp>` (cap=512) for commands TO the session actor.
4//! EQ (Event Queue): unbounded `mpsc::UnboundedSender<SessionEvent>` for events FROM the session actor.
5
6use std::path::PathBuf;
7
8use serde::{Deserialize, Serialize};
9use tokio::sync::mpsc;
10
11use crate::message::{AgentEvent, Message};
12pub use crate::submission::{
13    MAX_PENDING_SUBMISSION_BYTES, MAX_PENDING_SUBMISSIONS, MAX_STEERING_QUEUE_BYTES,
14    MAX_STEERING_QUEUE_IMAGE_BYTES, MAX_STEERING_QUEUE_IMAGES, MAX_STEERING_QUEUE_ITEMS,
15    MAX_SUBMISSION_BATCH_BYTES, MAX_SUBMISSION_BATCH_ITEMS, MAX_SUBMISSION_IMAGE_BYTES,
16    MAX_SUBMISSION_IMAGE_COUNT, MAX_SUBMISSION_ITEM_BYTES, MAX_SUBMISSION_TOTAL_IMAGE_BYTES,
17    PendingSubmissionState, StructuredSubmission, SubmissionItem, SubmissionKind,
18    SubmissionReceipt, SubmissionReceiptDisposition, SubmissionRejectionReason, SubmissionSource,
19};
20
21/// Commands sent to the session actor via the bounded SQ.
22#[derive(Debug, Clone, Serialize, Deserialize)]
23#[serde(tag = "type", rename_all = "snake_case")]
24pub enum SessionOp {
25    Submit {
26        message: String,
27    },
28    SubmitMultimodal {
29        text: String,
30        attachments: Vec<crate::message::ContentPart>,
31    },
32    PreviewRequest {
33        message: String,
34    },
35    SubmitStructured {
36        submission: StructuredSubmission,
37    },
38    SubmitStructuredTracked {
39        submission: StructuredSubmission,
40        #[serde(skip)]
41        receipt_tx: Option<mpsc::UnboundedSender<SubmissionReceipt>>,
42    },
43    /// Stable public compatibility operation.
44    ReconcileStructured {
45        submission: StructuredSubmission,
46    },
47    /// Stable public compatibility operation with a direct receipt channel.
48    ReconcileStructuredTracked {
49        submission: StructuredSubmission,
50        #[serde(skip)]
51        receipt_tx: Option<mpsc::UnboundedSender<SubmissionReceipt>>,
52    },
53    /// Transitional additive alias routed to the same observational reconcile path.
54    SubmitStructuredReconcile {
55        submission: StructuredSubmission,
56    },
57    /// Transitional additive alias routed to the same observational reconcile path.
58    SubmitStructuredReconcileTracked {
59        submission: StructuredSubmission,
60        #[serde(skip)]
61        receipt_tx: Option<mpsc::UnboundedSender<SubmissionReceipt>>,
62    },
63    SetSkillContext {
64        name: Option<String>,
65        content: Option<String>,
66    },
67    /// Legacy unqualified compatibility interrupt.
68    Interrupt,
69    /// Generation- and Turn-targeted interrupt for new interactive paths.
70    InterruptTurn {
71        session_generation: u64,
72        turn_id: String,
73    },
74    /// Explicitly terminalizes one durably accepted submission that paused
75    /// before Provider execution. The immutable submission is never replayed
76    /// or rewritten; only the exact current generation may resolve it.
77    CancelPausedSubmission {
78        session_generation: u64,
79        submission_id: String,
80    },
81    Shutdown,
82}
83
84#[derive(Debug, Clone, Serialize, Deserialize)]
85#[serde(tag = "type", rename_all = "snake_case")]
86#[non_exhaustive]
87pub enum SessionEvent {
88    SubmissionQueued {
89        session_id: String,
90        submission_id: String,
91        sender_generation: u64,
92        source: SubmissionSource,
93        item_count: usize,
94        total_text_bytes: usize,
95    },
96    SubmissionStarted {
97        session_id: String,
98        submission_id: String,
99        sender_generation: u64,
100        turn_id: String,
101    },
102    /// Authoritative model-visible projection emitted immediately before a
103    /// structured Turn starts. The Actor, not the receipt observer, defines
104    /// the visible ordering when retained work precedes a new resume item.
105    StructuredSubmissionStarted {
106        session_id: String,
107        session_generation: u64,
108        submission: StructuredSubmission,
109        receipt_id: String,
110        turn_id: String,
111    },
112    SubmissionRejected {
113        session_id: String,
114        submission_id: String,
115        sender_generation: u64,
116        reason: SubmissionRejectionReason,
117    },
118    SubmissionPaused {
119        session_id: String,
120        session_generation: u64,
121        submission_id: String,
122        receipt_id: String,
123        reason: SubmissionRejectionReason,
124    },
125    /// A pre-start durable submission was explicitly resolved without
126    /// Provider execution.
127    SubmissionResolved {
128        session_id: String,
129        session_generation: u64,
130        submission_id: String,
131        receipt_id: String,
132        state: PendingSubmissionState,
133    },
134    SubmissionReceipt {
135        session_id: String,
136        session_generation: u64,
137        submission_id: String,
138        reservation_id: String,
139        receipt_id: String,
140        source: SubmissionSource,
141        item_count: usize,
142        total_text_bytes: usize,
143        disposition: SubmissionReceiptDisposition,
144    },
145    StructuredTurnEvent {
146        session_id: String,
147        session_generation: u64,
148        source: SubmissionSource,
149        submission_id: String,
150        receipt_id: String,
151        turn_id: String,
152        sequence: u64,
153        payload: TurnEventPayload,
154    },
155    EntriesCommitted {
156        session_id: String,
157        turn_id: String,
158        entry_ids: Vec<String>,
159    },
160    TurnEvent {
161        session_id: String,
162        turn_id: String,
163        sequence: u64,
164        payload: TurnEventPayload,
165    },
166    AgentEvent {
167        event: AgentEvent,
168    },
169    ApprovalRequired {
170        tool_name: String,
171        arguments: String,
172        call_id: String,
173    },
174    TurnStarted {
175        turn_id: String,
176    },
177    TurnCompleted {
178        turn_id: String,
179        status: TurnCompletionStatus,
180    },
181    Error {
182        message: String,
183    },
184}
185
186#[derive(Debug, Clone, Serialize, Deserialize)]
187#[serde(tag = "kind", rename_all = "snake_case")]
188#[non_exhaustive]
189pub enum TurnEventPayload {
190    Started,
191    Progress { event: AgentEvent },
192    Completed { status: TurnCompletionStatus },
193}
194
195#[derive(Debug, Clone, Serialize, Deserialize)]
196#[serde(tag = "status", rename_all = "snake_case")]
197pub enum TurnCompletionStatus {
198    Success {
199        #[serde(default)]
200        final_text: String,
201        #[serde(default, skip_serializing_if = "Vec::is_empty")]
202        new_messages: Vec<crate::message::Message>,
203    },
204    Cancelled,
205    Error {
206        message: String,
207    },
208}
209
210pub struct SessionHandle {
211    pub sq_tx: mpsc::Sender<SessionOp>,
212    pub eq_rx: mpsc::UnboundedReceiver<SessionEvent>,
213}
214
215#[derive(Debug, Clone, Serialize, Deserialize)]
216pub struct SessionConfig {
217    #[serde(default)]
218    pub runtime_policy: RuntimePolicy,
219    pub workspace_root: PathBuf,
220    #[serde(default)]
221    pub initial_history: Vec<Message>,
222    #[serde(default = "default_model_context_limit")]
223    pub model_context_limit: u32,
224}
225
226#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
227#[serde(default)]
228pub struct RuntimePolicy {
229    pub approval_mode: ApprovalMode,
230}
231
232impl RuntimePolicy {
233    #[must_use]
234    pub fn interactive() -> Self {
235        Self {
236            approval_mode: ApprovalMode::Interactive,
237        }
238    }
239
240    #[must_use]
241    pub fn headless_deny() -> Self {
242        Self {
243            approval_mode: ApprovalMode::HeadlessDeny,
244        }
245    }
246}
247
248impl Default for RuntimePolicy {
249    fn default() -> Self {
250        Self::interactive()
251    }
252}
253
254#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
255#[serde(rename_all = "snake_case")]
256pub enum ApprovalMode {
257    #[default]
258    Interactive,
259    HeadlessDeny,
260}
261
262fn default_model_context_limit() -> u32 {
263    128_000
264}
265
266#[cfg(test)]
267#[allow(warnings)]
268mod tests {
269    use super::*;
270
271    fn structured_submission() -> StructuredSubmission {
272        StructuredSubmission {
273            id: "batch_1".into(),
274            source: SubmissionSource::User,
275            sender_generation: 7,
276            items: vec![SubmissionItem {
277                id: "item_1".into(),
278                enqueue_sequence: 1,
279                kind: SubmissionKind::UserTurn,
280                text: "structured".into(),
281                attachments: Vec::new(),
282            }],
283        }
284    }
285
286    #[test]
287    fn session_op_serde_roundtrip() {
288        let submission = structured_submission();
289        let ops = vec![
290            SessionOp::Submit {
291                message: "hello".into(),
292            },
293            SessionOp::PreviewRequest {
294                message: "diagnostic".into(),
295            },
296            SessionOp::SubmitStructured {
297                submission: submission.clone(),
298            },
299            SessionOp::SubmitStructuredTracked {
300                submission: submission.clone(),
301                receipt_tx: None,
302            },
303            SessionOp::ReconcileStructured {
304                submission: submission.clone(),
305            },
306            SessionOp::ReconcileStructuredTracked {
307                submission: submission.clone(),
308                receipt_tx: None,
309            },
310            SessionOp::SubmitStructuredReconcile {
311                submission: submission.clone(),
312            },
313            SessionOp::SubmitStructuredReconcileTracked {
314                submission,
315                receipt_tx: None,
316            },
317            SessionOp::Interrupt,
318            SessionOp::InterruptTurn {
319                session_generation: 7,
320                turn_id: "turn_7".into(),
321            },
322            SessionOp::CancelPausedSubmission {
323                session_generation: 7,
324                submission_id: "batch_1".into(),
325            },
326            SessionOp::Shutdown,
327        ];
328        for op in &ops {
329            let json = serde_json::to_string(op).expect("operation should succeed");
330            let back: SessionOp = serde_json::from_str(&json).expect("operation should succeed");
331            assert_eq!(
332                serde_json::to_value(op).expect("operation should succeed"),
333                serde_json::to_value(&back).expect("operation should succeed")
334            );
335        }
336    }
337
338    #[test]
339    fn session_event_serde_roundtrip() {
340        let submission = structured_submission();
341        let events = vec![
342            SessionEvent::SubmissionQueued {
343                session_id: "session_1".into(),
344                submission_id: "batch_1".into(),
345                sender_generation: 7,
346                source: SubmissionSource::User,
347                item_count: 1,
348                total_text_bytes: 10,
349            },
350            SessionEvent::SubmissionPaused {
351                session_id: "session_1".into(),
352                session_generation: 7,
353                submission_id: "batch_1".into(),
354                receipt_id: "receipt_1".into(),
355                reason: SubmissionRejectionReason::ContextBudgetExceeded,
356            },
357            SessionEvent::SubmissionResolved {
358                session_id: "session_1".into(),
359                session_generation: 7,
360                submission_id: "batch_1".into(),
361                receipt_id: "receipt_1".into(),
362                state: PendingSubmissionState::TerminalCancelled,
363            },
364            SessionEvent::SubmissionReceipt {
365                session_id: "session_1".into(),
366                session_generation: 7,
367                submission_id: "batch_1".into(),
368                reservation_id: "reservation:batch_1".into(),
369                receipt_id: "receipt_1".into(),
370                source: SubmissionSource::User,
371                item_count: 1,
372                total_text_bytes: 10,
373                disposition: SubmissionReceiptDisposition::AlreadyAccepted {
374                    state: PendingSubmissionState::AcceptedPending,
375                    turn_id: None,
376                },
377            },
378            SessionEvent::StructuredSubmissionStarted {
379                session_id: "session_1".into(),
380                session_generation: 7,
381                submission: submission.clone(),
382                receipt_id: "receipt_1".into(),
383                turn_id: "turn_1".into(),
384            },
385            SessionEvent::StructuredTurnEvent {
386                session_id: "session_1".into(),
387                session_generation: 7,
388                source: SubmissionSource::User,
389                submission_id: "batch_1".into(),
390                receipt_id: "receipt_1".into(),
391                turn_id: "turn_1".into(),
392                sequence: 0,
393                payload: TurnEventPayload::Started,
394            },
395            SessionEvent::TurnEvent {
396                session_id: "session_1".into(),
397                turn_id: "turn_1".into(),
398                sequence: 1,
399                payload: TurnEventPayload::Progress {
400                    event: AgentEvent::TextDelta {
401                        delta: "hello".into(),
402                    },
403                },
404            },
405            SessionEvent::TurnCompleted {
406                turn_id: "2".into(),
407                status: TurnCompletionStatus::Cancelled,
408            },
409            SessionEvent::Error {
410                message: "fail".into(),
411            },
412        ];
413        for event in &events {
414            let json = serde_json::to_string(event).expect("operation should succeed");
415            let back: SessionEvent = serde_json::from_str(&json).expect("operation should succeed");
416            assert_eq!(
417                serde_json::to_value(event).expect("operation should succeed"),
418                serde_json::to_value(&back).expect("operation should succeed")
419            );
420        }
421    }
422
423    #[test]
424    fn session_config_serde_roundtrip() {
425        let config = SessionConfig {
426            runtime_policy: RuntimePolicy::headless_deny(),
427            workspace_root: PathBuf::from("/tmp/test"),
428            initial_history: vec![],
429            model_context_limit: 128_000,
430        };
431        let json = serde_json::to_string(&config).expect("operation should succeed");
432        let back: SessionConfig = serde_json::from_str(&json).expect("operation should succeed");
433        assert_eq!(config.runtime_policy, back.runtime_policy);
434        assert_eq!(config.workspace_root, back.workspace_root);
435        assert_eq!(config.initial_history, back.initial_history);
436        assert_eq!(config.model_context_limit, back.model_context_limit);
437    }
438}
talos_core/session.rs

talos_core/
session.rs
