lash/advanced

use std::sync::Arc;

let factory = lash::rlm::RlmProtocolPluginFactory::new(
    lash::rlm::RlmProtocolPluginConfig::builder()
        .instruction_limit(lash::rlm::InstructionBound::instructions(1_000_000))
        .wall_clock(lash::rlm::WallClockBound::secs(30))
        .memory_limit(lash::rlm::MemoryBound::mebibytes(64))
        .build(),
    Arc::new(lash::persistence::InMemoryLashlangArtifactStore::new()),
);
let core = lash::LashCore::rlm_builder(lash::TurnBudget::Unbounded, factory)
    .provider(provider)
    .model(model)
    .effect_host(Arc::new(lash::durability::InlineEffectHost::default()))
    .attachment_store(Arc::new(lash::persistence::InMemoryAttachmentStore::new()))
    .process_env_store(Arc::new(
        lash::persistence::InMemoryProcessExecutionEnvStore::new(),
    ))
    // Start bounded; tune both limits for your backend's latency envelope.
    .commit_budget(lash::CommitBudget::bounded(1024 * 1024, 512))
    .queued_work_batching(lash::QueuedWorkBatchingConfig::new(1024))
    .build(crate::example_process_owner())?;

The supplied owner is the host identity: stable owner_id, fresh per-boot incarnation_id. Lash mints a fresh executor_id for every runtime open. Lease reentry requires all three values to match, so resuming the same parked open reenters but a second concurrent open under the same core observes Busy and cannot revoke the first executor's fence.

use std::sync::Arc;

use lash::persistence::FileAttachmentStore;
use lash_sqlite_store::{SqliteSessionStoreFactory, Store};

let store_factory = Arc::new(SqliteSessionStoreFactory::new(data_dir.join("sessions")));
let artifact_store = Arc::new(Store::open(&data_dir.join("artifacts.db")).await?);

let factory = lash::rlm::RlmProtocolPluginFactory::new(
    lash::rlm::RlmProtocolPluginConfig::builder()
        .instruction_limit(lash::rlm::InstructionBound::instructions(1_000_000))
        .wall_clock(lash::rlm::WallClockBound::secs(30))
        .memory_limit(lash::rlm::MemoryBound::mebibytes(64))
        .build(),
    artifact_store.clone(),
);
let core = lash::LashCore::rlm_builder(lash::TurnBudget::Unbounded, factory)
    .provider(provider)
    .model(model)
    .store_factory(store_factory)
    .effect_host(Arc::new(lash::durability::InlineEffectHost::default()))
    .attachment_store(Arc::new(FileAttachmentStore::new(
        data_dir.join("attachments"),
    )))
    .process_env_store(artifact_store)
    // Start bounded; tune both limits for your backend's latency envelope.
    .commit_budget(lash::CommitBudget::bounded(1024 * 1024, 512))
    .queued_work_batching(lash::QueuedWorkBatchingConfig::new(1024))
    .build(crate::example_process_owner())?;

Sessions park automatically when the handle drops. Long-running servers can hold thousands of session ids cheaply. Reopening a session loads its one durable leaf-to-root history.

Plugin background work (host-managed process handles, post-turn observation summaries, MCP warm-ups) continues past turn return. To drain it before returning (one-shot processes that need observations persisted before exit):

session.refresh_background_graph().await?;

Exposed by the CLI as --await-background-work on lash --print. Embedders rarely need it outside short-lived processes.

