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

lash_sansio/sansio/sections/
turn_protocol.rs
