car-eventlog 0.27.0

//! Event log with JSONL persistence for Common Agent Runtime.
//!
//! Append-only event log. Every runtime operation is recorded here.
//! Supports optional JSONL journal persistence for replay and audit.

pub mod harness_metrics;

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::fs::{self, OpenOptions};
use std::io::{BufRead, BufReader, BufWriter, Write};
use std::path::{Path, PathBuf};
use std::sync::mpsc;
use std::thread;
use uuid::Uuid;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EventLogStats {
    pub events: usize,
    pub spans: usize,
    pub approx_event_bytes: usize,
    pub approx_span_bytes: usize,
}

/// Event kinds matching the Python EventKind enum.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EventKind {
    ProposalReceived,
    ActionValidated,
    ActionRejected,
    ActionExecuting,
    ActionSucceeded,
    ActionFailed,
    ActionSkipped,
    ActionRetrying,
    ActionDeduplicated,
    PolicyViolation,
    StateChanged,
    StateSnapshot,
    StateRollback,
    // Skill lifecycle events (SkillRL-inspired)
    SkillDistilled,
    SkillEvolved,
    SkillDeprecated,
    EvolutionTriggered,
    /// A provisional skill candidate passed the validation gate and was promoted
    /// to Active, superseding its incumbent (SkillOpt-inspired — see
    /// `docs/solutions/gated-skill-optimization.md`).
    CandidatePromoted,
    /// A provisional skill candidate failed the validation gate and was rejected
    /// (recorded in the rejected-edit buffer so it isn't regenerated).
    CandidateRejected,
    // Memory consolidation ("dream") events
    Consolidated,
    // Replanning events
    ReplanAttempted,
    ReplanProposalReceived,
    ReplanRejected,
    ReplanExhausted,
    // Voice turn telemetry — emitted by car-engine's voice_turn dispatch
    // and the orchestrator. `data` carries `turn_id` (u64) plus
    // event-specific fields like `text_len`, `error`, `timeout_ms`.
    VoiceFastTurnStarted,
    VoiceFastTurnEnded,
    VoiceSidecarResolved,
    VoiceSidecarFailed,
    VoiceSidecarTimedOut,
    VoiceTurnCancelled,
    VoiceBridgePlayed,
    // Foreman merge-verify gate (verified-parallel-coding-orchestrator).
    // Emitted by car-multi's foreman gate when a farmed-out worktree is
    // verified before integration. `data` carries `subtask`, `changed_symbols`,
    // `containment_violations`, `semantic_conflicts`, and `build_test`. This is
    // the audit trail that makes the gate policy-aware rather than a bare merge.
    GateAccepted,
    GateRejected,
    // Per-execution caller / tenant scope (Parslee-ai/car#187 phase 3).
    // Emitted by Runtime::execute_scoped* once per proposal when the
    // RuntimeScope carries any identity. `data` carries `caller_id`,
    // `tenant_id`, and `claims` — exact set depends on what the
    // dispatcher forwarded. Audit / log analysis correlates actions
    // back to the caller / tenant that triggered them.
    SessionScope,
    // Permission-tier gate decisions (survey "Code as Agent Harness"
    // §3.4.3, §5.2.5 — the harness as safety governor). Emitted by
    // car-engine's TierPermissionHandler when the permission gate
    // evaluates an action. `data` carries `gate_decision` (allow /
    // needs_approval / deny), `required_tier`, `granted_tier`, and (for
    // escalation/deny) `fingerprint` + `reason`. The audit trail that
    // makes permission tiers inspectable rather than implicit.
    PermissionDecision,
    // A durable human-in-the-loop approval/rejection was recorded
    // (§5.2.5 — "approvals should be auditable state transitions").
    // `data` carries `fingerprint`, `approval` (approved / rejected),
    // `required_tier`, `reviewer`, `reason`, and optional `evidence`.
    // The auditable counterpart to the ApprovalLedger's durable record.
    ApprovalRecorded,
    // Deep-telemetry breadcrumbs (survey §3.5.1 — deep telemetry as the
    // optimization substrate; "decision-tree traces show where the agent
    // repeatedly chooses unproductive paths"). A BranchDecision records a
    // fork the harness took and why; `data` carries `branch` (the chosen
    // path), `reason`, and any decision-specific context. The substrate an
    // Evolution Agent (§3.5.2) replays to find where the loop wastes work.
    BranchDecision,
    // An alternative the harness considered and discarded — a failed
    // attempt superseded by a retry/replan, a candidate not selected.
    // `data` carries `alternative` (what was rejected) and `reason`.
    // Without this, telemetry shows only the path taken, not the paths
    // pruned, which is exactly what failure-mode diagnosis needs.
    AlternativeRejected,
    // An inference call's token/cost telemetry (§3.5.1). Carries the
    // standardized metric keys (`tokens_in`, `tokens_out`, `cost_usd`) via
    // `append_metered`. A dedicated kind so model cost feeds
    // `metrics_totals` without inflating action-success counts.
    InferenceMetered,
    // A transactional conflict the harness detected before executing a
    // proposal against the versioned shared state (survey §4.3/§5.2.4).
    // Emitted by the executor's pre-execution transaction check. `data`
    // carries `kind` (write_write / read_write / stale_assumption), `key`,
    // `actions`, `explanation`, and `resolution`. Under strict mode the
    // proposal is rejected; under warn mode it is only recorded.
    TransactionConflict,
}