let factory = lash::rlm::RlmProtocolPluginFactory::new(
    lash::rlm::RlmProtocolPluginConfig::builder()
        .instruction_limit(lash::rlm::InstructionBound::instructions(1_000_000))
        .wall_clock(lash::rlm::WallClockBound::secs(30))
        .memory_limit(lash::rlm::MemoryBound::mebibytes(64))
        .build(),
    artifact_store,
);
let core = lash::LashCore::rlm_builder(lash::TurnBudget::Unbounded, factory)
    .provider(provider)
    .model(model)
    .store_factory(store_factory)
    .process_registry(process_registry)
    .effect_host(Arc::new(lash::durability::InlineEffectHost::default()))
    .attachment_store(attachment_store)
    .process_env_store(process_env_store)
    // Start bounded; tune both limits for your backend's latency envelope.
    .commit_budget(lash::CommitBudget::bounded(1024 * 1024, 512))
    .queued_work_batching(lash::QueuedWorkBatchingConfig::new(1024))
    .build(crate::example_process_owner())?;

use std::collections::BTreeMap;

let mut resources = lashlang::LashlangHostCatalog::new();
resources.add_trigger_source_constructor(
    ["app", "button"],
    lashlang::TypeExpr::Object(Vec::new()),
    lashlang::NamedDataType::object(
        "app.ButtonPressed",
        vec![lashlang::TypeField {
            name: "color".into(),
            ty: lashlang::TypeExpr::Str,
            optional: false,
        }],
    )?,
)?;

let environment = lashlang::LashlangHostEnvironment {
    resources,
    abilities: lashlang::LashlangAbilities::default().with_processes(),
    ..lashlang::LashlangHostEnvironment::default()
};

let compiled = lashlang::compile_module(lashlang::ModuleCompileRequest {
    source: r#"
process on_button(event: app.ButtonPressed) {
  finish event.color
}
source = app.button({})
finish source
"#,
    environment: &environment,
    artifact_store: Some(artifact_store.as_ref()),
})
.await?;

let process = compiled
    .introspection
    .exported_processes
    .iter()
    .find(|process| process.definition.process_name == "on_button")
    .expect("compiled module exports on_button");

let inputs = lashlang::TriggerInputTemplate::new(BTreeMap::from([(
    "event".to_string(),
    lashlang::TriggerInputBinding::Event,
)]));
let compatibility =
    lashlang::check_trigger_compatibility(lashlang::TriggerCompatibilityRequest {
        artifact: &compiled.artifact,
        definition: &process.definition,
        source_type: "app.button",
        inputs: &inputs,
    })?;

println!(
    "compiled {} and trigger emits {}",
    compiled.module_ref,
    compatibility.event_type.name()
);

Shape: the runtime driver polls TurnMachine. For each reply-producing turn effect it builds a RuntimeEffectEnvelope with a typed RuntimeInvocation, invokes the active ScopedEffectController, and feeds the returned outcome back as the matching response. The InlineEffectHost runs locally and is not in-flight durable. A durable workflow EffectHost such as Restate records effects in workflow history, uses durable timers for sleeps, and reruns the handler with the same turn id before Lash retries the final idempotent commit.

Do not wrap a durable handler turn in another persistent submitted/running work-item state machine. The workflow key and turn_id are the durable in-flight identity; Lash's session execution lease, TurnWorkDriver keyed promises, and idempotent commit are the runtime correctness boundary; host tables should store product rows, reconnect outbox rows, exact turn addresses, or terminal output.

Full contract: Architecture → Durability. Embedders without workflow orchestration use the regular session-store path.

use std::sync::Arc;

use lash::{SessionSpec, plugins::PluginFactory};
use lash_subagents::{SubagentsPluginFactory, default_registry};

let registry = Arc::new(default_registry(&tier_models));

let child_spec = SessionSpec::inherit().turn_budget(lash::TurnBudget::bounded(8));
let _configured_turn_budget = child_spec.turn_budget;
let subagents = SubagentsPluginFactory::new(registry).with_session_spec(child_spec);

