lash_core/runtime/
mod.rs

1mod assembly;
2mod builder;
3pub(crate) mod causal;
4mod clock;
5mod config_ops;
6mod effect;
7mod environment;
8mod error;
9mod host;
10mod in_memory_store;
11mod io;
12mod lifecycle;
13mod observation;
14mod process;
15mod process_work_driver;
16mod process_worker;
17mod queued_work_driver;
18mod session_api;
19mod session_manager;
20mod session_ops;
21mod state;
22#[cfg(test)]
23pub(crate) mod tests;
24mod turn_boundary;
25mod turn_commit_draft;
26mod turn_driver;
27mod turn_graph_editor;
28mod turn_loop;
29mod turn_queue;
30mod usage;
31
32use std::any::Any;
33use std::collections::HashMap;
34use std::fmt;
35use std::sync::Arc;
36use std::sync::Mutex as StdMutex;
37use std::sync::atomic::{AtomicBool, Ordering};
38
39use tokio::sync::{Mutex, mpsc};
40use tokio_util::sync::CancellationToken;
41
42use crate::llm::types::{
43    LlmOutputPart, LlmProviderTraceEvent, LlmProviderTraceSender, LlmRequest, LlmResponse,
44    LlmStreamEvent, LlmUsage,
45};
46use crate::plugin::{
47    CheckpointHookContext, PrepareTurnRequest, SessionConfigChangedContext, SessionRelation,
48};
49use crate::sansio::{LlmCallError, Response};
50use crate::session_model::{
51    Message, MessageRole, Part, PartKind, PruneState, RuntimeSessionPolicy, SessionEvent,
52    SessionPolicy, TokenUsage, fresh_message_id, make_error_event, reassign_part_ids, shared_parts,
53    transport_stream_events,
54};
55use crate::{
56    CheckpointKind, PersistentRuntimeServices, PluginActionInvokeError, PromptHookContext,
57    RuntimeServices, SandboxMessage, Session, SessionCreateRequest, SessionError, SessionHandle,
58    SessionSnapshot, SessionStartPoint, ToolCallRecord, TurnFinish, TurnOutcome, TurnStop,
59};
60use crate::{Effect, TurnMachine};
61
62use host::*;
63use session_manager::*;
64use turn_boundary::*;
65use turn_commit_draft::*;
66use turn_driver::*;
67
68pub(crate) const RUNTIME_TURN_LEASE_TTL_MS: u64 = 15 * 60 * 1000;
69
70// `PromptUsage` is re-exported below alongside the runtime's own types.
71pub use lash_sansio::PromptUsage;
72
73use assembly::{
74    LlmDebugText, LlmDebugToolCall, LlmStreamAccumulator, LlmStreamDebugState, LlmStreamEventLog,
75    LlmStreamState, LlmStreamSummary, TurnAssembler,
76};
77#[cfg(test)]
78#[allow(unused_imports)]
79use assembly::{classify_output_state, sanitize_assistant_output};
80pub use builder::EmbeddedRuntimeBuilder;
81pub use causal::process_event_invocation;
82pub(crate) use causal::tool_retry_sleep_invocation;
83pub use clock::{Clock, SystemClock};
84pub(crate) use effect::RuntimeEffectControllerHandle;
85pub use effect::{
86    AwaitEventKey, AwaitEventWaitIdentity, CausalRef, EffectHost, ExecutionScope,
87    ExternalCompletionError, InlineEffectHost, InlineRuntimeEffectController, LlmAttachmentSpec,
88    LlmRequestSpec, ProcessCommand, ProcessEffectOutcome, Resolution, ResolveOutcome,
89    RuntimeEffectCommand, RuntimeEffectController, RuntimeEffectControllerError,
90    RuntimeEffectEnvelope, RuntimeEffectKind, RuntimeEffectLocalExecutor, RuntimeEffectOutcome,
91    RuntimeInvocation, RuntimeReplay, RuntimeScope, RuntimeSubject, ScopedEffectController,
92    ToolBatchEffectOutcome, ToolCallLaunch,
93};
94pub use environment::{ParkedSession, Residency, RuntimeEnvironment, RuntimeEnvironmentBuilder};
95pub use error::{DurableStoreFacet, RuntimeError, RuntimeErrorCode};
96pub use host::{EmbeddedRuntimeHost, ProcessRuntimeHost, RuntimeHostConfig};
97pub use in_memory_store::{InMemorySessionStore, InMemorySessionStoreFactory};
98use io::normalize_input_items;
99pub use observation::{
100    InMemoryLiveReplayStore, InMemoryLiveReplayStoreConfig, LiveReplayGap, LiveReplayGapReason,
101    LiveReplayResult, LiveReplayStore, LiveReplayStoreError, LiveReplaySubscribeResult,
102    LiveReplaySubscription, RuntimeHandle, RuntimeObservation, SessionCursor, SessionCursorError,
103    SessionObservation, SessionObservationEvent, SessionObservationEventPayload,
104    SessionObservationSubscription, SessionProcessEventKind, SessionQueueEventKind, SessionResume,
105    SessionRevision,
106};
107#[cfg(any(test, feature = "testing"))]
108pub use process::TestLocalProcessRegistry;
109pub use process::{
110    DefaultProcessCancelAbility, InMemoryProcessExecutionEnvStore, ObservedProcess,
111    ObservedProcessEvent, ObservedWorkItem, PROCESS_LEASE_SCHEMA_VERSION, ProcessAwaitOutput,
112    ProcessCancelAbility, ProcessCancelAllRequest, ProcessCancelRequest, ProcessCancelSource,
113    ProcessCancelSummary, ProcessEngine, ProcessEngineRegistry, ProcessEngineRunContext,
114    ProcessEngineRunGuard, ProcessEngineRuntimeContext, ProcessEngineValidationContext,
115    ProcessEvent, ProcessEventAppendPlan, ProcessEventAppendRequest, ProcessEventAppendResult,
116    ProcessEventSemantics, ProcessEventSemanticsSpec, ProcessEventType, ProcessExecutionContext,
117    ProcessExecutionEnvRef, ProcessExecutionEnvSpec, ProcessExecutionEnvStore, ProcessExternalRef,
118    ProcessHandleDescriptor, ProcessHandleGrant, ProcessHandleGrantEntry, ProcessHandleSummary,
119    ProcessId, ProcessIdentity, ProcessInput, ProcessLease, ProcessLeaseCompletion,
120    ProcessLifecycleStatus, ProcessListFilter, ProcessListMode, ProcessOpScope, ProcessOriginator,
121    ProcessProvenance, ProcessRecord, ProcessRegistration, ProcessRegistry, ProcessService,
122    ProcessSessionDeleteReport, ProcessSpawnProvenance, ProcessStartGrant, ProcessStartOptions,
123    ProcessStartRequest, ProcessStatus, ProcessStatusFilter, ProcessTerminalSemantics,
124    ProcessTerminalSpec, ProcessTerminalState, ProcessValueSelector, ProcessWake,
125    ProcessWakeDedupeKey, ProcessWakeDelivery, ProcessWakeSpec, ProcessWorkObserver,
126    ProcessWorkSnapshot, SessionScope, SessionScopeId, UnavailableProcessService, WaitKind,
127    WaitState, apply_process_status_projection, current_epoch_ms, epoch_ms_from_system_time,
128    load_process_execution_env, materialize_process_event_semantics, persist_process_execution_env,
129    prepare_process_event_append, prepare_process_registration, process_event_payload_hash,
130    process_signal_event_type, process_signal_name_from_event_type, process_signal_wait_key,
131    process_wake_delivery, process_wake_input_from_event_payload, process_wake_turn_cause,
132    process_wake_turn_text, require_event_replay, system_time_from_epoch_ms,
133    validate_process_signal_name,
134};
135pub use process_work_driver::{InlineProcessRunHandle, ProcessRunHandle, ProcessWorkDriver};
136pub use process_worker::{DurableProcessWorker, DurableProcessWorkerConfig};
137pub use queued_work_driver::{QueuedWorkDriver, QueuedWorkRunHandle, QueuedWorkRunRequest};
138pub use session_manager::DirectCompletionClient;
139pub use state::RuntimeSessionState;
140use state::{
141    append_session_nodes_to_state_with_clock, apply_residency_on_load, apply_session_checkpoint,
142    apply_session_head, normalize_session_graph, open_agent_frame_in_state_with_clock,
143};
144pub use turn_loop::ensure_durable_effect_input;
145pub use turn_queue::{
146    DeliveryPolicy, MergeKey, QUEUED_WORK_CLAIM_TTL_MS, QueuedCheckpointWork, QueuedTurnWork,
147    QueuedWorkBatch, QueuedWorkBatchDraft, QueuedWorkClaim, QueuedWorkClaimBoundary,
148    QueuedWorkCompletion, QueuedWorkItem, QueuedWorkPayload, SessionCommand, SessionCommandReceipt,
149    SlotPolicy, process_wake_batch_draft,
150};
151pub use usage::{
152    SessionUsageReport, TokenLedgerEntry, UsageReportRow, UsageTotals, diff_token_ledger,
153    diff_usage_reports,
154};
155use usage::{merge_ledger_entry, merge_usage_delta_entries, normalize_prompt_usage};
156
157#[doc(hidden)]
158#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
159pub enum RuntimeTurnPhase {
160    ContextTransform,
161    BeforeTurnHooks,
162    PromptBuild,
163    EffectLoop,
164    FinalizeTurn,
165    PersistTurn,
166    FinalCommit,
167    PostPersistHooks,
168}
169
170#[doc(hidden)]
171pub trait RuntimeTurnPhaseProbe: Send + Sync {
172    fn begin(&self, phase: RuntimeTurnPhase);
173    fn end(&self, phase: RuntimeTurnPhase);
174    fn begin_named(&self, _phase: &str) {}
175    fn end_named(&self, _phase: &str) {}
176}
177
178#[doc(hidden)]
179#[derive(Clone, Default)]
180pub struct RuntimeTurnPhaseProbeSlot {
181    probes: Arc<StdMutex<HashMap<crate::SessionScopeId, Arc<dyn RuntimeTurnPhaseProbe>>>>,
182}
183
184impl RuntimeTurnPhaseProbeSlot {
185    pub fn set_for_session(
186        &self,
187        session_id: impl Into<String>,
188        probe: Arc<dyn RuntimeTurnPhaseProbe>,
189    ) {
190        self.set_for_scope(&crate::SessionScope::new(session_id), probe);
191    }
192
193    pub fn set_for_scope(
194        &self,
195        scope: &crate::SessionScope,
196        probe: Arc<dyn RuntimeTurnPhaseProbe>,
197    ) {
198        self.probes
199            .lock()
200            .expect("runtime phase probe slot")
201            .insert(scope.id(), probe);
202    }
203
204    pub fn get_for_scope(
205        &self,
206        scope: &crate::SessionScope,
207    ) -> Option<Arc<dyn RuntimeTurnPhaseProbe>> {
208        let probes = self.probes.lock().expect("runtime phase probe slot");
209        probes.get(&scope.id()).cloned().or_else(|| {
210            probes
211                .get(&crate::SessionScope::new(&scope.session_id).id())
212                .cloned()
213        })
214    }
215}
216
217#[doc(hidden)]
218pub struct RuntimeNamedPhase {
219    probe: Option<Arc<dyn RuntimeTurnPhaseProbe>>,
220    phase: &'static str,
221}
222
223impl RuntimeNamedPhase {
224    pub fn begin(
225        probe: Option<Arc<dyn RuntimeTurnPhaseProbe>>,
226        phase: &'static str,
227    ) -> RuntimeNamedPhase {
228        if let Some(probe) = probe.as_ref() {
229            probe.begin_named(phase);
230        }
231        RuntimeNamedPhase { probe, phase }
232    }
233}
234
235impl Drop for RuntimeNamedPhase {
236    fn drop(&mut self) {
237        if let Some(probe) = self.probe.as_ref() {
238            probe.end_named(self.phase);
239        }
240    }
241}
242
243/// Host-provided per-turn input.
244#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
245#[serde(tag = "type", rename_all = "snake_case")]
246pub enum InputItem {
247    Text { text: String },
248    ImageRef { id: String },
249}
250
251impl InputItem {
252    pub fn text(text: impl Into<String>) -> Self {
253        Self::Text { text: text.into() }
254    }
255
256    pub fn image_ref(id: impl Into<String>) -> Self {
257        Self::ImageRef { id: id.into() }
258    }
259}
260
261/// Host-provided per-turn input.
262#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
263pub struct TurnInput {
264    pub items: Vec<InputItem>,
265    #[serde(default)]
266    pub image_blobs: HashMap<String, Vec<u8>>,
267    /// Per-turn override for protocol-owned turn options.
268    #[serde(default, skip_serializing_if = "Option::is_none")]
269    pub protocol_turn_options: Option<crate::ProtocolTurnOptions>,
270    /// Optional externally-stable trace turn id. Normal runtime callers leave
271    /// this empty and the runtime generates one per outer turn.
272    #[serde(default, skip_serializing_if = "Option::is_none")]
273    pub trace_turn_id: Option<String>,
274    #[serde(skip)]
275    pub protocol_extension: Option<ProtocolTurnExtensionHandle>,
276    #[serde(skip)]
277    pub turn_context: TurnContext,
278}
279
280impl TurnInput {
281    pub fn empty() -> Self {
282        Self::items(std::iter::empty())
283    }
284
285    pub fn text(text: impl Into<String>) -> Self {
286        Self::items([InputItem::text(text)])
287    }
288
289    pub fn items(items: impl IntoIterator<Item = InputItem>) -> Self {
290        Self {
291            items: items.into_iter().collect(),
292            image_blobs: HashMap::new(),
293            protocol_turn_options: None,
294            trace_turn_id: None,
295            protocol_extension: None,
296            turn_context: TurnContext::default(),
297        }
298    }
299
300    pub fn with_image_blob(mut self, id: impl Into<String>, bytes: Vec<u8>) -> Self {
301        self.image_blobs.insert(id.into(), bytes);
302        self
303    }
304
305    pub fn with_image_blobs<I, K>(mut self, image_blobs: I) -> Self
306    where
307        I: IntoIterator<Item = (K, Vec<u8>)>,
308        K: Into<String>,
309    {
310        self.image_blobs.extend(
311            image_blobs
312                .into_iter()
313                .map(|(id, bytes)| (id.into(), bytes)),
314        );
315        self
316    }
317
318    pub fn with_image_ref(mut self, id: impl Into<String>, bytes: Vec<u8>) -> Self {
319        let id = id.into();
320        self.items.push(InputItem::image_ref(id.clone()));
321        self.image_blobs.insert(id, bytes);
322        self
323    }
324
325    pub fn with_protocol_turn_options(mut self, options: crate::ProtocolTurnOptions) -> Self {
326        self.protocol_turn_options = Some(options);
327        self
328    }
329
330    pub fn with_trace_turn_id(mut self, trace_turn_id: impl Into<String>) -> Self {
331        self.trace_turn_id = Some(trace_turn_id.into());
332        self
333    }
334}
335
336/// Per-turn, in-process side channel of typed plugin inputs.
337///
338/// This is an `Any`-keyed map of live Rust values handed to plugins for a
339/// single turn. It is deliberately **not** serializable: the values never
340/// survive a process boundary, so durable effect-host runs explicitly reject a
341/// turn that carries any live inputs (see
342/// [`LiveTurnInputs::durable_effect_rejection`]). Durable callers must instead
343/// encode replayable data in
344/// `protocol_turn_options` or persisted plugin state.
345#[derive(Clone, Default)]
346pub struct LiveTurnInputs {
347    inputs: HashMap<&'static str, Arc<dyn Any + Send + Sync>>,
348}
349
350impl LiveTurnInputs {
351    fn insert<T>(&mut self, plugin_id: &'static str, input: T)
352    where
353        T: Send + Sync + 'static,
354    {
355        self.inputs.insert(plugin_id, Arc::new(input));
356    }
357
358    fn get<T>(&self, plugin_id: &'static str) -> Option<&T>
359    where
360        T: 'static,
361    {
362        self.inputs
363            .get(plugin_id)
364            .and_then(|input| input.downcast_ref::<T>())
365    }
366
367    fn contains(&self, plugin_id: &'static str) -> bool {
368        self.inputs.contains_key(plugin_id)
369    }
370
371    fn is_empty(&self) -> bool {
372        self.inputs.is_empty()
373    }
374
375    pub fn plugin_ids(&self) -> Vec<&'static str> {
376        self.inputs.keys().copied().collect()
377    }
378
379    /// Returns an error when live per-turn inputs would make a durable effect
380    /// host replay depend on process-local values.
381    pub(crate) fn durable_effect_rejection(&self) -> Result<(), RuntimeError> {
382        if self.is_empty() {
383            return Ok(());
384        }
385        Err(RuntimeError::new(
386            RuntimeErrorCode::DurableEffectLivePluginInput,
387            "durable effect hosts do not support live TurnContext plugin inputs; encode replayable data in protocol_turn_options or persisted plugin state",
388        ))
389    }
390}
391
392#[derive(Clone, Default)]
393pub struct TurnContext {
394    plugin_inputs: LiveTurnInputs,
395    provider: Option<crate::ProviderHandle>,
396    model: Option<crate::ModelSpec>,
397    prompt: crate::PromptLayer,
398}
399
400impl TurnContext {
401    pub fn new() -> Self {
402        Self::default()
403    }
404
405    pub fn insert_plugin_input<T>(&mut self, plugin_id: &'static str, input: T)
406    where
407        T: Send + Sync + 'static,
408    {
409        self.plugin_inputs.insert(plugin_id, input);
410    }
411
412    pub fn set_provider(&mut self, provider: crate::ProviderHandle) {
413        self.provider = Some(provider);
414    }
415
416    pub fn provider(&self) -> Option<&crate::ProviderHandle> {
417        self.provider.as_ref()
418    }
419
420    pub fn set_model(&mut self, model: crate::ModelSpec) {
421        self.model = Some(model);
422    }
423
424    pub fn model_spec(&self) -> Option<&crate::ModelSpec> {
425        self.model.as_ref()
426    }
427
428    pub fn plugin_input<T>(&self, plugin_id: &'static str) -> Option<&T>
429    where
430        T: 'static,
431    {
432        self.plugin_inputs.get(plugin_id)
433    }
434
435    pub fn has_plugin_input(&self, plugin_id: &'static str) -> bool {
436        self.plugin_inputs.contains(plugin_id)
437    }
438
439    pub fn has_live_plugin_inputs(&self) -> bool {
440        !self.plugin_inputs.is_empty()
441    }
442
443    pub fn live_plugin_input_ids(&self) -> Vec<&'static str> {
444        self.plugin_inputs.plugin_ids()
445    }
446
447    /// Live plugin inputs for this turn. The durable boundary inspects this to
448    /// reject turns carrying non-serializable live state.
449    pub(crate) fn live_plugin_inputs(&self) -> &LiveTurnInputs {
450        &self.plugin_inputs
451    }
452
453    pub fn set_prompt_template(&mut self, template: crate::PromptTemplate) {
454        self.prompt.template = Some(template);
455    }
456
457    pub fn add_prompt_contribution(&mut self, contribution: crate::PromptContribution) {
458        self.prompt.add_contribution(contribution);
459    }
460
461    pub fn replace_prompt_slot(
462        &mut self,
463        slot: crate::PromptSlot,
464        contributions: impl IntoIterator<Item = crate::PromptContribution>,
465    ) {
466        self.prompt.replace_slot(slot, contributions);
467    }
468
469    pub fn clear_prompt_slot(&mut self, slot: crate::PromptSlot) {
470        self.prompt.clear_slot(slot);
471    }
472
473    pub fn set_prompt_layer(&mut self, prompt: crate::PromptLayer) {
474        self.prompt = prompt;
475    }
476
477    pub fn prompt_layer(&self) -> &crate::PromptLayer {
478        &self.prompt
479    }
480}
481
482impl fmt::Debug for TurnContext {
483    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
484        f.debug_struct("TurnContext")
485            .field("plugin_inputs", &self.plugin_inputs.plugin_ids())
486            .field("has_provider", &self.provider.is_some())
487            .field("has_model", &self.model.is_some())
488            .field("has_prompt_layer", &(!self.prompt.is_empty()))
489            .finish()
490    }
491}
492
493#[derive(Clone)]
494pub struct ProtocolTurnExtensionHandle(Arc<dyn ProtocolTurnExtension>);
495
496impl ProtocolTurnExtensionHandle {
497    pub fn new(extension: impl ProtocolTurnExtension + 'static) -> Self {
498        Self(Arc::new(extension))
499    }
500
501    pub fn as_any(&self) -> &dyn Any {
502        self.0.as_any()
503    }
504
505    pub fn prompt_contributions(&self) -> Vec<crate::PromptContribution> {
506        self.0.prompt_contributions()
507    }
508}
509
510impl fmt::Debug for ProtocolTurnExtensionHandle {
511    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
512        f.write_str("ProtocolTurnExtensionHandle(..)")
513    }
514}
515
516pub trait ProtocolTurnExtension: Send + Sync {
517    fn as_any(&self) -> &dyn Any;
518
519    fn prompt_contributions(&self) -> Vec<crate::PromptContribution> {
520        Vec::new()
521    }
522}
523
524#[derive(Clone)]
525pub struct ProtocolSessionExtensionHandle(Arc<dyn ProtocolSessionExtension>);
526
527impl ProtocolSessionExtensionHandle {
528    pub fn new(extension: impl ProtocolSessionExtension + 'static) -> Self {
529        Self(Arc::new(extension))
530    }
531
532    pub fn as_any(&self) -> &dyn Any {
533        self.0.as_any()
534    }
535}
536
537impl fmt::Debug for ProtocolSessionExtensionHandle {
538    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
539        f.write_str("ProtocolSessionExtensionHandle(..)")
540    }
541}
542
543pub trait ProtocolSessionExtension: Send + Sync {
544    fn as_any(&self) -> &dyn Any;
545}
546
547#[derive(Clone, Debug)]
548pub(super) enum NormalizedItem {
549    Text(String),
550    Image(crate::AttachmentRef),
551}
552
553/// Canonical assistant output payload.
554#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
555pub struct AssistantOutput {
556    pub safe_text: String,
557    pub raw_text: String,
558    pub state: OutputState,
559}
560
561/// Quality and usability of assembled terminal output.
562#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
563#[serde(rename_all = "snake_case")]
564pub enum OutputState {
565    Usable,
566    EmptyOutput,
567    TracebackOnly,
568    RecoveredFromError,
569}
570
571/// Code execution output observed during a turn.
572#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
573pub struct CodeOutputRecord {
574    pub output: String,
575    #[serde(default, skip_serializing_if = "Option::is_none")]
576    pub error: Option<String>,
577}
578
579/// High-level execution summary for a completed turn.
580#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
581pub struct ExecutionSummary {
582    #[serde(default)]
583    pub had_tool_calls: bool,
584    #[serde(default)]
585    pub had_code_execution: bool,
586}
587
588/// Structured issue surfaced during turn execution.
589#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
590pub struct TurnIssue {
591    pub kind: String,
592    #[serde(default, skip_serializing_if = "Option::is_none")]
593    pub code: Option<String>,
594    #[serde(default, skip_serializing_if = "Option::is_none")]
595    pub terminal_reason: Option<crate::LlmTerminalReason>,
596    pub message: String,
597    #[serde(default, skip_serializing_if = "Option::is_none")]
598    pub raw: Option<String>,
599}
600
601/// Canonical high-level turn result returned to hosts.
602#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
603pub struct AssembledTurn {
604    pub state: SessionSnapshot,
605    pub outcome: crate::TurnOutcome,
606    pub assistant_output: AssistantOutput,
607    pub execution: ExecutionSummary,
608    #[serde(default)]
609    pub token_usage: TokenUsage,
610    /// Per-(session, source, model) ledger entries for child sessions whose
611    /// LLM calls completed during this turn. `token_usage` above is the
612    /// parent's own LLM tokens; `total_usage` (on the embed-facing
613    /// `TurnResult`) sums both.
614    #[serde(default)]
615    pub children_usage: Vec<TokenLedgerEntry>,
616    #[serde(default)]
617    pub tool_calls: Vec<ToolCallRecord>,
618    #[serde(default)]
619    pub errors: Vec<TurnIssue>,
620}
621
622/// Result of driving one logical host turn through any AgentFrame switches.
623///
624/// A frame switch is an internal runtime continuation, similar to compaction
625/// from a host's perspective. Callers that need a final answer can use
626/// [`LashRuntime::stream_turn_with_agent_frames`] and inspect `final_turn()`.
627#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
628pub struct AgentFrameRun {
629    pub turns: Vec<AssembledTurn>,
630}
631
632impl AgentFrameRun {
633    pub fn final_turn(&self) -> Option<&AssembledTurn> {
634        self.turns.last()
635    }
636
637    pub fn into_final_turn(mut self) -> Option<AssembledTurn> {
638        self.turns.pop()
639    }
640
641    pub fn frame_switch_count(&self) -> usize {
642        self.turns
643            .iter()
644            .filter(|turn| matches!(turn.outcome, crate::TurnOutcome::AgentFrameSwitch { .. }))
645            .count()
646    }
647}
648
649/// Termination policy knobs.
650#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
651pub struct TerminationPolicy {
652    #[serde(default)]
653    pub treat_missing_done_as_failure: bool,
654}
655
656impl Default for TerminationPolicy {
657    fn default() -> Self {
658        Self {
659            treat_missing_done_as_failure: true,
660        }
661    }
662}
663
664/// Host application sink for low-level streaming runtime events.
665/// `SessionEvent` is protocol-specific preview/progress data.
666#[async_trait::async_trait]
667pub trait EventSink: Send + Sync {
668    fn is_noop(&self) -> bool {
669        false
670    }
671
672    async fn emit(&self, event: SessionEvent);
673}
674
675/// No-op sink useful for callers that only care about final state.
676pub struct NoopEventSink;
677
678/// Static no-op event sink for callers that need a `&dyn EventSink` default.
679pub static NOOP_EVENT_SINK: NoopEventSink = NoopEventSink;
680
681#[async_trait::async_trait]
682impl EventSink for NoopEventSink {
683    fn is_noop(&self) -> bool {
684        true
685    }
686
687    async fn emit(&self, _event: SessionEvent) {}
688}
689
690/// Stable identifier for a semantic turn activity.
691#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
692#[serde(transparent)]
693pub struct TurnActivityId(pub String);
694
695impl TurnActivityId {
696    pub fn new(id: impl Into<String>) -> Self {
697        Self(id.into())
698    }
699
700    pub fn fresh() -> Self {
701        Self(uuid::Uuid::new_v4().to_string())
702    }
703}
704
705/// App-facing semantic activity emitted during a turn.
706///
707/// `id` is unique per emitted activity event. `correlation_id` groups related
708/// events in the same logical activity, such as code start/completion, tool
709/// start/completion, or text deltas from one output block.
710#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
711pub struct TurnActivity {
712    pub id: TurnActivityId,
713    pub correlation_id: TurnActivityId,
714    #[serde(flatten)]
715    pub event: TurnEvent,
716}
717
718impl TurnActivity {
719    pub fn new(correlation_id: TurnActivityId, event: TurnEvent) -> Self {
720        Self {
721            id: TurnActivityId::fresh(),
722            correlation_id,
723            event,
724        }
725    }
726
727    pub fn independent(event: TurnEvent) -> Self {
728        let correlation_id = TurnActivityId::fresh();
729        Self::new(correlation_id, event)
730    }
731}
732
733/// App-facing semantic event payload for a turn activity.
734///
735/// Unlike [`SessionEvent`], these events are stable application signals rather
736/// than low-level runtime/debug events. Public streams carry these payloads
737/// inside [`TurnActivity`] so every emitted item has identity.
738#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
739#[serde(tag = "type", rename_all = "snake_case")]
740#[allow(clippy::large_enum_variant)]
741pub enum TurnEvent {
742    QueuedWorkStarted {
743        boundary: crate::QueuedWorkClaimBoundary,
744        batch_ids: Vec<String>,
745        causes: Vec<crate::TurnCause>,
746    },
747    ModelRequestStarted {
748        protocol_iteration: usize,
749    },
750    AssistantProseDelta {
751        text: String,
752    },
753    ReasoningDelta {
754        text: String,
755    },
756    CodeBlockStarted {
757        language: String,
758        code: String,
759        #[serde(default, skip_serializing_if = "Option::is_none")]
760        graph_key: Option<String>,
761    },
762    CodeBlockCompleted {
763        language: String,
764        output: String,
765        error: Option<String>,
766        success: bool,
767        duration_ms: u64,
768        tool_call_ids: Vec<String>,
769        #[serde(default, skip_serializing_if = "Option::is_none")]
770        graph_key: Option<String>,
771    },
772    ToolCallStarted {
773        call_id: Option<String>,
774        name: String,
775        args: serde_json::Value,
776    },
777    ToolCallCompleted {
778        call_id: Option<String>,
779        name: String,
780        args: serde_json::Value,
781        output: crate::ToolCallOutput,
782        duration_ms: u64,
783    },
784    SubmittedValue {
785        value: serde_json::Value,
786    },
787    ToolValue {
788        tool_name: String,
789        value: serde_json::Value,
790    },
791    Usage {
792        protocol_iteration: usize,
793        usage: TokenUsage,
794        cumulative: TokenUsage,
795    },
796    ChildUsage {
797        session_id: String,
798        source: String,
799        model: String,
800        protocol_iteration: usize,
801        usage: TokenUsage,
802        cumulative: TokenUsage,
803    },
804    RetryStatus {
805        wait_seconds: u64,
806        attempt: usize,
807        max_attempts: usize,
808        reason: String,
809    },
810    PluginRuntime {
811        plugin_id: String,
812        event: crate::PluginRuntimeEvent,
813    },
814    QueuedInputAccepted {
815        checkpoint: crate::CheckpointKind,
816        inputs: Vec<crate::AcceptedInjectedTurnInput>,
817    },
818    QueuedMessagesCommitted {
819        messages: Vec<crate::PluginMessage>,
820        checkpoint: crate::CheckpointKind,
821    },
822    Error {
823        message: String,
824    },
825}
826
827#[async_trait::async_trait]
828pub trait TurnActivitySink: Send + Sync {
829    fn is_noop(&self) -> bool {
830        false
831    }
832
833    async fn emit(&self, activity: TurnActivity);
834}
835
836pub struct NoopTurnActivitySink;
837
838/// Static no-op turn-activity sink for callers that need a `&dyn TurnActivitySink` default.
839pub static NOOP_TURN_ACTIVITY_SINK: NoopTurnActivitySink = NoopTurnActivitySink;
840
841#[async_trait::async_trait]
842impl TurnActivitySink for NoopTurnActivitySink {
843    fn is_noop(&self) -> bool {
844        true
845    }
846
847    async fn emit(&self, _activity: TurnActivity) {}
848}
849
850/// Optional sinks and scoped effect controller passed to one of [`LashRuntime`]'s
851/// turn-driving entry points (`stream_turn`,
852/// `stream_turn_with_agent_frames`).
853///
854/// Construct via [`TurnOptions::new`] and chain `with_*` builders. Event sinks
855/// default to no-op sinks. Execution scope is explicit and required at every
856/// runtime boundary that can execute nondeterministic work.
857pub struct TurnOptions<'a> {
858    events: Option<&'a dyn EventSink>,
859    turn_events: Option<&'a dyn TurnActivitySink>,
860    scoped_effect_controller: ScopedEffectController<'a>,
861    cancel: CancellationToken,
862}
863
864impl<'a> TurnOptions<'a> {
865    pub fn new(
866        cancel: CancellationToken,
867        scoped_effect_controller: ScopedEffectController<'a>,
868    ) -> Self {
869        Self {
870            events: None,
871            turn_events: None,
872            scoped_effect_controller,
873            cancel,
874        }
875    }
876
877    pub fn with_events(mut self, events: &'a dyn EventSink) -> Self {
878        self.events = Some(events);
879        self
880    }
881
882    pub fn with_turn_events(mut self, turn_events: &'a dyn TurnActivitySink) -> Self {
883        self.turn_events = Some(turn_events);
884        self
885    }
886
887    pub(crate) fn events_or_noop(&self) -> &'a dyn EventSink {
888        self.events.unwrap_or(&NOOP_EVENT_SINK)
889    }
890
891    pub(crate) fn turn_events_or_noop(&self) -> &'a dyn TurnActivitySink {
892        self.turn_events.unwrap_or(&NOOP_TURN_ACTIVITY_SINK)
893    }
894
895    pub(crate) fn execution_scope_id(&self) -> &str {
896        self.scoped_effect_controller.scope_id()
897    }
898
899    pub(crate) fn scoped_effect_controller(&self) -> ScopedEffectController<'a> {
900        self.scoped_effect_controller.clone()
901    }
902}
903
904enum RuntimeStreamEvent {
905    Session(SessionEvent),
906    Turn(TurnActivity),
907}
908
909#[derive(Clone)]
910pub struct SessionStoreCreateRequest {
911    pub session_id: String,
912    pub relation: SessionRelation,
913    pub policy: SessionPolicy,
914}
915
916impl SessionStoreCreateRequest {
917    pub fn parent_session_id(&self) -> Option<&str> {
918        self.relation.parent_session_id()
919    }
920}
921
922#[async_trait::async_trait]
923pub trait SessionStoreFactory: Send + Sync {
924    /// Durability tier the stores produced by this factory provide; defaults to
925    /// [`DurabilityTier::Inline`].
926    fn durability_tier(&self) -> crate::DurabilityTier {
927        crate::DurabilityTier::Inline
928    }
929
930    async fn create_store(
931        &self,
932        request: &SessionStoreCreateRequest,
933    ) -> Result<Arc<dyn crate::store::RuntimePersistence>, String>;
934
935    async fn open_existing_store(
936        &self,
937        _request: &SessionStoreCreateRequest,
938    ) -> Result<Option<Arc<dyn crate::store::RuntimePersistence>>, String> {
939        Ok(None)
940    }
941
942    async fn delete_session(&self, session_id: &str) -> Result<(), String>;
943}
944
945/// Generic runtime for CLI or programmatic embedding.
946pub struct LashRuntime {
947    pub(in crate::runtime) session: Option<Session>,
948    pub(in crate::runtime) policy: SessionPolicy,
949    pub(in crate::runtime) host: RuntimeHost,
950    pub(in crate::runtime) services: RuntimeServices,
951    pub(in crate::runtime) state: RuntimeSessionState,
952    pub(in crate::runtime) runtime_scope_id: Arc<str>,
953    pub(in crate::runtime) managed_sessions: Arc<Mutex<HashMap<String, RuntimeHandle>>>,
954    pub(in crate::runtime) managed_turns: Arc<Mutex<HashMap<String, ManagedSessionTurn>>>,
955    /// Protocol-owned turn options for this session.
956    pub(in crate::runtime) protocol_turn_options: crate::ProtocolTurnOptions,
957    /// Session-scoped token cost ledger. Shared by ALL
958    /// `RuntimeSessionServices` instances created from this runtime
959    /// (both per-turn and async maintenance). Entries accumulate here
960    /// and are drained into `state.token_ledger` at turn-commit time.
961    pub(in crate::runtime) shared_token_ledger: Arc<std::sync::Mutex<Vec<TokenLedgerEntry>>>,
962    pub(in crate::runtime) process_sync_needed: Arc<AtomicBool>,
963    pub(in crate::runtime) turn_phase_probe: Option<Arc<dyn RuntimeTurnPhaseProbe>>,
964    /// Resident-graph policy chosen by the host. Controls whether
965    /// [`LashRuntime::refresh_session_graph_from_store`] reloads the full
966    /// graph or just the active path, matching the trimming behavior set at
967    /// load time via [`apply_residency_on_load`](crate::runtime::apply_residency_on_load).
968    pub(in crate::runtime) residency: Residency,
969}
lash_core/runtime/mod.rs

lash_core/runtime/
mod.rs
