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

lash_core/runtime/
mod.rs