/// Status of a trace span.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum SpanStatus {
    Ok,
    Error,
    Unset,
}

/// A trace span representing a unit of work.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Span {
    pub trace_id: String,
    pub span_id: String,
    pub parent_span_id: Option<String>,
    pub name: String,
    pub start_time: DateTime<Utc>,
    pub end_time: Option<DateTime<Utc>>,
    pub status: SpanStatus,
    pub attributes: HashMap<String, Value>,
}

/// Standardized `Event.data` keys for cross-cutting telemetry metrics, so
/// every emit site records them under the same name and aggregation can
/// rely on it (survey §3.5.1: deep telemetry "records the decision process
/// in greater detail: token usage and cost, model/tool latency …").
pub mod metric_keys {
    /// Wall-clock duration of the unit of work, milliseconds (f64).
    pub const DURATION_MS: &str = "duration_ms";
    /// Input/prompt tokens consumed (u64).
    pub const TOKENS_IN: &str = "tokens_in";
    /// Output/completion tokens produced (u64).
    pub const TOKENS_OUT: &str = "tokens_out";
    /// Estimated cost in USD (f64).
    pub const COST_USD: &str = "cost_usd";
}

/// Cross-cutting telemetry metrics attachable to any event. All optional —
/// a tool call has latency but no tokens; an inference has all four. Merged
/// into `Event.data` under [`metric_keys`] by [`EventLog::append_metered`],
/// and read back via the `Event` accessors, so downstream aggregation
/// (harness-level metrics, the Evolution Agent) has a uniform source.
#[derive(Debug, Clone, Copy, Default, PartialEq, Serialize, Deserialize)]
pub struct Metrics {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub duration_ms: Option<f64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tokens_in: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tokens_out: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cost_usd: Option<f64>,
}

impl Metrics {
    /// Latency-only metrics (the common tool/action case).
    pub fn latency(duration_ms: f64) -> Self {
        Self {
            duration_ms: Some(duration_ms),
            ..Default::default()
        }
    }

    /// Token + cost metrics for an inference call.
    pub fn inference(tokens_in: u64, tokens_out: u64, cost_usd: Option<f64>) -> Self {
        Self {
            duration_ms: None,
            tokens_in: Some(tokens_in),
            tokens_out: Some(tokens_out),
            cost_usd,
        }
    }

    pub fn with_duration(mut self, duration_ms: f64) -> Self {
        self.duration_ms = Some(duration_ms);
        self
    }

    /// Merge these metrics into an event `data` map under [`metric_keys`].
    fn merge_into(&self, data: &mut HashMap<String, Value>) {
        if let Some(d) = self.duration_ms {
            data.insert(metric_keys::DURATION_MS.into(), Value::from(d));
        }
        if let Some(t) = self.tokens_in {
            data.insert(metric_keys::TOKENS_IN.into(), Value::from(t));
        }
        if let Some(t) = self.tokens_out {
            data.insert(metric_keys::TOKENS_OUT.into(), Value::from(t));
        }
        if let Some(c) = self.cost_usd {
            data.insert(metric_keys::COST_USD.into(), Value::from(c));
        }
    }
}

/// A single event in the log.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Event {
    pub kind: EventKind,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub action_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub proposal_id: Option<String>,
    #[serde(default)]
    pub data: HashMap<String, Value>,
    #[serde(default = "Utc::now")]
    pub timestamp: DateTime<Utc>,
}

impl Event {
    /// Wall-clock duration recorded on this event, if any.
    pub fn duration_ms(&self) -> Option<f64> {
        self.data.get(metric_keys::DURATION_MS).and_then(Value::as_f64)
    }

    /// Input tokens recorded on this event, if any.
    pub fn tokens_in(&self) -> Option<u64> {
        self.data.get(metric_keys::TOKENS_IN).and_then(Value::as_u64)
    }

