gateway: trace-id propagation + coordinator session JSONL (Rust parity)

Cross-runtime parity with the Go-side observability wave (commits
d6d2fdf + 1a3a82a in golangLAKEHOUSE). The two layers J flagged:
the LIVE per-call view (Langfuse) and the LONGITUDINAL forensic view
(JSONL queryable via DuckDB). Hard correctness gate (FillValidator
phantom-rejection) was already in place; this is the observability
on top.

## Trace-id propagation

X-Lakehouse-Trace-Id header constant declared in
crates/gateway/src/v1/iterate.rs (matches Go's shared.TraceIDHeader
byte-for-byte). When set on an inbound /v1/iterate request, the
handler reuses it; the chat + validate self-loopback hops forward
the same header so chatd's trace emit nests under the parent rather
than minting a fresh top-level trace per call.

ChatTrace gains a parent_trace_id field. emit_chat_inner skips the
trace-create event when parent is set, only emits the
generation-create which attaches to the existing trace tree. Result:
an iterate session with N retries shows in Langfuse as ONE tree, not
N+1 disconnected traces.

emit_attempt_span (new) writes one Langfuse span per iteration
attempt with input={iteration, model, provider, prompt} and
output={verdict, raw, error}. WARNING level on non-accepted
verdicts. The returned span id is stamped on the corresponding
SessionRecord attempt for cross-log correlation.

## Coordinator session JSONL

crates/gateway/src/v1/session_log.rs — new writer matching Go's
internal/validator/session_log.go schema byte-for-byte:
  - SessionRecord with schema=session.iterate.v1
  - SessionAttemptRecord per retry
  - SessionLogger.append: tokio Mutex serialized append-only
  - Best-effort posture (slog.Warn on error, never blocks request)

iterate.rs builds + appends a row on EVERY code path:
  - accepted: write_session_accepted with grounded_in_roster bool
    derived from validate_workers WorkerLookup (matches Go's
    handlers.rosterCheckFor("fill") semantics)
  - max-iter-exhausted: write_session_failure
  - infra-error: write_infra_error (so a missing /v1/iterate event
    never silently disappears from the longitudinal log)

[gateway].session_log_path config field (empty = disabled).
Production: /var/lib/lakehouse/gateway/sessions.jsonl. Operators who
want a unified longitudinal stream can point both Rust and Go
loggers at the same path — write-append is safe at the row sizes we
produce.

## Cross-runtime parity probe

crates/gateway/src/bin/parity_session_log: tiny stdin/stdout helper
that round-trips a fixture through SessionRecord serde.
golangLAKEHOUSE/scripts/cutover/parity/session_log_parity.sh feeds
4 fixtures through both helpers and diffs the rows after stripping
timestamp + daemon (the two fields that legitimately differ between
producers).

Result: **4/4 byte-equal** including the unicode-prompt fixture
("Café résumé ⭐ 你好"). Schema parity holds. The non-trivial-equal
guard in the probe rejects the case where both sides fail
identically — protecting against a regression where one side
silently stops producing valid JSON.

## Verification

- cargo test -p gateway --lib: 90/90 PASS (3 new session_log tests
  including concurrent-append safety)
- cargo check --workspace: clean
- session_log_parity.sh: 4/4 fixtures byte-equal
- Both runtimes can append to the same path; DuckDB sees one stream
- The Go-side validatord smoke remains 5/5 (unchanged)

## Architecture invariant

Don't propose to "wire trace-id propagation in Rust" or "add Rust
session log" — both are now shipped on the demo/post-pr11-polish
branch. The longitudinal log + Langfuse tree together cover the
multi-call observability concern J flagged 2026-05-02.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/gateway/src/v1/iterate.rs

@@ -21,12 +21,19 @@
 //! re-implementation. Staffing executors, agent loops, and future
 //! validators all reach the same code path.
 
-use axum::{extract::State, http::StatusCode, response::IntoResponse, Json};
+use axum::{extract::State, http::{HeaderMap, StatusCode}, response::IntoResponse, Json};
 use serde::{Deserialize, Serialize};
 
 const DEFAULT_MAX_ITERATIONS: u32 = 3;
 const LOOPBACK_TIMEOUT_SECS: u64 = 240;
 
