harness/graph(langfuse): nest observations into a run tree + add scores API

docs/modules/graph/observability.md

@@ -313,14 +313,17 @@ The batch maps the graph run onto Langfuse's trace/observation model:
 - one `trace-create` (id defaults to the run's `root_run_id`; name defaults to
   the `graph_id`; session defaults to the run's `thread_id`) with the
   `GraphHealthSummary` folded into trace metadata
+- one structural `span-create` for the graph run (`{trace}:run:{run_id}`, named
+  for the `graph_id`), parented to the trace, bracketing the whole run
 - one timed `span-create` per superstep (`{trace}:step:{n}`), parented to the
-  trace
+  graph-run span
 - one timed `span-create` per node handler (`{trace}:node:{name}:{step}`),
   parented to its superstep span; `node.failed` promotes the span to `ERROR`
   level with the rendered error as `statusMessage`
-- one timed `span-create` per embedded subgraph
+- one timed `span-create` per embedded subgraph, parented to the graph-run span
 - an `event-create` for every remaining observation (routes, checkpoints,
-  interrupts, custom writes, run lifecycle), with `run.failed` mapped to `ERROR`
+  interrupts, custom writes, run lifecycle), parented to the graph-run span,
+  with `run.failed` mapped to `ERROR`
 
 Still-running work (a `node.started` with no terminal event) is exported as an
 open span with a start time but no end time.
@@ -334,6 +337,14 @@ every graph step, node, model generation, and tool call under one Langfuse
 trace. This is what makes full end-to-end telemetry — including tool health and
 tool timing — visible in a single trace tree.
 
+The nesting is exact, not just co-located: the graph-run span
+(`{trace}:run:{run_id}`) uses the same id scheme the harness exporter parents
+its agent run spans to. A sub-agent a node spawns carries `parent_run_id` equal
+to the graph run id, so its harness-exported run span resolves to
+`{trace}:run:{graph_run_id}` and nests directly under the graph-run span rather
+than floating at the trace root — the graph, its nodes, and their agents form
+one contiguous tree.
+
 ## Debug Payloads
 
 Debug streams are for executor introspection. They should be structured enough

docs/modules/harness/observability.md

@@ -343,12 +343,42 @@ replays them verbatim.
 
 Downstream, the Langfuse exporter reads these fields to populate the
 Input/Output panels of a generation (`generation-create`) and a tool
-observation (`tool-create`). With capture disabled the fields are absent and the
+observation (`span-create`). With capture disabled the fields are absent and the
 observation renders with ids, timing, and usage only. The harness exporter also
 defaults the trace-level metadata from the run lineage (root/first run ids and
 parent), mirroring the graph exporter, so a trace is correlatable even when the
 caller passes no metadata.
 
+## Langfuse Observation Tree
+
+The harness exporter mirrors LangChain's callback **run tree** rather than a
+flat event list. For each distinct `run_id` in a batch it emits one
+`span-create` (`{trace_id}:run:{run_id}`, named `agent` for the top-level run
+and `sub-agent` for any spawned child), folding the `RunStarted`/`RunCompleted`/
+`RunFailed` events into that span's start/end window and ERROR status. Every
+model generation, tool span, and remaining event then nests under its run span
+via `parentObservationId`.
+
+Because run-span ids are deterministic and trace-namespaced, the tree spans
+**separately exported batches**: a sub-agent run whose observations are shipped
+in their own batch carries `parent_run_id`, so its run span resolves to
+`{trace_id}:run:{parent_run_id}` and nests under the parent agent's span in the
+same trace. The graph exporter emits a matching `{trace_id}:run:{run_id}` span
+for the graph run, so a graph run and the agent runs its nodes spawn
+reconstruct one unified nested tree.
+
+## Scores
+
+`LangfuseClient::create_score` (and the offline `build_score_batch`) mirrors
+Langfuse's `createScore` — the way a LangChain/LangGraph trace is graded after
+the fact (human ratings, LLM-as-judge, regression metrics). A `LangfuseScore`
+is numeric, categorical, or boolean; scoped to a `trace_id` and optionally to a
+single observation id; and carries an optional comment. Score ids default to a
+deterministic value derived from the target and metric name, so re-scoring the
+same target upserts rather than duplicating. Point the score's `trace_id` at an
+exported trace (recall both exporters default the trace id to the run's
+`root_run_id`) to attach the evaluation to that run.
+
 ## Graph Events
 
 State-graph event payloads should be rich enough for a UI to render run

src/graph/observability/README.md

@@ -43,10 +43,12 @@ so existing runs are unchanged.
   `from_status`).
 - `GraphLangfuseExporter` (`langfuse/`) — exports a run's observations to
   Langfuse, turning supersteps and nodes into timed spans (failures promoted
-  to `ERROR`) and attaching the health summary to the trace. It shares the
-  harness `LangfuseClient` transport and defaults its `traceId` to the run's
-  `root_run_id`, so a graph run and the agent/tool runs its nodes spawn land
-  under one trace.
+  to `ERROR`) and attaching the health summary to the trace. It emits a
+  structural graph-run span (`{trace}:run:{run_id}`) that steps, subgraphs, and
+  point events nest under. It shares the harness `LangfuseClient` transport and
+  defaults its `traceId` to the run's `root_run_id`; because the graph-run span
+  uses the same id scheme the harness exporter parents its agent run spans to,
+  a graph run and the agent/tool runs its nodes spawn nest into one trace tree.
 
 ## Persistence bridge
 