let factory = lash::rlm::RlmProtocolPluginFactory::new(
    lash::rlm::RlmProtocolPluginConfig::builder()
        .instruction_limit(lash::rlm::InstructionBound::instructions(1_000_000))
        .wall_clock(lash::rlm::WallClockBound::secs(30))
        .memory_limit(lash::rlm::MemoryBound::mebibytes(64))
        .build(),
    Arc::new(lash::persistence::InMemoryLashlangArtifactStore::new()),
);
let core = lash::LashCore::rlm_builder(lash::TurnBudget::Unbounded, factory)
    .provider(provider)
    .model(
        lash::ModelSpec::builder(model.clone())
            .context_window_tokens(200_000)
            .build()
            .expect("valid model metadata"),
    )
    .effect_host(Arc::new(lash::durability::InlineEffectHost::default()))
    .attachment_store(Arc::new(lash::persistence::InMemoryAttachmentStore::new()))
    .process_env_store(Arc::new(
        lash::persistence::InMemoryProcessExecutionEnvStore::new(),
    ))
    // Start bounded; tune both limits for your backend's latency envelope.
    .commit_budget(lash::CommitBudget::bounded(1024 * 1024, 512))
    .queued_work_batching(lash::QueuedWorkBatchingConfig::new(1024))
    .plugin(Arc::new(subagents) as Arc<dyn PluginFactory>)
    .build(crate::example_process_owner())?;

Capabilities return SessionSpec overlays. StaticCapability pins exact child authority; TierCapability backs the built-in explore and peer tiers. Do not construct SessionPolicy directly; it's the resolved runtime artifact.

Built-in tiers

Both tiers consult tier_models for an override, then fall back to the parent's model. Pass overrides through default_registry(&tier_models).

Capability resolution

Capability::build_session_request(ctx) sees the live parent SessionPolicy and returns a SessionCreateRequest. The runtime composes the capability's SessionSpec against parent policy to produce effective child config. The spec is what's persisted, so a resumed subagent re-derives the same authority.

Interactive-only tools such as ask and showcase would block a subagent indefinitely; nothing is listening on the other side. Hide them from every child surface with SubagentsPluginFactory::with_hidden_tools(...) or a custom SessionToolAccess; the CLI hides its five interactive tools this way. Recursion is bounded separately: at the maximum spawn depth (5) the runtime hides spawn_agent itself.

Usage attribution

Child sessions emit TurnEvent::ChildUsage on the parent's stream, tagged with source ("subagent", "compaction", "observer") and child session_id. TurnReport.children_usage rolls up per (source, model); total_usage() sums parent + children. Finer splits: fold ChildUsage from the stream directly.

use std::collections::BTreeMap;
use std::time::Duration;

use lash_plugin_mcp::{MCP_PROTOCOL_VERSION, McpPluginFactory, McpServerConfig};

println!("Lash MCP handshake protocol: {MCP_PROTOCOL_VERSION}");

let mut servers = BTreeMap::new();
servers.insert(
    "docs".to_string(),
    McpServerConfig::stdio("uvx", vec!["mcp-server-docs".into()])
        .with_env([("DOCS_INDEX", "/srv/docs")]),
);
servers.insert(
    "web".to_string(),
    // Headers are static across reconnects: Lash does not perform OAuth
    // or token refresh, so the host must rotate credentials itself.
    McpServerConfig::streamable_http("https://mcp.example.com/rpc")
        .with_headers([("Authorization", "Bearer host-managed-token")])
        .with_timeouts(
            Duration::from_secs(10),
            Duration::from_secs(60),
            Duration::from_secs(600),
        ),
);

// Capabilities are advertised only for handlers installed here. Lash
// routes server requests; the host still owns model, UI, and root policy.
let mcp = McpPluginFactory::builder(servers)
    .sampling_handler(sampling)
    .elicitation_handler(elicitation)
    .roots_provider(roots)
    .build()
    .await?;

let core = configured_mcp_core(provider, model, mcp)?;
Ok(())
}