    /// Output tokens recorded on this event, if any.
    pub fn tokens_out(&self) -> Option<u64> {
        self.data.get(metric_keys::TOKENS_OUT).and_then(Value::as_u64)
    }

    /// Estimated cost (USD) recorded on this event, if any.
    pub fn cost_usd(&self) -> Option<f64> {
        self.data.get(metric_keys::COST_USD).and_then(Value::as_f64)
    }

    /// All metrics carried on this event, gathered into a [`Metrics`].
    pub fn metrics(&self) -> Metrics {
        Metrics {
            duration_ms: self.duration_ms(),
            tokens_in: self.tokens_in(),
            tokens_out: self.tokens_out(),
            cost_usd: self.cost_usd(),
        }
    }
}

/// Summed telemetry metrics across a set of events — the trajectory-level
/// totals harness-level evaluation (§5.2.1) and the Evolution Agent
/// (§3.5.2) reason over. `tokens` is the sum of in + out.
#[derive(Debug, Clone, Copy, Default, PartialEq, Serialize, Deserialize)]
pub struct MetricsTotals {
    pub duration_ms: f64,
    pub tokens_in: u64,
    pub tokens_out: u64,
    pub tokens: u64,
    pub cost_usd: f64,
    /// Number of events that carried at least one metric.
    pub metered_events: usize,
}

/// Sum the telemetry metrics across a slice of events. The single
/// implementation behind both [`EventLog::metrics_totals`] and the
/// harness-metrics computation, so the two can never drift on the metric
/// contract (neo review: avoid a duplicated copy).
pub fn metrics_totals_of(events: &[Event]) -> MetricsTotals {
    let mut totals = MetricsTotals::default();
    for ev in events {
        let m = ev.metrics();
        let mut metered = false;
        if let Some(d) = m.duration_ms {
            totals.duration_ms += d;
            metered = true;
        }
        if let Some(t) = m.tokens_in {
            totals.tokens_in = totals.tokens_in.saturating_add(t);
            metered = true;
        }
        if let Some(t) = m.tokens_out {
            totals.tokens_out = totals.tokens_out.saturating_add(t);
            metered = true;
        }
        if let Some(c) = m.cost_usd {
            totals.cost_usd += c;
            metered = true;
        }
        if metered {
            totals.metered_events += 1;
        }
    }
    totals.tokens = totals.tokens_in.saturating_add(totals.tokens_out);
    totals
}

/// Background JSONL journal writer. `EventLog::append` hands a serialized event
/// line to this over a channel; a dedicated thread owns the file and does the
/// actual write. So `append` never does file I/O while a caller holds the log
/// mutex — the head-of-line blocking that bites when many concurrent tasks
/// (e.g. Foreman gate verifications running under one shared, journaled session
/// log) each re-opened and wrote the file under the lock.
///
/// Best-effort, like the journal it replaces: an open/write failure drops the
/// line (the in-memory event vec is unaffected) — but unlike the old silent
/// journal, the hard failures (can't spawn the thread, can't open the file) are
/// surfaced via `tracing::warn!`, since this carries the gate audit trail and a
/// silently-broken audit log is worse than a noisy one.
///
/// The channel is unbounded so a burst never blocks the hot path. This relies on
/// an envelope: low per-session journal volume and a writer that keeps up, so the
/// backlog stays small. It is not a *new* unbounded-growth risk — the in-memory
/// `events` vec already grows without bound under the same pathological
/// hot-loop-`append` workload, so the channel is not the first thing to OOM.
struct JournalWriter {
    /// `None` only if the writer thread could not be spawned (journaling then
    /// silently disabled — still best-effort).
    tx: Option<mpsc::Sender<String>>,
    handle: Option<thread::JoinHandle<()>>,
}

impl JournalWriter {
    fn spawn(path: PathBuf) -> Self {
        let (tx, rx) = mpsc::channel::<String>();
        match thread::Builder::new()
            .name("car-eventlog-journal".into())
            .spawn(move || journal_loop(path, rx))
        {
            Ok(handle) => Self {
                tx: Some(tx),
                handle: Some(handle),
            },
            // Drop tx (rx dies with it); journaling becomes a no-op.
            Err(e) => {
                tracing::warn!(error = %e, "car-eventlog: failed to spawn journal writer thread — journaling disabled for this log");
                Self {
                    tx: None,
                    handle: None,
                }
            }
        }
    }

    fn send(&self, line: String) {
        if let Some(tx) = &self.tx {
            // Best-effort: if the writer thread has gone, drop the line.
            let _ = tx.send(line);
        }
    }
}