src/graph/observability/langfuse.rs

@@ -128,18 +128,29 @@ impl GraphLangfuseExporter {
         let trace_ts = iso_ms(first.ts_ms);
         let health = GraphHealthSummary::from_observations(observations);
 
-        let mut batch = Vec::with_capacity(observations.len() + 1);
+        let mut batch = Vec::with_capacity(observations.len() + 2);
         batch.push(trace_create(&trace_id, &trace_ts, &trace, first, &health));
 
+        // A structural span for the graph run itself, sitting directly under the
+        // trace. Every step, subgraph, and point event nests under it, and — key
+        // for a unified trace — its id follows the same `{trace}:run:{run_id}`
+        // scheme the harness exporter parents its run spans to. A sub-agent that
+        // a graph node spawns carries `parent_run_id == graph run id`, so its
+        // harness-exported run span nests directly under this graph-run span
+        // rather than floating at the trace root.
+        let run_parent = run_span_id(&trace_id, first.root_run_id.as_str());
+        batch.push(graph_run_span(&trace_id, &run_parent, observations, first));
+
         let mut consumed = vec![false; observations.len()];
         push_span_events(
             &trace_id,
+            &run_parent,
             observations,
             &mut consumed,
             &mut batch,
             self.span_metadata_fn.as_deref(),
         );
-        push_point_events(&trace_id, observations, &consumed, &mut batch);
+        push_point_events(&trace_id, &run_parent, observations, &consumed, &mut batch);
 
         Ok(json!({ "batch": batch }))
     }
@@ -155,6 +166,59 @@ impl GraphLangfuseExporter {
     }
 }
 
