lash_core/runtime/
mod.rs

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

lash_core/runtime/
mod.rs
