1use crate::support::*;
2use lash_core::runtime::{
3 ProcessCommand, ProcessEffectOutcome, RuntimeEffectCommand, RuntimeEffectEnvelope,
4 RuntimeEffectKind, RuntimeEffectLocalExecutor, RuntimeEffectOutcome, RuntimeInvocation,
5 RuntimeScope,
6};
7
8#[derive(Clone)]
9pub struct LashCore {
10 pub(crate) env: RuntimeEnvironment,
11 pub(crate) policy: SessionPolicy,
12 pub(crate) protocol_factory: Option<Arc<dyn PluginFactory>>,
13 pub(crate) store_factory: Option<Arc<dyn SessionStoreFactory>>,
14 pub(crate) plugin_factories: Arc<Vec<Arc<dyn PluginFactory>>>,
15 pub(crate) provider: Option<ProviderHandle>,
16 pub(crate) live_replay_store: Arc<dyn LiveReplayStore>,
17 pub(crate) process_lifecycle_available: bool,
21 pub(crate) work_driver: Arc<InlineWorkDriverSlot>,
24}
25
26pub(crate) enum ProcessWorkDriverSetup {
29 None,
31 LazyDefault {
37 config: Box<DurableProcessWorkerConfig>,
38 },
39 External { driver: ProcessWorkDriver },
41}
42
43#[derive(Clone, Default)]
44pub(crate) enum ProcessWorkSource {
45 #[default]
46 None,
47 Inline {
48 registry: Arc<dyn ProcessRegistry>,
49 hub: Option<lash_core::ProcessChangeHub>,
50 },
51 External(ProcessWorkDriver),
52}
53
54impl ProcessWorkSource {
55 fn process_registry(&self) -> Option<Arc<dyn ProcessRegistry>> {
56 match self {
57 Self::None => None,
58 Self::Inline { registry, .. } => Some(Arc::clone(registry)),
59 Self::External(driver) => Some(driver.process_registry()),
60 }
61 }
62
63 fn has_registry(&self) -> bool {
64 !matches!(self, Self::None)
65 }
66
67 fn watched(self, sink: Option<Arc<dyn lash_core::ProcessEventSink>>) -> Self {
68 match self {
69 Self::Inline {
70 registry,
71 hub: None,
72 } => {
73 let (registry, hub) = lash_core::watch_process_registry_with_sink(registry, sink);
74 Self::Inline {
75 registry,
76 hub: Some(hub),
77 }
78 }
79 other => other,
83 }
84 }
85}
86
87#[derive(Clone, Default)]
88pub(crate) enum QueuedWorkSource {
89 None,
90 #[default]
91 LazyDefault,
92 External(QueuedWorkDriver),
93}
94
95pub(crate) enum QueuedWorkDriverSetup {
96 None,
97 LazyDefault {
98 config: Arc<InlineQueuedWorkRunConfig>,
99 },
100 External {
101 driver: QueuedWorkDriver,
102 },
103}
104
105pub(crate) struct InlineWorkDriverSetup {
106 process: ProcessWorkDriverSetup,
107 queued: QueuedWorkDriverSetup,
108}
109
110#[derive(Clone, Default)]
111pub(crate) struct ResolvedWorkDrivers {
112 pub(crate) process: Option<ProcessWorkDriver>,
113 pub(crate) queued: Option<QueuedWorkDriver>,
114 pub(crate) drive_process_on_open: bool,
115}
116
117pub(crate) struct InlineWorkDriverSlot {
123 setup: InlineWorkDriverSetup,
124 drivers: tokio::sync::OnceCell<ResolvedWorkDrivers>,
125 phase_probe_slot: Option<lash_core::runtime::RuntimeTurnPhaseProbeSlot>,
126}
127
128impl InlineWorkDriverSlot {
129 fn new(setup: InlineWorkDriverSetup) -> Self {
130 let phase_probe_slot = match &setup.process {
131 ProcessWorkDriverSetup::LazyDefault { config } => {
132 Some(config.turn_phase_probe_slot.clone())
133 }
134 ProcessWorkDriverSetup::None | ProcessWorkDriverSetup::External { .. } => None,
135 };
136 Self {
137 setup,
138 drivers: tokio::sync::OnceCell::new(),
139 phase_probe_slot,
140 }
141 }
142
143 pub(crate) async fn drivers(&self) -> ResolvedWorkDrivers {
146 self.drivers
147 .get_or_init(|| async {
148 let queued = match &self.setup.queued {
149 QueuedWorkDriverSetup::None => None,
150 QueuedWorkDriverSetup::External { driver } => Some(driver.clone()),
151 QueuedWorkDriverSetup::LazyDefault { config } => Some(QueuedWorkDriver::new(
152 Arc::new(InlineQueuedWorkRunHandle::new(Arc::clone(config))),
153 )),
154 };
155 let (process, drive_process_on_open) = match &self.setup.process {
156 ProcessWorkDriverSetup::None => (None, false),
157 ProcessWorkDriverSetup::External { driver } => (Some(driver.clone()), false),
158 ProcessWorkDriverSetup::LazyDefault { config } => {
159 let mut config = (**config).clone();
160 if let Some(driver) = queued.clone() {
161 config = config.with_queued_work_driver(driver);
162 }
163 let registry = Arc::clone(&config.process_registry);
164 let hub = config.process_change_hub.clone();
165 let worker = DurableProcessWorker::new(config);
166 let driver = if let Some(hub) = hub {
167 ProcessWorkDriver::from_watched(
168 registry,
169 hub,
170 Arc::new(lash_core::InlineProcessRunHandle::new(worker)),
171 )
172 } else {
173 ProcessWorkDriver::inline(registry, worker)
174 };
175 (Some(driver), true)
176 }
177 };
178 ResolvedWorkDrivers {
179 process,
180 queued,
181 drive_process_on_open,
182 }
183 })
184 .await
185 .clone()
186 }
187
188 pub(crate) fn phase_probe_slot(&self) -> Option<lash_core::runtime::RuntimeTurnPhaseProbeSlot> {
189 self.phase_probe_slot.clone()
190 }
191
192 fn configured_process_work_driver(&self) -> Option<ProcessWorkDriver> {
193 match &self.setup.process {
194 ProcessWorkDriverSetup::External { driver } => Some(driver.clone()),
195 ProcessWorkDriverSetup::None | ProcessWorkDriverSetup::LazyDefault { .. } => None,
196 }
197 }
198
199 fn configured_queued_work_driver(&self) -> Option<QueuedWorkDriver> {
200 match &self.setup.queued {
201 QueuedWorkDriverSetup::External { driver } => Some(driver.clone()),
202 QueuedWorkDriverSetup::None | QueuedWorkDriverSetup::LazyDefault { .. } => None,
203 }
204 }
205}
206
207pub(crate) struct InlineQueuedWorkRunConfig {
208 env: RuntimeEnvironment,
209 policy: SessionPolicy,
210 protocol_factory: Option<Arc<dyn PluginFactory>>,
211 plugin_factories: Arc<Vec<Arc<dyn PluginFactory>>>,
212 store_factory: Arc<dyn SessionStoreFactory>,
213 live_replay_store: Arc<dyn LiveReplayStore>,
214 process_lifecycle_available: bool,
215}
216
217impl InlineQueuedWorkRunConfig {
218 fn new(
219 env: RuntimeEnvironment,
220 policy: SessionPolicy,
221 protocol_factory: Option<Arc<dyn PluginFactory>>,
222 plugin_factories: Arc<Vec<Arc<dyn PluginFactory>>>,
223 store_factory: Arc<dyn SessionStoreFactory>,
224 live_replay_store: Arc<dyn LiveReplayStore>,
225 process_lifecycle_available: bool,
226 ) -> Self {
227 Self {
228 env,
229 policy,
230 protocol_factory,
231 plugin_factories,
232 store_factory,
233 live_replay_store,
234 process_lifecycle_available,
235 }
236 }
237}
238
239struct InlineQueuedWorkRunHandle {
240 config: Arc<InlineQueuedWorkRunConfig>,
241}
242
243impl InlineQueuedWorkRunHandle {
244 fn new(config: Arc<InlineQueuedWorkRunConfig>) -> Self {
245 Self { config }
246 }
247}
248
249#[async_trait]
250impl QueuedWorkRunHandle for InlineQueuedWorkRunHandle {
251 async fn run_queued_work(
252 &self,
253 request: QueuedWorkRunRequest,
254 ) -> std::result::Result<(), lash_core::PluginError> {
255 let Some(session_id) = request.session_id else {
256 return Ok(());
257 };
258 let reason = request.reason;
259 let mut policy = self.config.policy.clone();
260 policy.session_id = Some(session_id.clone());
261 let store = self
262 .config
263 .store_factory
264 .create_store(&SessionStoreCreateRequest {
265 session_id: session_id.clone(),
266 relation: SessionRelation::default(),
267 policy: policy.clone(),
268 })
269 .await
270 .map_err(lash_core::PluginError::Session)?;
271 let state = crate::session::load_state_for_residency(
272 self.config.env.residency,
273 &session_id,
274 &policy,
275 store.as_ref(),
276 )
277 .await
278 .map_err(|err| lash_core::PluginError::Session(err.to_string()))?;
279 let plugin_host = build_plugin_host(
280 self.config.protocol_factory.as_ref(),
281 self.config.plugin_factories.as_ref(),
282 Vec::new(),
283 )
284 .map_err(|err| lash_core::PluginError::Session(err.to_string()))?;
285 let mut env = self.config.env.clone();
286 env.core = plugin_host
287 .install_process_engine_contributions(
288 env.core.clone(),
289 self.config.process_lifecycle_available,
290 )
291 .map_err(|err| lash_core::PluginError::Session(err.to_string()))?;
292 env.plugin_host = Some(Arc::new(plugin_host));
293 let effect_host = Arc::clone(&env.core.control.effect_host);
294 let runtime = LashRuntime::from_environment(&env, policy, state, Some(store))
295 .await
296 .map_err(|err| lash_core::PluginError::Session(err.to_string()))?;
297 let handle = RuntimeHandle::with_live_replay_store(
298 runtime,
299 Arc::clone(&self.config.live_replay_store),
300 );
301 let scope = lash_core::ExecutionScope::queue_drain(session_id, reason);
302 let scoped = effect_host
303 .scoped(scope)
304 .map_err(|err| lash_core::PluginError::Session(err.to_string()))?;
305 crate::turn::stream_next_queued_prepared_turn(
306 &handle,
307 crate::turn::TurnSinks::default(),
308 scoped,
309 CancellationToken::new(),
310 lash_core::TurnCancelOriginHint::default(),
311 &[],
312 )
313 .await
314 .map_err(|err| lash_core::PluginError::Session(err.to_string()))?;
315 Ok(())
316 }
317}
318
319#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
320pub struct SessionDeleteReport {
321 pub session_id: String,
322 pub process: Option<lash_core::ProcessSessionDeleteReport>,
323}
324
325impl LashCore {
326 pub fn builder() -> LashCoreBuilder {
327 LashCoreBuilder::default()
328 }
329
330 pub fn standard_builder() -> LashCoreBuilder {
333 LashCore::builder()
334 .protocol_plugin(Arc::new(
335 lash_protocol_standard::StandardProtocolPluginFactory::new(),
336 ))
337 .plugins(default_runtime_stack())
338 }
339
340 #[cfg(feature = "rlm")]
348 pub fn rlm_builder(factory: crate::rlm::RlmProtocolPluginFactory) -> LashCoreBuilder {
349 LashCore::builder()
350 .protocol_plugin(Arc::new(factory))
351 .plugins(default_runtime_stack())
352 }
353
354 pub fn session(&self, session_id: impl Into<String>) -> SessionBuilder {
355 SessionBuilder {
356 core: self.clone(),
357 session_id: session_id.into(),
358 spec: SessionSpec::inherit(),
359 parent_session_id: None,
360 session_execution_owner: None,
361 store: None,
362 provider: None,
363 active_plugins: Vec::new(),
364 plugin_factories: Vec::new(),
365 plugin_options: PluginOptions::default(),
366 }
367 }
368
369 pub async fn resume(&self, parked: ParkedSession) -> Result<LashSession> {
382 let plugin_host = build_plugin_host(
387 self.protocol_factory.as_ref(),
388 self.plugin_factories.as_ref(),
389 Vec::new(),
390 )?;
391 let mut env = self.env.clone();
392 env.core = plugin_host.install_process_engine_contributions(
393 env.core.clone(),
394 self.process_lifecycle_available,
395 )?;
396 env.plugin_host = Some(Arc::new(plugin_host));
397 let effect_host = Arc::clone(&env.core.control.effect_host);
398 let drivers = self.work_driver.drivers().await;
399 env.process_work_driver = drivers.process.clone();
400 env.queued_work_driver = drivers.queued.clone();
401 let runtime = LashRuntime::resume(parked.inner, &env).await?;
402 let handle =
403 RuntimeHandle::with_live_replay_store(runtime, Arc::clone(&self.live_replay_store));
404 Ok(LashSession {
405 runtime: handle,
406 effect_host,
407 parent_session_id: None,
408 active_plugins: Vec::new(),
409 process_phase_probe_slot: self.work_driver.phase_probe_slot(),
410 turn_cancels: crate::turn::TurnCancelRegistry::default(),
411 })
412 }
413
414 pub fn flush_trace_sink(&self) -> Result<()> {
425 if let Some(sink) = self.env.core.tracing.trace_sink.as_ref() {
426 sink.flush()?;
427 }
428 Ok(())
429 }
430
431 pub fn triggers(&self) -> crate::admin::CoreTriggerAdmin {
432 crate::admin::CoreTriggerAdmin { core: self.clone() }
433 }
434
435 pub fn processes(&self) -> crate::process_admin::Processes {
436 crate::process_admin::Processes { core: self.clone() }
437 }
438
439 pub fn completions(&self) -> crate::admin::Completions {
440 crate::admin::Completions { core: self.clone() }
441 }
442
443 pub fn effect_host(&self) -> Arc<dyn EffectHost> {
444 Arc::clone(&self.env.core.control.effect_host)
445 }
446
447 pub fn turn_work_driver(&self) -> lash_core::TurnWorkDriver {
453 lash_core::TurnWorkDriver::new(self.effect_host())
454 }
455
456 pub async fn enqueue_turn_input(
463 &self,
464 session_id: impl Into<String>,
465 input: lash_core::TurnInput,
466 ingress: lash_core::TurnInputIngress,
467 id: Option<String>,
468 ) -> Result<lash_core::PendingTurnInput> {
469 lash_core::ensure_durable_effect_input(&input).map_err(EmbedError::Runtime)?;
470 let session_id = session_id.into();
471 let Some(store_factory) = self.store_factory.as_ref() else {
472 return Err(EmbedError::MissingSessionStoreFactory);
473 };
474 let mut policy = self.policy.clone();
475 policy.session_id = Some(session_id.clone());
476 let store = store_factory
477 .create_store(&SessionStoreCreateRequest {
478 session_id: session_id.clone(),
479 relation: SessionRelation::default(),
480 policy,
481 })
482 .await
483 .map_err(|message| EmbedError::StoreFactory {
484 session_id: session_id.clone(),
485 message,
486 })?;
487 let is_next_turn = matches!(ingress, lash_core::TurnInputIngress::NextTurn);
488 let mut draft = lash_core::PendingTurnInputDraft::new(session_id, ingress, input);
489 draft.source_key = id.map(|id| format!("host:{id}"));
490 let enqueued = store
491 .enqueue_pending_turn_input(draft)
492 .await
493 .map_err(|err| {
494 EmbedError::Runtime(lash_core::RuntimeError::new(
495 lash_core::RuntimeErrorCode::StoreCommitFailed,
496 err.to_string(),
497 ))
498 })?;
499 if is_next_turn && let Some(driver) = self.work_driver.drivers().await.queued.as_ref() {
500 driver
501 .claim_and_run_pending(Some(&enqueued.session_id), "queued_turn_input")
502 .await
503 .map_err(|err| {
504 EmbedError::Runtime(lash_core::RuntimeError::new(
505 lash_core::RuntimeErrorCode::Other("queued_work".to_string()),
506 err.to_string(),
507 ))
508 })?;
509 }
510 Ok(enqueued)
511 }
512
513 pub async fn delete_session(
514 &self,
515 session_id: impl AsRef<str>,
516 scoped_effect_controller: ScopedEffectController<'_>,
517 ) -> Result<SessionDeleteReport> {
518 let session_id = session_id.as_ref().to_string();
519 let Some(store_factory) = self.store_factory.as_ref() else {
520 return Err(EmbedError::MissingSessionStoreFactory);
521 };
522 let process = if let Some(process_registry) = self.env.process_registry.as_ref() {
523 let invocation = RuntimeInvocation::effect(
524 RuntimeScope::new(session_id.clone()),
525 format!("process:delete-session:{session_id}"),
526 RuntimeEffectKind::Process,
527 format!("{session_id}:delete-session"),
528 );
529 let outcome = scoped_effect_controller
530 .controller()
531 .execute_effect(
532 RuntimeEffectEnvelope::new(
533 invocation,
534 RuntimeEffectCommand::process(ProcessCommand::DeleteSession {
535 session_id: session_id.clone(),
536 }),
537 ),
538 RuntimeEffectLocalExecutor::processes(
539 Arc::clone(process_registry),
540 self.env.process_work_driver.clone(),
541 ),
542 )
543 .await
544 .map_err(|err| EmbedError::SessionDeleteProcess {
545 session_id: session_id.clone(),
546 message: err.to_string(),
547 })?;
548 match outcome {
549 RuntimeEffectOutcome::Process {
550 result: ProcessEffectOutcome::DeleteSession { report },
551 } => Some(report),
552 other => {
553 return Err(EmbedError::SessionDeleteProcess {
554 session_id,
555 message: format!(
556 "process delete returned the wrong outcome: {}",
557 other.kind().as_str()
558 ),
559 });
560 }
561 }
562 } else {
563 None
564 };
565 if let Some(trigger_store) = self.env.trigger_store.as_ref() {
566 trigger_store
567 .delete_session_subscriptions(&session_id)
568 .await
569 .map_err(|err| EmbedError::SessionDeleteProcess {
570 session_id: session_id.clone(),
571 message: err.to_string(),
572 })?;
573 }
574 self.env
575 .core
576 .control
577 .effect_host
578 .revoke_await_events_for_session(&session_id)
579 .await
580 .map_err(|err| EmbedError::SessionDeleteProcess {
581 session_id: session_id.clone(),
582 message: err.to_string(),
583 })?;
584 store_factory
585 .delete_session(&session_id)
586 .await
587 .map_err(|message| EmbedError::StoreFactory {
588 session_id: session_id.clone(),
589 message,
590 })?;
591 Ok(SessionDeleteReport {
592 session_id,
593 process,
594 })
595 }
596
597 pub fn process_registry(&self) -> Option<Arc<dyn ProcessRegistry>> {
598 self.env.process_registry.as_ref().cloned()
599 }
600
601 pub fn durable_process_worker_config(&self) -> Result<DurableProcessWorkerConfig> {
602 self.durable_process_worker_config_with_plugins(std::iter::empty::<Arc<dyn PluginFactory>>())
603 }
604
605 pub fn durable_process_worker_config_with_plugins(
606 &self,
607 extra_plugin_factories: impl IntoIterator<Item = Arc<dyn PluginFactory>>,
608 ) -> Result<DurableProcessWorkerConfig> {
609 let Some(process_registry) = self.process_registry() else {
610 return Err(EmbedError::MissingProcessRegistry);
611 };
612 let Some(store_factory) = self.store_factory.as_ref() else {
613 return Err(EmbedError::MissingProcessWorkerStoreFactory);
614 };
615 let plugin_host = build_plugin_host(
616 self.protocol_factory.as_ref(),
617 self.plugin_factories.as_ref(),
618 extra_plugin_factories.into_iter().collect(),
619 )?;
620 let runtime_host = plugin_host.install_process_engine_contributions(
621 self.env.core.clone(),
622 self.process_lifecycle_available,
623 )?;
624 let mut config = DurableProcessWorkerConfig::new(
625 Arc::new(plugin_host),
626 runtime_host,
627 Arc::clone(store_factory),
628 process_registry,
629 )
630 .with_session_policy(self.policy.clone())
631 .with_residency(self.env.residency);
632 if let Some(trigger_store) = self.env.trigger_store.as_ref() {
633 config = config.with_trigger_store(Arc::clone(trigger_store));
634 }
635 if let Some(driver) = self.work_driver.configured_process_work_driver() {
636 config = config
637 .with_change_hub(driver.change_hub())
638 .with_process_work_driver(driver);
639 }
640 if let Some(driver) = self.work_driver.configured_queued_work_driver() {
641 config = config.with_queued_work_driver(driver);
642 }
643 Ok(config)
644 }
645}
646
647fn default_runtime_stack() -> PluginStack {
648 lash_plugin_tool_output_budget::tool_output_budget_stack()
649}
650
651#[derive(Default)]
652pub struct LashCoreBuilder {
653 pub(crate) protocol_factory: Option<Arc<dyn PluginFactory>>,
654 session_spec: SessionSpec,
655 provider: Option<ProviderHandle>,
656 pub(crate) store_factory: Option<Arc<dyn SessionStoreFactory>>,
657 child_store_factory: Option<Arc<dyn SessionStoreFactory>>,
658 effect_host: Option<Arc<dyn EffectHost>>,
662 attachment_store: Option<Arc<dyn AttachmentStore>>,
663 process_env_store: Option<Arc<dyn ProcessExecutionEnvStore>>,
664 trigger_store: Option<Arc<dyn lash_core::TriggerStore>>,
665 prompt: Option<PromptLayer>,
667 trace_sink: Option<Arc<dyn lash_trace::TraceSink>>,
668 trace_level: Option<lash_trace::TraceLevel>,
669 trace_context: Option<lash_trace::TraceContext>,
670 termination: Option<TerminationPolicy>,
671 runtime_host_config: Option<RuntimeHostConfig>,
673 tool_providers: Vec<Arc<dyn ToolProvider>>,
674 plugin_stack: PluginStack,
675 plugin_host: Option<PluginHost>,
676 residency: Option<Residency>,
677 lease_timings: Option<lash_core::LeaseTimings>,
678 clock: Option<Arc<dyn lash_core::Clock>>,
679 process_work_source: ProcessWorkSource,
682 process_event_sink: Option<Arc<dyn lash_core::ProcessEventSink>>,
685 queued_work_source: QueuedWorkSource,
686 live_replay_store: Option<Arc<dyn LiveReplayStore>>,
687}
688
689impl LashCoreBuilder {
690 pub fn protocol_plugin(mut self, plugin: Arc<dyn PluginFactory>) -> Self {
691 self.protocol_factory = Some(plugin);
692 self
693 }
694
695 pub fn provider(mut self, provider: ProviderHandle) -> Self {
696 self.session_spec = self.session_spec.provider_id(provider.kind());
697 self.provider = Some(provider);
698 self
699 }
700
701 pub fn model(mut self, model: lash_core::ModelSpec) -> Self {
702 self.session_spec = self.session_spec.model(model);
703 self
704 }
705
706 pub fn max_turns(mut self, max_turns: usize) -> Self {
707 self.session_spec = self.session_spec.max_turns(max_turns);
708 self
709 }
710
711 pub fn session_spec(mut self, spec: SessionSpec) -> Self {
712 self.session_spec = spec;
713 self
714 }
715
716 pub fn store_factory(mut self, store_factory: Arc<dyn SessionStoreFactory>) -> Self {
724 self.store_factory = Some(store_factory);
725 self
726 }
727
728 pub fn child_store_factory(mut self, store_factory: Arc<dyn SessionStoreFactory>) -> Self {
736 self.child_store_factory = Some(store_factory);
737 self
738 }
739
740 pub fn attachment_store(mut self, attachment_store: Arc<dyn AttachmentStore>) -> Self {
741 self.attachment_store = Some(attachment_store);
742 self
743 }
744
745 pub fn process_env_store(
746 mut self,
747 process_env_store: Arc<dyn ProcessExecutionEnvStore>,
748 ) -> Self {
749 self.process_env_store = Some(process_env_store);
750 self
751 }
752
753 pub fn effect_host(mut self, effect_host: Arc<dyn EffectHost>) -> Self {
758 self.effect_host = Some(effect_host);
759 self
760 }
761
762 pub fn tools(mut self, tools: Arc<dyn ToolProvider>) -> Self {
763 self.tool_providers.push(tools);
764 self
765 }
766
767 pub fn plugin(mut self, plugin: Arc<dyn PluginFactory>) -> Self {
768 self.plugin_stack.push(plugin);
769 self
770 }
771
772 pub fn plugins(mut self, stack: PluginStack) -> Self {
773 self.plugin_stack = stack;
774 self
775 }
776
777 pub fn configure_plugins(mut self, configure: impl FnOnce(&mut PluginStack)) -> Self {
778 configure(&mut self.plugin_stack);
779 self
780 }
781
782 pub fn trace_sink(mut self, trace_sink: Arc<dyn lash_trace::TraceSink>) -> Self {
783 self.trace_sink = Some(trace_sink);
784 self
785 }
786
787 pub fn trace_jsonl_path(mut self, path: impl Into<std::path::PathBuf>) -> Self {
788 self.trace_sink = Some(Arc::new(lash_trace::JsonlTraceSink::new(path.into())));
789 self
790 }
791
792 pub fn trace_level(mut self, trace_level: lash_trace::TraceLevel) -> Self {
793 self.trace_level = Some(trace_level);
794 self
795 }
796
797 pub fn trace_context(mut self, trace_context: lash_trace::TraceContext) -> Self {
798 self.trace_context = Some(trace_context);
799 self
800 }
801
802 pub fn termination(mut self, termination: TerminationPolicy) -> Self {
803 self.termination = Some(termination);
804 self
805 }
806
807 pub fn residency(mut self, residency: Residency) -> Self {
808 self.residency = Some(residency);
809 self
810 }
811
812 pub fn lease_timings(mut self, lease_timings: lash_core::LeaseTimings) -> Self {
825 self.lease_timings = Some(lease_timings);
826 self
827 }
828
829 pub fn clock(mut self, clock: Arc<dyn lash_core::Clock>) -> Self {
831 self.clock = Some(clock);
832 self
833 }
834
835 pub fn live_replay_store(mut self, live_replay_store: Arc<dyn LiveReplayStore>) -> Self {
839 self.live_replay_store = Some(live_replay_store);
840 self
841 }
842
843 fn resolve_runtime_host_config(&mut self) -> Result<RuntimeHostConfig> {
846 if let Some(base) = self.runtime_host_config.take() {
847 return Ok(self.apply_core_overrides(base));
848 }
849 let effect_host = self
850 .effect_host
851 .take()
852 .ok_or(EmbedError::MissingEffectHost)?;
853 let attachment_store = self
854 .attachment_store
855 .take()
856 .ok_or(EmbedError::MissingAttachmentStore)?;
857 let process_env_store = self
858 .process_env_store
859 .take()
860 .ok_or(EmbedError::MissingProcessEnvStore)?;
861 let core = RuntimeHostConfig::new(effect_host, attachment_store, process_env_store);
862 Ok(self.apply_core_overrides(core))
863 }
864
865 fn apply_core_overrides(&mut self, mut core: RuntimeHostConfig) -> RuntimeHostConfig {
867 if let Some(effect_host) = self.effect_host.take() {
868 core.control.effect_host = effect_host;
869 }
870 if let Some(attachment_store) = self.attachment_store.take() {
871 core.durability.attachment_store = Arc::new(
872 lash_core::SessionAttachmentStore::ephemeral(attachment_store),
873 );
874 }
875 if let Some(process_env_store) = self.process_env_store.take() {
876 core.durability.process_env_store = process_env_store;
877 }
878 if let Some(prompt) = self.prompt.take() {
879 core.prompt.prompt = prompt;
880 }
881 if let Some(trace_sink) = self.trace_sink.take() {
882 core.tracing.trace_sink = Some(trace_sink);
883 }
884 if let Some(trace_level) = self.trace_level.take() {
885 core.tracing.trace_level = trace_level;
886 }
887 if let Some(trace_context) = self.trace_context.take() {
888 core.tracing.trace_context = trace_context;
889 }
890 if let Some(termination) = self.termination.take() {
891 core.control.termination = termination;
892 }
893 if let Some(lease_timings) = self.lease_timings.take() {
894 core.control.lease_timings = lease_timings;
895 }
896 if let Some(clock) = self.clock.take() {
897 core.clock = clock;
898 }
899 core
900 }
901
902 fn effective_session_store_tier(&self) -> Option<DurabilityTier> {
915 self.child_store_factory
916 .as_ref()
917 .or(self.store_factory.as_ref())
918 .map(|factory| factory.durability_tier())
919 }
920
921 fn ensure_store_peer_coherence(
930 session_store_tier: Option<DurabilityTier>,
931 trigger_store_tier: Option<DurabilityTier>,
932 process_registry_tier: Option<DurabilityTier>,
933 core: &RuntimeHostConfig,
934 ) -> Result<()> {
935 let attachment_tier = Some(
936 core.durability
937 .attachment_store
938 .persistence()
939 .durability_tier(),
940 );
941 let process_env_tier = Some(core.durability.process_env_store.durability_tier());
942 let effect_host_tier = Some(core.control.effect_host.durability_tier());
943
944 if session_store_tier == Some(DurabilityTier::Durable) {
945 if attachment_tier == Some(DurabilityTier::Inline) {
946 return Err(EmbedError::DurableStorePeerRequired {
947 facet: "attachment store",
948 });
949 }
950 if process_env_tier == Some(DurabilityTier::Inline) {
951 return Err(EmbedError::DurableStorePeerRequired {
952 facet: "process execution environment store",
953 });
954 }
955 for engine in core.process_engines.engines() {
958 if engine.durability_tier() == DurabilityTier::Inline {
959 return Err(EmbedError::DurableStorePeerRequired {
960 facet: engine.kind(),
961 });
962 }
963 }
964 }
965
966 if process_registry_tier == Some(DurabilityTier::Durable) {
967 if session_store_tier != Some(DurabilityTier::Durable) {
968 return Err(EmbedError::DurableProcessRegistryRequiresStoreFactory);
969 }
970 if trigger_store_tier != Some(DurabilityTier::Durable) {
971 return Err(EmbedError::DurableStorePeerRequired {
972 facet: "trigger store",
973 });
974 }
975 if process_env_tier != Some(DurabilityTier::Durable) {
976 return Err(EmbedError::DurableStorePeerRequired {
977 facet: "process execution environment store",
978 });
979 }
980 }
981
982 if trigger_store_tier == Some(DurabilityTier::Durable) {
983 if session_store_tier != Some(DurabilityTier::Durable) {
984 return Err(EmbedError::DurableStorePeerRequired {
985 facet: "session store factory",
986 });
987 }
988 if process_env_tier != Some(DurabilityTier::Durable) {
989 return Err(EmbedError::DurableStorePeerRequired {
990 facet: "process execution environment store",
991 });
992 }
993 if process_registry_tier == Some(DurabilityTier::Inline) {
994 return Err(EmbedError::DurableStorePeerRequired {
995 facet: "process registry",
996 });
997 }
998 }
999
1000 if effect_host_tier == Some(DurabilityTier::Durable) {
1001 if attachment_tier != Some(DurabilityTier::Durable) {
1002 return Err(EmbedError::DurableStorePeerRequired {
1003 facet: "attachment store",
1004 });
1005 }
1006 if process_env_tier != Some(DurabilityTier::Durable) {
1007 return Err(EmbedError::DurableStorePeerRequired {
1008 facet: "process execution environment store",
1009 });
1010 }
1011 }
1012
1013 Ok(())
1014 }
1015
1016 pub fn build(mut self) -> Result<LashCore> {
1017 let protocol_factory = self.protocol_factory.clone();
1018 if protocol_factory.is_none() && self.plugin_host.is_none() {
1019 return Err(EmbedError::MissingProtocolPlugin);
1020 }
1021 let provider_id = self
1022 .session_spec
1023 .provider_id
1024 .clone()
1025 .or_else(|| {
1026 self.provider
1027 .as_ref()
1028 .map(|provider| provider.kind().to_string())
1029 })
1030 .unwrap_or_default();
1031 let model = self
1032 .session_spec
1033 .model
1034 .clone()
1035 .ok_or(EmbedError::MissingModelSpec)?;
1036
1037 let base_policy = SessionPolicy {
1038 provider_id,
1039 model,
1040 max_turns: self.session_spec.max_turns.flatten(),
1041 ..SessionPolicy::default()
1042 };
1043 let policy = self.session_spec.resolve_against(&base_policy);
1044
1045 let session_store_tier = self.effective_session_store_tier();
1048 let trigger_store_tier = self
1049 .trigger_store
1050 .as_ref()
1051 .map(|store| store.durability_tier());
1052 let process_work_source = self
1053 .process_work_source
1054 .clone()
1055 .watched(self.process_event_sink.clone());
1056 let process_registry_tier = process_work_source
1057 .process_registry()
1058 .map(|registry| registry.durability_tier());
1059
1060 let mut core = self.resolve_runtime_host_config()?;
1061 if let Some(provider) = self.provider.clone() {
1062 core.providers.provider_resolver =
1063 Arc::new(lash_core::SingleProviderResolver::new(provider));
1064 }
1065 let plugin_factories = if let Some(plugin_host) = self.plugin_host {
1066 plugin_host.factories().to_vec()
1067 } else {
1068 let mut factories = Vec::new();
1069 if !self.tool_providers.is_empty() {
1070 let spec = self
1071 .tool_providers
1072 .into_iter()
1073 .fold(PluginSpec::new(), PluginSpec::with_tool_provider);
1074 factories.push(Arc::new(StaticPluginFactory::new("embed_tools", spec))
1075 as Arc<dyn PluginFactory>);
1076 }
1077 factories.extend(self.plugin_stack.into_factories());
1078 factories
1079 };
1080 let default_plugin_host =
1081 build_plugin_host(protocol_factory.as_ref(), &plugin_factories, Vec::new())?;
1082 let process_lifecycle_available = process_work_source.has_registry();
1086 let core_with_engines = default_plugin_host
1094 .install_process_engine_contributions(core.clone(), process_lifecycle_available)?;
1095 Self::ensure_store_peer_coherence(
1098 session_store_tier,
1099 trigger_store_tier,
1100 process_registry_tier,
1101 &core_with_engines,
1102 )?;
1103
1104 let process_registry = process_work_source.process_registry();
1105
1106 let process_work_driver = Self::resolve_process_work_driver(
1112 &process_work_source,
1113 &default_plugin_host,
1114 &core,
1115 process_lifecycle_available,
1116 self.child_store_factory
1119 .as_ref()
1120 .or(self.store_factory.as_ref()),
1121 &policy,
1122 self.residency.unwrap_or_default(),
1123 self.trigger_store.as_ref(),
1124 )?;
1125
1126 let live_replay_clock = Arc::clone(&core.clock);
1127 let mut env_builder = RuntimeEnvironment::builder()
1128 .with_plugin_host(Arc::new(default_plugin_host))
1129 .with_runtime_host_config(core);
1130 if let Some(process_registry) = process_registry.as_ref() {
1131 env_builder = env_builder.with_process_registry(Arc::clone(process_registry));
1132 }
1133 if let Some(residency) = self.residency {
1134 env_builder = env_builder.with_residency(residency);
1135 }
1136 if let Some(child_store_factory) = self
1137 .child_store_factory
1138 .as_ref()
1139 .or(self.store_factory.as_ref())
1140 {
1141 env_builder = env_builder.with_session_store_factory(Arc::clone(child_store_factory));
1142 }
1143 if let Some(trigger_store) = self.trigger_store.as_ref() {
1144 env_builder = env_builder.with_trigger_store(Arc::clone(trigger_store));
1145 }
1146 let live_replay_store = self.live_replay_store.take().unwrap_or_else(|| {
1147 Arc::new(InMemoryLiveReplayStore::with_clock(
1148 lash_core::InMemoryLiveReplayStoreConfig::default(),
1149 live_replay_clock,
1150 ))
1151 });
1152 let env = env_builder.build();
1153 let queued_work_driver = Self::resolve_queued_work_driver(
1154 &self.queued_work_source,
1155 env.clone(),
1156 policy.clone(),
1157 protocol_factory.clone(),
1158 Arc::new(plugin_factories.clone()),
1159 self.child_store_factory
1160 .as_ref()
1161 .or(self.store_factory.as_ref()),
1162 Arc::clone(&live_replay_store),
1163 process_lifecycle_available,
1164 );
1165 let work_driver = InlineWorkDriverSetup {
1166 process: process_work_driver,
1167 queued: queued_work_driver,
1168 };
1169
1170 Ok(LashCore {
1171 env,
1172 policy,
1173 store_factory: self.store_factory,
1174 plugin_factories: Arc::new(plugin_factories),
1175 provider: self.provider,
1176 live_replay_store,
1177 protocol_factory,
1178 process_lifecycle_available,
1179 work_driver: Arc::new(InlineWorkDriverSlot::new(work_driver)),
1180 })
1181 }
1182
1183 #[allow(clippy::too_many_arguments)]
1194 fn resolve_process_work_driver(
1195 process_work_source: &ProcessWorkSource,
1196 worker_plugin_host: &PluginHost,
1197 core: &RuntimeHostConfig,
1198 process_lifecycle_available: bool,
1199 store_factory: Option<&Arc<dyn SessionStoreFactory>>,
1200 policy: &SessionPolicy,
1201 residency: lash_core::Residency,
1202 trigger_store: Option<&Arc<dyn lash_core::TriggerStore>>,
1203 ) -> Result<ProcessWorkDriverSetup> {
1204 let (process_registry, process_change_hub) = match process_work_source {
1205 ProcessWorkSource::None => return Ok(ProcessWorkDriverSetup::None),
1206 ProcessWorkSource::External(driver) => {
1207 return Ok(ProcessWorkDriverSetup::External {
1208 driver: driver.clone(),
1209 });
1210 }
1211 ProcessWorkSource::Inline { registry, hub } => (Arc::clone(registry), hub.clone()),
1212 };
1213 let Some(store_factory) = store_factory else {
1217 return Err(EmbedError::ProcessRegistryRequiresStoreFactory);
1218 };
1219 let runtime_host = worker_plugin_host
1224 .install_process_engine_contributions(core.clone(), process_lifecycle_available)?;
1225 let phase_probe_slot = lash_core::runtime::RuntimeTurnPhaseProbeSlot::default();
1226 let mut config = DurableProcessWorkerConfig::new(
1227 Arc::new(worker_plugin_host.clone()),
1228 runtime_host,
1229 Arc::clone(store_factory),
1230 process_registry,
1231 )
1232 .with_session_policy(policy.clone())
1233 .with_trigger_store(trigger_store.cloned().unwrap_or_else(|| {
1234 Arc::new(lash_core::InMemoryTriggerStore::with_clock(Arc::clone(
1235 &core.clock,
1236 )))
1237 }))
1238 .with_residency(residency)
1239 .with_turn_phase_probe_slot(phase_probe_slot);
1240 if let Some(hub) = process_change_hub {
1241 config = config.with_change_hub(hub);
1242 }
1243 let config = Box::new(config);
1244 Ok(ProcessWorkDriverSetup::LazyDefault { config })
1245 }
1246
1247 #[allow(clippy::too_many_arguments)]
1248 fn resolve_queued_work_driver(
1249 queued_work_source: &QueuedWorkSource,
1250 env: RuntimeEnvironment,
1251 policy: SessionPolicy,
1252 protocol_factory: Option<Arc<dyn PluginFactory>>,
1253 plugin_factories: Arc<Vec<Arc<dyn PluginFactory>>>,
1254 store_factory: Option<&Arc<dyn SessionStoreFactory>>,
1255 live_replay_store: Arc<dyn LiveReplayStore>,
1256 process_lifecycle_available: bool,
1257 ) -> QueuedWorkDriverSetup {
1258 match queued_work_source {
1259 QueuedWorkSource::None => QueuedWorkDriverSetup::None,
1260 QueuedWorkSource::External(driver) => QueuedWorkDriverSetup::External {
1261 driver: driver.clone(),
1262 },
1263 QueuedWorkSource::LazyDefault => match store_factory {
1264 Some(store_factory) => QueuedWorkDriverSetup::LazyDefault {
1265 config: Arc::new(InlineQueuedWorkRunConfig::new(
1266 env,
1267 policy,
1268 protocol_factory,
1269 plugin_factories,
1270 Arc::clone(store_factory),
1271 live_replay_store,
1272 process_lifecycle_available,
1273 )),
1274 },
1275 None => QueuedWorkDriverSetup::None,
1276 },
1277 }
1278 }
1279
1280 pub fn advanced(self) -> AdvancedLashCoreBuilder {
1281 AdvancedLashCoreBuilder { builder: self }
1282 }
1283
1284 pub fn process_registry(mut self, process_registry: Arc<dyn ProcessRegistry>) -> Self {
1285 self.process_work_source = ProcessWorkSource::Inline {
1286 registry: process_registry,
1287 hub: None,
1288 };
1289 self
1290 }
1291
1292 pub fn process_event_sink(mut self, sink: Arc<dyn lash_core::ProcessEventSink>) -> Self {
1307 self.process_event_sink = Some(sink);
1308 self
1309 }
1310
1311 pub fn trigger_store(mut self, store: Arc<dyn lash_core::TriggerStore>) -> Self {
1312 self.trigger_store = Some(store);
1313 self
1314 }
1315
1316 pub fn process_work_driver(mut self, driver: ProcessWorkDriver) -> Self {
1323 self.process_work_source = ProcessWorkSource::External(driver);
1324 self
1325 }
1326
1327 pub fn queued_work_driver(mut self, driver: QueuedWorkDriver) -> Self {
1329 self.queued_work_source = QueuedWorkSource::External(driver);
1330 self
1331 }
1332
1333 pub fn disable_queued_work_driver(mut self) -> Self {
1334 self.queued_work_source = QueuedWorkSource::None;
1335 self
1336 }
1337}
1338
1339pub(crate) fn build_plugin_host(
1340 protocol_factory: Option<&Arc<dyn PluginFactory>>,
1341 common_factories: &[Arc<dyn PluginFactory>],
1342 extra_factories: Vec<Arc<dyn PluginFactory>>,
1343) -> Result<PluginHost> {
1344 let mut factories = Vec::with_capacity(
1345 usize::from(protocol_factory.is_some()) + common_factories.len() + extra_factories.len(),
1346 );
1347 if let Some(protocol_factory) = protocol_factory {
1348 factories.push(Arc::clone(protocol_factory));
1349 }
1350 factories.extend(common_factories.iter().cloned());
1351 factories.extend(extra_factories);
1352 Ok(PluginHost::new(factories))
1353}
1354
1355impl PromptLayerSink for LashCoreBuilder {
1356 fn prompt_layer_mut(&mut self) -> &mut PromptLayer {
1357 self.prompt.get_or_insert_with(PromptLayer::new)
1358 }
1359}
1360
1361pub struct AdvancedLashCoreBuilder {
1362 builder: LashCoreBuilder,
1363}
1364
1365impl AdvancedLashCoreBuilder {
1366 pub fn runtime_host_config(mut self, core: lash_core::RuntimeHostConfig) -> Self {
1367 self.builder.runtime_host_config = Some(core);
1368 self
1369 }
1370
1371 pub fn plugin_host(mut self, plugin_host: PluginHost) -> Self {
1372 self.builder.plugin_host = Some(plugin_host);
1373 self
1374 }
1375
1376 pub fn build(self) -> Result<LashCore> {
1377 self.builder.build()
1378 }
1379}