impl Drop for JournalWriter {
    fn drop(&mut self) {
        // Close the channel so the writer drains its backlog, flushes, and
        // exits; join so buffered lines are durable by the time the log is gone.
        self.tx.take();
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }
}

/// The journal thread's body: own the file, write each line, flush when the
/// channel goes momentarily idle (batches bursts, keeps durability prompt).
fn journal_loop(path: PathBuf, rx: mpsc::Receiver<String>) {
    let file = match OpenOptions::new().create(true).append(true).open(&path) {
        Ok(file) => file,
        // Can't open — surface it (this is the audit journal), then block-drain
        // so the channel doesn't accumulate if senders keep trying, and exit.
        // `recv()` blocks (it is not a spin loop) and returns Err once every
        // sender drops. Matches the prior fail-soft journal, but no longer silent.
        Err(e) => {
            tracing::warn!(path = %path.display(), error = %e, "car-eventlog: cannot open journal file — events for this log will not be persisted");
            while rx.recv().is_ok() {}
            return;
        }
    };
    let mut writer = BufWriter::new(file);
    while let Ok(line) = rx.recv() {
        let _ = writeln!(writer, "{line}");
        // Drain whatever is already queued without blocking, then flush once —
        // one fsync-free flush amortized over a burst instead of per line.
        while let Ok(more) = rx.try_recv() {
            let _ = writeln!(writer, "{more}");
        }
        let _ = writer.flush();
    }
    let _ = writer.flush();
}

/// Append-only event log with optional JSONL journal.
pub struct EventLog {
    events: Vec<Event>,
    spans: Vec<Span>,
    journal: Option<JournalWriter>,
}

impl EventLog {
    pub fn new() -> Self {
        Self {
            events: Vec::new(),
            spans: Vec::new(),
            journal: None,
        }
    }

    pub fn with_journal(path: PathBuf) -> Self {
        if let Some(parent) = path.parent() {
            let _ = fs::create_dir_all(parent);
        }
        Self {
            events: Vec::new(),
            spans: Vec::new(),
            journal: Some(JournalWriter::spawn(path)),
        }
    }

    pub fn append(
        &mut self,
        kind: EventKind,
        action_id: Option<&str>,
        proposal_id: Option<&str>,
        data: HashMap<String, Value>,
    ) -> &Event {
        let event = Event {
            kind,
            action_id: action_id.map(|s| s.to_string()),
            proposal_id: proposal_id.map(|s| s.to_string()),
            data,
            timestamp: Utc::now(),
        };

        // Hand the serialized line to the background writer — no file I/O here,
        // so a caller holding the log mutex is never blocked on disk.
        if let Some(journal) = &self.journal {
            if let Ok(json) = serde_json::to_string(&event) {
                journal.send(json);
            }
        }

        self.events.push(event);
        self.events.last().unwrap()
    }

    /// Append an event with cross-cutting [`Metrics`] (duration, tokens,
    /// cost) merged into its `data` under [`metric_keys`]. Use this for any
    /// event whose latency or token cost should feed trajectory-level
    /// aggregation (`metrics_totals`) — the deep-telemetry substrate of
    /// §3.5.1. Metric keys present in both `data` and `metrics` take the
    /// `metrics` value (the metrics argument wins).
    pub fn append_metered(
        &mut self,
        kind: EventKind,
        action_id: Option<&str>,
        proposal_id: Option<&str>,
        mut data: HashMap<String, Value>,
        metrics: Metrics,
    ) -> &Event {
        metrics.merge_into(&mut data);
        self.append(kind, action_id, proposal_id, data)
    }

    /// Sum the telemetry metrics across every event in the log — the
    /// trajectory-level totals (tokens, cost, wall-clock) that harness-level
    /// evaluation (§5.2.1) and the Evolution Agent (§3.5.2) reason over.
    ///
    /// Contract: this sums **every** event carrying a [`metric_keys`] value,
    /// regardless of which append path emitted it. A duration recorded once
    /// per action (e.g. `ActionSucceeded`) is counted once; the standardized
    /// keys mean there is a single value per metric per event, so there is no
    /// double-count as long as each unit of work meters itself once. Token
    /// metrics from `InferenceMetered` and latency from action events sum
    /// into the same totals — that is intended (total cost = model + tools).
    pub fn metrics_totals(&self) -> MetricsTotals {
        metrics_totals_of(&self.events)
    }

    pub fn events(&self) -> &[Event] {
        &self.events
    }

    pub fn len(&self) -> usize {
        self.events.len()
    }

    pub fn span_len(&self) -> usize {
        self.spans.len()
    }

