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

lash_core/runtime/
mod.rs
