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

lash_core/runtime/
mod.rs
