session/persistence

This identity cutover also applies to external durable substrates. Before deploying, drain and recreate Restate LashDurableWaitIndex and LashDurableWaitWorkflow state. The v3 await-event key epoch is stored under the v2 wait-index namespace. Post-cutover index operations and workflows that wake cross the epoch gate and reject pre-cutover state with a recreate instruction. A fully parked pre-cutover workflow cannot run new code and v3 resolutions do not address its v2 key, so it never self-terminates; draining and purging those invocations before deployment is the only remedy.

use std::sync::Arc;

use lash_sqlite_store::{SqliteSessionStoreFactory, Store};

let data_dir = std::path::PathBuf::from("./.lash-data");
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::new(
        lash::rlm::ExecutionBound::instructions(1_000_000),
        lash::rlm::ExecutionBound::secs(30),
    ),
    artifact_store,
);
let core = lash::LashCore::rlm_builder(factory)
    .provider(provider)
    .model(
        lash::ModelSpec::builder(model.clone())
            .context_window_tokens(200_000)
            .build()
            .expect("valid model metadata"),
    )
    .store_factory(store_factory)
    .effect_host(Arc::new(lash::durability::InlineEffectHost::default()))
    .attachment_store(Arc::new(lash::persistence::FileAttachmentStore::new(
        data_dir.join("attachments"),
    )))
    .build()?;

Per-session override:

let session = core
    .session(chat_id)
    .store(Arc::new(my_custom_persistence))
    .open()
    .await?;

The examples use the first-party SQLite adapter because it is the smallest local durable setup. The runtime contract is SessionStoreFactory / RuntimePersistence; a host can provide another database-backed implementation when it preserves the same advisory session execution lease, head-CAS commits, checkpoint/blob references, pending turn-input claims, queued work, and idempotency semantics.

Without .store_factory(...) or .store(...), turns still succeed but state stays in memory. You still provide explicit in-memory effect, artifact, and attachment facets.

use std::sync::Arc;

use lash_postgres_store::PostgresStorage;
use lash_s3_store::S3AttachmentStore;

let storage = PostgresStorage::connect(&database_url).await?;
let attachments = S3AttachmentStore::builder("lash-attachments", "us-east-1")
    .endpoint_url("http://localhost:9000") // omit for AWS S3
    .access_key_id("minioadmin")
    .secret_access_key("minioadmin")
    .path_style(true)
    .prefix("prod/lash")
    .build()?;

let factory = lash::rlm::RlmProtocolPluginFactory::new(
    lash::rlm::RlmProtocolPluginConfig::new(
        lash::rlm::ExecutionBound::instructions(1_000_000),
        lash::rlm::ExecutionBound::secs(30),
    ),
    Arc::new(storage.lashlang_artifact_store()),
);
let core = lash::LashCore::rlm_builder(factory)
    .store_factory(Arc::new(storage.session_store_factory()))
    .process_registry(Arc::new(storage.process_registry()))
    .trigger_store(Arc::new(storage.trigger_store()))
    .attachment_store(Arc::new(attachments))
    // provider, model, effect host...
    .build()?;

PostgresStorage::connect creates the exact supported schema on an empty database and rejects mismatched component versions. Runtime checkpoints, pending turn inputs, queued work, process rows, triggers, attachment manifests, and Lashlang artifacts live in Postgres. Attachment bytes live in S3-compatible object storage under content-addressed keys; MinIO uses the same implementation with an endpoint URL and path-style mode.

PostgresStoreConfig applies connection-level backstops by default: a 10 second lock_timeout and a 30 second statement_timeout. Mutating session work first claims the advisory session execution lease; session commits perform the head-revision CAS and any claim-ownership checks in one transaction. Postgres serialization failures, deadlocks, and lock-acquisition timeouts on the backstop write surface as retryable conflicts so hosts can reload and retry instead of treating ordinary write contention as an opaque backend failure.

The distributed integration harness is just restate-postgres-workers-e2e. It starts Postgres, MinIO, Restate, a deterministic OpenAI-compatible mock provider, two worker processes behind a Caddy h2c proxy, and a runner. The workers build a normal downstream-style LashCore with PostgresStorage, S3AttachmentStore, PostgresLashlangArtifactStore, PostgresTriggerStore, PostgresProcessRegistry, RestateRuntimeEffectController, RestateProcessDeployment, and JSONL trace sinks.

The scenario runs real session.turn(input).stream_to(...) turns through the public facade. It covers foreground tools, attachment creation through the configured attachment store, parent/nested/parallel Lashlang processes, durable sleep and wake paths, public trigger registration and delivery, live replay from a stored observation cursor, MinIO byte/metadata assertions, and failover where worker A exits from the real turn path after terminal finish or another durable effect and worker B completes through Restate replay. The runner asserts one provider call per journaled effect, one committed turn, no duplicate runtime rows, and no active Lash Restate invocations left in pending, ready, running, backing-off, or suspended.

