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

lash_core/runtime/
mod.rs