    pub fn is_empty(&self) -> bool {
        self.events.is_empty()
    }

    pub fn stats(&self) -> EventLogStats {
        EventLogStats {
            events: self.events.len(),
            spans: self.spans.len(),
            approx_event_bytes: approx_json_bytes(&self.events),
            approx_span_bytes: approx_json_bytes(&self.spans),
        }
    }

    pub fn truncate_events_keep_last(&mut self, keep_last: usize) -> usize {
        truncate_vec_keep_last(&mut self.events, keep_last)
    }

    pub fn truncate_spans_keep_last(&mut self, keep_last: usize) -> usize {
        truncate_vec_keep_last(&mut self.spans, keep_last)
    }

    pub fn clear(&mut self) -> EventLogStats {
        let removed = self.stats();
        self.events.clear();
        self.events.shrink_to_fit();
        self.spans.clear();
        self.spans.shrink_to_fit();
        removed
    }

    pub fn filter(&self, kind: Option<&EventKind>, action_id: Option<&str>) -> Vec<&Event> {
        self.events
            .iter()
            .filter(|e| {
                if let Some(k) = kind {
                    if &e.kind != k {
                        return false;
                    }
                }
                if let Some(aid) = action_id {
                    if e.action_id.as_deref() != Some(aid) {
                        return false;
                    }
                }
                true
            })
            .collect()
    }

    /// Begin a new trace span. Returns the generated span_id.
    pub fn begin_span(
        &mut self,
        name: &str,
        trace_id: &str,
        parent_span_id: Option<&str>,
        attributes: HashMap<String, Value>,
    ) -> String {
        let span_id = Uuid::new_v4().to_string();
        let span = Span {
            trace_id: trace_id.to_string(),
            span_id: span_id.clone(),
            parent_span_id: parent_span_id.map(|s| s.to_string()),
            name: name.to_string(),
            start_time: Utc::now(),
            end_time: None,
            status: SpanStatus::Unset,
            attributes,
        };
        self.spans.push(span);
        span_id
    }

    /// End an open span by setting its status and end time.
    pub fn end_span(&mut self, span_id: &str, status: SpanStatus) {
        if let Some(span) = self.spans.iter_mut().find(|s| s.span_id == span_id) {
            span.end_time = Some(Utc::now());
            span.status = status;
        }
    }

    /// Return all spans.
    pub fn spans(&self) -> Vec<Span> {
        self.spans.clone()
    }

    /// Export traces as OTLP-compatible JSON.
    pub fn export_traces(&self) -> String {
        // Group spans by trace_id
        let mut traces: HashMap<&str, Vec<&Span>> = HashMap::new();
        for span in &self.spans {
            traces.entry(span.trace_id.as_str()).or_default().push(span);
        }

        let resource_spans: Vec<Value> = traces
            .into_iter()
            .map(|(_trace_id, spans)| {
                let scope_spans = spans
                    .iter()
                    .map(|s| {
                        let mut span_obj = serde_json::json!({
                            "traceId": s.trace_id,
                            "spanId": s.span_id,
                            "name": s.name,
                            "startTimeUnixNano": s.start_time.timestamp_nanos_opt().unwrap_or(0).to_string(),
                            "status": {
                                "code": match s.status {
                                    SpanStatus::Ok => 1,
                                    SpanStatus::Error => 2,
                                    SpanStatus::Unset => 0,
                                }
                            },
                            "attributes": s.attributes.iter().map(|(k, v)| {
                                serde_json::json!({
                                    "key": k,
                                    "value": { "stringValue": v.to_string() }
                                })
                            }).collect::<Vec<_>>(),
                        });

                        if let Some(ref parent) = s.parent_span_id {
                            span_obj.as_object_mut().unwrap().insert(
                                "parentSpanId".to_string(),
                                Value::from(parent.as_str()),
                            );
                        }
                        if let Some(end) = s.end_time {
                            span_obj.as_object_mut().unwrap().insert(
                                "endTimeUnixNano".to_string(),
                                Value::from(end.timestamp_nanos_opt().unwrap_or(0).to_string()),
                            );
                        }

                        span_obj
                    })
                    .collect::<Vec<_>>();

                serde_json::json!({
                    "resource": {
                        "attributes": [
                            { "key": "service.name", "value": { "stringValue": "car-runtime" } }
                        ]
                    },
                    "scopeSpans": [{
                        "scope": { "name": "car-eventlog" },
                        "spans": scope_spans
                    }]
                })
            })
            .collect();

        serde_json::to_string(&serde_json::json!({
            "resourceSpans": resource_spans
        }))
        .unwrap_or_else(|_| "{}".to_string())
    }