A live session tip is forkable through its head checkpoint at no extra retention cost. To preserve a turn after the source advances, pin it while it is still the live tip. fork_points() enumerates pinned past turns and current live tips; unpin() releases only the explicit pin.

let snapshot = session.admin().state().export().await;
let turn_node_id = snapshot
    .session_graph
    .leaf_node_id
    .clone()
    .expect("completed turn");

core.pin(&turn_node_id).await?;
// The source may now run more turns.
core.fork_at(&turn_node_id, branch_chat_id).await?;
core.unpin(&turn_node_id).await?;

The fork inherits the retained checkpoint and AgentFrame lineage. It starts with an empty usage ledger and no execution-scoped queue, lease, receipt, or wake-subscription state. Its explicit observer selector (All, None, or Only) and chosen process ids are recorded in the durable fork relation; session open idempotently reconciles an unfinished observer intent after a crash. Wake delivery remains addressed only to the subscribed source session.

History ownership is reachability, not the session id that first wrote a node. Child edges, live session heads, and explicit pins are counted references. Deleting one sibling stops reclamation at the first shared node still referenced by another head or pin.

A host-level rewind is explicit composition: pin the target, fork there under a new id, switch the product UI to that id, then delete the superseded session. The superseded host-facing id is permanently retired and cannot be reopened in that store. The agent-service example also snapshots its app-owned messages and board when the user pins a turn; Lash does not copy host tables.

let effect_host = core.effect_host();
let scope = effect_host.scoped(core.session_delete_scope(chat_id).await?)?;
let report = core.delete_session(chat_id, scope).await?;

if let Some(process_report) = report.process {
    audit_process_cleanup(process_report)?;
}

Deletion requires a core-level SessionStoreFactory; per-session stores cannot be rediscovered from a bare id. If a ProcessRegistry is installed, Lash removes that session's weak observer edges, clears subscriptions addressed to it, discards its pending wake deliveries, and deletes its trigger subscriptions. Process records and event logs remain byte-identical; deletion never cancels a process.

A materialized host-facing session id is single-use: deletion writes a permanent tombstone, and later create, fork, queue, attachment, lease, metadata, or commit writes fail with StoreError::SessionDeleted. Deleting an id that was never materialized is a no-op. A reset must rotate to a new id. Lash-minted runtime-internal process session ids are reclaimed during process pruning without tombstones.

The tombstone, await-event revocation ledger, effect journal, and Restate state form one trust domain and must be reset together. SQLite deployments must not wipe the catalog and effect databases independently. The v3 await-event promise-key cutover also requires draining and purging in-flight Restate invocations and recreating Restate wait-index/workflow state before upgrade. Operations that execute after deployment reject the old epoch, but a fully parked v2 invocation executes no new guard and cannot be reached by a v3 resolution; it remains parked until the operator purges it.

Writes are the sharper edge. A hand-written UPDATE lands outside the revision fence and the idempotency receipt the store commits in the same transaction, so it can leave trigger, process, or session state that no retry, replay, or receipt read can reconstruct. Lash has no repair path for state changed that way, and the damage surfaces later rather than at the write.

Every question worth asking a lash_* table has a supported answer:

Re-enabling a disabled trigger subscription is the case hosts most often reach for SQL to solve. It is a first-class fenced verb: read the live revision, then Enable at that revision under an idempotent operation id.

use lash::triggers::{TriggerCommand, TriggerCommandOutcome, TriggerSubscriptionFilter};

// Read the live revision through the command surface, not with SQL.
let TriggerCommandOutcome::List { records } = store
    .execute_command(
        "reenable-read",
        TriggerCommand::List {
            owner_scope: owner_scope.clone(),
            filter: TriggerSubscriptionFilter {
                subscription_key: Some(subscription_key.to_string()),
                ..TriggerSubscriptionFilter::default()
            },
        },
    )
    .await??
else {
    unreachable!("List returns list records")
};
let Some(record) = records.first() else {
    anyhow::bail!("no live subscription for `{subscription_key}`");
};

// Enable is fenced on the revision just read, and the operation id makes the
// retry idempotent.
let outcome = store
    .execute_command(
        &format!("reenable:{subscription_key}:{}", record.revision),
        TriggerCommand::Enable {
            owner_scope,
            actor,
            subscription_key: subscription_key.to_string(),
            expected_revision: record.revision,
        },
    )
    .await?;
match outcome {
    Ok(TriggerCommandOutcome::Mutation { receipt }) => {
        assert!(receipt.enabled);
    }
    Ok(_) => unreachable!("Enable returns one mutation receipt"),
    // A concurrent writer moved the row first: re-read and retry. Never
    // patch the row by hand.
    Err(conflict) => retry_reenable_later(conflict)?,
}

A stale expected_revision returns a typed conflict carrying the current revision, so the correct response is re-read and retry. Hosts upgrading away from hand-written trigger SQL will find the operation-by-operation mapping under retiring host SQL.