+/// The stable Langfuse observation id for a run's span, shared with the harness
+/// exporter (`{trace_id}:run:{run_id}`) so a graph run and the harness agent
+/// runs its nodes spawn resolve to one nested tree under the same trace.
+fn run_span_id(trace_id: &str, run_id: &str) -> String {
+    format!("{trace_id}:run:{run_id}")
+}
+
+/// Builds the structural `span-create` for the graph run, parented to the trace.
+///
+/// The span brackets the run: it starts at the first observation and ends at the
+/// last (or at a terminal `RunCompleted`/`RunFailed` when present). It is a plain
+/// structural anchor — run failure is still surfaced by the dedicated
+/// `run.failed` point event — so its only job is to give steps, subgraphs, and
+/// any child agent runs a single parent to nest under.
+fn graph_run_span(
+    trace_id: &str,
+    run_span_id: &str,
+    observations: &[GraphObservation],
+    first: &GraphObservation,
+) -> Value {
+    let start_ts = first.ts_ms;
+    // Prefer a terminal run event's timestamp; else close at the last observation.
+    let end_ts = observations
+        .iter()
+        .rev()
+        .find(|o| {
+            matches!(
+                o.event,
+                GraphEvent::RunCompleted { .. } | GraphEvent::RunFailed { .. }
+            )
+        })
+        .or_else(|| observations.last())
+        .map(|o| o.ts_ms)
+        .unwrap_or(start_ts);
+    json!({
+        "id": run_span_id,
+        "timestamp": iso_ms(end_ts),
+        "type": "span-create",
+        "body": clean_nulls(json!({
+            "id": run_span_id,
+            "traceId": trace_id,
+            "name": first.graph_id.as_str(),
+            "startTime": iso_ms(start_ts),
+            "endTime": iso_ms(end_ts),
+            "metadata": json!({
+                "run_id": first.run_id.as_str(),
+                "root_run_id": first.root_run_id.as_str(),
+                "graph_id": first.graph_id.as_str(),
+            }),
+        })),
+    })
+}
+
 /// Resolves the Langfuse trace id: the configured id when set, else the first
 /// observation's root run id so it aligns with the harness agent exporter.
 fn resolve_trace_id(trace: &LangfuseTraceConfig, observations: &[GraphObservation]) -> String {
@@ -225,6 +289,7 @@ fn trace_create(
 /// re-emitted as point events. Unpaired starts become open spans (start only).
 fn push_span_events(
     trace_id: &str,
+    run_parent: &str,
     observations: &[GraphObservation],
     consumed: &mut [bool],
     batch: &mut Vec<Value>,
@@ -249,6 +314,7 @@ fn push_span_events(
                     consumed[idx] = true;
                     batch.push(step_span(
                         trace_id,
+                        run_parent,
                         *step,
                         start_ts,
                         Some(obs.ts_ms),
@@ -309,6 +375,7 @@ fn push_span_events(
                     consumed[idx] = true;
                     batch.push(subgraph_span(
                         trace_id,
+                        run_parent,
                         node,
                         namespace,
                         start_ts,
@@ -327,7 +394,7 @@ fn push_span_events(
         while let Some((start_idx, start_ts)) = queue.pop_front() {
             let obs = &observations[start_idx];
             batch.push(step_span(
-                trace_id, step, start_ts, None, obs, obs, injector,
+                trace_id, run_parent, step, start_ts, None, obs, obs, injector,
             ));
         }
     }
@@ -349,6 +416,7 @@ fn push_span_events(
         while let Some((start_idx, start_ts)) = queue.pop_front() {
             batch.push(subgraph_span(
                 trace_id,
+                run_parent,
                 &node,
                 &namespace,
                 start_ts,
@@ -364,6 +432,7 @@ fn push_span_events(
 /// span, mapping `run.failed` to `ERROR` level with the rendered error.
 fn push_point_events(
     trace_id: &str,
+    run_parent: &str,
     observations: &[GraphObservation],
     consumed: &[bool],
     batch: &mut Vec<Value>,
@@ -388,6 +457,7 @@ fn push_point_events(
             "body": clean_nulls(json!({
                 "id": obs.event_id.as_str(),
                 "traceId": trace_id,
+                "parentObservationId": run_parent,
                 "name": obs.event.kind(),
                 "startTime": ts,
                 "level": level,
@@ -402,6 +472,7 @@ fn push_point_events(
 #[allow(clippy::too_many_arguments)]
 fn step_span(
     trace_id: &str,
+    run_parent: &str,
     step: usize,
     start_ts: u64,
     end_ts: Option<u64>,
@@ -413,7 +484,7 @@ fn step_span(
     span_event(
         trace_id,
         &format!("{trace_id}:step:{step}"),
-        None,
+        Some(run_parent.to_string()),
         &format!("step {step}"),
         start_ts,
         end_ts,
@@ -454,8 +525,10 @@ fn node_span(
 }
 
 /// Builds a `span-create` for an embedded subgraph, parented to the trace.
+#[allow(clippy::too_many_arguments)]
 fn subgraph_span(
     trace_id: &str,
+    run_parent: &str,
     node: &NodeId,
     namespace: &[String],
     start_ts: u64,
@@ -467,7 +540,7 @@ fn subgraph_span(
     span_event(
         trace_id,
         &format!("{trace_id}:subgraph:{}", namespace.join("/")),
-        None,
+        Some(run_parent.to_string()),
         &format!("subgraph {}", node.as_str()),
         start_ts,
         end_ts,

src/graph/observability/langfuse/test.rs

@@ -172,13 +172,22 @@ fn steps_and_nodes_become_timed_spans() {
         .unwrap();
     let events = batch["batch"].as_array().unwrap();
 
+    // The graph run itself is a structural span under the trace, named for the
+    // graph, that every step nests beneath. Its id uses the shared
+    // `{trace}:run:{run}` scheme so harness sub-agent runs (parent_run_id = the
+    // graph run) nest under it too, unifying graph + agent into one trace tree.
+    let run_span = find(events, "span-create", "root-1:run:root-1").expect("graph run span");
+    assert_eq!(run_span["body"]["name"], "demo-graph");
+    assert!(run_span["body"].get("parentObservationId").is_none());
+    assert_eq!(run_span["body"]["startTime"], "2024-01-01T00:00:01.000Z");
+
     // Superstep span with real start/end times.
     let step = find(events, "span-create", "root-1:step:1").expect("step span");
     assert_eq!(step["body"]["name"], "step 1");
     assert_eq!(step["body"]["startTime"], "2024-01-01T00:00:01.010Z");
     assert_eq!(step["body"]["endTime"], "2024-01-01T00:00:01.090Z");
-    // Steps parent to the trace, not another span.
-    assert!(step["body"].get("parentObservationId").is_none());
+    // Steps parent to the graph-run span, not straight to the trace.
+    assert_eq!(step["body"]["parentObservationId"], "root-1:run:root-1");
 
     // Node a completed cleanly and is parented to its step span.
     let node_a = find(events, "span-create", "root-1:node:a:1").expect("node a span");

src/harness/observability/README.md

@@ -40,7 +40,14 @@ surface and operational constraints.
     `AgentLatencyMetrics::from_observations(&[AgentObservation])`.
 - `LangfuseAuth`, `LangfuseClient`, `LangfuseTraceConfig` (re-exported from the
   private `langfuse` submodule) — the Langfuse exporter used to ship harness
-  (and, via shared helpers, graph) traces to Langfuse.
+  (and, via shared helpers, graph) traces to Langfuse. The exporter emits a
+  per-run span (`{trace}:run:{run_id}`, `agent`/`sub-agent`) and nests every
+  generation, tool span, and event under it, so a run — and sub-agent recursion
+  across separately-exported batches — surfaces as a tree, matching LangChain's
+  callback run hierarchy.
+- `LangfuseScore` / `LangfuseScoreValue` — an evaluation score (numeric,
+  categorical, or boolean; trace- or observation-scoped) attached to an exported
+  trace via `LangfuseClient::create_score`, mirroring Langfuse's `createScore`.
 
 ## Persistence bridge