    /// Load an event log from a JSONL journal file.
    pub fn load(path: &Path) -> std::io::Result<Self> {
        let file = fs::File::open(path)?;
        let reader = BufReader::new(file);
        let mut events = Vec::new();

        for line in reader.lines() {
            let line = line?;
            let line = line.trim();
            if !line.is_empty() {
                if let Ok(event) = serde_json::from_str::<Event>(line) {
                    events.push(event);
                }
            }
        }

        Ok(Self {
            events,
            spans: Vec::new(),
            // Subsequent appends journal back to the same file (append mode
            // preserves the loaded content) via the background writer.
            journal: Some(JournalWriter::spawn(path.to_path_buf())),
        })
    }
}

fn approx_json_bytes<T: Serialize>(value: &T) -> usize {
    serde_json::to_vec(value)
        .map(|bytes| bytes.len())
        .unwrap_or(0)
}

fn truncate_vec_keep_last<T>(items: &mut Vec<T>, keep_last: usize) -> usize {
    let len = items.len();
    if len <= keep_last {
        return 0;
    }
    let removed = len - keep_last;
    items.drain(..removed);
    items.shrink_to_fit();
    removed
}

impl Default for EventLog {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn append_and_read() {
        let mut log = EventLog::new();
        log.append(
            EventKind::ProposalReceived,
            None,
            Some("p1"),
            [("source".to_string(), Value::from("test"))].into(),
        );
        assert_eq!(log.len(), 1);
        assert_eq!(log.events()[0].kind, EventKind::ProposalReceived);
    }

    #[test]
    fn metered_event_carries_metrics_in_data() {
        let mut log = EventLog::new();
        log.append_metered(
            EventKind::ActionSucceeded,
            Some("a1"),
            Some("p1"),
            [("tool".to_string(), Value::from("search"))].into(),
            Metrics::inference(120, 45, Some(0.0012)).with_duration(83.0),
        );
        let ev = &log.events()[0];
        // Original data preserved; metrics merged under standardized keys.
        assert_eq!(ev.data.get("tool").unwrap(), "search");
        assert_eq!(ev.duration_ms(), Some(83.0));
        assert_eq!(ev.tokens_in(), Some(120));
        assert_eq!(ev.tokens_out(), Some(45));
        assert_eq!(ev.cost_usd(), Some(0.0012));
    }

    #[test]
    fn metrics_totals_sum_across_events() {
        let mut log = EventLog::new();
        log.append_metered(
            EventKind::ActionSucceeded,
            Some("a1"),
            None,
            HashMap::new(),
            Metrics::latency(50.0),
        );
        log.append_metered(
            EventKind::ActionSucceeded,
            Some("a2"),
            None,
            HashMap::new(),
            Metrics::inference(100, 20, Some(0.5)).with_duration(70.0),
        );
        // An un-metered event must not affect totals.
        log.append(EventKind::ProposalReceived, None, None, HashMap::new());

        let t = log.metrics_totals();
        assert_eq!(t.duration_ms, 120.0);
        assert_eq!(t.tokens_in, 100);
        assert_eq!(t.tokens_out, 20);
        assert_eq!(t.tokens, 120);
        assert_eq!(t.cost_usd, 0.5);
        assert_eq!(t.metered_events, 2);
    }

    #[test]
    fn metrics_totals_counts_raw_appended_duration_key() {
        // Contract: metrics_totals sums any event carrying a metric key,
        // regardless of append path. A legacy raw `append` that puts
        // "duration_ms" in data must still be counted (locks the contract
        // documented on metrics_totals).
        let mut log = EventLog::new();
        log.append(
            EventKind::ActionSucceeded,
            Some("a1"),
            None,
            [(metric_keys::DURATION_MS.to_string(), Value::from(42.0))].into(),
        );
        let t = log.metrics_totals();
        assert_eq!(t.duration_ms, 42.0);
        assert_eq!(t.metered_events, 1);
    }

    #[test]
    fn new_telemetry_event_kinds_serialize_snake_case() {
        // The new kinds must round-trip as snake_case for the JSON wire.
        let json = serde_json::to_string(&EventKind::BranchDecision).unwrap();
        assert_eq!(json, "\"branch_decision\"");
        let json = serde_json::to_string(&EventKind::AlternativeRejected).unwrap();
        assert_eq!(json, "\"alternative_rejected\"");
        let json = serde_json::to_string(&EventKind::InferenceMetered).unwrap();
        assert_eq!(json, "\"inference_metered\"");
    }

