lash_sansio/sansio/sections/
turn_protocol.rs

1impl TurnProtocol for UnitTurnProtocol {
2    type Event = ();
3    type Termination = ();
4    type DriverState = serde_json::Value;
5}
6
7/// Opaque identifier linking an effect to its response.
8#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, serde::Deserialize)]
9pub struct EffectId(pub u64);
10
11#[derive(Clone, Debug, Serialize, serde::Deserialize)]
12pub struct PendingToolCall {
13    pub call_id: String,
14    pub tool_name: String,
15    pub args: Value,
16    /// Opaque provider replay state carried through for the next request.
17    pub replay: Option<ProviderReplayMeta>,
18}
19
20#[derive(Clone, Debug, Serialize, serde::Deserialize)]
21pub struct CompletedToolCall {
22    pub call_id: String,
23    pub tool_name: String,
24    pub args: Value,
25    pub output: ToolCallOutput,
26    pub model_return: ModelToolReturn,
27    pub duration_ms: u64,
28    /// See [`PendingToolCall::replay`].
29    pub replay: Option<ProviderReplayMeta>,
30}
31
32#[derive(Clone, Debug, PartialEq, Eq, Serialize, serde::Deserialize)]
33pub struct TurnCause {
34    pub id: String,
35    pub event_type: String,
36    pub origin: MessageOrigin,
37    pub text: String,
38}
39
40impl TurnCause {
41    pub fn to_event_message(&self) -> Message {
42        Message {
43            id: self.id.clone(),
44            role: MessageRole::Event,
45            parts: Arc::new(vec![Part {
46                id: format!("{}.p0", self.id),
47                kind: PartKind::Text,
48                content: self.text.clone(),
49                attachment: None,
50                tool_call_id: None,
51                tool_name: None,
52                tool_replay: None,
53                prune_state: PruneState::Intact,
54                reasoning_meta: None,
55                response_meta: None,
56            }]),
57            origin: Some(self.origin.clone()),
58        }
59    }
60}
61
62#[derive(Clone, Debug, Default, Serialize, serde::Deserialize)]
63pub struct CheckpointDelivery {
64    /// Normal user messages admitted by durable turn-input ingress.
65    ///
66    /// These are already fully materialized rather than plugin messages so
67    /// they retain the same `origin: None` representation as user input that
68    /// starts a turn.
69    #[serde(default)]
70    pub committed_user_messages: Vec<Message>,
71    pub messages: Vec<PluginMessage>,
72    pub transient_messages: Vec<PluginMessage>,
73    pub turn_causes: Vec<TurnCause>,
74}
75
76pub fn render_turn_causes_prompt(causes: &[TurnCause]) -> Option<String> {
77    if causes.is_empty() {
78        return None;
79    }
80
81    let mut rendered = String::from("=== TURN EVENTS ===");
82    for (index, cause) in causes.iter().enumerate() {
83        rendered.push_str("\n\n");
84        rendered.push_str(&format!(
85            "--- event[{index}] · {} · {} ---\n",
86            cause.event_type, cause.id
87        ));
88        rendered.push_str("Origin: ");
89        rendered.push_str(&render_message_origin(&cause.origin));
90        rendered.push_str("\n\n");
91        rendered.push_str(cause.text.trim());
92    }
93    Some(rendered)
94}
95
96fn render_message_origin(origin: &MessageOrigin) -> String {
97    match origin {
98        MessageOrigin::Plugin {
99            plugin_id,
100            transient,
101        } => {
102            if *transient {
103                format!("plugin {plugin_id} (transient)")
104            } else {
105                format!("plugin {plugin_id}")
106            }
107        }
108        MessageOrigin::Process {
109            process_id,
110            event_type,
111            sequence,
112            wake_id,
113            ..
114        } => match wake_id {
115            Some(wake_id) => {
116                format!("process {process_id} {event_type} #{sequence} ({wake_id})")
117            }
118            None => format!("process {process_id} {event_type} #{sequence}"),
119        },
120    }
121}
122
123#[derive(Clone, Debug, Serialize, serde::Deserialize)]
124pub enum LogEvent {
125    LlmDebug {
126        session_id: String,
127        protocol_iteration: usize,
128        usage: TokenUsage,
129        provider_usage: Option<Value>,
130        request_body: Option<String>,
131        response_text: String,
132        response_parts: Option<Value>,
133    },
134    LlmError {
135        session_id: String,
136        protocol_iteration: usize,
137        request_body: Option<String>,
138        message: String,
139        retryable: bool,
140        raw: Option<String>,
141        code: Option<String>,
142        terminal_reason: LlmTerminalReason,
143    },
144}
145
146/// An effect the host must fulfil.
147//
148// `Clone` is implemented by hand below rather than derived: the derive would
149// demand `M: Clone`, but only `M::Event` is ever cloned (and `TurnProtocol`
150// already guarantees `Event: Clone`), so a manual impl keeps `Effect<M>`
151// cloneable for every protocol — which the turn checkpoint relies on.
152#[derive(Debug, Serialize, serde::Deserialize)]
153// justification: effects are short-lived machine states whose generic protocol payload remains inline for checkpoint cloning.
154#[allow(clippy::large_enum_variant)]
155pub enum Effect<M: TurnProtocol = UnitTurnProtocol> {
156    /// Sync the live execution environment before the turn proceeds.
157    ///
158    /// `update_machine_config` is only needed after the turn has
159    /// already advanced at least once and the host may need to swap in
160    /// a refreshed system prompt or tool schema for the next
161    /// protocol iteration. Initial syncs are host-only because the machine was
162    /// already constructed from a fresh execution environment.
163    SyncExecutionEnvironment {
164        id: EffectId,
165        update_machine_config: bool,
166    },
167    /// Start an LLM call.
168    LlmCall {
169        id: EffectId,
170        request: Arc<LlmRequest>,
171    },
172    /// Cancel an in-progress LLM stream.
173    CancelLlm { id: EffectId },
174    /// Execute one or more driver-scheduled tool calls.
175    ToolCalls {
176        id: EffectId,
177        calls: Vec<PendingToolCall>,
178    },
179    /// Execute a protocol-owned code block.
180    ExecCode {
181        id: EffectId,
182        language: String,
183        code: String,
184    },
185    /// Run a host/plugin checkpoint before the machine continues or completes.
186    Checkpoint {
187        id: EffectId,
188        checkpoint: CheckpointKind,
189    },
190    /// Host-implemented fire-and-forget logging.
191    Log { event: LogEvent },
192    /// Fire-and-forget event (no response needed).
193    Emit(SessionStreamEvent),
194    /// Prompt-history progress that may be durably persisted by the host.
195    ///
196    /// This is separate from [`SessionStreamEvent`]: UI stream events can be partial,
197    /// duplicated, or display-only, while `Progress` is emitted only after the
198    /// state machine has applied semantic message or protocol-step changes.
199    Progress {
200        messages: MessageSequence,
201        event_delta: Vec<SessionHistoryRecord<M::Event>>,
202        protocol_iteration: usize,
203    },
204    /// Turn is done.
205    Done {
206        messages: MessageSequence,
207        event_delta: Vec<SessionHistoryRecord<M::Event>>,
208        protocol_iteration: usize,
209    },
210}
211
212impl<M: TurnProtocol> Clone for Effect<M> {
213    fn clone(&self) -> Self {
214        match self {
215            Self::SyncExecutionEnvironment {
216                id,
217                update_machine_config,
218            } => Self::SyncExecutionEnvironment {
219                id: *id,
220                update_machine_config: *update_machine_config,
221            },
222            Self::LlmCall { id, request } => Self::LlmCall {
223                id: *id,
224                request: Arc::clone(request),
225            },
226            Self::CancelLlm { id } => Self::CancelLlm { id: *id },
227            Self::ToolCalls { id, calls } => Self::ToolCalls {
228                id: *id,
229                calls: calls.clone(),
230            },
231            Self::ExecCode { id, language, code } => Self::ExecCode {
232                id: *id,
233                language: language.clone(),
234                code: code.clone(),
235            },
236            Self::Checkpoint { id, checkpoint } => Self::Checkpoint {
237                id: *id,
238                checkpoint: *checkpoint,
239            },
240            Self::Log { event } => Self::Log {
241                event: event.clone(),
242            },
243            Self::Emit(event) => Self::Emit(event.clone()),
244            Self::Progress {
245                messages,
246                event_delta,
247                protocol_iteration,
248            } => Self::Progress {
249                messages: messages.clone(),
250                event_delta: event_delta.clone(),
251                protocol_iteration: *protocol_iteration,
252            },
253            Self::Done {
254                messages,
255                event_delta,
256                protocol_iteration,
257            } => Self::Done {
258                messages: messages.clone(),
259                event_delta: event_delta.clone(),
260                protocol_iteration: *protocol_iteration,
261            },
262        }
263    }
264}
265
266impl<M: TurnProtocol> Effect<M> {
267    fn id(&self) -> Option<EffectId> {
268        match self {
269            Self::SyncExecutionEnvironment { id, .. }
270            | Self::LlmCall { id, .. }
271            | Self::CancelLlm { id }
272            | Self::ToolCalls { id, .. }
273            | Self::ExecCode { id, .. }
274            | Self::Checkpoint { id, .. } => Some(*id),
275            Self::Log { .. } | Self::Emit(_) | Self::Progress { .. } | Self::Done { .. } => None,
276        }
277    }
278}
279
280/// Error details from a failed LLM call.
281#[derive(Clone, Debug, Serialize, serde::Deserialize)]
282pub struct LlmCallError {
283    pub message: String,
284    pub retryable: bool,
285    /// Typed transport classification of the failure. Defaults to
286    /// [`ProviderFailureKind::Unknown`] for wrappers that are not provider
287    /// failures (and when decoding effect journals written before the field
288    /// existed).
289    #[serde(default)]
290    pub kind: crate::llm::types::ProviderFailureKind,
291    pub raw: Option<String>,
292    pub code: Option<String>,
293    pub terminal_reason: LlmTerminalReason,
294    pub request_body: Option<String>,
295    /// Output and usage observed before the failed stream ended. Partial tool
296    /// calls in this response are retained for diagnosis but never executed.
297    #[serde(default, skip_serializing_if = "Option::is_none")]
298    pub partial_response: Option<Box<LlmResponse>>,
299}
300
301/// A response to a previously emitted effect.
302pub enum Response {
303    /// Live execution environment sync completed.
304    ExecutionEnvironmentSynced {
305        id: EffectId,
306        result: Result<Option<ExecutionEnvironmentSync>, String>,
307    },
308    /// Full LLM response.
309    LlmComplete {
310        id: EffectId,
311        result: Result<LlmResponse, LlmCallError>,
312        /// When true, text deltas were already emitted during streaming,
313        /// so the driver should skip emitting `TextDelta` events.
314        text_streamed: bool,
315    },
316    /// Native tool results.
317    ToolResults {
318        id: EffectId,
319        results: Vec<CompletedToolCall>,
320    },
321    /// Mode code execution result.
322    ExecResult {
323        id: EffectId,
324        result: Result<crate::ExecResponse, String>,
325    },
326    /// Checkpoint result with optional injected messages.
327    Checkpoint {
328        id: EffectId,
329        delivery: CheckpointDelivery,
330    },
331}
332
333#[derive(Clone, Debug, Serialize, serde::Deserialize)]
334pub struct ExecutionEnvironmentSync {
335    pub system_prompt: Arc<str>,
336    pub tool_specs: Arc<Vec<LlmToolSpec>>,
337}
338
339pub struct WaitingLlmState<M: TurnProtocol = UnitTurnProtocol> {
340    pub request: Arc<LlmRequest>,
341    driver_state: Option<M::DriverState>,
342}
343
344impl<M: TurnProtocol> WaitingLlmState<M> {
345    pub fn take_driver_state(&mut self) -> Option<M::DriverState> {
346        self.driver_state.take()
347    }
348}
349
350pub struct WaitingExecState<M: TurnProtocol = UnitTurnProtocol> {
351    driver_state: M::DriverState,
352}
353
354impl<M: TurnProtocol> WaitingExecState<M> {
355    pub fn into_driver_state(self) -> M::DriverState {
356        self.driver_state
357    }
358}
359
360#[derive(Clone, Debug, PartialEq, Serialize, serde::Deserialize)]
361pub enum CheckpointResumeAction {
362    PrepareIteration,
363    Finish(TurnOutcome),
364}
365
366// justification: driver actions are single-step machine values and boxing generic driver state would add allocation to every iteration.
367#[allow(clippy::large_enum_variant)]
368pub enum DriverAction<M: TurnProtocol = UnitTurnProtocol> {
369    Emit(SessionStreamEvent),
370    AppendEvents(Vec<SessionHistoryRecord<M::Event>>),
371    StartLlm {
372        request: Arc<LlmRequest>,
373        driver_state: Option<M::DriverState>,
374    },
375    StartTools {
376        calls: Vec<PendingToolCall>,
377    },
378    StartExec {
379        language: String,
380        code: String,
381        driver_state: M::DriverState,
382    },
383    StartCheckpoint {
384        checkpoint: CheckpointKind,
385        on_empty: CheckpointResumeAction,
386    },
387    AdvanceProtocolIteration,
388    ScheduleTurnLimitFinal {
389        message: Message,
390    },
391    Finish(TurnOutcome),
392}
393
394pub struct DriverContextView<'a, M: TurnProtocol = UnitTurnProtocol> {
395    config: &'a TurnMachineConfig<M>,
396    messages: &'a MessageSequence,
397    events: &'a [SessionHistoryRecord<M::Event>],
398    turn_causes: &'a [TurnCause],
399    protocol_iteration: usize,
400    protocol_run_offset: usize,
401    termination: &'a TurnTerminationPolicyState,
402}
403
404impl<'a, M: TurnProtocol> DriverContextView<'a, M> {
405    pub fn project_llm_request(&self, use_tools: bool) -> Arc<LlmRequest> {
406        self.config.projector.project(ProjectorContext {
407            config: self.config,
408            messages: self.messages,
409            events: self.events,
410            turn_causes: self.turn_causes,
411            protocol_iteration: self.protocol_iteration,
412            use_tools,
413        })
414    }
415
416    pub fn protocol_iteration(&self) -> usize {
417        self.protocol_iteration
418    }
419
420    pub fn protocol_run_offset(&self) -> usize {
421        self.protocol_run_offset
422    }
423
424    pub fn max_turns(&self) -> Option<usize> {
425        self.config.max_turns
426    }
427
428    pub fn termination(&self) -> &M::Termination {
429        &self.config.termination
430    }
431
432    pub fn autonomous(&self) -> bool {
433        self.config.autonomous
434    }
435
436    pub fn should_force_exit_after_grace_turn(&self) -> bool {
437        self.termination.should_force_exit_after_grace_turn()
438    }
439
440    pub fn turn_limit_final_to_schedule(&self) -> Option<usize> {
441        self.termination.turn_limit_final_to_schedule(
442            self.protocol_iteration,
443            self.protocol_run_offset,
444            self.config.max_turns,
445        )
446    }
447
448    pub fn messages(&self) -> &MessageSequence {
449        self.messages
450    }
451
452    pub fn events(&self) -> &[SessionHistoryRecord<M::Event>] {
453        self.events
454    }
455
456    pub fn turn_causes(&self) -> &[TurnCause] {
457        self.turn_causes
458    }
459}
460
461pub struct ProjectorContext<'a, M: TurnProtocol = UnitTurnProtocol> {
462    pub config: &'a TurnMachineConfig<M>,
463    pub messages: &'a MessageSequence,
464    pub events: &'a [SessionHistoryRecord<M::Event>],
465    pub turn_causes: &'a [TurnCause],
466    pub protocol_iteration: usize,
467    pub use_tools: bool,
468}
469
470pub trait ContextProjector<M: TurnProtocol = UnitTurnProtocol>: Send + Sync {
471    fn project(&self, ctx: ProjectorContext<'_, M>) -> Arc<LlmRequest>;
472}
473
474#[derive(Clone, Debug, Default)]
475pub struct ChatContextProjector;
476
477impl<M: TurnProtocol> ContextProjector<M> for ChatContextProjector {
478    fn project(&self, ctx: ProjectorContext<'_, M>) -> Arc<LlmRequest> {
479        let rendered_prompt = render_messages_for_projector(ctx.messages, ctx.turn_causes);
480        let attachments: Vec<AttachmentSource> = rendered_prompt.attachments;
481        let mut messages = rendered_prompt.messages;
482        if let Some(turn_events) = render_turn_causes_prompt(ctx.turn_causes) {
483            messages.push(crate::llm::types::LlmMessage::text(
484                crate::llm::types::LlmRole::User,
485                Arc::from(turn_events),
486            ));
487        }
488        if !ctx.config.system_prompt.trim().is_empty() {
489            messages.insert(
490                0,
491                crate::llm::types::LlmMessage::text(
492                    crate::llm::types::LlmRole::System,
493                    Arc::clone(&ctx.config.system_prompt),
494                ),
495            );
496        }
497
498        Arc::new(LlmRequest {
499            model: ctx.config.model.clone(),
500            messages,
501            attachments,
502            resolved_stored: Default::default(),
503            tools: if ctx.use_tools {
504                Arc::clone(&ctx.config.tool_specs)
505            } else {
506                Arc::new(Vec::new())
507            },
508            tool_choice: if ctx.use_tools {
509                LlmToolChoice::Auto
510            } else {
511                LlmToolChoice::None
512            },
513            model_variant: ctx.config.model_variant.clone(),
514            model_capability: ctx.config.model_capability.clone(),
515            generation: ctx.config.generation.clone(),
516            scope: crate::llm::types::LlmRequestScope::new(
517                ctx.config.session_id.clone(),
518                format!("{}:frame:sansio", ctx.config.session_id),
519                format!(
520                    "{}:sansio:llm:{}",
521                    ctx.config.session_id, ctx.protocol_iteration
522                ),
523            ),
524            output_spec: None,
525            stream_events: None,
526            provider_trace: None,
527        })
528    }
529}
530
531fn render_messages_for_projector(
532    messages: &MessageSequence,
533    turn_causes: &[TurnCause],
534) -> crate::RenderedPrompt {
535    if turn_causes.is_empty() {
536        return messages.render_prompt();
537    }
538
539    let active_cause_ids = turn_causes
540        .iter()
541        .map(|cause| cause.id.as_str())
542        .collect::<HashSet<_>>();
543    let filtered = messages
544        .iter()
545        .filter(|message| {
546            !(matches!(message.role, MessageRole::Event)
547                && active_cause_ids.contains(message.id.as_str()))
548        })
549        .cloned()
550        .collect::<Vec<_>>();
551    render_prompt(filtered.as_slice())
552}
553
554pub trait ProtocolDriverHandle<M: TurnProtocol = UnitTurnProtocol>: Send + Sync {
555    fn prepare_protocol_iteration(&self, ctx: DriverContextView<'_, M>) -> Vec<DriverAction<M>>;
556    fn handle_llm_success(
557        &self,
558        ctx: DriverContextView<'_, M>,
559        waiting: WaitingLlmState<M>,
560        llm_response: LlmResponse,
561        text_streamed: bool,
562    ) -> Vec<DriverAction<M>>;
563    fn handle_tool_results(
564        &self,
565        ctx: DriverContextView<'_, M>,
566        completed: Vec<CompletedToolCall>,
567    ) -> Vec<DriverAction<M>>;
568    fn handle_exec_result(
569        &self,
570        ctx: DriverContextView<'_, M>,
571        waiting: WaitingExecState<M>,
572        result: Result<crate::ExecResponse, String>,
573    ) -> Vec<DriverAction<M>>;
574}
575
576/// Configuration for a `TurnMachine` instance.
577pub struct TurnMachineConfig<M: TurnProtocol = UnitTurnProtocol> {
578    pub protocol_driver: Arc<dyn ProtocolDriverHandle<M>>,
579    pub projector: Arc<dyn ContextProjector<M>>,
580    pub sync_execution_environment: bool,
581    pub model: String,
582    /// Model context-window size in tokens, if known. Lets the kernel
583    /// reclassify a zero-output `OutputLimit` terminal reason as
584    /// `ContextOverflow` when the prompt nearly filled the window. `None`
585    /// disables that refinement.
586    pub max_context_tokens: Option<usize>,
587    pub max_turns: Option<usize>,
588    pub model_variant: crate::ReasoningSelection,
589    pub model_capability: crate::llm::capability::ModelCapability,
590    pub generation: crate::llm::types::GenerationOptions,
591    pub autonomous: bool,
592    pub tool_specs: Arc<Vec<LlmToolSpec>>,
593    pub system_prompt: Arc<str>,
594    pub session_id: String,
595    pub emit_llm_trace: bool,
596    pub termination: M::Termination,
597    pub turn_limit_final_message: crate::TurnLimitFinalMessage,
598}
599
600#[cfg(test)]
601mod llm_call_error_tests {
602    use super::LlmCallError;
603    use crate::llm::types::ProviderFailureKind;
604
605    #[test]
606    fn llm_call_error_decodes_journal_entries_that_predate_kind() {
607        // `LlmCallError` is serialized inside durable effect journals
608        // (`RuntimeEffectOutcome::LlmCall`). Entries written before the typed
609        // `kind` field existed must decode with `Unknown`.
610        let legacy = r#"{
611            "message":"rate limited",
612            "retryable":true,
613            "raw":null,
614            "code":"429",
615            "terminal_reason":"provider_error",
616            "request_body":null
617        }"#;
618        let decoded: LlmCallError = serde_json::from_str(legacy).expect("legacy call error");
619        assert!(decoded.retryable);
620        assert_eq!(decoded.kind, ProviderFailureKind::Unknown);
621    }
622}
623
624// ─── Internal state ───
lash_sansio/sansio/sections/turn_protocol.rs

lash_sansio/sansio/sections/
turn_protocol.rs
