lash_core/runtime/effect/
envelope.rs

1use std::sync::Arc;
2
3use serde::{Deserialize, Serialize};
4
5use crate::CheckpointKind;
6use crate::llm::types::{
7    LlmAttachment, LlmEventSender, LlmMessage, LlmOutputSpec, LlmProviderTraceSender,
8    LlmToolChoice, LlmToolSpec,
9};
10use crate::runtime::ProcessHandleGrantEntry;
11use crate::sansio::{CompletedToolCall, ExecutionEnvironmentSync, LlmCallError};
12use crate::tool_dispatch::ToolTriggerEffectOutcome;
13use crate::{
14    AttachmentCreateMeta, AttachmentRef, CausalRef, CheckpointDelivery, ExecResponse,
15    LlmRequest as CoreLlmRequest, LlmResponse, MediaType, ProcessAwaitOutput,
16    ProcessExecutionContext, ProcessListMode, ProcessRecord, ProcessRegistration,
17    ProcessStartGrant, SessionScope,
18};
19
20use super::executor::RuntimeEffectControllerError;
21
22/// Durable category for a runtime effect.
23#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
24#[serde(rename_all = "snake_case")]
25pub enum RuntimeEffectKind {
26    LlmCall,
27    Direct,
28    ToolAttempt,
29    ToolBatch,
30    Process,
31    ExecCode,
32    Checkpoint,
33    SyncExecutionEnvironment,
34    Sleep,
35    AwaitEvent,
36    DurableStep,
37}
38
39impl RuntimeEffectKind {
40    pub fn as_str(self) -> &'static str {
41        match self {
42            Self::LlmCall => "llm_call",
43            Self::Direct => "direct",
44            Self::ToolAttempt => "tool_attempt",
45            Self::ToolBatch => "tool_batch",
46            Self::Process => "process",
47            Self::ExecCode => "exec_code",
48            Self::Checkpoint => "checkpoint",
49            Self::SyncExecutionEnvironment => "sync_execution_environment",
50            Self::Sleep => "sleep",
51            Self::AwaitEvent => "await_event",
52            Self::DurableStep => "durable_step",
53        }
54    }
55}
56
57/// Canonical lineage for a runtime-side invocation.
58#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
59pub struct RuntimeInvocation {
60    pub scope: RuntimeScope,
61    pub subject: RuntimeSubject,
62    #[serde(default, skip_serializing_if = "Option::is_none")]
63    pub caused_by: Option<CausalRef>,
64    #[serde(default, skip_serializing_if = "Option::is_none")]
65    pub replay: Option<RuntimeReplay>,
66}
67
68impl RuntimeInvocation {
69    pub fn effect(
70        scope: RuntimeScope,
71        effect_id: impl Into<String>,
72        kind: RuntimeEffectKind,
73        replay_key: impl Into<String>,
74    ) -> Self {
75        Self {
76            scope,
77            subject: RuntimeSubject::Effect {
78                effect_id: effect_id.into(),
79                kind,
80            },
81            caused_by: None,
82            replay: Some(RuntimeReplay {
83                key: replay_key.into(),
84            }),
85        }
86    }
87
88    pub fn with_caused_by(mut self, caused_by: Option<CausalRef>) -> Self {
89        self.caused_by = caused_by;
90        self
91    }
92
93    pub fn effect_id(&self) -> Option<&str> {
94        match &self.subject {
95            RuntimeSubject::Effect { effect_id, .. } => Some(effect_id),
96            _ => None,
97        }
98    }
99
100    pub fn effect_kind(&self) -> Option<RuntimeEffectKind> {
101        match &self.subject {
102            RuntimeSubject::Effect { kind, .. } => Some(*kind),
103            _ => None,
104        }
105    }
106
107    pub fn replay_key(&self) -> Option<&str> {
108        self.replay.as_ref().map(|replay| replay.key.as_str())
109    }
110
111    pub fn causal_ref(&self) -> Option<CausalRef> {
112        match &self.subject {
113            RuntimeSubject::Effect { effect_id, .. } => Some(CausalRef::Effect {
114                session_id: self.scope.session_id.clone(),
115                turn_id: self.scope.turn_id.clone(),
116                effect_id: effect_id.clone(),
117            }),
118            RuntimeSubject::Process { process_id } => Some(CausalRef::Process {
119                process_id: process_id.clone(),
120            }),
121            RuntimeSubject::ProcessEvent {
122                process_id,
123                sequence,
124                ..
125            } => Some(CausalRef::ProcessEvent {
126                process_id: process_id.clone(),
127                sequence: *sequence,
128            }),
129            RuntimeSubject::TriggerOccurrence { occurrence_id } => {
130                Some(CausalRef::TriggerOccurrence {
131                    occurrence_id: occurrence_id.clone(),
132                    subscription_id: None,
133                })
134            }
135            RuntimeSubject::SessionNode { node_id } => Some(CausalRef::SessionNode {
136                session_id: self.scope.session_id.clone(),
137                node_id: node_id.clone(),
138            }),
139        }
140    }
141}
142
143#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
144pub struct RuntimeScope {
145    pub session_id: String,
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    pub turn_id: Option<String>,
148    #[serde(default, skip_serializing_if = "Option::is_none")]
149    pub turn_index: Option<usize>,
150    #[serde(default, skip_serializing_if = "Option::is_none")]
151    pub protocol_iteration: Option<usize>,
152}
153
154impl RuntimeScope {
155    pub fn new(session_id: impl Into<String>) -> Self {
156        Self {
157            session_id: session_id.into(),
158            turn_id: None,
159            turn_index: None,
160            protocol_iteration: None,
161        }
162    }
163
164    pub fn for_turn(
165        session_id: impl Into<String>,
166        turn_id: impl Into<String>,
167        turn_index: usize,
168        protocol_iteration: usize,
169    ) -> Self {
170        Self {
171            session_id: session_id.into(),
172            turn_id: Some(turn_id.into()),
173            turn_index: Some(turn_index),
174            protocol_iteration: Some(protocol_iteration),
175        }
176    }
177}
178
179#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
180pub struct RuntimeReplay {
181    pub key: String,
182}
183
184#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
185#[serde(tag = "type", rename_all = "snake_case")]
186pub enum RuntimeSubject {
187    Effect {
188        effect_id: String,
189        kind: RuntimeEffectKind,
190    },
191    Process {
192        process_id: String,
193    },
194    ProcessEvent {
195        process_id: String,
196        sequence: u64,
197        event_type: String,
198    },
199    TriggerOccurrence {
200        occurrence_id: String,
201    },
202    SessionNode {
203        node_id: String,
204    },
205}
206
207/// Fully serializable envelope emitted at Lash's nondeterministic boundary.
208#[derive(Clone, Debug, Serialize, Deserialize)]
209pub struct RuntimeEffectEnvelope {
210    pub invocation: RuntimeInvocation,
211    pub command: RuntimeEffectCommand,
212}
213
214impl RuntimeEffectEnvelope {
215    pub fn new(invocation: RuntimeInvocation, command: RuntimeEffectCommand) -> Self {
216        Self::try_new(invocation, command).expect("valid runtime effect invocation")
217    }
218
219    pub fn try_new(
220        invocation: RuntimeInvocation,
221        command: RuntimeEffectCommand,
222    ) -> Result<Self, RuntimeEffectControllerError> {
223        validate_effect_invocation(&invocation, command.kind())?;
224        validate_effect_command(&command)?;
225        Ok(Self {
226            invocation,
227            command,
228        })
229    }
230
231    pub fn stable_hash(&self) -> Result<String, RuntimeEffectControllerError> {
232        crate::stable_hash::stable_json_sha256_hex(self).map_err(|err| {
233            RuntimeEffectControllerError::new(
234                "runtime_effect_envelope_hash",
235                format!("failed to serialize runtime effect envelope: {err}"),
236            )
237        })
238    }
239}
240
241fn validate_effect_invocation(
242    invocation: &RuntimeInvocation,
243    command_kind: RuntimeEffectKind,
244) -> Result<(), RuntimeEffectControllerError> {
245    let RuntimeSubject::Effect { effect_id, kind } = &invocation.subject else {
246        return Err(RuntimeEffectControllerError::new(
247            "runtime_effect_invocation_subject",
248            "runtime effect envelope subject must be an effect",
249        ));
250    };
251    if effect_id.trim().is_empty() {
252        return Err(RuntimeEffectControllerError::new(
253            "runtime_effect_invocation_subject",
254            "runtime effect envelope effect id must be non-empty",
255        ));
256    }
257    if *kind != command_kind {
258        return Err(RuntimeEffectControllerError::new(
259            "runtime_effect_invocation_kind",
260            format!(
261                "runtime effect invocation kind {} does not match command kind {}",
262                kind.as_str(),
263                command_kind.as_str()
264            ),
265        ));
266    }
267    if invocation
268        .replay
269        .as_ref()
270        .is_none_or(|replay| replay.key.is_empty())
271    {
272        return Err(RuntimeEffectControllerError::new(
273            "runtime_effect_replay_required",
274            "runtime effect envelope requires replay.key",
275        ));
276    }
277    Ok(())
278}
279
280fn validate_effect_command(
281    command: &RuntimeEffectCommand,
282) -> Result<(), RuntimeEffectControllerError> {
283    if let RuntimeEffectCommand::DurableStep { step_id, .. } = command
284        && step_id.trim().is_empty()
285    {
286        return Err(RuntimeEffectControllerError::new(
287            "runtime_effect_durable_step_id",
288            "runtime effect durable step id must be non-empty",
289        ));
290    }
291    if let RuntimeEffectCommand::ToolAttempt {
292        call,
293        execution_grant: _,
294        attempt,
295        max_attempts,
296    } = command
297    {
298        if call.call_id.trim().is_empty() {
299            return Err(RuntimeEffectControllerError::new(
300                "runtime_effect_tool_attempt_call_id",
301                "runtime effect tool attempt requires a non-empty call id",
302            ));
303        }
304        if *attempt == 0 || *max_attempts == 0 || *attempt > *max_attempts {
305            return Err(RuntimeEffectControllerError::new(
306                "runtime_effect_tool_attempt_index",
307                format!(
308                    "runtime effect tool attempt must satisfy 1 <= attempt <= max_attempts, got {attempt}/{max_attempts}"
309                ),
310            ));
311        }
312    }
313    if let RuntimeEffectCommand::ToolBatch { batch } = command {
314        if batch.batch_id.trim().is_empty() {
315            return Err(RuntimeEffectControllerError::new(
316                "runtime_effect_tool_batch_id",
317                "runtime effect tool batch id must be non-empty",
318            ));
319        }
320        if batch.calls.is_empty() {
321            return Err(RuntimeEffectControllerError::new(
322                "runtime_effect_tool_batch_empty",
323                "runtime effect tool batch must contain at least one prepared call",
324            ));
325        }
326        for (index, call) in batch.calls.iter().enumerate() {
327            if call.call.call_id.trim().is_empty() {
328                return Err(RuntimeEffectControllerError::new(
329                    "runtime_effect_tool_batch_call_id",
330                    format!("runtime effect tool batch call {index} has an empty call id"),
331                ));
332            }
333            if call.replay_suffix.trim().is_empty() {
334                return Err(RuntimeEffectControllerError::new(
335                    "runtime_effect_tool_batch_call_replay",
336                    format!("runtime effect tool batch call {index} has an empty replay suffix"),
337                ));
338            }
339        }
340    }
341    Ok(())
342}
343
344/// Serializable command emitted at Lash's nondeterministic runtime boundary.
345#[derive(Clone, Debug, Serialize, Deserialize)]
346#[serde(tag = "type", rename_all = "snake_case")]
347pub enum RuntimeEffectCommand {
348    LlmCall {
349        request: Box<LlmRequestSpec>,
350    },
351    Direct {
352        request: Box<LlmRequestSpec>,
353        usage_source: String,
354    },
355    ToolAttempt {
356        call: crate::PreparedToolCall,
357        #[serde(default, skip_serializing_if = "Option::is_none")]
358        execution_grant: Option<Box<crate::ToolExecutionGrant>>,
359        attempt: u32,
360        max_attempts: u32,
361    },
362    ToolBatch {
363        batch: crate::PreparedToolBatch,
364    },
365    Process {
366        command: Box<ProcessCommand>,
367    },
368    ExecCode {
369        language: String,
370        code: String,
371    },
372    Checkpoint {
373        checkpoint: CheckpointKind,
374    },
375    SyncExecutionEnvironment {
376        update_machine_config: bool,
377    },
378    Sleep {
379        duration_ms: u64,
380    },
381    AwaitEvent {
382        key: crate::AwaitEventKey,
383    },
384    DurableStep {
385        step_id: String,
386        input: serde_json::Value,
387    },
388}
389
390impl RuntimeEffectCommand {
391    pub fn process(command: ProcessCommand) -> Self {
392        Self::Process {
393            command: Box::new(command),
394        }
395    }
396
397    pub fn kind(&self) -> RuntimeEffectKind {
398        match self {
399            Self::LlmCall { .. } => RuntimeEffectKind::LlmCall,
400            Self::Direct { .. } => RuntimeEffectKind::Direct,
401            Self::ToolAttempt { .. } => RuntimeEffectKind::ToolAttempt,
402            Self::ToolBatch { .. } => RuntimeEffectKind::ToolBatch,
403            Self::Process { .. } => RuntimeEffectKind::Process,
404            Self::ExecCode { .. } => RuntimeEffectKind::ExecCode,
405            Self::Checkpoint { .. } => RuntimeEffectKind::Checkpoint,
406            Self::SyncExecutionEnvironment { .. } => RuntimeEffectKind::SyncExecutionEnvironment,
407            Self::Sleep { .. } => RuntimeEffectKind::Sleep,
408            Self::AwaitEvent { .. } => RuntimeEffectKind::AwaitEvent,
409            Self::DurableStep { .. } => RuntimeEffectKind::DurableStep,
410        }
411    }
412}
413
414/// Serializable operation against the process admin plane.
415#[derive(Clone, Debug, Serialize, Deserialize)]
416#[serde(tag = "op", rename_all = "snake_case")]
417#[allow(clippy::large_enum_variant)]
418pub enum ProcessCommand {
419    Start {
420        registration: ProcessRegistration,
421        #[serde(default, skip_serializing_if = "Option::is_none")]
422        grant: Option<ProcessStartGrant>,
423        #[serde(
424            default,
425            skip_serializing_if = "boxed_process_execution_context_is_empty"
426        )]
427        execution_context: Box<ProcessExecutionContext>,
428    },
429    List {
430        session_scope: SessionScope,
431        #[serde(default)]
432        mode: ProcessListMode,
433    },
434    Transfer {
435        from_scope: SessionScope,
436        to_scope: SessionScope,
437        process_ids: Vec<String>,
438    },
439    DeleteSession {
440        session_id: String,
441    },
442    Await {
443        process_id: String,
444    },
445    Cancel {
446        process_id: String,
447        reason: Option<String>,
448    },
449    Signal {
450        process_id: String,
451        signal_name: String,
452        signal_id: String,
453        request: crate::ProcessEventAppendRequest,
454    },
455}
456
457fn boxed_process_execution_context_is_empty(context: &ProcessExecutionContext) -> bool {
458    context.is_empty()
459}
460
461type CheckpointOutcome = Result<CheckpointDelivery, RuntimeEffectControllerError>;
462
463impl ProcessCommand {
464    pub fn effect_id(&self) -> String {
465        match self {
466            Self::Start { registration, .. } => format!("process:start:{}", registration.id),
467            Self::List {
468                session_scope,
469                mode,
470            } => {
471                format!("process:list:{}:{}", session_scope.id(), mode.as_str())
472            }
473            Self::Transfer {
474                from_scope,
475                to_scope,
476                process_ids,
477            } => {
478                let digest = crate::stable_hash::stable_json_sha256_hex(process_ids)
479                    .unwrap_or_else(|_| "unhashable".to_string());
480                format!(
481                    "process:transfer:{}:{}:{digest}",
482                    from_scope.id(),
483                    to_scope.id()
484                )
485            }
486            Self::DeleteSession { session_id } => format!("process:delete-session:{session_id}"),
487            Self::Await { process_id } => format!("process:await:{process_id}"),
488            Self::Cancel { process_id, .. } => format!("process:cancel:{process_id}"),
489            Self::Signal {
490                process_id,
491                signal_name,
492                signal_id,
493                ..
494            } => {
495                format!("process:signal:{process_id}:signal.{signal_name}:{signal_id}")
496            }
497        }
498    }
499}
500
501/// Serializable result of a process operation.
502#[derive(Clone, Debug, Serialize, Deserialize)]
503#[serde(tag = "op", rename_all = "snake_case")]
504pub enum ProcessEffectOutcome {
505    Start {
506        // Boxed so the fat durable record does not size the whole outcome enum
507        // (and the runtime effect enum wrapping it) inline through the recursive
508        // effect executor.
509        record: Box<ProcessRecord>,
510    },
511    List {
512        entries: Vec<ProcessHandleGrantEntry>,
513    },
514    Transfer,
515    DeleteSession {
516        report: crate::ProcessSessionDeleteReport,
517    },
518    Await {
519        output: ProcessAwaitOutput,
520    },
521    Cancel {
522        record: Box<ProcessRecord>,
523    },
524    Signal {
525        // Boxed for the same reason as the record variants: a fat event should
526        // not size the outcome enum inline through the recursive executor.
527        event: Box<crate::ProcessEvent>,
528    },
529}
530
531#[derive(Clone, Debug, Serialize, Deserialize)]
532pub struct ToolAttemptEffectOutcome {
533    pub launch: ToolAttemptLaunch,
534    #[serde(default, skip_serializing_if = "Vec::is_empty")]
535    pub triggers: Vec<ToolTriggerEffectOutcome>,
536}
537
538#[derive(Clone, Debug, Serialize, Deserialize)]
539pub struct ToolBatchEffectOutcome {
540    pub launches: Vec<ToolCallLaunch>,
541    #[serde(default, skip_serializing_if = "Vec::is_empty")]
542    pub triggers: Vec<ToolTriggerEffectOutcome>,
543}
544
545#[derive(Clone, Debug, Serialize, Deserialize)]
546#[serde(tag = "status", rename_all = "snake_case")]
547#[allow(clippy::large_enum_variant)]
548pub enum ToolCallLaunch {
549    Done {
550        result: CompletedToolCall,
551    },
552    Pending {
553        key: crate::AwaitEventKey,
554        pending: crate::PendingCompletion,
555        duration_ms: u64,
556    },
557}
558
559#[derive(Clone, Debug, Serialize, Deserialize)]
560#[serde(tag = "status", rename_all = "snake_case")]
561pub enum ToolAttemptLaunch {
562    Done {
563        record: crate::ToolCallRecord,
564    },
565    Pending {
566        key: crate::AwaitEventKey,
567        pending: crate::PendingCompletion,
568        duration_ms: u64,
569    },
570}
571
572pub type RuntimeLlmCallOutcome = (
573    Result<LlmResponse, LlmCallError>,
574    bool,
575    Option<crate::LlmCallRecord>,
576);
577
578pub type RuntimeDirectLlmOutcome = (
579    Result<LlmResponse, LlmCallError>,
580    Option<crate::LlmCallRecord>,
581);
582
583/// Serializable result of a runtime effect command.
584#[derive(Clone, Debug, Serialize, Deserialize)]
585#[serde(tag = "type", rename_all = "snake_case")]
586#[allow(clippy::large_enum_variant)]
587pub enum RuntimeEffectOutcome {
588    LlmCall {
589        result: Result<LlmResponse, LlmCallError>,
590        text_streamed: bool,
591        /// Sealed provider-attempt history. Older journal entries and calls
592        /// interrupted before the provider handle returns have no record.
593        #[serde(default, skip_serializing_if = "Option::is_none")]
594        call_record: Option<crate::LlmCallRecord>,
595    },
596    Direct {
597        result: Result<LlmResponse, LlmCallError>,
598        /// Sealed provider-attempt history for this single direct call.
599        #[serde(default, skip_serializing_if = "Option::is_none")]
600        call_record: Option<crate::LlmCallRecord>,
601    },
602    ToolAttempt {
603        launch: ToolAttemptLaunch,
604        #[serde(default, skip_serializing_if = "Vec::is_empty")]
605        triggers: Vec<ToolTriggerEffectOutcome>,
606    },
607    ToolBatch {
608        launches: Vec<ToolCallLaunch>,
609        #[serde(default, skip_serializing_if = "Vec::is_empty")]
610        triggers: Vec<ToolTriggerEffectOutcome>,
611    },
612    Process {
613        result: ProcessEffectOutcome,
614    },
615    ExecCode {
616        result: Result<ExecResponse, String>,
617    },
618    Checkpoint {
619        result: CheckpointOutcome,
620    },
621    SyncExecutionEnvironment {
622        result: Result<Option<ExecutionEnvironmentSync>, String>,
623    },
624    Sleep,
625    AwaitEvent {
626        resolution: crate::Resolution,
627    },
628    DurableStep {
629        value: serde_json::Value,
630    },
631}
632
633// =============================================================================
634// Request specs (serializable forms of LLM/Direct requests)
635// =============================================================================
636
637/// Serializable attachment data for runtime effect envelopes.
638///
639/// Effect envelopes carry attachment references only. Local executors resolve
640/// bytes from the configured attachment store when a provider request is
641/// actually executed.
642#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
643pub struct LlmAttachmentSpec {
644    pub reference: AttachmentRef,
645}
646
647impl LlmAttachmentSpec {
648    fn into_attachment(self) -> LlmAttachment {
649        LlmAttachment::reference(self.reference)
650    }
651}
652
653/// Serializable LLM request data. Live stream and provider-trace callbacks are
654/// attached by the local executor, and attachment bytes are resolved locally
655/// from refs rather than persisted in the effect envelope.
656#[derive(Clone, Debug, Serialize, Deserialize)]
657pub struct LlmRequestSpec {
658    pub model: String,
659    pub messages: Vec<LlmMessage>,
660    pub attachments: Vec<LlmAttachmentSpec>,
661    pub tools: Arc<Vec<LlmToolSpec>>,
662    pub tool_choice: LlmToolChoice,
663    pub model_variant: crate::ReasoningSelection,
664    #[serde(default)]
665    pub model_capability: crate::ModelCapability,
666    #[serde(default)]
667    pub generation: crate::GenerationOptions,
668    pub scope: crate::LlmRequestScope,
669    pub output_spec: Option<LlmOutputSpec>,
670}
671
672impl LlmRequestSpec {
673    pub async fn from_request(
674        request: &CoreLlmRequest,
675        attachment_store: &crate::SessionAttachmentStore,
676    ) -> Result<Self, RuntimeEffectControllerError> {
677        Ok(Self {
678            model: request.model.clone(),
679            messages: request.messages.clone(),
680            attachments: attachment_specs_from_attachments(&request.attachments, attachment_store)
681                .await?,
682            tools: Arc::clone(&request.tools),
683            tool_choice: request.tool_choice.clone(),
684            model_variant: request.model_variant.clone(),
685            model_capability: request.model_capability.clone(),
686            generation: request.generation.clone(),
687            scope: request.scope.clone(),
688            output_spec: request.output_spec.clone(),
689        })
690    }
691
692    pub fn into_request(
693        self,
694        stream_events: Option<LlmEventSender>,
695        provider_trace: Option<LlmProviderTraceSender>,
696    ) -> CoreLlmRequest {
697        CoreLlmRequest {
698            model: self.model,
699            messages: self.messages,
700            attachments: self
701                .attachments
702                .into_iter()
703                .map(LlmAttachmentSpec::into_attachment)
704                .collect(),
705            tools: self.tools,
706            tool_choice: self.tool_choice,
707            model_variant: self.model_variant,
708            model_capability: self.model_capability,
709            generation: self.generation,
710            scope: self.scope,
711            output_spec: self.output_spec,
712            stream_events,
713            provider_trace,
714        }
715    }
716}
717
718async fn attachment_specs_from_attachments(
719    attachments: &[LlmAttachment],
720    attachment_store: &crate::SessionAttachmentStore,
721) -> Result<Vec<LlmAttachmentSpec>, RuntimeEffectControllerError> {
722    let mut specs = Vec::with_capacity(attachments.len());
723    for attachment in attachments {
724        specs.push(attachment_spec_from_attachment(attachment, attachment_store).await?);
725    }
726    Ok(specs)
727}
728
729async fn attachment_spec_from_attachment(
730    attachment: &LlmAttachment,
731    attachment_store: &crate::SessionAttachmentStore,
732) -> Result<LlmAttachmentSpec, RuntimeEffectControllerError> {
733    if let Some(reference) = attachment.reference.as_ref() {
734        return Ok(LlmAttachmentSpec {
735            reference: reference.clone(),
736        });
737    }
738    if attachment.data.is_empty() {
739        return Err(RuntimeEffectControllerError::new(
740            "runtime_effect_attachment_missing_reference",
741            "runtime effect attachment has neither a durable reference nor inline bytes",
742        ));
743    }
744    let media_type = MediaType::from_mime(&attachment.mime).ok_or_else(|| {
745        RuntimeEffectControllerError::new(
746            "runtime_effect_attachment_media_type",
747            format!(
748                "attachment media type `{}` cannot be represented durably",
749                attachment.mime
750            ),
751        )
752    })?;
753    let reference = attachment_store
754        .put(
755            attachment.data.clone(),
756            AttachmentCreateMeta::new(media_type, None, None, None),
757        )
758        .await
759        .map_err(|err| {
760            RuntimeEffectControllerError::new(
761                "runtime_effect_attachment_store",
762                format!("failed to store attachment before runtime effect invocation: {err}"),
763            )
764        })?;
765    Ok(LlmAttachmentSpec { reference })
766}
767
768impl RuntimeEffectOutcome {
769    pub fn into_llm_call(self) -> Result<RuntimeLlmCallOutcome, RuntimeEffectControllerError> {
770        match self {
771            Self::LlmCall {
772                result,
773                text_streamed,
774                call_record,
775            } => Ok((result, text_streamed, call_record)),
776            other => Err(RuntimeEffectControllerError::wrong_outcome(
777                RuntimeEffectKind::LlmCall,
778                other.kind(),
779            )),
780        }
781    }
782
783    pub fn into_direct_response(
784        self,
785    ) -> Result<RuntimeDirectLlmOutcome, RuntimeEffectControllerError> {
786        match self {
787            Self::Direct {
788                result,
789                call_record,
790            } => Ok((result, call_record)),
791            other => Err(RuntimeEffectControllerError::wrong_outcome(
792                RuntimeEffectKind::Direct,
793                other.kind(),
794            )),
795        }
796    }
797
798    pub fn into_tool_attempt_effect(
799        self,
800    ) -> Result<ToolAttemptEffectOutcome, RuntimeEffectControllerError> {
801        match self {
802            Self::ToolAttempt { launch, triggers } => {
803                Ok(ToolAttemptEffectOutcome { launch, triggers })
804            }
805            other => Err(RuntimeEffectControllerError::wrong_outcome(
806                RuntimeEffectKind::ToolAttempt,
807                other.kind(),
808            )),
809        }
810    }
811
812    pub fn into_tool_batch_effect(
813        self,
814    ) -> Result<ToolBatchEffectOutcome, RuntimeEffectControllerError> {
815        match self {
816            Self::ToolBatch { launches, triggers } => {
817                Ok(ToolBatchEffectOutcome { launches, triggers })
818            }
819            other => Err(RuntimeEffectControllerError::wrong_outcome(
820                RuntimeEffectKind::ToolBatch,
821                other.kind(),
822            )),
823        }
824    }
825
826    pub fn into_process(self) -> Result<ProcessEffectOutcome, RuntimeEffectControllerError> {
827        match self {
828            Self::Process { result } => Ok(result),
829            other => Err(RuntimeEffectControllerError::wrong_outcome(
830                RuntimeEffectKind::Process,
831                other.kind(),
832            )),
833        }
834    }
835
836    pub fn into_exec_code(
837        self,
838    ) -> Result<Result<ExecResponse, String>, RuntimeEffectControllerError> {
839        match self {
840            Self::ExecCode { result } => Ok(result),
841            other => Err(RuntimeEffectControllerError::wrong_outcome(
842                RuntimeEffectKind::ExecCode,
843                other.kind(),
844            )),
845        }
846    }
847
848    pub(crate) fn into_checkpoint(self) -> Result<CheckpointOutcome, RuntimeEffectControllerError> {
849        match self {
850            Self::Checkpoint { result } => Ok(result),
851            other => Err(RuntimeEffectControllerError::wrong_outcome(
852                RuntimeEffectKind::Checkpoint,
853                other.kind(),
854            )),
855        }
856    }
857
858    pub fn into_sync_execution_environment(
859        self,
860    ) -> Result<Result<Option<ExecutionEnvironmentSync>, String>, RuntimeEffectControllerError>
861    {
862        match self {
863            Self::SyncExecutionEnvironment { result } => Ok(result),
864            other => Err(RuntimeEffectControllerError::wrong_outcome(
865                RuntimeEffectKind::SyncExecutionEnvironment,
866                other.kind(),
867            )),
868        }
869    }
870
871    pub fn into_await_event(self) -> Result<crate::Resolution, RuntimeEffectControllerError> {
872        match self {
873            Self::AwaitEvent { resolution } => Ok(resolution),
874            other => Err(RuntimeEffectControllerError::wrong_outcome(
875                RuntimeEffectKind::AwaitEvent,
876                other.kind(),
877            )),
878        }
879    }
880
881    pub fn into_durable_step(self) -> Result<serde_json::Value, RuntimeEffectControllerError> {
882        match self {
883            Self::DurableStep { value } => Ok(value),
884            other => Err(RuntimeEffectControllerError::wrong_outcome(
885                RuntimeEffectKind::DurableStep,
886                other.kind(),
887            )),
888        }
889    }
890
891    pub fn kind(&self) -> RuntimeEffectKind {
892        match self {
893            Self::LlmCall { .. } => RuntimeEffectKind::LlmCall,
894            Self::Direct { .. } => RuntimeEffectKind::Direct,
895            Self::ToolAttempt { .. } => RuntimeEffectKind::ToolAttempt,
896            Self::ToolBatch { .. } => RuntimeEffectKind::ToolBatch,
897            Self::Process { .. } => RuntimeEffectKind::Process,
898            Self::ExecCode { .. } => RuntimeEffectKind::ExecCode,
899            Self::Checkpoint { .. } => RuntimeEffectKind::Checkpoint,
900            Self::SyncExecutionEnvironment { .. } => RuntimeEffectKind::SyncExecutionEnvironment,
901            Self::Sleep => RuntimeEffectKind::Sleep,
902            Self::AwaitEvent { .. } => RuntimeEffectKind::AwaitEvent,
903            Self::DurableStep { .. } => RuntimeEffectKind::DurableStep,
904        }
905    }
906}
lash_core/runtime/effect/envelope.rs

lash_core/runtime/effect/
envelope.rs