    #[test]
    fn filter_by_kind() {
        let mut log = EventLog::new();
        log.append(
            EventKind::ProposalReceived,
            None,
            Some("p1"),
            HashMap::new(),
        );
        log.append(
            EventKind::ActionValidated,
            Some("a1"),
            Some("p1"),
            HashMap::new(),
        );
        log.append(
            EventKind::ActionSucceeded,
            Some("a1"),
            Some("p1"),
            HashMap::new(),
        );

        let validated = log.filter(Some(&EventKind::ActionValidated), None);
        assert_eq!(validated.len(), 1);
    }

    #[test]
    fn filter_by_action_id() {
        let mut log = EventLog::new();
        log.append(EventKind::ActionValidated, Some("a1"), None, HashMap::new());
        log.append(EventKind::ActionValidated, Some("a2"), None, HashMap::new());

        let a1_events = log.filter(None, Some("a1"));
        assert_eq!(a1_events.len(), 1);
    }

    #[test]
    fn journal_write_and_reload() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("events.jsonl");

        {
            let mut log = EventLog::with_journal(journal.clone());
            log.append(
                EventKind::ProposalReceived,
                None,
                Some("p1"),
                HashMap::new(),
            );
            log.append(
                EventKind::ActionSucceeded,
                Some("a1"),
                Some("p1"),
                HashMap::new(),
            );
        }

        assert!(journal.exists());