fn configured_mcp_core(
provider: ProviderHandle,
model: String,
mcp: McpPluginFactory,
) -> lash::Result<LashCore> {
let factory = lash::rlm::RlmProtocolPluginFactory::new(
    lash::rlm::RlmProtocolPluginConfig::builder()
        .instruction_limit(lash::rlm::InstructionBound::instructions(1_000_000))
        .wall_clock(lash::rlm::WallClockBound::secs(30))
        .memory_limit(lash::rlm::MemoryBound::mebibytes(64))
        .build(),
    std::sync::Arc::new(lash::persistence::InMemoryLashlangArtifactStore::new()),
);
lash::LashCore::rlm_builder(lash::TurnBudget::Unbounded, factory)
    .provider(provider)
    .model(
        lash::ModelSpec::builder(model.clone())
            .context_window_tokens(200_000)
            .build()
            .expect("valid model metadata"),
    )
    .effect_host(std::sync::Arc::new(
        lash::durability::InlineEffectHost::default(),
    ))
    .attachment_store(std::sync::Arc::new(
        lash::persistence::InMemoryAttachmentStore::new(),
    ))
    .process_env_store(std::sync::Arc::new(
        lash::persistence::InMemoryProcessExecutionEnvStore::new(),
    ))
    // Start bounded; tune both limits for your backend's latency envelope.
    .commit_budget(lash::CommitBudget::bounded(1024 * 1024, 512))
    .queued_work_batching(lash::QueuedWorkBatchingConfig::new(1024))
    .plugin(std::sync::Arc::new(mcp))
    .build(crate::example_process_owner())
}

Tools appear as mcp__<server>__<tool> with original schemas preserved. attach_server / detach_server mutate the live pool without rebuilding the core.

Server-to-client requests: host owns policy

MCP 2025-11-25 sampling, elicitation, and roots are available through three dyn-compatible host seams: McpSamplingHandler, McpElicitationHandler, and McpRootsProvider. Each callback receives one request struct whose sealed context identifies the server and exposes a cancellation token; interactive UI and model work must stop when that token fires. Lash validates accepted form content against the server's requested schema before writing it to the wire. Decline and cancel carry no content. URL-capable elicitation handlers also receive the matching notifications/elicitation/complete id.

Lash never picks a model, fabricates user input, or guesses workspace roots. The initialize handshake advertises each capability only when its handler was installed on McpPluginFactory::builder; an elicitation handler declares its exact form and URL modes. Call notify_roots_changed() after a dynamic roots provider changes. Connected servers receive notifications/roots/list_changed; a server that reconnects reads the current value through roots/list.

Sampling and elicitation execute as ordinary host I/O inside the open MCP tool attempt. Handler seams must not emit journaled effects or submit nested Lash commands: the outer attempt is the single durable journal unit. A sampling handler owns its model and billing. Its spend belongs in the host/provider trace only; it is not added to the Lash session usage ledger or TurnReport usage.

Lifecycle and failure isolation

Servers connect once at McpPluginFactory::new(...). Built session plugins clone the shared Arc<McpConnectionPool>, so startup cost is one-per-process, not one-per-chat.

Each pool entry has one lifecycle actor that exclusively owns its running service, stdio child, generation counter, reconnect backoff, and liveness timers. Tool calls do not pass through that actor: they clone the currently published peer and generation from a read-optimized cell. A stale failure can therefore report its generation without disconnecting a newer service. This is per-connection resource plumbing, not a pool-wide scheduler or host drain policy (ADR 0014).

After attach_server, new tools are visible to sessions on their next tool-catalog refresh (start of next turn). detach_server shuts the server down and removes its tools; in-flight calls fail rather than hang.

// Hot-swap a server at runtime.
mcp.attach_server(
    "new-tool".to_string(),
    McpServerConfig::stdio("uvx", vec!["mcp-server-new".into()]),
)
.await?;

mcp.detach_server("old-tool").await?;

// Call this after the host's dynamic roots change.
mcp.notify_roots_changed().await?;

