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

lash_core/runtime/
mod.rs