        let reloaded = EventLog::load(&journal).unwrap();
        assert_eq!(reloaded.len(), 2);
        assert_eq!(reloaded.events()[0].kind, EventKind::ProposalReceived);
        assert_eq!(reloaded.events()[1].kind, EventKind::ActionSucceeded);
    }

    #[test]
    fn journal_preserves_order_and_count_under_burst() {
        // The background writer must not lose or reorder events under a tight
        // append burst; drop-join guarantees the backlog is flushed before the
        // log is gone.
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("burst.jsonl");
        {
            let mut log = EventLog::with_journal(journal.clone());
            for i in 0..500 {
                log.append(
                    EventKind::ActionSucceeded,
                    Some(&format!("a{i}")),
                    None,
                    HashMap::new(),
                );
            }
        } // drop joins the writer thread → all 500 lines flushed.

        let reloaded = EventLog::load(&journal).unwrap();
        assert_eq!(reloaded.len(), 500, "no events lost");
        for (i, event) in reloaded.events().iter().enumerate() {
            assert_eq!(
                event.action_id.as_deref(),
                Some(format!("a{i}").as_str()),
                "order preserved at {i}"
            );
        }
    }

    #[test]
    fn unopenable_journal_is_best_effort_not_fatal() {
        // The whole "best-effort" promise rests on this branch: a journal path
        // that can't be opened (here: the path IS an existing directory) must not
        // panic or block append — the in-memory log keeps working.
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("a-directory");
        fs::create_dir(&journal).unwrap(); // open(append) on a dir fails

        let mut log = EventLog::with_journal(journal);
        log.append(EventKind::ProposalReceived, None, Some("p1"), HashMap::new());
        log.append(EventKind::ActionSucceeded, Some("a1"), None, HashMap::new());
        assert_eq!(log.len(), 2, "in-memory log unaffected by an unwritable journal");
        // Drop must still terminate cleanly (writer thread drained and joined).
    }

    #[test]
    fn load_then_append_preserves_existing_and_adds() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("resume.jsonl");
        {
            let mut log = EventLog::with_journal(journal.clone());
            log.append(EventKind::ProposalReceived, None, Some("p1"), HashMap::new());
        }
        // Resume: load, append more, drop → both old and new are on disk.
        {
            let mut log = EventLog::load(&journal).unwrap();
            assert_eq!(log.len(), 1);
            log.append(EventKind::ActionSucceeded, Some("a1"), None, HashMap::new());
        }
        let reloaded = EventLog::load(&journal).unwrap();
        assert_eq!(reloaded.len(), 2, "append-mode preserved the loaded line");
        assert_eq!(reloaded.events()[0].kind, EventKind::ProposalReceived);
        assert_eq!(reloaded.events()[1].kind, EventKind::ActionSucceeded);
    }

    #[test]
    fn event_kind_serializes_snake_case() {
        assert_eq!(
            serde_json::to_string(&EventKind::ProposalReceived).unwrap(),
            "\"proposal_received\""
        );
        assert_eq!(
            serde_json::to_string(&EventKind::StateSnapshot).unwrap(),
            "\"state_snapshot\""
        );
    }

    #[test]
    fn stats_truncate_and_clear_release_retained_entries() {
        let mut log = EventLog::new();
        for idx in 0..5 {
            log.append(
                EventKind::ActionSucceeded,
                Some(&format!("a{idx}")),
                Some("p1"),
                [("payload".to_string(), Value::from("x".repeat(16)))].into(),
            );
            log.begin_span("action.tool_call", "trace", None, HashMap::new());
        }

        let stats = log.stats();
        assert_eq!(stats.events, 5);
        assert_eq!(stats.spans, 5);
        assert!(stats.approx_event_bytes > 0);
        assert!(stats.approx_span_bytes > 0);

        assert_eq!(log.truncate_events_keep_last(2), 3);
        assert_eq!(log.truncate_spans_keep_last(1), 4);
        assert_eq!(log.len(), 2);
        assert_eq!(log.span_len(), 1);
        assert_eq!(log.events()[0].action_id.as_deref(), Some("a3"));

        let removed = log.clear();
        assert_eq!(removed.events, 2);
        assert_eq!(removed.spans, 1);
        assert_eq!(log.len(), 0);
        assert_eq!(log.span_len(), 0);
    }

    #[test]
    fn span_begin_end_lifecycle() {
        let mut log = EventLog::new();
        let trace_id = "trace-1".to_string();

        let span_id = log.begin_span(
            "test.operation",
            &trace_id,
            None,
            [("key".to_string(), Value::from("value"))].into(),
        );

        let spans = log.spans();
        assert_eq!(spans.len(), 1);
        assert_eq!(spans[0].name, "test.operation");
        assert_eq!(spans[0].trace_id, "trace-1");
        assert!(spans[0].parent_span_id.is_none());
        assert!(spans[0].end_time.is_none());
        assert_eq!(spans[0].status, SpanStatus::Unset);

        log.end_span(&span_id, SpanStatus::Ok);

        let spans = log.spans();
        assert!(spans[0].end_time.is_some());
        assert_eq!(spans[0].status, SpanStatus::Ok);
    }

    #[test]
    fn span_parent_child_relationship() {
        let mut log = EventLog::new();
        let trace_id = "trace-2".to_string();

        let parent_id = log.begin_span("parent.op", &trace_id, None, HashMap::new());
        let child_id = log.begin_span("child.op", &trace_id, Some(&parent_id), HashMap::new());

        let spans = log.spans();
        assert_eq!(spans.len(), 2);

        let child = spans.iter().find(|s| s.span_id == child_id).unwrap();
        assert_eq!(child.parent_span_id.as_deref(), Some(parent_id.as_str()));
        assert_eq!(child.trace_id, trace_id);

        let parent = spans.iter().find(|s| s.span_id == parent_id).unwrap();
        assert!(parent.parent_span_id.is_none());
    }

    #[test]
    fn export_traces_produces_valid_json() {
        let mut log = EventLog::new();
        let trace_id = "trace-3".to_string();

        let root = log.begin_span(
            "proposal.execute",
            &trace_id,
            None,
            [("proposal_id".to_string(), Value::from("p1"))].into(),
        );
        let child = log.begin_span(
            "action.tool_call",
            &trace_id,
            Some(&root),
            [("tool".to_string(), Value::from("read_file"))].into(),
        );
        log.end_span(&child, SpanStatus::Ok);
        log.end_span(&root, SpanStatus::Ok);

        let json_str = log.export_traces();
        let parsed: Value =
            serde_json::from_str(&json_str).expect("export_traces must produce valid JSON");

        let resource_spans = parsed["resourceSpans"].as_array().unwrap();
        assert_eq!(resource_spans.len(), 1);

        let scope_spans = &resource_spans[0]["scopeSpans"][0]["spans"];
        let spans_arr = scope_spans.as_array().unwrap();
        assert_eq!(spans_arr.len(), 2);

        // Verify OTLP structure
        for span in spans_arr {
            assert!(span.get("traceId").is_some());
            assert!(span.get("spanId").is_some());
            assert!(span.get("name").is_some());
            assert!(span.get("startTimeUnixNano").is_some());
            assert!(span.get("endTimeUnixNano").is_some());
            assert!(span.get("status").is_some());
        }

        // Verify the child has parentSpanId
        let child_span = spans_arr
            .iter()
            .find(|s| s["name"] == "action.tool_call")
            .unwrap();
        assert!(child_span.get("parentSpanId").is_some());
    }

    #[test]
    fn span_status_set_on_error() {
        let mut log = EventLog::new();
        let trace_id = "trace-4".to_string();

        let span_id = log.begin_span("failing.op", &trace_id, None, HashMap::new());
        log.end_span(&span_id, SpanStatus::Error);

        let spans = log.spans();
        assert_eq!(spans[0].status, SpanStatus::Error);
        assert!(spans[0].end_time.is_some());
    }
}