car-eventlog 0.34.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_adapt;
pub mod harness_metrics;
pub mod observability;
pub mod tool_receipts;

pub use observability::{
    evaluate_alerts, summarize, summarize_log, Alert, AlertKind, AlertThresholds, MetricsSummary,
};

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,
    // A proposal-admission gate decision (EPIC A / task A1 — the
    // executor's pre-execution safety seam). Emitted once per registered
    // `AdmissionGate` that runs during proposal admission. `data` carries
    // `gate` (the gate name, e.g. information_flow / concurrency / policy),
    // `decision` (allow / reject / needs_approval), and — when the gate
    // objects — `reason`, `blocked` (the offending action ids), and an
    // optional `fingerprint` for approval escalations. The audit trail
    // that makes the verified safety checks inspectable as live
    // enforcement rather than dormant library functions.
    AdmissionGateDecision,
    // A tool-use hallucination caught by cross-checking the model's claims
    // against the runtime's own execution receipts (EPIC A / A6 — arXiv
    // 2603.10060). `data` carries `count` and `hallucinations` (each with
    // kind/tool/explanation). Deterministic and zero-inference: the runtime
    // ran the tools, so it holds unforgeable ground truth.
    ToolReceiptHallucination,
}

/// 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>,
    /// Hash of the previous event in the chain (EPIC A / A9 tamper-
    /// evidence). `None` when hash chaining is disabled (the default) —
    /// the field is skipped in serialization, so logs without chaining are
    /// byte-identical to before this was added.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prev_hash: Option<String>,
    /// This event's own content hash, computed over its fields plus
    /// `prev_hash`. Present only when hash chaining is enabled.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub hash: Option<String>,
}

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
}

/// Per-agent cost/token attribution (EPIC G / G3). Folded from
/// `InferenceMetered` events that carry an `agent` field in `data`.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct AgentCost {
    pub agent: String,
    /// Number of metered inference events attributed to this agent.
    pub calls: u64,
    pub tokens_in: u64,
    pub tokens_out: u64,
    pub cost_usd: f64,
}

/// Attribute token/cost totals per agent by folding `InferenceMetered` events
/// grouped by their `data["agent"]` field (EPIC G / G3). Events with no `agent`
/// field are grouped under `"unknown"`. Ordered by agent name (BTreeMap) so the
/// report is deterministic. This is how a multi-agent run reports cost per agent
/// (e.g. Researcher $2, Coordinator $0.5) and, joined with the `tools`/`workflow`
/// provenance fields the emit sites stamp, how a tool call is traceable to its
/// agent.
pub fn cost_by_agent_of(events: &[Event]) -> Vec<AgentCost> {
    use std::collections::BTreeMap;
    let mut map: BTreeMap<String, AgentCost> = BTreeMap::new();
    for e in events {
        if e.kind != EventKind::InferenceMetered {
            continue;
        }
        let agent = e
            .data
            .get("agent")
            .and_then(|v| v.as_str())
            .unwrap_or("unknown")
            .to_string();
        let entry = map.entry(agent.clone()).or_insert_with(|| AgentCost {
            agent,
            ..Default::default()
        });
        entry.calls += 1;
        entry.tokens_in = entry.tokens_in.saturating_add(e.tokens_in().unwrap_or(0));
        entry.tokens_out = entry.tokens_out.saturating_add(e.tokens_out().unwrap_or(0));
        entry.cost_usd += e.cost_usd().unwrap_or(0.0);
    }
    map.into_values().collect()
}

/// 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>,
    /// When true, each appended event is hash-chained to its predecessor
    /// (EPIC A / A9). Off by default — enabling it is opt-in so existing
    /// JSONL output stays byte-identical for consumers that don't need
    /// tamper-evidence.
    hash_chaining: bool,
    /// The hash of the most recently appended event, threaded into the
    /// next event's `prev_hash`. The genesis link uses the empty string.
    last_hash: Option<String>,
    /// Auto-retention policy (EPIC G / G2). When set, `append` caps the
    /// in-memory event count at `max_events` (dropping oldest) so the log
    /// can't grow unbounded. Age-based trimming is applied by
    /// `enforce_retention`. `None` = keep everything (unchanged default).
    retention: Option<RetentionPolicy>,
    /// Path of the JSONL journal, kept so retention trims can compact
    /// (rewrite) the file — the background [`JournalWriter`] only appends.
    journal_path: Option<PathBuf>,
    /// Approximate number of event lines currently in the journal file:
    /// incremented per journaled append, seeded from the parsed event count
    /// on [`EventLog::load`], reset to the retained count after a
    /// compaction. Drives the compaction throttle.
    journal_lines: usize,
    /// Total events ever dropped from the in-memory log (retention trims,
    /// manual truncation, `clear`). Monotonic. Lets consumers that project
    /// over `events()` — e.g. the tool-receipt verifier (A6) — know the
    /// retained window is incomplete instead of mistaking an evicted event
    /// for one that never happened.
    trimmed_events: u64,
    /// Monotonic cumulative cost (USD) across every event ever appended
    /// (EPIC G / G1). Updated at append time and **never** decremented by
    /// retention trims, truncation, or `clear`, so a cumulative budget check
    /// can't slide backward when old events are evicted. Seeded from the
    /// journal on [`EventLog::load`].
    cumulative_cost_usd: f64,
}

