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

lash_sansio/sansio/sections/
turn_protocol.rs
