Skip to main content

lash/
core.rs

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    /// Whether this deployment has process lifecycle available (a process
18    /// registry is wired). Threaded to every plugin host so core can install the
19    /// same plugin-contributed process engines when it rebuilds a runtime.
20    pub(crate) process_lifecycle_available: bool,
21    /// Shared resolution of host-owned work drivers. Shared across `LashCore`
22    /// clones so inline process and queued drivers are constructed at most once.
23    pub(crate) work_driver: Arc<InlineWorkDriverSlot>,
24}
25
26/// How a [`LashCore`] resolves its process work driver, decided at `build()`
27/// and shared across clones.
28pub(crate) enum ProcessWorkDriverSetup {
29    /// No process registry is wired; there is nothing to run.
30    None,
31    /// Lazily construct the default inline process driver on first
32    /// `session().open()`. A store factory is required to build the config (the
33    /// worker rebuilds a session runtime per process); a registry with no store
34    /// factory is rejected at build with
35    /// [`EmbedError::ProcessRegistryRequiresStoreFactory`].
36    LazyDefault {
37        config: Box<DurableProcessWorkerConfig>,
38    },
39    /// The host wired an external driver.
40    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            // An external driver was wrapped by its host, which installs any
80            // sink through the driver constructor; the inline sink does not
81            // apply here. Already-watched inline sources keep their wrap.
82            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
117/// Shared, lazily-initialized host-work state for a [`LashCore`].
118///
119/// The once-guard ([`tokio::sync::OnceCell`]) constructs inline drivers exactly
120/// once across `LashCore` clones, on the first `session().open()` or admin path
121/// that needs them.
122pub(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    /// Resolve host work drivers for a session host. Idempotent: the once-guard
144    /// ensures inline drivers are constructed once.
145    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    /// Sugar entry point: a [`LashCoreBuilder`] pre-seeded with the standard
331    /// protocol plugin and the default runtime plugin stack.
332    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    /// Sugar entry point: a [`LashCoreBuilder`] pre-seeded with a
341    /// host-configured RLM protocol factory and the default runtime plugin
342    /// stack.
343    ///
344    /// The host configures the factory (projection resolver, deferred tool
345    /// resolver, execution sink/jsonl path, and — required at construction — the
346    /// Lashlang artifact store) before passing it in.
347    #[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    /// Rebuild a live session from a [`ParkedSession`](crate::ParkedSession)
370    /// handle produced by [`LashSession::park`](crate::LashSession::park).
371    ///
372    /// Resume reloads the flushed state from the parked store (honoring this
373    /// core's residency), reinstalls this core's plugin configuration and work
374    /// drivers, and returns a ready [`LashSession`]. The parked store instance
375    /// is reused directly, so the transcript the session flushed at park time is
376    /// visible again after resume.
377    ///
378    /// This restores the core-level plugin stack. Session-specific plugins added
379    /// per open via [`SessionBuilder::plugin`] are not re-applied here; parking
380    /// is the round-trip for the core's own configuration.
381    pub async fn resume(&self, parked: ParkedSession) -> Result<LashSession> {
382        // Build the per-session env exactly like `SessionBuilder::open_resolved`
383        // (minus builder-scoped plugins): a fresh plugin host with this core's
384        // factories, the shared work drivers, and the core provider resolver
385        // already carried on `self.env`.
386        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    /// Flush this core's configured trace sink, if any.
415    ///
416    /// Hosts that hand `lash` a trace sink via
417    /// [`LashCoreBuilder::trace_sink`] already hold their own `Arc` and can
418    /// flush it directly; this is the equivalent lever for hosts that did not
419    /// retain the handle. It flushes the core's copy — for a
420    /// [`JsonlTraceSink`](lash_trace::JsonlTraceSink) that fsyncs the file, and
421    /// for an OTel sink it is a no-op (the host still owns provider flush; see
422    /// the tracing docs). Call it before process exit alongside the host's own
423    /// exporter/provider shutdown.
424    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    /// Exact-turn cooperative control for this deployment's effect host.
448    ///
449    /// The returned driver is independently usable from any session handle.
450    /// Session and turn ids are routing identity, not authorization; authorize
451    /// requests in the host API before forwarding them to Lash.
452    pub fn turn_work_driver(&self) -> lash_core::TurnWorkDriver {
453        lash_core::TurnWorkDriver::new(self.effect_host())
454    }
455
456    /// Persist host input without opening a competing session writer.
457    ///
458    /// This is the downstream-host ingress seam for input submitted while a
459    /// durable turn may already own the session execution lease. Active-turn
460    /// input is claimed by that exact turn at a checkpoint; next-turn input is
461    /// handed to the configured queued-work driver after it is durably stored.
462    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    // `RuntimeHostConfig` has no `Default`: the generic host-owned durability
659    // dependencies must be named. They are collected here and resolved in
660    // `build()`, which errors if any is unset.
661    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    // Benign core overrides applied on top of the resolved core.
666    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    // Advanced full-config override; used as the base core when present.
672    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    // Single source of truth for process lifecycle support and process-work
680    // consumption.
681    process_work_source: ProcessWorkSource,
682    // Optional host-facing best-effort feed of appended process events,
683    // installed on the inline process-registry decorator at build time.
684    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    /// Configure a factory that can create a persistence store for any root
717    /// session opened from this core.
718    ///
719    /// The factory must honor `SessionStoreCreateRequest::session_id` and
720    /// return a store for that specific session. Do not use this to wrap one
721    /// pre-opened root store; pass root-only stores with
722    /// `LashCore::session(...).store(store)` instead.
723    pub fn store_factory(mut self, store_factory: Arc<dyn SessionStoreFactory>) -> Self {
724        self.store_factory = Some(store_factory);
725        self
726    }
727
728    /// Configure the persistence factory used by managed child sessions, such
729    /// as local subagents.
730    ///
731    /// Child factories must return a distinct store bound to the requested
732    /// child session id. Hosts that pass an explicit root store with
733    /// `SessionBuilder::store` should set this when child sessions need
734    /// persistence.
735    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    /// Set the deployment effect host — the durability boundary every operation
754    /// crosses. Pass [`InlineEffectHost`](crate::durability::InlineEffectHost)
755    /// for in-process execution, or a workflow-backed host for durable
756    /// execution.
757    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    /// Configure the lease timing capability for every durable single-writer
813    /// lane this deployment claims: session execution leases, turn-input and
814    /// queued-work claims, and process leases.
815    ///
816    /// This is the failover-latency vs false-takeover-risk knob. Like
817    /// [`residency`](Self::residency) it is an operational deployment decision,
818    /// so it lives on the main builder tier rather than behind
819    /// [`advanced`](Self::advanced). Construct the value with
820    /// [`LeaseTimings::new`](lash_core::LeaseTimings::new), which enforces
821    /// `ttl >= 3 * renew_interval`. Effect hosts accept the same type at
822    /// construction (e.g. SQLite/Postgres effect-replay options), so a host can
823    /// share one timing decision across both boundaries.
824    pub fn lease_timings(mut self, lease_timings: lash_core::LeaseTimings) -> Self {
825        self.lease_timings = Some(lease_timings);
826        self
827    }
828
829    /// Use one host clock for runtime sleeps and embedded-store time.
830    pub fn clock(mut self, clock: Arc<dyn lash_core::Clock>) -> Self {
831        self.clock = Some(clock);
832        self
833    }
834
835    /// Configure the bounded live replay buffer used by session observation
836    /// cursors. This is best-effort reconnect recovery only; durable state
837    /// still comes from the session store and [`SessionReadView`].
838    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    /// Resolve the runtime host config, requiring the generic host-owned
844    /// durability dependencies to have been named.
845    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    /// Apply benign + still-set dependency overrides on top of a base core.
866    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    /// Validate store peer-coherence of the wired durability dependencies.
903    ///
904    /// Durability is established by what the host wired; the per-invocation
905    /// durable controller is not visible here (the build-time controller is
906    /// inline by construction), so this checks the stores against each other
907    /// only — never the controller (see A5 in the durable-first wiring spec):
908    ///
909    /// - a durable session store factory requires a durable attachment and
910    ///   artifact store (they back the same session state);
911    /// - a durable process registry requires a session store factory that is
912    ///   itself durable (the registry's process records are meaningless without
913    ///   a durable session behind them).
914    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    /// Validate store peer-coherence, sweeping every registered process engine.
922    ///
923    /// Runs after the runtime host is resolved and process-engine contributions
924    /// are installed, so it reads the effective effect host / attachment /
925    /// process-env tiers off `core`, and the session-store / trigger-store /
926    /// process-registry tiers captured from the builder before its plugin stack
927    /// was consumed. A durable session store requires every registered engine to
928    /// be durable too.
929    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            // Every registered process engine must be durable behind a durable
956            // session store, regardless of how it was contributed.
957            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        // Capture the store-peer tiers that live on builder fields before the
1046        // plugin stack is consumed below; the engine sweep happens after install.
1047        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        // Whether process lifecycle is available (a process registry is wired).
1083        // Threaded to every plugin host so core installs the same
1084        // plugin-contributed process engines wherever it rebuilds a runtime.
1085        let process_lifecycle_available = process_work_source.has_registry();
1086        // Install onto a throwaway clone purely to sweep every registered
1087        // engine's durability tier for the coherence check. `env.core` stays
1088        // free of plugin-contributed engines so that each runtime-construction
1089        // site (session open, queued-work drain, durable process worker)
1090        // installs them fresh onto a clean registry — keeping the unique-kind
1091        // enforcement in `try_with_engine` a genuine cross-factory check rather
1092        // than a self-collision on a registry that already carries them.
1093        let core_with_engines = default_plugin_host
1094            .install_process_engine_contributions(core.clone(), process_lifecycle_available)?;
1095        // Coherence runs after engines are installed so it can sweep every
1096        // registered engine's durability tier.
1097        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        // Resolve process work before the process source is moved into the
1107        // environment. The default inline driver's config is built
1108        // eagerly so a missing store factory fails loudly at build, not at
1109        // first open. It is built from the same single-protocol plugin host the
1110        // live runtime uses, so the worker can rebuild a runtime for a process.
1111        let process_work_driver = Self::resolve_process_work_driver(
1112            &process_work_source,
1113            &default_plugin_host,
1114            &core,
1115            process_lifecycle_available,
1116            // The worker rebuilds sessions with the same factory `build()` wires
1117            // below: `child_store_factory.or(store_factory)`.
1118            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    /// Decide how a built [`LashCore`] sources its process work driver.
1184    ///
1185    /// - no registry => nothing to run ([`ProcessWorkDriverSetup::None`]);
1186    /// - external driver wired => use it ([`ProcessWorkDriverSetup::External`]);
1187    /// - inline registry wired => lazily construct the default inline driver on first open. Its
1188    ///   [`DurableProcessWorkerConfig`] is built eagerly when a store factory is
1189    ///   present; without one the inline worker cannot rebuild session runtimes.
1190    // Mirrors the sibling `resolve_queued_work_driver`: a builder helper whose
1191    // inputs are the heterogeneous, all-required driver-resolution state and
1192    // have no cohesive sub-grouping.
1193    #[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        // The worker rebuilds a session runtime per process, so it needs a store
1214        // factory; without one the default runner could not execute anything, so
1215        // fail loudly rather than silently leave processes unexecuted.
1216        let Some(store_factory) = store_factory else {
1217            return Err(EmbedError::ProcessRegistryRequiresStoreFactory);
1218        };
1219        // The worker rebuilds with the same plugin host the live runtime uses,
1220        // including the protocol plugin that supplies the protocol session
1221        // capability. Install its plugin-contributed process engines onto a
1222        // clean copy of the base host — `core` deliberately carries none.
1223        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    /// Install a best-effort, host-facing [`ProcessEventSink`] on the inline
1293    /// process registry.
1294    ///
1295    /// Each appended process event is pushed to the sink after its durable
1296    /// write, in per-process append order. This is freshness, not truth: it
1297    /// never buffers or retries, terminal events are not emitted through it
1298    /// (observe completion via the await seam), and consumers reconcile from
1299    /// the durable event log. See [`ProcessEventSink`] for the full contract.
1300    ///
1301    /// Applies to the inline registry path ([`Self::process_registry`]); a host
1302    /// that supplies its own [`ProcessWorkDriver`](lash_core::ProcessWorkDriver)
1303    /// installs the sink through the driver's constructor instead.
1304    ///
1305    /// [`ProcessEventSink`]: lash_core::ProcessEventSink
1306    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    /// Configure an externally owned process work runner.
1317    ///
1318    /// Durable hosts construct a [`ProcessWorkDriver`] from the same process
1319    /// registry and wake handle used by their deployment runner, then pass it
1320    /// here. The driver registry becomes the core's process registry and no
1321    /// inline runner is spawned.
1322    pub fn process_work_driver(mut self, driver: ProcessWorkDriver) -> Self {
1323        self.process_work_source = ProcessWorkSource::External(driver);
1324        self
1325    }
1326
1327    /// Configure an externally owned queued-work driver.
1328    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}