/// Journal-compaction throttle floor (G2): a retention trim only triggers a
/// journal rewrite once the journal holds at least this many more lines than
/// the retained set (and the rewrite would shrink it by ≥25% — see
/// [`EventLog::maybe_compact_journal`]). Keeps frequent small trims from
/// rewriting the file on every append.
const JOURNAL_COMPACT_MIN_EXCESS: usize = 1024;

/// Compute the content hash of an event for the tamper-evidence chain.
///
/// Hashes `prev_hash` plus a canonical rendering of the event's content
/// (kind, action/proposal ids, sorted `data`, timestamp). The top-level
/// `data` map is sorted by key so the digest is stable across a
/// serialize/deserialize round-trip (serde_json already emits nested object
/// keys in sorted order). Any after-the-fact edit to a chained event — or an
/// interior deletion/reordering — breaks the chain from that point on. The
/// chain has no anchored head hash, so truncation at either end (dropping a
/// prefix or a suffix of the log wholesale) is NOT detectable; see
/// [`EventLog::verify_chain`] for the precise guarantee.
fn event_digest(
    prev_hash: &str,
    kind: &EventKind,
    action_id: Option<&str>,
    proposal_id: Option<&str>,
    data: &HashMap<String, Value>,
    timestamp: &DateTime<Utc>,
) -> String {
    use sha2::{Digest, Sha256};
    let mut sorted: Vec<(&String, &Value)> = data.iter().collect();
    sorted.sort_by(|a, b| a.0.cmp(b.0));
    let data_canon: String = sorted
        .iter()
        .map(|(k, v)| format!("{k}={}", v))
        .collect::<Vec<_>>()
        .join("\u{1f}");
    let kind_str = serde_json::to_string(kind).unwrap_or_default();
    let mut hasher = Sha256::new();
    hasher.update(prev_hash.as_bytes());
    hasher.update(b"\x1e");
    hasher.update(kind_str.as_bytes());
    hasher.update(b"\x1e");
    hasher.update(action_id.unwrap_or("").as_bytes());
    hasher.update(b"\x1e");
    hasher.update(proposal_id.unwrap_or("").as_bytes());
    hasher.update(b"\x1e");
    hasher.update(data_canon.as_bytes());
    hasher.update(b"\x1e");
    hasher.update(timestamp.to_rfc3339().as_bytes());
    let digest = hasher.finalize();
    digest.iter().map(|b| format!("{b:02x}")).collect()
}

/// Retention policy for an [`EventLog`] (EPIC G / G2). Bounds the log by
/// **size** (`max_events`, enforced automatically on append — oldest dropped)
/// and by **age** (`max_age_secs`, applied by [`EventLog::enforce_retention`]).
/// Both `None` = keep everything.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct RetentionPolicy {
    /// Cap the in-memory event count; on overflow the oldest are dropped.
    #[serde(default)]
    pub max_events: Option<usize>,
    /// Drop events older than this many seconds when `enforce_retention` runs.
    #[serde(default)]
    pub max_age_secs: Option<i64>,
}

/// A structured audit query over the event log (EPIC G / G2). Every field is
/// an AND-conjoined filter; empty/`None` fields don't constrain. Answers
/// "who ran what tool when, and which approvals applied" by filtering the
/// `SessionScope` / `PermissionDecision` / `ApprovalRecorded` / action trail.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct EventQuery {
    /// Restrict to these event kinds (empty = any kind).
    #[serde(default)]
    pub kinds: Vec<EventKind>,
    /// Exact match on `action_id`.
    #[serde(default)]
    pub action_id: Option<String>,
    /// Exact match on `proposal_id`.
    #[serde(default)]
    pub proposal_id: Option<String>,
    /// Inclusive lower time bound.
    #[serde(default)]
    pub since: Option<DateTime<Utc>>,
    /// Exclusive upper time bound.
    #[serde(default)]
    pub until: Option<DateTime<Utc>>,
    /// Match events whose `data` contains ALL these key→value pairs (compared
    /// as strings). Covers caller/tenant/tool/gate/decision, which live in
    /// `data` on the audit events.
    #[serde(default)]
    pub data_matches: std::collections::HashMap<String, String>,
    /// Cap the number of results (most-recent-first). `None`/0 = unlimited.
    #[serde(default)]
    pub limit: Option<usize>,
}