Concurrent turns from two workers on the same session id normally resolve at the lease boundary. A busy session returns session_execution_busy before the turn starts; a runner may return session_execution_lease_lost when an operation observes handoff. Lease loss alone does not veto the turn driver's current-head commit, but it does veto a nested plugin or observer commit borrowing that held lane: the borrowed fence is checked before receipt replay or mutation. If the head CAS backstop fires, the loser receives a turn-level error such as store_commit_failed; these are not TurnStop variants.

Durable workflow hosts that already serialize one logical invocation can open the session with SessionBuilder::session_execution_owner(LeaseOwnerIdentity::opaque(...)). Reentry requires the same owner id and the same incarnation id; a retry that represents a new process incarnation must claim or reclaim through the fenced lease path instead of clearing the old owner. Local CLI runtimes attach process liveness metadata so a crashed same-host holder can be reclaimed quickly, while opaque or cross-host holders fall back to the TTL backstop. Composing owner identity, lease timings, and drain into a failover policy is covered in running in production.

use lash::runtime::RuntimeErrorCode;

match session.turn(input).run().await {
    Ok(turn) => persist(turn)?,
    Err(lash::EmbedError::Runtime(err))
        if err.code == RuntimeErrorCode::SessionExecutionBusy =>
    {
        retry_later(err)?;
    }
    Err(lash::EmbedError::Runtime(err))
        if err.code == RuntimeErrorCode::SessionExecutionLeaseLost =>
    {
        // The durable lane moved to another owner before commit: reopen and retry.
        let session = core.session(chat_id).open().await?;
        retry_or_report(err, session)?;
    }
    Err(lash::EmbedError::Runtime(err))
        if err.code == RuntimeErrorCode::StoreCommitContended =>
    {
        // The failed commit published nothing: retry the same operation unchanged.
        retry_later(err)?;
    }
    Err(lash::EmbedError::Runtime(err)) if err.code == RuntimeErrorCode::StoreCommitFailed => {
        // The CAS backstop fired: reload and retry.
        let session = core.session(chat_id).open().await?;
        retry_or_report(err, session)?;
    }
    Err(other) => bail!(other),
}

Hosts expecting concurrent access should serialize at their layer or re-open the session after conflict and decide how to merge intent.

The process registry is a separate store with its own retention lever. After the host advances the mandatory projection watermark, core.processes().prune(cutoff_epoch_ms, ProjectionWatermark::UpTo(cursor)) replaces eligible terminal rows with payload-free tombstones, removes their events, wakes, observer edges, and leases, and reconciles the exact trigger-delivery rows those tombstones guard. A deployment without a projector must choose ProjectionWatermark::NoProjector explicitly. The deletion change feed lets host projections drop their rows; non-terminal rows are never touched. Run core.processes().compact_tombstones(cutoff_epoch_ms, watermark) on the same maintenance cadence. It refuses to compact tombstones still referenced by outstanding trigger deliveries; if a configured trigger store is unhealthy, compaction stops and retains tombstones until it recovers. Keep the retention window comfortably longer than any in-flight ProcessWorkDriver::await_terminal. A late await receives the typed ProcessNoLongerRetained information outcome. For Restate process and turn attachments, an attach-ceiling-elapsed error ends only that transport connection: the durable wait remains live, so hosts re-attach with the same process id or turn address rather than reporting terminal failure. See docs/adr/0017-process-observation-is-best-effort-push-over-state-truth.md and ADR 0021.

// One factory at boot, shared across every chat.
let store_factory = Arc::new(SqliteSessionStoreFactory::new(
    data_dir.join("lash-sessions"),
));
let artifact_store =
    Arc::new(lash_sqlite_store::Store::open(&data_dir.join("lash-artifacts.db")).await?);

let factory = lash::rlm::RlmProtocolPluginFactory::new(
    lash::rlm::RlmProtocolPluginConfig::new(
        lash::rlm::ExecutionBound::instructions(1_000_000),
        lash::rlm::ExecutionBound::secs(30),
    ),
    artifact_store,
);
let core = lash::LashCore::rlm_builder(factory)
    .provider(provider)
    .model(
        lash::ModelSpec::builder(model.clone())
            .variant(lash::provider::ReasoningSelection::Effort(
                model_variant.clone(),
            ))
            .context_window_tokens(200_000)
            .build()
            .expect("valid model metadata")
            .with_capability(adaptive_reasoning_capability()),
    )
    .store_factory(store_factory)
    .effect_host(Arc::new(lash::durability::InlineEffectHost::default()))
    .attachment_store(Arc::new(lash::persistence::FileAttachmentStore::new(
        data_dir.join("attachments"),
    )))
    .build()?;

// Per request: open a session keyed by the app's chat id.
let session = core.session(chat_id).open().await?;

Full source: examples/agent-service.