Explicit pool shutdown sends every entry actor Shutdown, then joins all entries concurrently. Shutdown preempts handshake, liveness-probe, and host-request waits, leaving each entry's five-second total deadline to cover a three-second graceful child close, kill, a one-second bounded reap, and scheduling margin. A child can be abandoned if it survives that preemptive kill and reap or if the entry deadline expires mid-reap; the abort branch reports the live active_pid, and its literal PID and reason are recorded in status and tracing. There is deliberately no background waitpid sweep. Ordinary drop remains kill-only and cannot promise reap, so hosts should use the factory/core shutdown lever during an orderly exit.

Failures are isolated: an exited child, a 500ing endpoint, or a dropped SSE stream surfaces as a structured unavailable ToolOutcome on the affected tool; other servers keep working. Both initial construction and attach_server register an unreachable server and retry it in the background. Configuration errors still fail immediately, and server_statuses() exposes connectivity and the latest error.

let artifact_store =
    std::sync::Arc::new(lash_sqlite_store::Store::open(&data_dir.join("artifacts.db")).await?);
let factory = lash::rlm::RlmProtocolPluginFactory::new(
    lash::rlm::RlmProtocolPluginConfig::builder()
        .instruction_limit(lash::rlm::InstructionBound::instructions(1_000_000))
        .wall_clock(lash::rlm::WallClockBound::secs(30))
        .memory_limit(lash::rlm::MemoryBound::mebibytes(64))
        .build(),
    artifact_store.clone(),
);
let core = lash::LashCore::rlm_builder(lash::TurnBudget::Unbounded, factory)
    .provider(provider)
    .model(
        lash::ModelSpec::builder("anthropic/claude-sonnet-4.6")
            .context_window_tokens(200_000)
            .build()
            .expect("valid model metadata"),
    )
    .store_factory(store_factory)
    .effect_host(std::sync::Arc::new(
        lash::durability::InlineEffectHost::default(),
    ))
    .attachment_store(std::sync::Arc::new(
        lash::persistence::FileAttachmentStore::new(data_dir.join("attachments")),
    ))
    .process_env_store(artifact_store)
    // Start bounded; tune both limits for your backend's latency envelope.
    .commit_budget(lash::CommitBudget::bounded(1024 * 1024, 512))
    .queued_work_batching(lash::QueuedWorkBatchingConfig::new(1024))
    .turn_budget(lash::TurnBudget::bounded(50))
    // The outer bound on one turn's work; the inner bound is how long a
    // turn may fail to do any. This one defaults to bounded — name it to
    // change it, or to opt out with `NoProgressBudget::Unbounded`.
    .no_progress_budget(lash::NoProgressBudget::bounded(12))
    .build(crate::example_process_owner())?;

Streaming is semantic: TurnBuilder::stream emits TurnActivity and resolves to a TurnReport. Raw runtime telemetry belongs in tracing, not the app surface.

QueuedWorkBatchingConfig also names how much queued work one wake drains. Lash decides which rows may share a turn — merge key, delivery boundary, authority, kind, row and age bounds — and the host decides how many of that strictly FIFO prefix actually drain. The default is DrainMode::OneAtATime: one queued row per drain, no token arithmetic. with_drain_mode(DrainMode::All) drains every compatible pending row for large-window hosts that want throughput, and with_drain_policy(...) takes a custom QueuedDrainPolicy for anything else; the policy sees each row's conservative projection and the window budget. A single row that renders larger than the whole context window can never drain, so it is refused with a typed SelectedQueuedWorkDrainRefusalCause::QueuedItemExceedsContextWindow naming the row and the window it needs, and stays queued for host recovery. The policy sizes automatic drains only: a selection your host names explicitly (queued_turn().batch_ids(...)) already states the composition, so it drains exactly the rows you asked for.

OPENROUTER_API_KEY=... cargo run -p agent-service
# then open http://127.0.0.1:3000

Source: examples/agent-service. Walkthrough: Agent Service.