/// Does a JSON `data` value equal the query string? Compares strings directly
/// and stringifies scalars so `{"count": 3}` matches `"3"`.
fn data_value_matches(v: &Value, want: &str) -> bool {
    match v {
        Value::String(s) => s == want,
        Value::Null => false,
        other => *other == want,
    }
}

impl EventQuery {
    /// Does `e` satisfy every constraint in this query?
    pub fn matches(&self, e: &Event) -> bool {
        if !self.kinds.is_empty() && !self.kinds.contains(&e.kind) {
            return false;
        }
        if let Some(aid) = &self.action_id {
            if e.action_id.as_deref() != Some(aid.as_str()) {
                return false;
            }
        }
        if let Some(pid) = &self.proposal_id {
            if e.proposal_id.as_deref() != Some(pid.as_str()) {
                return false;
            }
        }
        if let Some(since) = self.since {
            if e.timestamp < since {
                return false;
            }
        }
        if let Some(until) = self.until {
            if e.timestamp >= until {
                return false;
            }
        }
        for (k, want) in &self.data_matches {
            match e.data.get(k) {
                Some(v) if data_value_matches(v, want) => {}
                _ => return false,
            }
        }
        true
    }
}

impl EventLog {
    pub fn new() -> Self {
        Self {
            events: Vec::new(),
            spans: Vec::new(),
            journal: None,
            hash_chaining: false,
            last_hash: None,
            retention: None,
            journal_path: None,
            journal_lines: 0,
            trimmed_events: 0,
            cumulative_cost_usd: 0.0,
        }
    }

    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.clone())),
            hash_chaining: false,
            last_hash: None,
            retention: None,
            journal_path: Some(path),
            journal_lines: 0,
            trimmed_events: 0,
            cumulative_cost_usd: 0.0,
        }
    }

    /// Enable tamper-evident hash chaining for events appended from now on
    /// (EPIC A / A9). The chain continues from the last already-appended
    /// event's hash if one exists (re-enabling after a load), else from the
    /// genesis link. Returns `self` for builder-style use.
    pub fn with_hash_chaining(mut self) -> Self {
        self.enable_hash_chaining();
        self
    }

    /// Turn on hash chaining in place. Idempotent.
    pub fn enable_hash_chaining(&mut self) {
        self.hash_chaining = true;
        // Continue the chain from whatever the last event already carries.
        if self.last_hash.is_none() {
            self.last_hash = self.events.last().and_then(|e| e.hash.clone());
        }
    }

    /// Whether hash chaining is currently enabled.
    pub fn hash_chaining_enabled(&self) -> bool {
        self.hash_chaining
    }

    pub fn append(
        &mut self,
        kind: EventKind,
        action_id: Option<&str>,
        proposal_id: Option<&str>,
        data: HashMap<String, Value>,
    ) -> &Event {
        let timestamp = Utc::now();
        let (prev_hash, hash) = if self.hash_chaining {
            let prev = self.last_hash.clone().unwrap_or_default();
            let h = event_digest(&prev, &kind, action_id, proposal_id, &data, &timestamp);
            self.last_hash = Some(h.clone());
            (Some(prev), Some(h))
        } else {
            (None, None)
        };
        let event = Event {
            kind,
            action_id: action_id.map(|s| s.to_string()),
            proposal_id: proposal_id.map(|s| s.to_string()),
            data,
            timestamp,
            prev_hash,
            hash,
        };

        // 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.journal_lines += 1;
            }
        }

        // Monotonic cumulative cost (G1): fold cost in at append time so a
        // budget check survives retention trims of the underlying events.
        if let Some(c) = event.cost_usd() {
            self.cumulative_cost_usd += c;
        }

        self.events.push(event);
        // Auto-retention (EPIC G / G2): cap the in-memory log at max_events so
        // it can't grow unbounded. Cheap — a bounded pop from the front only
        // when over the cap. Age-based trimming is on-demand via
        // enforce_retention (walking every event on each append would be O(n)).
        if let Some(max) = self.retention.as_ref().and_then(|p| p.max_events) {
            if self.events.len() > max {
                let removed = truncate_vec_keep_last(&mut self.events, max);
                self.trimmed_events += removed as u64;
                // The journal keeps the dropped events until the (throttled)
                // compaction rewrites it to the retained set.
                self.maybe_compact_journal();
            }
        }
        self.events.last().unwrap()
    }

    /// Verify the tamper-evidence hash chain over the currently-loaded
    /// events (EPIC A / A9). Walks every event that carries a `hash`,
    /// recomputing it from its content + the running `prev_hash` and
    /// checking the links join up. Returns `Ok(n)` with the number of
    /// chained events verified, or `Err(index)` naming the first event
    /// whose hash or linkage doesn't match — i.e. the point at which a
    /// chained event was edited, or an interior event was deleted or
    /// reordered.
    ///
    /// **Scope of the guarantee:** the chain detects *interior*
    /// edits/reorderings/deletions only. It cannot detect truncation at
    /// either end: there is no anchored head hash, so the first chained
    /// event's `prev_hash` is taken on trust (dropping a prefix goes
    /// unnoticed), and nothing pins the tail (dropping a suffix goes
    /// unnoticed). Detecting head/tail truncation requires anchoring the
    /// chain head (and a trusted latest-hash witness), which is out of
    /// scope until that anchor exists.
    ///
    /// Events without a `hash` (appended before chaining was enabled) are
    /// skipped, so a partially-chained log verifies its chained suffix.
    pub fn verify_chain(&self) -> Result<usize, usize> {
        let mut prev = String::new();
        let mut verified = 0usize;
        let mut chain_started = false;
        for (i, ev) in self.events.iter().enumerate() {
            let Some(stored) = &ev.hash else {
                // Once the chain has started, a gap is a break.
                if chain_started {
                    return Err(i);
                }
                continue;
            };
            // The recorded prev_hash must match the running hash.
            let recorded_prev = ev.prev_hash.clone().unwrap_or_default();
            if chain_started && recorded_prev != prev {
                return Err(i);
            }
            let recomputed = event_digest(
                &recorded_prev,
                &ev.kind,
                ev.action_id.as_deref(),
                ev.proposal_id.as_deref(),
                &ev.data,
                &ev.timestamp,
            );
            if &recomputed != stored {
                return Err(i);
            }
            prev = stored.clone();
            chain_started = true;
            verified += 1;
        }
        Ok(verified)
    }

    /// 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)
    }

    /// Per-agent cost/token report (EPIC G / G3) — see [`cost_by_agent_of`].
    pub fn cost_by_agent(&self) -> Vec<AgentCost> {
        cost_by_agent_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 {
        let removed = truncate_vec_keep_last(&mut self.events, keep_last);
        self.trimmed_events += removed as u64;
        if removed > 0 {
            self.maybe_compact_journal();
        }
        removed
    }

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

    /// Drop every retained event and span, releasing their memory. The
    /// JSONL journal is left untouched (it is the audit trail); the
    /// monotonic counters (`trimmed_events`, `cumulative_cost_usd`) are
    /// preserved — `clear` frees memory, it doesn't reset the log's history.
    pub fn clear(&mut self) -> EventLogStats {
        let removed = self.stats();
        self.trimmed_events += removed.events as u64;
        self.events.clear();
        self.events.shrink_to_fit();
        self.spans.clear();
        self.spans.shrink_to_fit();
        removed
    }

    /// Total events ever dropped from the in-memory log (retention trims,
    /// manual truncation, `clear`). Monotonic; `> 0` means the retained
    /// window is incomplete — consumers projecting over [`Self::events`]
    /// (e.g. the A6 tool-receipt verifier) must treat an absent event as
    /// possibly-evicted, not as never-happened.
    pub fn trimmed_events(&self) -> u64 {
        self.trimmed_events
    }

    /// Monotonic cumulative cost (USD) across every event ever appended
    /// (EPIC G / G1). Unlike folding `cost_usd` over [`Self::events`] — which
    /// slides backward when retention trims metered events — this counter
    /// only grows, so it is the correct denominator for a cumulative budget
    /// (`AlertThresholds::max_cost_usd`). Seeded from the journal on
    /// [`Self::load`]; survives trims and [`Self::clear`].
    pub fn cumulative_cost_usd(&self) -> f64 {
        self.cumulative_cost_usd
    }

    /// Current size of the JSONL journal file in bytes, if a journal is
    /// configured and stat-able. The background writer batches, so this may
    /// momentarily lag the last few appends.
    pub fn journal_size_bytes(&self) -> Option<u64> {
        let path = self.journal_path.as_ref()?;
        fs::metadata(path).ok().map(|m| m.len())
    }

    /// Journal-compaction throttle (G2): rewrite only when the journal holds
    /// at least [`JOURNAL_COMPACT_MIN_EXCESS`] more lines than the retained
    /// set AND the rewrite would shrink it by ≥25%. Frequent small trims
    /// therefore cost nothing; each compaction rewrites at most the retained
    /// set and is amortized O(1) per append.
    fn maybe_compact_journal(&mut self) {
        if self.journal_path.is_none() {
            return;
        }
        let excess = self.journal_lines.saturating_sub(self.events.len());
        if excess >= JOURNAL_COMPACT_MIN_EXCESS && excess.saturating_mul(4) >= self.journal_lines {
            self.compact_journal();
        }
    }

    /// Rewrite the JSONL journal to contain exactly the currently-retained
    /// events (G2 journal compaction — before this, retention trimmed the
    /// in-memory log only and the journal grew unbounded). Atomic: writes a
    /// sibling temp file and renames it over the journal. The background
    /// writer is joined first (draining its backlog and closing its handle —
    /// renaming under a live append-mode handle would orphan subsequent
    /// writes to the old inode), then respawned on the compacted file.
    ///
    /// Hash chaining (A9) survives: [`Self::verify_chain`] anchors the first
    /// hashed event on its *stored* `prev_hash`, so the retained tail of a
    /// chained log still verifies after a compact + reload. Corollary: a
    /// head-trim by retention is indistinguishable from compaction — tamper
    /// evidence covers the retained tail only.
    ///
    /// Returns `true` if the journal was rewritten. Failure is best-effort
    /// like the journal itself: a warning is logged, the old (uncompacted)
    /// journal stays in place, and appending resumes against it.
    pub fn compact_journal(&mut self) -> bool {
        let Some(path) = self.journal_path.clone() else {
            return false;
        };
        // Join the writer so pending lines are flushed and its handle closed.
        self.journal = None;
        let tmp = path.with_extension("compact-tmp");
        let rewrite = (|| -> std::io::Result<()> {
            {
                let file = fs::File::create(&tmp)?;
                let mut w = BufWriter::new(file);
                for ev in &self.events {
                    let line = serde_json::to_string(ev).map_err(std::io::Error::other)?;
                    writeln!(w, "{line}")?;
                }
                w.flush()?;
            }
            fs::rename(&tmp, &path)
        })();
        let ok = match rewrite {
            Ok(()) => {
                self.journal_lines = self.events.len();
                true
            }
            Err(e) => {
                let _ = fs::remove_file(&tmp);
                tracing::warn!(
                    path = %path.display(), error = %e,
                    "car-eventlog: journal compaction failed — journal keeps growing until the next successful compaction"
                );
                false
            }
        };
        self.journal = Some(JournalWriter::spawn(path));
        ok
    }

    /// Run a structured audit [`EventQuery`], returning matching events
    /// most-recent-first, capped at `query.limit` (EPIC G / G2).
    pub fn query(&self, query: &EventQuery) -> Vec<&Event> {
        let mut out: Vec<&Event> = self.events.iter().filter(|e| query.matches(e)).collect();
        out.reverse(); // most recent first for audit review
        if let Some(limit) = query.limit.filter(|l| *l > 0) {
            out.truncate(limit);
        }
        out
    }

    /// Install an auto-retention policy (EPIC G / G2). `max_events` is then
    /// enforced on every `append`; call [`Self::enforce_retention`] to also
    /// apply the age bound.
    pub fn set_retention(&mut self, policy: Option<RetentionPolicy>) {
        self.retention = policy;
    }

    /// The active retention policy, if any.
    pub fn retention(&self) -> Option<&RetentionPolicy> {
        self.retention.as_ref()
    }

    /// Apply a retention policy now: drop events older than `max_age_secs`
    /// and cap the count at `max_events` (keeping the most recent). Returns
    /// the number of events removed. Independent of the installed policy, so a
    /// caller can run a one-off sweep. When a journal is configured, a trim
    /// also triggers the throttled journal compaction (see
    /// [`Self::compact_journal`]) so the JSONL file tracks retention instead
    /// of growing unbounded.
    pub fn enforce_retention(&mut self, policy: &RetentionPolicy, now: DateTime<Utc>) -> usize {
        let before = self.events.len();
        if let Some(age) = policy.max_age_secs {
            let cutoff = now - chrono::Duration::seconds(age);
            self.events.retain(|e| e.timestamp >= cutoff);
        }
        if let Some(max) = policy.max_events {
            truncate_vec_keep_last(&mut self.events, max);
        }
        let removed = before.saturating_sub(self.events.len());
        self.trimmed_events += removed as u64;
        if removed > 0 {
            self.maybe_compact_journal();
        }
        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_values().map(|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);
                }
            }
        }

        // If the loaded tail is chained, keep chaining ENABLED and continue
        // from the last hash. Restoring `last_hash` but leaving chaining off
        // (the pre-fix behaviour) permanently broke the chain: one unchained
        // append before a manual re-enable left a gap that made every future
        // `verify_chain` report tampering, with no repair path (review C-9b).
        let last_hash = events.last().and_then(|e| e.hash.clone());
        let hash_chaining = last_hash.is_some();
        // Seed the monotonic counters from what the journal preserved: the
        // cumulative cost restarts from the journaled spend (G1), and the
        // journal line count from the parsed events (unparseable lines are
        // undercounted — the first compaction re-baselines exactly).
        let cumulative_cost_usd = events.iter().filter_map(Event::cost_usd).sum();
        let journal_lines = events.len();
        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())),
            hash_chaining,
            last_hash,
            retention: None,
            journal_path: Some(path.to_path_buf()),
            journal_lines,
            trimmed_events: 0,
            cumulative_cost_usd,
        })
    }
}

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 query_filters_by_kind_data_and_time() {
        let mut log = EventLog::new();
        log.append(
            EventKind::PermissionDecision,
            Some("a1"),
            None,
            [
                ("caller".to_string(), Value::from("alice")),
                ("tool".to_string(), Value::from("shell")),
            ]
            .into(),
        );
        log.append(
            EventKind::PermissionDecision,
            Some("a2"),
            None,
            [
                ("caller".to_string(), Value::from("bob")),
                ("tool".to_string(), Value::from("shell")),
            ]
            .into(),
        );
        log.append(
            EventKind::StateChanged,
            Some("a3"),
            None,
            Default::default(),
        );

        // Filter by kind.
        let q = EventQuery {
            kinds: vec![EventKind::PermissionDecision],
            ..Default::default()
        };
        assert_eq!(log.query(&q).len(), 2);

        // Filter by a data field (who ran the tool).
        let q = EventQuery {
            data_matches: [("caller".to_string(), "alice".to_string())].into(),
            ..Default::default()
        };
        let hits = log.query(&q);
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].action_id.as_deref(), Some("a1"));

        // Combined tool + kind.
        let q = EventQuery {
            kinds: vec![EventKind::PermissionDecision],
            data_matches: [("tool".to_string(), "shell".to_string())].into(),
            limit: Some(1),
            ..Default::default()
        };
        // Most-recent-first + limit → the bob decision.
        let hits = log.query(&q);
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].action_id.as_deref(), Some("a2"));
    }

    #[test]
    fn cost_by_agent_folds_metered_events() {
        let mut log = EventLog::new();
        log.append_metered(
            EventKind::InferenceMetered,
            None,
            None,
            [("agent".to_string(), Value::from("researcher"))].into(),
            Metrics {
                tokens_in: Some(100),
                tokens_out: Some(50),
                cost_usd: Some(2.0),
                ..Default::default()
            },
        );
        log.append_metered(
            EventKind::InferenceMetered,
            None,
            None,
            [("agent".to_string(), Value::from("researcher"))].into(),
            Metrics {
                tokens_in: Some(10),
                tokens_out: Some(5),
                cost_usd: Some(0.2),
                ..Default::default()
            },
        );
        log.append_metered(
            EventKind::InferenceMetered,
            None,
            None,
            [("agent".to_string(), Value::from("coordinator"))].into(),
            Metrics {
                cost_usd: Some(0.5),
                ..Default::default()
            },
        );
        let report = log.cost_by_agent();
        assert_eq!(report.len(), 2);
        // BTreeMap order: coordinator, researcher.
        assert_eq!(report[0].agent, "coordinator");
        assert_eq!(report[0].cost_usd, 0.5);
        assert_eq!(report[1].agent, "researcher");
        assert_eq!(report[1].calls, 2);
        assert_eq!(report[1].tokens_in, 110);
        assert_eq!(report[1].tokens_out, 55);
        assert!((report[1].cost_usd - 2.2).abs() < 1e-9);
    }

    #[test]
    fn auto_retention_caps_event_count() {
        let mut log = EventLog::new();
        log.set_retention(Some(RetentionPolicy {
            max_events: Some(3),
            max_age_secs: None,
        }));
        for i in 0..10 {
            log.append(
                EventKind::StateChanged,
                Some(&format!("a{i}")),
                None,
                Default::default(),
            );
        }
        // Only the last 3 survive.
        assert_eq!(log.len(), 3);
        assert_eq!(log.events()[0].action_id.as_deref(), Some("a7"));
        assert_eq!(log.events()[2].action_id.as_deref(), Some("a9"));
    }

    #[test]
    fn enforce_retention_drops_old_by_age() {
        let mut log = EventLog::new();
        // Two events; backdate the first well past the age bound.
        log.append(
            EventKind::StateChanged,
            Some("old"),
            None,
            Default::default(),
        );
        log.events[0].timestamp = Utc::now() - chrono::Duration::seconds(3600);
        log.append(
            EventKind::StateChanged,
            Some("fresh"),
            None,
            Default::default(),
        );

        let removed = log.enforce_retention(
            &RetentionPolicy {
                max_events: None,
                max_age_secs: Some(60),
            },
            Utc::now(),
        );
        assert_eq!(removed, 1);
        assert_eq!(log.len(), 1);
        assert_eq!(log.events()[0].action_id.as_deref(), Some("fresh"));
    }

    #[test]
    fn retention_trims_are_counted() {
        let mut log = EventLog::new();
        log.set_retention(Some(RetentionPolicy {
            max_events: Some(2),
            max_age_secs: None,
        }));
        for i in 0..5 {
            log.append(
                EventKind::StateChanged,
                Some(&format!("a{i}")),
                None,
                Default::default(),
            );
        }
        assert_eq!(log.trimmed_events(), 3);
        assert_eq!(log.truncate_events_keep_last(1), 1);
        assert_eq!(log.trimmed_events(), 4);
        log.clear();
        assert_eq!(log.trimmed_events(), 5);
    }

    #[test]
    fn cumulative_cost_is_monotonic_across_trims_and_reload() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("cost.jsonl");
        {
            let mut log = EventLog::with_journal(journal.clone());
            log.set_retention(Some(RetentionPolicy {
                max_events: Some(1),
                max_age_secs: None,
            }));
            for _ in 0..4 {
                log.append_metered(
                    EventKind::InferenceMetered,
                    None,
                    None,
                    Default::default(),
                    Metrics {
                        cost_usd: Some(2.5),
                        ..Default::default()
                    },
                );
            }
            // Trims dropped 3 metered events; the counter never slid back.
            assert_eq!(log.len(), 1);
            assert!((log.cumulative_cost_usd() - 10.0).abs() < 1e-9);
        }
        // Reload seeds the counter from what the journal preserved (here the
        // journal was never compacted, so the full spend survives).
        let reloaded = EventLog::load(&journal).unwrap();
        assert!((reloaded.cumulative_cost_usd() - 10.0).abs() < 1e-9);
    }

    #[test]
    fn journal_compaction_rewrites_to_retained_set() {
        // Real journal compaction (review G2): once the excess clears the
        // throttle (≥ JOURNAL_COMPACT_MIN_EXCESS lines AND ≥25% shrink), the
        // retention trim rewrites the JSONL file to exactly the retained
        // events instead of letting it grow forever.
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("compact.jsonl");
        let keep = 16usize;
        let total = keep + JOURNAL_COMPACT_MIN_EXCESS + 8;
        {
            let mut log = EventLog::with_journal(journal.clone());
            log.set_retention(Some(RetentionPolicy {
                max_events: Some(keep),
                max_age_secs: None,
            }));
            for i in 0..total {
                log.append(
                    EventKind::ActionSucceeded,
                    Some(&format!("a{i}")),
                    None,
                    HashMap::new(),
                );
            }
            assert_eq!(log.len(), keep);
            assert!(log.journal_size_bytes().unwrap_or(0) > 0);
        } // drop joins the writer → file settled.

        // The journal holds only the retained tail, not all `total` lines.
        let reloaded = EventLog::load(&journal).unwrap();
        assert!(
            reloaded.len() < total,
            "journal must have been compacted (got {} lines)",
            reloaded.len()
        );
        // The newest events survived, contiguously up to the last append.
        assert_eq!(
            reloaded.events().last().unwrap().action_id.as_deref(),
            Some(format!("a{}", total - 1).as_str())
        );
    }

    #[test]
    fn compact_journal_preserves_hash_chain_of_retained_tail() {
        // A9 × G2: verify_chain anchors the first hashed event on its stored
        // prev_hash, so a compacted journal's retained tail must still verify
        // after reload even though the chain's head was dropped.
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("chained.jsonl");
        {
            let mut log = EventLog::with_journal(journal.clone()).with_hash_chaining();
            for i in 0..20 {
                log.append(
                    EventKind::ActionSucceeded,
                    Some(&format!("a{i}")),
                    None,
                    HashMap::new(),
                );
            }
            // Trim to the last 5 and force an (unthrottled) compaction.
            log.truncate_events_keep_last(5);
            assert!(log.compact_journal(), "compaction must succeed");
            // Appends after compaction land in the compacted file and keep
            // chaining from the retained tail.
            log.append(
                EventKind::ActionSucceeded,
                Some("post"),
                None,
                HashMap::new(),
            );
        }
        let reloaded = EventLog::load(&journal).unwrap();
        assert_eq!(reloaded.len(), 6);
        assert_eq!(reloaded.verify_chain(), Ok(6), "retained tail must verify");
        assert_eq!(reloaded.events()[0].action_id.as_deref(), Some("a15"));
        assert_eq!(reloaded.events()[5].action_id.as_deref(), Some("post"));
    }

    #[test]
    fn compact_journal_without_journal_is_noop() {
        let mut log = EventLog::new();
        log.append(EventKind::StateChanged, Some("a"), None, Default::default());
        assert!(!log.compact_journal());
        assert_eq!(log.journal_size_bytes(), None);
    }

    #[test]
    fn chaining_off_by_default_no_hashes() {
        let mut log = EventLog::new();
        log.append(
            EventKind::ActionSucceeded,
            Some("a1"),
            Some("p1"),
            HashMap::new(),
        );
        assert!(!log.hash_chaining_enabled());
        assert!(log.events()[0].hash.is_none());
        assert!(log.events()[0].prev_hash.is_none());
        // verify_chain over an unchained log is vacuously ok (0 verified).
        assert_eq!(log.verify_chain(), Ok(0));
    }

    #[test]
    fn hash_chain_verifies_clean_log() {
        let mut log = EventLog::new().with_hash_chaining();
        for i in 0..5 {
            log.append(
                EventKind::ActionSucceeded,
                Some(&format!("a{i}")),
                Some("p"),
                [("i".to_string(), Value::from(i))].into(),
            );
        }
        // Every event hashed, links join.
        assert!(log.events().iter().all(|e| e.hash.is_some()));
        assert_eq!(log.verify_chain(), Ok(5));
        // First event's prev_hash is the genesis (empty) link.
        assert_eq!(log.events()[0].prev_hash.as_deref(), Some(""));
        // Each subsequent prev_hash equals the prior event's hash.
        for w in log.events().windows(2) {
            assert_eq!(w[1].prev_hash, w[0].hash);
        }
    }

    #[test]
    fn tampering_with_data_breaks_chain() {
        let mut log = EventLog::new().with_hash_chaining();
        for i in 0..4 {
            log.append(
                EventKind::ActionSucceeded,
                Some(&format!("a{i}")),
                Some("p"),
                [("v".to_string(), Value::from(i))].into(),
            );
        }
        assert_eq!(log.verify_chain(), Ok(4));
        // Tamper with event #2's data after the fact.
        log.events[2].data.insert("v".to_string(), Value::from(999));
        // The chain breaks exactly at the edited event.
        assert_eq!(log.verify_chain(), Err(2));
    }

    #[test]
    fn deleting_an_event_breaks_chain() {
        let mut log = EventLog::new().with_hash_chaining();
        for i in 0..4 {
            log.append(
                EventKind::ActionSucceeded,
                Some(&format!("a{i}")),
                Some("p"),
                HashMap::new(),
            );
        }
        // Remove the second event — the next event's prev_hash no longer
        // matches the running hash.
        log.events.remove(1);
        assert_eq!(log.verify_chain(), Err(1));
    }

    #[test]
    fn chain_survives_serialize_roundtrip() {
        let mut log = EventLog::new().with_hash_chaining();
        for i in 0..3 {
            log.append(
                EventKind::PermissionDecision,
                Some(&format!("a{i}")),
                Some("p"),
                [
                    ("decision".to_string(), Value::from("allow")),
                    ("nested".to_string(), serde_json::json!({"z": 1, "a": 2})),
                ]
                .into(),
            );
        }
        // Serialize each event to JSON and back, then re-verify — the
        // digest must be stable across the round-trip.
        let lines: Vec<String> = log
            .events()
            .iter()
            .map(|e| serde_json::to_string(e).unwrap())
            .collect();
        let mut rebuilt = EventLog::new();
        for line in &lines {
            rebuilt.events.push(serde_json::from_str(line).unwrap());
        }
        assert_eq!(rebuilt.verify_chain(), Ok(3));
    }

    #[test]
    fn chain_survives_journal_load_and_append() {
        // Regression (review C-9b): `load` used to restore `last_hash` but
        // hard-set `hash_chaining: false`, so the first append after a load
        // produced an unchained event mid-chain — a permanent, unrepairable
        // verify_chain failure. Loading a chained tail must keep chaining on.
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("chain.jsonl");
        {
            let mut log = EventLog::with_journal(journal.clone());
            log.enable_hash_chaining();
            log.append(EventKind::ActionSucceeded, Some("a1"), None, HashMap::new());
            log.append(EventKind::ActionSucceeded, Some("a2"), None, HashMap::new());
        } // drop joins the writer thread → lines flushed.

        {
            let mut log = EventLog::load(&journal).unwrap();
            assert!(
                log.hash_chaining_enabled(),
                "loading a chained tail re-enables chaining"
            );
            log.append(EventKind::ActionSucceeded, Some("a3"), None, HashMap::new());
            assert_eq!(log.verify_chain(), Ok(3), "post-load append stays chained");
        }

        // And the whole thing still verifies after a second reload.
        let reloaded = EventLog::load(&journal).unwrap();
        assert_eq!(reloaded.len(), 3);
        assert_eq!(reloaded.verify_chain(), Ok(3));

        // An UNCHAINED journal must not turn chaining on.
        let plain = dir.path().join("plain.jsonl");
        {
            let mut log = EventLog::with_journal(plain.clone());
            log.append(EventKind::ActionSucceeded, Some("a1"), None, HashMap::new());
        }
        let loaded = EventLog::load(&plain).unwrap();
        assert!(!loaded.hash_chaining_enabled(), "unchained tail stays off");
    }

    #[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());
    }
}