tracing/observability

use std::sync::Arc;

use lash::{
    LashCore,
    tracing::{JsonlTraceSink, TraceLevel, TraceSink},
};

let trace_sink: Arc<dyn TraceSink> = Arc::new(JsonlTraceSink::new("./.lash-data/trace.jsonl"));

let core = lash::LashCore::standard_builder(lash::TurnBudget::Unbounded)
    .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))
    .trace_sink(trace_sink)
    .trace_level(TraceLevel::Extended)
    .build(crate::example_process_owner())?;

TraceLevel::Standard (default): turns, tool calls, LLM start/complete, token usage. TraceLevel::Extended adds provider stream chunks, prompt-component hashes, and runtime stream deltas. Extended is verbose; use for provider debugging.

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()),
)
.with_lashlang_execution_jsonl_path("./.lash-data/lashlang-execution.jsonl");
let core = lash::LashCore::rlm_builder(lash::TurnBudget::Unbounded, factory)
    .provider(provider)
    .model(model)
    .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))
    .build(crate::example_process_owner())?;

use std::sync::Arc;

use lash::tracing::{JsonlTraceSink, TeeTraceSink, TraceLashlangGraphStore, TraceSink};

let lashlang_graphs = Arc::new(TraceLashlangGraphStore::default());
let lashlang_execution_sink = Arc::new(TeeTraceSink::new([
    Arc::clone(&lashlang_graphs) as Arc<dyn TraceSink>,
    Arc::new(JsonlTraceSink::new("./.lash-data/lashlang-execution.jsonl"))
        as Arc<dyn TraceSink>,
]));

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()),
)
.with_lashlang_execution_sink(lashlang_execution_sink);
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())?;

let graph = lashlang_graphs.graph("process:process-id");

Tracking records use TraceEvent::LanguageExecution with typed payloads for execution start/finish, node start/complete/fail, branch selection, and child execution links. Producers stamp the language field as "lashlang". Every payload includes a deterministic event_key so TraceLashlangGraphStore dedupes replayed records. Graph keys are derived from runtime identity, such as effect:session:turn:effect-id for foreground code and process:process-id for durable processes. Graph snapshots are derived from trace events and safe for host UIs, dashboards, tests, and debugging; they are not canonical process state.

The v3→v4 rename also changed the OpenTelemetry span and attribute keys from lash.lashlang_execution* to lash.language_execution* and added lash.language_execution.language; update dashboards and alerts that key off those names.

When the linked Lashlang artifact contains static @label annotations, ExecutionStarted.execution_map.nodes carries each node's optional label_metadata with title and optional description. Lifecycle events continue to reference node_id only, and TraceLashlangGraphStore reduces the metadata from the initial execution map into graph nodes for host renderers.

Process wakes carry typed provenance through MessageOrigin::Process.caused_by. Hosts can inspect that semantic metadata to relate a wake back to a trigger occurrence, session node, or other runtime cause; rendering remains the host's responsibility.

Static map

ExecutionStarted carries TraceLanguageExecutionMap: graph identity, nodes, edges, node kinds, generated labels, and optional @label metadata. TraceLashlangGraphStore seeds those nodes as unobserved and branch edges as unknown. This is the data a UI uses to draw the full skeleton up front.

Dynamic updates

NodeStarted, NodeCompleted, and NodeFailed update one node by node_id, set timestamps, duration, occurrence, and latest error. BranchSelected marks the chosen then or else edge as selected, marks the sibling branch edge as rejected, and completes the selected branch-arm node. ChildStarted links a parent node to another graph key, such as a started process graph. ExecutionFinished sets the graph-level status.

@label

@label(title: "...", description: "...") is static authoring metadata only. It travels in the execution map as label_metadata, becomes part of the module artifact identity, and never appears as a runtime command or dynamic update. A renderer should display the title/description from the static node and overlay the live status from dynamic events.

process review_one(path: str) {
  @label(title: "Review file")
  report = await agents.default.spawn({
    task: format("Review {}", path),
    capability: "explore"
  })?
  finish report
}

@label(title: "Find candidates")
paths = await workspace.default.glob({ pattern: "src/**/*.rs" })?

@label(title: "Choose path")
if empty(paths) {
  @label(title: "Submit empty result")
  finish { reviewed: 0 }
} else {
  @label(title: "Start child review")
  child = start review_one(path: paths[0])

  @label(title: "Collect child result")
  report = (await child)?
  finish { reviewed: 1, report: report }
}

For the run shown above, the graph snapshot keeps every static node even when the branch is not taken. A host UI can render the rejected branch faintly, keep the child-process link attached to the Start child review node, and update Collect child result from running to completed or failed when the next node event arrives.