+/// Header name used to propagate a Langfuse parent trace id across
+/// daemon boundaries. Matches Go's `shared.TraceIDHeader` constant
+/// byte-for-byte (commit d6d2fdf in golangLAKEHOUSE) — same wire
+/// format means a Go caller can hit Rust's /v1/iterate (or vice
+/// versa) and the resulting Langfuse trees nest correctly.
+pub const TRACE_ID_HEADER: &str = "x-lakehouse-trace-id";
+
 #[derive(Deserialize)]
 pub struct IterateRequest {
     /// "fill" | "email" | "playbook" — picks which validator runs.
@@ -90,26 +97,47 @@ pub struct IterateFailure {
 }
 
 pub async fn iterate(
-    State(_state): State<super::V1State>,
+    State(state): State<super::V1State>,
+    headers: HeaderMap,
     Json(req): Json<IterateRequest>,
 ) -> impl IntoResponse {
     let max_iter = req.max_iterations.unwrap_or(DEFAULT_MAX_ITERATIONS).max(1);
     let temperature = req.temperature.unwrap_or(0.2);
     let max_tokens = req.max_tokens.unwrap_or(4096);
     let mut history: Vec<IterateAttempt> = Vec::with_capacity(max_iter as usize);
+    let mut attempt_records: Vec<super::session_log::SessionAttemptRecord> = Vec::with_capacity(max_iter as usize);
     let mut current_prompt = req.prompt.clone();
 
+    // Resolve the parent Langfuse trace id. Caller-forwarded header
+    // wins (cross-daemon tree linkage); otherwise mint a fresh id so
+    // the iterate session is its own tree. Same shape as the Go-side
+    // validatord trace propagation.
+    let trace_id: String = headers
+        .get(TRACE_ID_HEADER)
+        .and_then(|v| v.to_str().ok())
+        .filter(|s| !s.is_empty())
+        .map(|s| s.to_string())
+        .unwrap_or_else(new_trace_id);
+
     let client = match reqwest::Client::builder()
         .timeout(std::time::Duration::from_secs(LOOPBACK_TIMEOUT_SECS))
         .build() {
         Ok(c) => c,
-        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, format!("client build: {e}")).into_response(),
+        Err(e) => {
+            // Even infrastructure failures get a session row so a
+            // missing /v1/iterate event never silently disappears
+            // from the longitudinal log.
+            write_infra_error(&state, &req, &trace_id, max_iter, 0, format!("client build: {e}")).await;
+            return (StatusCode::INTERNAL_SERVER_ERROR, format!("client build: {e}")).into_response();
+        }
     };
     // Self-loopback to the gateway port. Carries gateway internal
     // calls through /v1/chat + /v1/validate so /v1/usage tracks them.
     let gateway = "http://127.0.0.1:3100";
+    let t0 = std::time::Instant::now();
 
     for iteration in 0..max_iter {
+        let attempt_started = chrono::Utc::now();
         // ── Generate ──
         let mut messages = Vec::with_capacity(2);
         if let Some(sys) = &req.system {
@@ -123,20 +151,33 @@ pub async fn iterate(
             "temperature": temperature,
             "max_tokens": max_tokens,
         });
-        let raw = match call_chat(&client, gateway, &chat_body).await {
+        let raw = match call_chat(&client, gateway, &chat_body, &trace_id).await {
             Ok(r) => r,
-            Err(e) => return (StatusCode::BAD_GATEWAY, format!("/v1/chat hop failed at iter {iteration}: {e}")).into_response(),
+            Err(e) => {
+                write_infra_error(&state, &req, &trace_id, max_iter, t0.elapsed().as_millis() as u64, format!("/v1/chat hop failed at iter {iteration}: {e}")).await;
+                return (StatusCode::BAD_GATEWAY, format!("/v1/chat hop failed at iter {iteration}: {e}")).into_response();
+            }
         };
 
         // ── Extract JSON ──
         let artifact = match extract_json(&raw) {
             Some(a) => a,
             None => {
+                let span_id = emit_attempt_span(
+                    &state, &trace_id, iteration, &req, &current_prompt, &raw, "no_json", None,
+                    attempt_started, chrono::Utc::now(),
+                );
                 history.push(IterateAttempt {
                     iteration,
                     raw: raw.chars().take(2000).collect(),
                     status: AttemptStatus::NoJson,
                 });
+                attempt_records.push(super::session_log::SessionAttemptRecord {
+                    iteration,
+                    verdict_kind: "no_json".to_string(),
+                    error: None,
+                    span_id,
+                });
                 current_prompt = format!(
                     "{}\n\nYour previous attempt did not contain a JSON object. Reply with ONLY a valid JSON object matching the requested artifact shape.",
                     req.prompt,
@@ -151,13 +192,26 @@ pub async fn iterate(
             "artifact": artifact,
             "context": req.context.clone().unwrap_or(serde_json::Value::Null),
         });
-        match call_validate(&client, gateway, &validate_body).await {
+        match call_validate(&client, gateway, &validate_body, &trace_id).await {
             Ok(report) => {
+                let span_id = emit_attempt_span(
+                    &state, &trace_id, iteration, &req, &current_prompt, &raw, "accepted", None,
+                    attempt_started, chrono::Utc::now(),
+                );
                 history.push(IterateAttempt {
                     iteration,
                     raw: raw.chars().take(2000).collect(),
                     status: AttemptStatus::Accepted,
                 });
+                attempt_records.push(super::session_log::SessionAttemptRecord {
+                    iteration,
+                    verdict_kind: "accepted".to_string(),
+                    error: None,
+                    span_id,
+                });
+                let duration_ms = t0.elapsed().as_millis() as u64;
+                let grounded = grounded_in_roster(&state, &req.kind, &artifact);
+                write_session_accepted(&state, &req, &trace_id, iteration + 1, max_iter, attempt_records, &artifact, grounded, duration_ms).await;
                 return (StatusCode::OK, Json(IterateResponse {
                     artifact,
                     validation: report,
@@ -167,6 +221,11 @@ pub async fn iterate(
             }
             Err(err) => {
                 let err_summary = err.to_string();
+                let span_id = emit_attempt_span(
+                    &state, &trace_id, iteration, &req, &current_prompt, &raw, "validation_failed",
+                    Some(err_summary.clone()),
+                    attempt_started, chrono::Utc::now(),
+                );
                 history.push(IterateAttempt {
                     iteration,
                     raw: raw.chars().take(2000).collect(),
@@ -174,6 +233,12 @@ pub async fn iterate(
                         error: serde_json::to_value(&err_summary).unwrap_or(serde_json::Value::Null),
                     },
                 });
+                attempt_records.push(super::session_log::SessionAttemptRecord {
+                    iteration,
+                    verdict_kind: "validation_failed".to_string(),
+                    error: Some(err_summary.clone()),
+                    span_id,
+                });
                 // Append validation feedback to prompt for next iter.
                 // The model sees concrete failure mode + retries with
                 // corrective context. This is the "observer correction"
@@ -188,6 +253,8 @@ pub async fn iterate(
         }
     }
 
+    let duration_ms = t0.elapsed().as_millis() as u64;
+    write_session_failure(&state, &req, &trace_id, max_iter, max_iter, attempt_records, duration_ms).await;
     (StatusCode::UNPROCESSABLE_ENTITY, Json(IterateFailure {
         error: format!("max iterations reached ({max_iter}) without passing validation"),
         iterations: max_iter,
@@ -195,12 +262,157 @@ pub async fn iterate(
     })).into_response()
 }
 
-async fn call_chat(client: &reqwest::Client, gateway: &str, body: &serde_json::Value) -> Result<String, String> {
-    let resp = client.post(format!("{gateway}/v1/chat"))
-        .json(body)
-        .send()
-        .await
-        .map_err(|e| format!("chat hop: {e}"))?;
+// ─── Helpers — Langfuse spans + session log + roster check ─────────
+
+fn emit_attempt_span(
+    state: &super::V1State,
+    trace_id: &str,
+    iteration: u32,
+    req: &IterateRequest,
+    prompt: &str,
+    raw: &str,
+    verdict: &str,
+    error: Option<String>,
+    started: chrono::DateTime<chrono::Utc>,
+    ended: chrono::DateTime<chrono::Utc>,
+) -> Option<String> {
+    let lf = state.langfuse.as_ref()?;
+    Some(lf.emit_attempt_span(super::langfuse_trace::AttemptSpan {
+        trace_id: trace_id.to_string(),
+        iteration,
+        model: req.model.clone(),
+        provider: req.provider.clone(),
+        prompt: prompt.to_string(),
+        raw: raw.to_string(),
+        verdict: verdict.to_string(),
+        error,
+        start_time: started.to_rfc3339(),
+        end_time: ended.to_rfc3339(),
+    }))
+}
+
+/// Verify every fill artifact's candidate IDs exist in the roster.
+/// Returns Some(true)/Some(false) on the fill kind, None otherwise
+/// (other kinds don't have worker IDs to ground). Same semantics as
+/// Go's `handlers.rosterCheckFor("fill")`.
+fn grounded_in_roster(
+    state: &super::V1State,
+    kind: &str,
+    artifact: &serde_json::Value,
+) -> Option<bool> {
+    if kind != "fill" {
+        return None;
+    }
+    let fills = artifact.get("fills").and_then(|v| v.as_array())?;
+    for f in fills {
+        let id = match f.get("candidate_id").and_then(|v| v.as_str()) {
+            Some(s) if !s.is_empty() => s,
+            _ => return Some(false),
+        };
+        if state.validate_workers.find(id).is_none() {
+            return Some(false);
+        }
+    }
+    Some(true)
+}
+
+async fn write_session_accepted(
+    state: &super::V1State,
+    req: &IterateRequest,
+    trace_id: &str,
+    iterations: u32,
+    max_iter: u32,
+    attempts: Vec<super::session_log::SessionAttemptRecord>,
+    artifact: &serde_json::Value,
+    grounded: Option<bool>,
+    duration_ms: u64,
+) {
+    let Some(logger) = state.session_log.as_ref() else { return };
+    let rec = build_session_record(req, trace_id, "accepted", iterations, max_iter, attempts, Some(artifact.clone()), grounded, duration_ms);
+    logger.append(rec).await;
+}
+
+async fn write_session_failure(
+    state: &super::V1State,
+    req: &IterateRequest,
+    trace_id: &str,
+    iterations: u32,
+    max_iter: u32,
+    attempts: Vec<super::session_log::SessionAttemptRecord>,
+    duration_ms: u64,
+) {
+    let Some(logger) = state.session_log.as_ref() else { return };
+    let rec = build_session_record(req, trace_id, "max_iter_exhausted", iterations, max_iter, attempts, None, None, duration_ms);
+    logger.append(rec).await;
+}
+
+async fn write_infra_error(
+    state: &super::V1State,
+    req: &IterateRequest,
+    trace_id: &str,
+    max_iter: u32,
+    duration_ms: u64,
+    error: String,
+) {
+    let Some(logger) = state.session_log.as_ref() else { return };
+    let attempts = vec![super::session_log::SessionAttemptRecord {
+        iteration: 0,
+        verdict_kind: "infra_error".to_string(),
+        error: Some(error),
+        span_id: None,
+    }];
+    let rec = build_session_record(req, trace_id, "infra_error", 0, max_iter, attempts, None, None, duration_ms);
+    logger.append(rec).await;
+}
+
+fn build_session_record(
+    req: &IterateRequest,
+    trace_id: &str,
+    final_verdict: &str,
+    iterations: u32,
+    max_iter: u32,
+    attempts: Vec<super::session_log::SessionAttemptRecord>,
+    artifact: Option<serde_json::Value>,
+    grounded: Option<bool>,
+    duration_ms: u64,
+) -> super::session_log::SessionRecord {
+    super::session_log::SessionRecord {
+        schema: super::session_log::SESSION_RECORD_SCHEMA.to_string(),
+        session_id: trace_id.to_string(),
+        timestamp: chrono::Utc::now().to_rfc3339(),
+        daemon: "gateway".to_string(),
+        kind: req.kind.clone(),
+        model: req.model.clone(),
+        provider: req.provider.clone(),
+        prompt: super::session_log::truncate(&req.prompt, 4000),
+        iterations,
+        max_iterations: max_iter,
+        final_verdict: final_verdict.to_string(),
+        attempts,
+        artifact,
+        grounded_in_roster: grounded,
+        duration_ms,
+    }
+}
+
+/// Generate a fresh trace id when no parent was forwarded. Same
+/// time-ordered hex shape Langfuse already accepts elsewhere in this
+/// crate (see `langfuse_trace::uuid_v7_like`).
+fn new_trace_id() -> String {
+    let ts = chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0);
+    let rand = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.subsec_nanos())
+        .unwrap_or(0);
+    format!("{:016x}-{:08x}", ts, rand)
+}
+
+async fn call_chat(client: &reqwest::Client, gateway: &str, body: &serde_json::Value, trace_id: &str) -> Result<String, String> {
+    let mut req = client.post(format!("{gateway}/v1/chat")).json(body);
+    if !trace_id.is_empty() {
+        req = req.header(TRACE_ID_HEADER, trace_id);
+    }
+    let resp = req.send().await.map_err(|e| format!("chat hop: {e}"))?;
     let status = resp.status();
     if !status.is_success() {
         let body = resp.text().await.unwrap_or_default();
@@ -213,12 +425,12 @@ async fn call_chat(client: &reqwest::Client, gateway: &str, body: &serde_json::V
         .to_string())
 }
 
-async fn call_validate(client: &reqwest::Client, gateway: &str, body: &serde_json::Value) -> Result<serde_json::Value, String> {
-    let resp = client.post(format!("{gateway}/v1/validate"))
-        .json(body)
-        .send()
-        .await
-        .map_err(|e| format!("validate hop: {e}"))?;
+async fn call_validate(client: &reqwest::Client, gateway: &str, body: &serde_json::Value, trace_id: &str) -> Result<serde_json::Value, String> {
+    let mut req = client.post(format!("{gateway}/v1/validate")).json(body);
+    if !trace_id.is_empty() {
+        req = req.header(TRACE_ID_HEADER, trace_id);
+    }
+    let resp = req.send().await.map_err(|e| format!("validate hop: {e}"))?;
     let status = resp.status();
     let parsed: serde_json::Value = resp.json().await.map_err(|e| format!("validate parse: {e}"))?;
     if status.is_success() {