{
  "graph_key": "effect:session-1:turn-7:exec-1",
  "status": "running",
  "nodes": [
    {
      "id": "n1",
      "kind": "operation",
      "label": "workspace.default.glob",
      "label_metadata": { "title": "Find candidates" },
      "status": "completed",
      "duration_ms": 12
    },
    {
      "id": "n2",
      "kind": "branch",
      "label": "if empty(paths)",
      "label_metadata": { "title": "Choose path" },
      "status": "completed"
    },
    {
      "id": "n3",
      "kind": "start",
      "label": "start review_one",
      "label_metadata": { "title": "Start child review" },
      "status": "completed"
    }
  ],
  "edges": [
    { "id": "e-then", "from": "n2", "to": "then", "label": "then", "selection": "rejected" },
    { "id": "e-else", "from": "n2", "to": "else", "label": "else", "selection": "selected" }
  ],
  "children": [
    {
      "parent_node_id": "n3",
      "child_graph_key": "process:review-process-id",
      "child_entry_name": "review_one"
    }
  ]
}

pub trait TraceSink: Send + Sync {
    fn append(&self, record: &TraceRecord) -> Result<(), TraceSinkError>;

    // Force buffered trace data to durable storage before exit.
    // Default: no-op. Override when the sink buffers or can fsync.
    fn flush(&self) -> Result<(), TraceSinkError> { Ok(()) }
}

Call flush before the process exits so records a sink has not yet committed are not lost. JsonlTraceSink::flush is honest about what it owns: each append already writes its record through to the OS (open, append, close — no in-process buffer), so flush only issues an fsync to push the OS page cache to disk. TeeTraceSink::flush fans out to every wrapped sink. StderrTraceSink and the default keep the no-op. A host that handed lash the sink already holds its own Arc and can flush it directly; LashCore::flush_trace_sink is the equivalent lever for hosts that did not retain the handle.

Fan out to multiple destinations by wrapping sinks:

struct FanoutTraceSink {
    sinks: Vec<Arc<dyn TraceSink>>,
}

impl TraceSink for FanoutTraceSink {
    fn append(&self, record: &TraceRecord) -> Result<(), TraceSinkError> {
        for sink in &self.sinks {
            // Treat errors per-sink; one failing destination shouldn't take the others down.
            let _ = sink.append(record);
        }
        Ok(())
    }
}

{
  "schema_version": 9,
  "id": "6621dfa7-2cd0-4296-ac20-15c0f2d3cec1",
  "timestamp": "2026-05-11T11:42:01.234+00:00",
  "context": {
    "session_id": "chat-123",
    "turn_id": "turn-7"
  },
  "type": "tool_call_completed",
  "call_id": "call-9",
  "name": "fetch_url",
  "args": { "url": "https://example.com" },
  "output": { "outcome": { "status": "success", "payload": { "...": "..." } } },
  "duration_ms": 8
}

Record type: lash_trace::TraceRecord. Event variants: lash_trace::TraceEvent. Both Deserialize; parse directly with serde_json::from_str::<TraceRecord>(). That documented decode path enforces the exact schema version structurally.

# In your downstream Cargo.toml, enable OpenTelemetry on the supported facade.
lash-runtime = { version = "=0.1.0-alpha.113", features = ["otel-trace"] }

lash_core::OtelTraceSink wraps an opentelemetry tracer and converts every event to a span with payload attributes. Use in place of (or alongside) JsonlTraceSink:

use std::sync::Arc;

use lash::{
    LashCore,
    tracing::{OtelTraceSink, TraceLevel, TraceSink},
};

// Exporter/provider setup stays with the host; this reads the
// process-global OpenTelemetry tracer provider.
let sink: Arc<dyn TraceSink> = Arc::new(OtelTraceSink::from_global_provider());
let core = lash::LashCore::standard_builder(lash::TurnBudget::Unbounded)
    .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))
    .trace_sink(sink)
    .trace_level(TraceLevel::Extended)
    .build(crate::example_process_owner())?;

OtelTraceSink::flush is deliberately a no-op: the sink only starts and ends spans on your tracer, while the buffering that risks span loss on exit lives in your BatchSpanProcessor / exporter. Flushing it is the host's duty — call force_flush() (or shutdown()) on your TracerProvider before the process exits. Lash never owns the provider, so it cannot do this for you; LashCore::flush_trace_sink flushes the sink only.

Span names are derived from the typed event. Turn, LLM, and tool lifecycles become nested spans (lash.turn, lash.llm, lash.tool); a tool produces one lash.tool span whether it was a standard native call or ran inside a code block, because both flow through the same emission seam. Composition snapshots become lash.composition.changed events with the fingerprint, prompt-character count, and tool count. Set include_payload_json only when the exporter should also receive the complete rendered prompt and ordered tool schemas. The typed ExecCodeStarted, ExecCodeCompleted, and ExecCodeFailed events collapse into a lash.exec_code family, with the precise event kind on the lash.protocol.diagnostic_phase attribute; ObservationProjection gets lash.observation_projection, and opaque host/plugin protocol steps stay lash.protocol_step. The channel map that ties these span names back to the other reporting surfaces is Reporting channels.