car_eventlog/

lib.rs

//! 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.

/// How long a test waits for something it does not control to happen: a
/// spawned process, a supervised agent, a file another task writes. The wait
/// ends as soon as the thing happens, so this costs nothing when the test
/// passes; it exists only to end a real hang. A full-suite run on a loaded
/// operator Mac has stretched a supervised agent's start past 10 s (car#3744),
/// and every shorter bound was a test of the machine's load as well as of the
/// behaviour it names (car#3763). `scripts/check-test-wait-bounds.sh` holds
/// test waits to it.
#[cfg(test)]
pub(crate) const LOADED_WAIT: std::time::Duration = std::time::Duration::from_secs(60);

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 car_secrets::{
    atomic_replace_private_file, create_private_file, open_private_append, revalidate_private_file,
    revalidate_private_path,
};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{HashMap, HashSet, VecDeque};
use std::fs;
use std::future::Future;
use std::io::{BufRead, BufReader, BufWriter, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::{mpsc, Arc, Condvar, Mutex, Weak};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::{Duration, Instant};
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, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum EventKind {
    /// The authenticated `runs.start` bracket reached its durable boundary.
    RunStarted,
    /// A body-free authenticated run cancellation request became durable.
    RunCancellationRequested,
    /// A deterministic run cancellation receipt became durable.
    RunCancellationResult,
    ProposalReceived,
    /// A proposal reached its deterministic terminal result. Emitted by the
    /// CAR server after the runtime has emitted every action transition.
    ProposalCompleted,
    ActionValidated,
    ActionRejected,
    ActionExecuting,
    ActionSucceeded,
    ActionFailed,
    ActionSkipped,
    ActionRetrying,
    ActionDeduplicated,
    PolicyViolation,
    StateChanged,
    StateSnapshot,
    /// Proposal-level aggregate of observed state mutations that survived the
    /// transaction boundary. Distinct from provisional per-action
    /// `state_changed` rows and from declared expected effects.
    StateCommitted,
    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,
    // Proactive memory intervention events (arXiv 2607.08716-inspired):
    // Phase 1 maintenance records compact bank edits derived from recent
    // trajectory telemetry; Phase 2 records whether the selector injected a
    // grounded reminder or explicitly remained silent.
    ProactiveMemoryMaintained,
    ProactiveMemoryIntervention,
    // 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,
    // The inference chain served a call on a model other than the preferred
    // candidate, mid-run. `data` carries `from`, `to` and `reason` (one of
    // `credential_rejected` / `credential_absent` / `rate_limited` /
    // `timed_out` / `failed`).
    //
    // car#1333 recorded WHO wrote each turn (`models_served`); this records
    // WHY the backbone changed, which is a different fact. Without it a run
    // that degraded mid-session had nothing on disk explaining a surprising
    // result, and mining would attribute it to the code under test rather than
    // to the model swap (car#1351).
    //
    // One row per HOP of a fallback chain. The live `model_fallback` event is
    // latched once per phase so the stream does not narrate every routing
    // decision; the journal wants each distinct transition, and not the same
    // one re-stated on all fifty turns of a run whose credential stayed dead.
    // Repeats are therefore collapsed CONSECUTIVELY, not globally — a lane
    // that fails, recovers, and fails again later is a new episode, and
    // suppressing it would leave a journal that cannot tell "changed once,
    // early" from "flapped all run".
    //
    // A candidate the ROUTER never offered uses the alternate payload
    // `{lane, reason, until}`: `rate_limited` after a prior 429, or
    // `circuit_open` while its breaker blocks dispatch. `until` is the Unix
    // cooldown end when known and null for a session-long exclusion or an
    // in-flight half-open probe. Those persistent conditions are deduplicated
    // once per lane per coder session rather than consecutively.
    ModelFallback,
    // 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.
    //
    // `data` also carries `reversibility` (reversible / compensable /
    // irreversible) on every variant — the SECOND, independent axis, from
    // car_policy::classify_reversibility. The tier answers "who may
    // authorize this?" and says nothing about whether the effect can be
    // undone: a `git push` and a charged card are both full_access /
    // needs_approval and have different rollback contracts. The gate does
    // not act on this field; it is recorded so an audit can tell those two
    // rows apart without re-deriving the classification later.
    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 model call refused on spend before anything was sent
    // (docs/proposals/spend-caps.md, "The records"). `data` carries `agent`
    // (the paying principal's meter key), `model`, `reason`
    // (`budget_reached` / `price_unknown` / `unpriced_call_in_flight` /
    // `spend_unrecordable`), `cause`,
    // and for a reached cap `cap` (`{usd, per}`), `holding`, `spent_usd`,
    // `resets_at` and `zone`. It carries no metrics: nothing was spent.
    ModelCallRefused,
    // 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,
    // Deterministic goal-loop verifier pass. Emitted after the runtime gathers
    // ground truth and `car-verify` evaluates a goal condition. `data` carries
    // `iteration`, `met`, `grounded`, `reason`, and `model_id`/`model_tier`
    // (local|cloud|unknown) so `/goal` outcomes — and which model tier produced
    // an ungrounded completion — are queryable from the same append-only
    // journal as the tool receipts they depend on.
    GoalEvaluated,
    // Terminal decision of an assistant/coder turn loop. `data` carries
    // `decision` ("empty_tool_calls" | "max_turns" | "stalled"), `stop_reason`,
    // `was_truncated`, and `turns`. The default (goal-less) loop declares success
    // the instant the model emits no tool calls, with no truncation/outcome
    // check — so this makes "why did the loop stop" (a clean finish vs a
    // truncated, turn-capped, or stalled one) queryable from the journal, on the
    // ungrounded default path that emits no `GoalEvaluated`.
    TurnCompleted,
    /// The active `runs.start` bracket reached one terminal state. This is
    /// distinct from `ProposalCompleted`: one run can contain many proposals.
    RunCompleted,
    /// A worker process started, on any start path
    /// (docs/proposals/orchestrator-workers.md). `data` carries `handle`, `at`,
    /// `seq` and `network_fence`: `none` while CAR starts workers without the
    /// network fence (CAR-544), with `warning` saying so, else `available`.
    /// `record_error` is present when the start could not be written to the
    /// worker's start records.
    WorkerStarted,
    /// A session's first read of a worker's start records replayed only the
    /// newest ones. `data` carries `handle`, `omitted` (how many older starts
    /// were not replayed) and `path` (the file that still holds them).
    WorkerStartsOmitted,
}

/// 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, schemars::JsonSchema)]
pub struct Event {
    pub kind: EventKind,
    /// Authenticated active-run identity stamped by CAR at append time.
    /// Historical journals omit this field and replay as `None`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub run_id: Option<String>,
    /// WebSocket client identity that opened `run_id` via `runs.start`.
    /// Never reconstructed from the journal filename during replay.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub client_id: Option<String>,
    /// CAR-minted policy-session identity used for this proposal, when the
    /// caller selected a live `session.policy.open` session. Unvalidated
    /// caller labels are never copied here.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub policy_session_id: Option<String>,
    #[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).
///
/// A row marked [`SUMMARY_OF`] is skipped whole: the calls it summarizes are
/// rows of their own in the same log, so summing it too would count them
/// twice.
pub fn metrics_totals_of(events: &[Event]) -> MetricsTotals {
    let mut totals = MetricsTotals::default();
    for ev in events {
        if ev.data.contains_key(SUMMARY_OF) {
            continue;
        }
        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,
    /// The dollars of the calls whose cost is known ([`Self::priced_calls`]).
    /// Never a stand-in for the rest: an unpriced call adds nothing here and
    /// is counted in [`Self::unpriced_calls`] instead (car#3400).
    pub cost_usd: f64,
    /// Calls with a known cost: a provider's figure, a list price, or a local
    /// model's $0.
    #[serde(default)]
    pub priced_calls: u64,
    /// Calls a subscription covers: no per-call dollars to add.
    #[serde(default)]
    pub subscription_calls: u64,
    /// Calls whose cost could not be established. Their dollars are unknown,
    /// not $0, so a total with any of these is a lower bound.
    #[serde(default)]
    pub unpriced_calls: u64,
    /// Sub-agent self-reports the daemon could not tie to one run
    /// ([`SUMMARY_AMBIGUOUS`]): kept and listed apart, never part of
    /// [`Self::calls`] or [`Self::cost_usd`], whose measured `infer` rows
    /// already count those calls once (car#3492).
    #[serde(default)]
    pub ambiguous_reports: u64,
    /// The dollars those [`Self::ambiguous_reports`] reported: what the
    /// runners said, not measured, and not added to [`Self::cost_usd`].
    #[serde(default)]
    pub ambiguous_reported_cost_usd: f64,
}

/// The [`AgentCost::agent`] of calls that name no agent: spend nobody is
/// recorded as paying, kept on its own line and never merged into an agent's
/// total (car#3400).
pub const UNATTRIBUTED_AGENT: &str = "unattributed";

/// The `data` key that marks an `InferenceMetered` row as a **summary** of
/// calls recorded on other rows of the same log; its value names the kind of
/// row that holds them (`"infer"`). The daemon marks a `multi.*` sub-agent
/// row so when that run's calls went through its `infer`: the daemon makes
/// no model call for a sub-agent, the client does, and each of the client's
/// calls that went through the daemon is its own `infer` row with its own
/// `call_id`. [`cost_by_agent_of`], [`metrics_totals_of`] and the harness
/// `model_calls` count skip a marked row, so a payer's calls are counted
/// once, on the rows that recorded them; the summary still carries its
/// runner's figures as metrics and in `reported_tokens_in`,
/// `reported_tokens_out` and `reported_cost_usd`. A row without the key
/// folds as it always has: every row an older CAR wrote, a daemon multi row
/// for a run that made no `infer` call, and a car-multi row written by an
/// embedder with no daemon meter — each the only record of its calls.
pub const SUMMARY_OF: &str = "summary_of";

/// The [`SUMMARY_OF`] value of a `multi.*` sub-agent row whose calls may or
/// may not be among the payer's `infer` rows: an untagged `infer` row arrived
/// while another callback on the session was open, so it could be either
/// run's (car#3492). [`cost_by_agent_of`] lists the row apart, in
/// [`AgentCost::ambiguous_reports`], and adds nothing of it to the measured
/// totals; [`metrics_totals_of`] skips it like any summary. No cost is
/// dropped and none is counted twice.
pub const SUMMARY_AMBIGUOUS: &str = "ambiguous";

/// 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 [`UNATTRIBUTED_AGENT`]; a row marked [`SUMMARY_OF`] is skipped,
/// because the calls it summarizes are rows of their own. Ordered by agent name
/// (BTreeMap) so the report is deterministic. Under the daemon's spend scope
/// `agent` is the paying principal's meter key (`agent:<id>`, `operator`), so
/// the report answers "how much did each principal spend"; a `multi.*`
/// sub-agent's row names the sub-agent in `sub_agent` beside it. Joined with
/// the `tools`/`workflow` provenance fields the emit sites stamp, it is 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 summary = e.data.get(SUMMARY_OF);
        let ambiguous = summary.and_then(Value::as_str) == Some(SUMMARY_AMBIGUOUS);
        if summary.is_some() && !ambiguous {
            continue;
        }
        let agent = e
            .data
            .get("agent")
            .and_then(|v| v.as_str())
            .unwrap_or(UNATTRIBUTED_AGENT)
            .to_string();
        let entry = map.entry(agent.clone()).or_insert_with(|| AgentCost {
            agent,
            ..Default::default()
        });
        if ambiguous {
            entry.ambiguous_reports += 1;
            entry.ambiguous_reported_cost_usd += e.cost_usd().unwrap_or(0.0);
            continue;
        }
        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));
        // The spend record's `cost_source` says which; a row without one (an
        // older CAR, an embedder with no meter) is priced when it has a figure.
        match (
            e.data.get("cost_source").and_then(|v| v.as_str()),
            e.cost_usd(),
        ) {
            (Some("subscription"), _) => entry.subscription_calls += 1,
            (Some("unknown"), _) | (None, None) => entry.unpriced_calls += 1,
            (_, cost) => {
                entry.priced_calls += 1;
                entry.cost_usd += cost.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.
enum JournalMessage {
    Async(String),
    Critical {
        line: String,
        known_existing: bool,
        ack: JournalAcknowledgement,
    },
    #[cfg(test)]
    Shutdown,
}

/// Hard upper bound for one asynchronous critical-journal acknowledgement.
/// Callers may choose a shorter deadline, but never create a timer entry that
/// lives longer than this process-wide contract.
pub const MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT: Duration = Duration::from_secs(30);

const DEFAULT_CRITICAL_ACKNOWLEDGEMENT_CAPACITY: usize = 64;

/// Deterministic failure seam for durable-journal tests. Production callers
/// use the default empty queue; embedders may inject one failure at an exact
/// write/flush/fsync boundary without replacing the filesystem.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JournalFailurePoint {
    AsyncWrite,
    Write,
    Flush,
    Fsync,
    /// Accept and durably write a critical row, but retain its acknowledgement.
    /// This models the ambiguous boundary where a caller cannot know whether
    /// the writer completed before its acknowledgement deadline.
    HoldAcknowledgement,
}

#[derive(Debug, Clone, Default)]
pub struct JournalFailureInjector {
    failures: Arc<Mutex<VecDeque<JournalFailurePoint>>>,
    held_acknowledgements: Arc<Mutex<Vec<JournalAcknowledgement>>>,
    critical_passes: Arc<Mutex<usize>>,
}

impl JournalFailureInjector {
    pub fn fail_next(&self, point: JournalFailurePoint) {
        self.failures
            .lock()
            .expect("journal failure injector mutex poisoned")
            .push_back(point);
    }

    /// Let the next `count` critical rows reach the writer untouched before
    /// any queued failure applies, so a test can fail the Nth row of a
    /// multi-row lifecycle transaction rather than only the first.
    #[doc(hidden)]
    pub fn pass_next_critical(&self, count: usize) {
        *self
            .critical_passes
            .lock()
            .expect("journal critical-pass mutex poisoned") += count;
    }

    fn take_critical_pass(&self) -> bool {
        let mut passes = self
            .critical_passes
            .lock()
            .expect("journal critical-pass mutex poisoned");
        if *passes == 0 {
            return false;
        }
        *passes -= 1;
        true
    }

    fn take(&self, point: JournalFailurePoint) -> bool {
        let mut failures = self
            .failures
            .lock()
            .expect("journal failure injector mutex poisoned");
        if failures.front() == Some(&point) {
            failures.pop_front();
            true
        } else {
            false
        }
    }

    fn hold_acknowledgement(&self, ack: JournalAcknowledgement) {
        self.held_acknowledgements
            .lock()
            .expect("journal held-acknowledgement mutex poisoned")
            .push(ack);
    }

    /// Number of acknowledgements retained by the explicit
    /// [`JournalFailurePoint::HoldAcknowledgement`] test seam.
    #[doc(hidden)]
    pub fn held_acknowledgement_count(&self) -> usize {
        self.held_acknowledgements
            .lock()
            .expect("journal held-acknowledgement mutex poisoned")
            .len()
    }

    /// Release acknowledgements retained by the explicit stalled-writer test
    /// seam. Production code never arms that seam.
    #[doc(hidden)]
    pub fn release_held_acknowledgements(&self) {
        let acknowledgements: Vec<_> = self
            .held_acknowledgements
            .lock()
            .expect("journal held-acknowledgement mutex poisoned")
            .drain(..)
            .collect();
        for acknowledgement in acknowledgements {
            acknowledgement.send(Ok(()));
        }
    }
}

#[derive(Debug)]
enum CriticalPreAcceptanceError {
    WriterUnavailable,
    WriterStopped,
    CoordinatorUnavailable(String),
    CapacityExhausted { capacity: usize },
    InvalidAcknowledgementTimeout { requested: Duration },
}

impl std::fmt::Display for CriticalPreAcceptanceError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::WriterUnavailable => write!(formatter, "journal writer thread is unavailable"),
            Self::WriterStopped => {
                write!(
                    formatter,
                    "journal writer stopped before accepting critical append"
                )
            }
            Self::CoordinatorUnavailable(reason) => write!(
                formatter,
                "critical acknowledgement coordinator is unavailable: {reason}"
            ),
            Self::CapacityExhausted { capacity } => write!(
                formatter,
                "critical acknowledgement capacity is exhausted ({capacity} in flight)"
            ),
            Self::InvalidAcknowledgementTimeout { requested } => write!(
                formatter,
                "critical acknowledgement timeout must be between 1ns and {}ms, got {}ms",
                MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT.as_millis(),
                requested.as_millis()
            ),
        }
    }
}

#[derive(Debug)]
enum CriticalPostAcceptanceError {
    DurabilityFailure(String),
    AcknowledgementTimedOut { timeout: Duration },
    CoordinatorStopped,
}

impl std::fmt::Display for CriticalPostAcceptanceError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::DurabilityFailure(reason) => write!(formatter, "{reason}"),
            Self::AcknowledgementTimedOut { timeout } => write!(
                formatter,
                "journal writer did not acknowledge within {}ms",
                timeout.as_millis()
            ),
            Self::CoordinatorStopped => {
                write!(
                    formatter,
                    "acknowledgement coordinator stopped after enqueue"
                )
            }
        }
    }
}

struct AsyncAcknowledgementState {
    terminal: bool,
    result: Option<Result<(), CriticalPostAcceptanceError>>,
    waker: Option<Waker>,
}

struct AsyncAcknowledgementEntry {
    id: u64,
    deadline: Instant,
    timeout: Duration,
    manager: Weak<AsyncAcknowledgementManagerInner>,
    state: Mutex<AsyncAcknowledgementState>,
}

impl AsyncAcknowledgementEntry {
    fn complete(&self, result: Result<(), CriticalPostAcceptanceError>) {
        self.complete_deciding(|| result);
    }

    fn complete_writer(&self, result: std::io::Result<()>) {
        self.complete_deciding(|| {
            if Instant::now() >= self.deadline {
                Err(CriticalPostAcceptanceError::AcknowledgementTimedOut {
                    timeout: self.timeout,
                })
            } else {
                result.map_err(|error| {
                    CriticalPostAcceptanceError::DurabilityFailure(error.to_string())
                })
            }
        });
    }

    fn complete_deciding(&self, decide: impl FnOnce() -> Result<(), CriticalPostAcceptanceError>) {
        let waker = {
            let mut state = self
                .state
                .lock()
                .expect("journal async-acknowledgement mutex poisoned");
            if state.terminal {
                return;
            }
            state.terminal = true;
            state.result = Some(decide());
            state.waker.take()
        };
        if let Some(manager) = self.manager.upgrade() {
            manager.remove(self.id);
        }
        if let Some(waker) = waker {
            waker.wake();
        }
    }

    fn cancel_preacceptance(&self) {
        {
            let mut state = self
                .state
                .lock()
                .expect("journal async-acknowledgement mutex poisoned");
            if state.terminal {
                return;
            }
            state.terminal = true;
        }
        if let Some(manager) = self.manager.upgrade() {
            manager.remove(self.id);
        }
    }
}

struct AsyncAcknowledgement {
    entry: Arc<AsyncAcknowledgementEntry>,
}

impl Future for AsyncAcknowledgement {
    type Output = Result<(), CriticalPostAcceptanceError>;

    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        let mut state = self
            .entry
            .state
            .lock()
            .expect("journal async-acknowledgement mutex poisoned");
        match state.result.take() {
            Some(result) => Poll::Ready(result),
            None => {
                state.waker = Some(context.waker().clone());
                Poll::Pending
            }
        }
    }
}

struct AsyncAcknowledgementManagerState {
    entries: HashMap<u64, Arc<AsyncAcknowledgementEntry>>,
    next_id: u64,
    shutting_down: bool,
}

struct AsyncAcknowledgementManagerInner {
    capacity: usize,
    state: Mutex<AsyncAcknowledgementManagerState>,
    changed: Condvar,
    #[cfg(test)]
    expiry_barrier: Mutex<Option<AsyncAcknowledgementExpiryBarrier>>,
}

#[cfg(test)]
#[derive(Clone)]
struct AsyncAcknowledgementExpiryBarrier {
    removed: Arc<std::sync::Barrier>,
    release: Arc<std::sync::Barrier>,
}

#[cfg(test)]
impl AsyncAcknowledgementExpiryBarrier {
    fn new() -> Self {
        Self {
            removed: Arc::new(std::sync::Barrier::new(2)),
            release: Arc::new(std::sync::Barrier::new(2)),
        }
    }

    fn pause_after_removal(&self) {
        self.removed.wait();
        self.release.wait();
    }

    fn wait_until_removed(&self) {
        self.removed.wait();
    }

    fn allow_timeout_completion(&self) {
        self.release.wait();
    }
}

impl AsyncAcknowledgementManagerInner {
    fn remove(&self, id: u64) {
        let removed = self
            .state
            .lock()
            .expect("journal acknowledgement-manager mutex poisoned")
            .entries
            .remove(&id)
            .is_some();
        if removed {
            self.changed.notify_all();
        }
    }
}

struct AsyncAcknowledgementManager {
    inner: Arc<AsyncAcknowledgementManagerInner>,
    worker: Mutex<Option<thread::JoinHandle<()>>>,
}

impl AsyncAcknowledgementManager {
    fn new(capacity: usize) -> Self {
        Self {
            inner: Arc::new(AsyncAcknowledgementManagerInner {
                capacity,
                state: Mutex::new(AsyncAcknowledgementManagerState {
                    entries: HashMap::new(),
                    next_id: 0,
                    shutting_down: false,
                }),
                changed: Condvar::new(),
                #[cfg(test)]
                expiry_barrier: Mutex::new(None),
            }),
            worker: Mutex::new(None),
        }
    }

    #[cfg(test)]
    fn pause_next_expiry_after_removal(&self, barrier: AsyncAcknowledgementExpiryBarrier) {
        *self
            .inner
            .expiry_barrier
            .lock()
            .expect("journal acknowledgement expiry-barrier mutex poisoned") = Some(barrier);
    }

    fn ensure_worker(&self) -> Result<(), CriticalPreAcceptanceError> {
        let mut worker = self
            .worker
            .lock()
            .expect("journal acknowledgement-worker mutex poisoned");
        if worker.is_some() {
            return Ok(());
        }
        let inner = self.inner.clone();
        let handle = thread::Builder::new()
            .name("car-eventlog-critical-ack".into())
            .spawn(move || async_acknowledgement_timer_loop(inner))
            .map_err(|error| {
                CriticalPreAcceptanceError::CoordinatorUnavailable(error.to_string())
            })?;
        *worker = Some(handle);
        Ok(())
    }

    fn reserve(
        &self,
        timeout: Duration,
    ) -> Result<AsyncAcknowledgementReservation, CriticalPreAcceptanceError> {
        if timeout.is_zero() || timeout > MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT {
            return Err(CriticalPreAcceptanceError::InvalidAcknowledgementTimeout {
                requested: timeout,
            });
        }
        self.ensure_worker()?;
        let mut state = self
            .inner
            .state
            .lock()
            .expect("journal acknowledgement-manager mutex poisoned");
        if state.shutting_down {
            return Err(CriticalPreAcceptanceError::CoordinatorUnavailable(
                "coordinator is shutting down".to_string(),
            ));
        }
        if state.entries.len() >= self.inner.capacity {
            return Err(CriticalPreAcceptanceError::CapacityExhausted {
                capacity: self.inner.capacity,
            });
        }
        let id = loop {
            let candidate = state.next_id;
            state.next_id = state.next_id.wrapping_add(1);
            if !state.entries.contains_key(&candidate) {
                break candidate;
            }
        };
        let entry = Arc::new(AsyncAcknowledgementEntry {
            id,
            deadline: Instant::now() + timeout,
            timeout,
            manager: Arc::downgrade(&self.inner),
            state: Mutex::new(AsyncAcknowledgementState {
                terminal: false,
                result: None,
                waker: None,
            }),
        });
        state.entries.insert(id, entry.clone());
        drop(state);
        self.inner.changed.notify_all();
        Ok(AsyncAcknowledgementReservation {
            entry,
            preaccepted: true,
        })
    }

    fn shutdown(&self) {
        let entries = {
            let mut state = self
                .inner
                .state
                .lock()
                .expect("journal acknowledgement-manager mutex poisoned");
            state.shutting_down = true;
            let entries = state
                .entries
                .drain()
                .map(|(_, entry)| entry)
                .collect::<Vec<_>>();
            self.inner.changed.notify_all();
            entries
        };
        for entry in entries {
            entry.complete(Err(CriticalPostAcceptanceError::CoordinatorStopped));
        }
        if let Some(worker) = self
            .worker
            .lock()
            .expect("journal acknowledgement-worker mutex poisoned")
            .take()
        {
            let _ = worker.join();
        }
    }
}

fn async_acknowledgement_timer_loop(inner: Arc<AsyncAcknowledgementManagerInner>) {
    loop {
        let expired = {
            let mut state = inner
                .state
                .lock()
                .expect("journal acknowledgement-manager mutex poisoned");
            loop {
                if state.shutting_down {
                    return;
                }
                let now = Instant::now();
                let expired_ids: Vec<_> = state
                    .entries
                    .iter()
                    .filter_map(|(id, entry)| (entry.deadline <= now).then_some(*id))
                    .collect();
                if !expired_ids.is_empty() {
                    break expired_ids
                        .into_iter()
                        .filter_map(|id| state.entries.remove(&id))
                        .collect::<Vec<_>>();
                }
                if let Some(deadline) = state.entries.values().map(|entry| entry.deadline).min() {
                    let wait = deadline.saturating_duration_since(now);
                    let (next, _) = inner
                        .changed
                        .wait_timeout(state, wait)
                        .expect("journal acknowledgement-manager mutex poisoned");
                    state = next;
                } else {
                    state = inner
                        .changed
                        .wait(state)
                        .expect("journal acknowledgement-manager mutex poisoned");
                }
            }
        };
        #[cfg(test)]
        if !expired.is_empty() {
            if let Some(barrier) = inner
                .expiry_barrier
                .lock()
                .expect("journal acknowledgement expiry-barrier mutex poisoned")
                .take()
            {
                barrier.pause_after_removal();
            }
        }
        for entry in expired {
            entry.complete(Err(CriticalPostAcceptanceError::AcknowledgementTimedOut {
                timeout: entry.timeout,
            }));
        }
    }
}

struct AsyncAcknowledgementReservation {
    entry: Arc<AsyncAcknowledgementEntry>,
    preaccepted: bool,
}

impl AsyncAcknowledgementReservation {
    fn sender(&self) -> AsyncAcknowledgementSender {
        AsyncAcknowledgementSender {
            entry: self.entry.clone(),
        }
    }

    fn into_future(mut self) -> AsyncAcknowledgement {
        self.preaccepted = false;
        AsyncAcknowledgement {
            entry: self.entry.clone(),
        }
    }
}

impl Drop for AsyncAcknowledgementReservation {
    fn drop(&mut self) {
        if self.preaccepted {
            self.entry.cancel_preacceptance();
        }
    }
}

struct AsyncAcknowledgementSender {
    entry: Arc<AsyncAcknowledgementEntry>,
}

impl AsyncAcknowledgementSender {
    fn send(&self, result: std::io::Result<()>) {
        self.entry.complete_writer(result);
    }
}

enum JournalAcknowledgement {
    Sync(mpsc::SyncSender<std::io::Result<()>>),
    Async(AsyncAcknowledgementSender),
}

impl JournalAcknowledgement {
    fn send(&self, result: std::io::Result<()>) {
        match self {
            Self::Sync(sender) => {
                let _ = sender.send(result);
            }
            Self::Async(sender) => sender.send(result),
        }
    }
}

impl std::fmt::Debug for JournalAcknowledgement {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Sync(_) => formatter.write_str("JournalAcknowledgement::Sync"),
            Self::Async(_) => formatter.write_str("JournalAcknowledgement::Async"),
        }
    }
}

/// Failure from a bounded critical append. A durability-unknown result retains
/// the exact serialized row and is safe to retry with the same lifecycle
/// identity and data.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CriticalAppendError {
    /// The append was rejected before an acknowledgement wait could begin.
    Rejected { reason: String },
    /// The writer accepted the request, but durable completion was not known
    /// before the acknowledgement bound. The exact row remains pending.
    DurabilityUnknown { reason: String },
}

impl CriticalAppendError {
    pub fn is_retry_safe(&self) -> bool {
        matches!(self, Self::DurabilityUnknown { .. })
    }
}

impl std::fmt::Display for CriticalAppendError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Rejected { reason } => {
                write!(formatter, "critical journal append rejected: {reason}")
            }
            Self::DurabilityUnknown { reason } => write!(
                formatter,
                "critical journal durability is unknown; retry the exact event safely: {reason}"
            ),
        }
    }
}

impl std::error::Error for CriticalAppendError {}

struct JournalWriter {
    /// `None` only if the writer thread could not be spawned (journaling then
    /// silently disabled — still best-effort).
    tx: Option<mpsc::Sender<JournalMessage>>,
    handle: Option<thread::JoinHandle<()>>,
    acknowledgements: AsyncAcknowledgementManager,
}

impl JournalWriter {
    fn spawn(path: PathBuf) -> Self {
        Self::spawn_with_injectors(path, JournalFailureInjector::default(), None)
    }

    fn spawn_with_injector(path: PathBuf, failures: JournalFailureInjector) -> Self {
        Self::spawn_with_injectors(path, failures, None)
    }

    fn spawn_with_private_path_injector(
        path: PathBuf,
        failures: car_secrets::PrivatePathDurabilityFailureInjector,
    ) -> Self {
        Self::spawn_with_injectors(path, JournalFailureInjector::default(), Some(failures))
    }

    fn spawn_with_injectors(
        path: PathBuf,
        failures: JournalFailureInjector,
        private_path_failures: Option<car_secrets::PrivatePathDurabilityFailureInjector>,
    ) -> Self {
        Self::spawn_with_injectors_and_ack_capacity(
            path,
            failures,
            private_path_failures,
            DEFAULT_CRITICAL_ACKNOWLEDGEMENT_CAPACITY,
        )
    }

    fn spawn_with_injectors_and_ack_capacity(
        path: PathBuf,
        failures: JournalFailureInjector,
        private_path_failures: Option<car_secrets::PrivatePathDurabilityFailureInjector>,
        acknowledgement_capacity: usize,
    ) -> Self {
        let (tx, rx) = mpsc::channel::<JournalMessage>();
        let acknowledgements = AsyncAcknowledgementManager::new(acknowledgement_capacity);
        match thread::Builder::new()
            .name("car-eventlog-journal".into())
            .spawn(move || journal_loop(path, rx, failures, private_path_failures))
        {
            Ok(handle) => Self {
                tx: Some(tx),
                handle: Some(handle),
                acknowledgements,
            },
            // 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,
                    acknowledgements,
                }
            }
        }
    }

    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(JournalMessage::Async(line));
        }
    }

    fn enqueue_critical_sync(
        &self,
        line: String,
        known_existing: bool,
    ) -> Result<mpsc::Receiver<std::io::Result<()>>, CriticalPreAcceptanceError> {
        let tx = self
            .tx
            .as_ref()
            .ok_or(CriticalPreAcceptanceError::WriterUnavailable)?;
        let (ack_tx, ack_rx) = mpsc::sync_channel(0);
        tx.send(JournalMessage::Critical {
            line,
            known_existing,
            ack: JournalAcknowledgement::Sync(ack_tx),
        })
        .map_err(|_| CriticalPreAcceptanceError::WriterStopped)?;
        Ok(ack_rx)
    }

    fn reserve_async_acknowledgement(
        &self,
        acknowledgement_timeout: Duration,
    ) -> Result<AsyncAcknowledgementReservation, CriticalPreAcceptanceError> {
        if self.tx.is_none() {
            return Err(CriticalPreAcceptanceError::WriterUnavailable);
        }
        self.acknowledgements.reserve(acknowledgement_timeout)
    }

    fn enqueue_critical_async(
        &self,
        line: String,
        known_existing: bool,
        reservation: AsyncAcknowledgementReservation,
    ) -> Result<AsyncAcknowledgement, CriticalPreAcceptanceError> {
        let tx = self
            .tx
            .as_ref()
            .ok_or(CriticalPreAcceptanceError::WriterUnavailable)?;
        tx.send(JournalMessage::Critical {
            line,
            known_existing,
            ack: JournalAcknowledgement::Async(reservation.sender()),
        })
        .map_err(|_| CriticalPreAcceptanceError::WriterStopped)?;
        Ok(reservation.into_future())
    }

    #[cfg(test)]
    fn remove_sender_for_test(&mut self) {
        self.tx.take();
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }

    #[cfg(test)]
    fn stop_receiver_for_test(&mut self) {
        if let Some(tx) = &self.tx {
            let _ = tx.send(JournalMessage::Shutdown);
        }
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }
}

impl Drop for JournalWriter {
    fn drop(&mut self) {
        self.acknowledgements.shutdown();
        // 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).
///
/// The file is opened **lazily on the first line to write**, not at thread
/// start. A session that never appends an event — health checks, heartbeats,
/// `agents.list` polls, and every other no-op connection — then leaves no
/// journal behind. Opening eagerly created a 0-byte `<client_id>.jsonl` per
/// connection that accumulated without bound (177K empties observed on a
/// long-lived daemon). Sessions that DO log are unaffected: the file is created
/// on their first event exactly as before.
fn journal_loop(
    path: PathBuf,
    rx: mpsc::Receiver<JournalMessage>,
    failures: JournalFailureInjector,
    private_path_failures: Option<car_secrets::PrivatePathDurabilityFailureInjector>,
) {
    let mut writer: Option<std::fs::File> = None;
    let mut written_critical_lines = HashSet::new();
    let mut blocked_critical: Option<String> = None;
    // Rows whose first asynchronous write failed remain ahead of the next
    // critical boundary. Rows received after a failed critical boundary stay
    // behind that exact row. Keeping the queues separate preserves the
    // producer order across retries.
    let mut failed_async = VecDeque::new();
    let mut after_blocked_critical = VecDeque::new();
    while let Ok(message) = rx.recv() {
        let existed_before_open = path.exists();
        let open = || match private_path_failures.as_ref() {
            Some(failures) => {
                car_secrets::open_private_append_with_failure_injector(&path, failures)
            }
            None => open_private_append(&path),
        };
        match message {
            #[cfg(test)]
            JournalMessage::Shutdown => break,
            JournalMessage::Async(line) => {
                if blocked_critical.is_some() {
                    after_blocked_critical.push_back(line);
                    continue;
                }
                if !failed_async.is_empty() {
                    failed_async.push_back(line);
                    continue;
                }
                if writer.is_none() {
                    writer = open().ok();
                }
                let result = match writer.as_mut() {
                    Some(file) => append_journal_line(
                        &path,
                        file,
                        &line,
                        false,
                        false,
                        existed_before_open,
                        &failures,
                        &mut written_critical_lines,
                    ),
                    None => Err(std::io::Error::other("cannot open journal file")),
                };
                if let Err(error) = result {
                    failed_async.push_back(line);
                    tracing::warn!(path = %path.display(), %error, "car-eventlog: asynchronous journal append failed and is awaiting ordered retry");
                }
                continue;
            }
            JournalMessage::Critical {
                line,
                known_existing,
                ack,
            } => {
                if blocked_critical
                    .as_deref()
                    .is_some_and(|pending| pending != line)
                {
                    ack.send(Err(std::io::Error::new(
                        std::io::ErrorKind::WouldBlock,
                        "another critical journal row is awaiting durability",
                    )));
                    continue;
                }
                if writer.is_none() {
                    writer = open().ok();
                }

                // A critical acknowledgement is an ordered durability
                // barrier. Repair every earlier asynchronous row first; if
                // any row is still unwritable, reserve this exact critical
                // line and fail closed instead of creating an audit gap.
                while let Some(pending) = failed_async.front() {
                    let replay = match writer.as_mut() {
                        Some(file) => append_journal_line(
                            &path,
                            file,
                            pending,
                            false,
                            false,
                            existed_before_open,
                            &failures,
                            &mut written_critical_lines,
                        ),
                        None => Err(std::io::Error::other("cannot open journal file")),
                    };
                    match replay {
                        Ok(()) => {
                            failed_async.pop_front();
                        }
                        Err(error) => {
                            blocked_critical = Some(line.clone());
                            ack.send(Err(std::io::Error::new(
                                error.kind(),
                                format!("prior asynchronous journal row is not durable: {error}"),
                            )));
                            break;
                        }
                    }
                }
                if !failed_async.is_empty() {
                    continue;
                }
                let line_failures = if failures.take_critical_pass() {
                    JournalFailureInjector::default()
                } else {
                    failures.clone()
                };
                let result = match writer.as_mut() {
                    Some(file) => append_journal_line(
                        &path,
                        file,
                        &line,
                        true,
                        known_existing,
                        existed_before_open,
                        &line_failures,
                        &mut written_critical_lines,
                    ),
                    None => Err(std::io::Error::other("cannot open journal file")),
                };
                match result {
                    Ok(()) => {
                        blocked_critical = None;
                        while let Some(queued) = after_blocked_critical.pop_front() {
                            if let Some(file) = writer.as_mut() {
                                if let Err(error) = append_journal_line(
                                    &path,
                                    file,
                                    &queued,
                                    false,
                                    false,
                                    true,
                                    &failures,
                                    &mut written_critical_lines,
                                ) {
                                    failed_async.push_back(queued);
                                    failed_async.append(&mut after_blocked_critical);
                                    tracing::warn!(path = %path.display(), %error, "car-eventlog: queued asynchronous append failed after critical recovery and is awaiting ordered retry");
                                    break;
                                }
                            }
                        }
                        if line_failures.take(JournalFailurePoint::HoldAcknowledgement) {
                            failures.hold_acknowledgement(ack);
                        } else {
                            ack.send(Ok(()));
                        }
                    }
                    Err(error) => {
                        blocked_critical = Some(line);
                        ack.send(Err(error));
                    }
                }
                continue;
            }
        };
    }
    if let Some(mut writer) = writer {
        let _ = writer.flush();
    }
}

fn append_journal_line(
    path: &Path,
    file: &mut std::fs::File,
    line: &str,
    critical: bool,
    known_existing: bool,
    existed_before_open: bool,
    failures: &JournalFailureInjector,
    written_critical_lines: &mut HashSet<String>,
) -> std::io::Result<()> {
    revalidate_private_path(path, file)?;
    let already_written = known_existing || written_critical_lines.contains(line);
    if !already_written {
        if (!critical && failures.take(JournalFailurePoint::AsyncWrite))
            || (critical && failures.take(JournalFailurePoint::Write))
        {
            return Err(std::io::Error::from_raw_os_error(28)); // ENOSPC
        }
        let mut bytes = Vec::with_capacity(line.len() + 2);
        let len = file.seek(SeekFrom::End(0))?;
        if len > 0 {
            file.seek(SeekFrom::End(-1))?;
            let mut tail = [0u8; 1];
            file.read_exact(&mut tail)?;
            if tail[0] != b'\n' {
                bytes.push(b'\n');
            }
        }
        bytes.extend_from_slice(line.as_bytes());
        bytes.push(b'\n');
        if let Err(error) = file.write_all(&bytes) {
            // `write_all` may have written a prefix before returning an
            // error. Restore the exact pre-row boundary so retry cannot leave
            // a truncated JSON fragment or duplicate suffix in the journal.
            let _ = file.set_len(len);
            let _ = file.seek(SeekFrom::End(0));
            return Err(error);
        }
        if critical {
            written_critical_lines.insert(line.to_string());
        }
    }
    if critical && failures.take(JournalFailurePoint::Flush) {
        return Err(std::io::Error::other("injected journal flush failure"));
    }
    file.flush()?;
    if critical {
        if failures.take(JournalFailurePoint::Fsync) {
            return Err(std::io::Error::other("injected journal fsync failure"));
        }
        file.sync_all()?;
        if !existed_before_open {
            sync_journal_parent(path)?;
        }
    }
    revalidate_private_path(path, file)
}

#[cfg(not(target_os = "windows"))]
fn sync_journal_parent(path: &Path) -> std::io::Result<()> {
    if let Some(parent) = path.parent() {
        std::fs::File::open(parent)?.sync_all()?;
    }
    Ok(())
}

#[cfg(target_os = "windows")]
fn sync_journal_parent(_path: &Path) -> std::io::Result<()> {
    // Windows has no portable directory-fsync primitive: FlushFileBuffers on
    // a directory handle returns ERROR_ACCESS_DENIED on supported NTFS
    // runners. open_private_append already completes the platform-specific
    // parent metadata boundary before returning the newly-created file.
    Ok(())
}

/// 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,
    /// Live producer binding for CAR-owned journal appends. This is runtime
    /// state, not replay state: loading historical rows never fabricates an
    /// active run from whatever happens to be at the journal tail.
    active_binding: Option<EventBinding>,
    /// Exact serialized critical events whose first durability attempt failed.
    /// A retry reuses the same timestamp/bytes and asks the writer to finish
    /// flush+fsync instead of minting a conflicting duplicate terminal.
    critical_pending: HashSet<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct EventBinding {
    run_id: String,
    client_id: String,
    policy_session_id: Option<String>,
}

struct PreparedCriticalAppend {
    existing_index: Option<usize>,
    event: Option<Event>,
    line: String,
    known_existing: bool,
}

/// 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,
    run_id: Option<&str>,
    client_id: Option<&str>,
    policy_session_id: Option<&str>,
    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());
    // Preserve historical hashes byte-for-byte when every binding field is
    // absent. Bound v0.51 events add one domain-separated identity segment.
    if run_id.is_some() || client_id.is_some() || policy_session_id.is_some() {
        hasher.update(b"\x1d");
        hasher.update(run_id.unwrap_or("").as_bytes());
        hasher.update(b"\x1f");
        hasher.update(client_id.unwrap_or("").as_bytes());
        hasher.update(b"\x1f");
        hasher.update(policy_session_id.unwrap_or("").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.
///
/// Fields also accept the camelCase spellings `index.d.ts` documented
/// (`maxEvents`, `maxAgeSecs`). Unknown keys are still ignored rather than
/// refused, so a caller sending harmless extras keeps working; before the
/// aliases a camelCase policy deserialized to "keep everything" (car#3395).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct RetentionPolicy {
    /// Cap the in-memory event count; on overflow the oldest are dropped.
    #[serde(default, alias = "maxEvents")]
    pub max_events: Option<usize>,
    /// Drop events older than this many seconds when `enforce_retention` runs.
    #[serde(default, alias = "maxAgeSecs")]
    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.
///
/// The multi-word fields also accept the camelCase spellings `index.d.ts`
/// documented (`actionId`, `proposalId`, `dataMatches`). Unknown keys are
/// ignored, so before the aliases a camelCase filter silently matched every
/// event (car#3395).
#[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, alias = "actionId")]
    pub action_id: Option<String>,
    /// Exact match on `proposal_id`.
    #[serde(default, alias = "proposalId")]
    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, alias = "dataMatches")]
    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,
            active_binding: None,
            critical_pending: HashSet::new(),
        }
    }

    pub fn with_journal(path: PathBuf) -> Self {
        let mut log = Self::with_lazy_journal(path.clone());
        log.journal = Some(JournalWriter::spawn(path));
        log
    }

    /// Create an empty journal-backed log without opening the file or spawning
    /// its writer until the first critical row actually needs an append.
    /// Startup reconciliation uses this for a missing per-client journal so a
    /// scan never owns one writer per historical client.
    pub fn with_lazy_journal(path: PathBuf) -> Self {
        Self {
            events: Vec::new(),
            spans: Vec::new(),
            journal: None,
            hash_chaining: false,
            last_hash: None,
            retention: None,
            journal_path: Some(path),
            journal_lines: 0,
            trimmed_events: 0,
            cumulative_cost_usd: 0.0,
            active_binding: None,
            critical_pending: HashSet::new(),
        }
    }

    /// Test/embedder seam for deterministic journal write/flush/fsync faults.
    pub fn with_journal_failure_injector(path: PathBuf, failures: JournalFailureInjector) -> Self {
        let mut log = Self::with_journal(path.clone());
        log.journal = Some(JournalWriter::spawn_with_injector(path, failures));
        log
    }

    #[cfg(test)]
    fn with_journal_failure_injector_and_ack_capacity(
        path: PathBuf,
        failures: JournalFailureInjector,
        acknowledgement_capacity: usize,
    ) -> Self {
        let mut log = Self::with_journal(path.clone());
        log.journal = Some(JournalWriter::spawn_with_injectors_and_ack_capacity(
            path,
            failures,
            None,
            acknowledgement_capacity,
        ));
        log
    }

    /// Test/embedder seam for deterministic first-use directory-entry faults.
    pub fn with_private_path_failure_injector(
        path: PathBuf,
        failures: car_secrets::PrivatePathDurabilityFailureInjector,
    ) -> Self {
        let mut log = Self::with_journal(path.clone());
        log.journal = Some(JournalWriter::spawn_with_private_path_injector(
            path, failures,
        ));
        log
    }

    /// Bind this log to one authenticated active run. Exact repeat binding is
    /// idempotent; a different run/client is rejected instead of silently
    /// re-attributing later action events.
    pub fn bind_run(&mut self, run_id: &str, client_id: &str) -> Result<(), String> {
        if run_id.is_empty() || client_id.is_empty() {
            return Err("active journal binding requires non-empty run_id and client_id".into());
        }
        match &self.active_binding {
            Some(binding) if binding.run_id == run_id && binding.client_id == client_id => Ok(()),
            Some(binding) => Err(format!(
                "journal is already bound to run_id `{}` and client_id `{}`",
                binding.run_id, binding.client_id
            )),
            None => {
                self.active_binding = Some(EventBinding {
                    run_id: run_id.to_string(),
                    client_id: client_id.to_string(),
                    policy_session_id: None,
                });
                Ok(())
            }
        }
    }

    /// Attach a CAR-minted policy session to the currently bound proposal.
    pub fn bind_policy_session(&mut self, policy_session_id: &str) -> Result<(), String> {
        if policy_session_id.is_empty() {
            return Err("policy_session_id must be non-empty".into());
        }
        let binding = self
            .active_binding
            .as_mut()
            .ok_or_else(|| "cannot bind a policy session without an active run".to_string())?;
        match binding.policy_session_id.as_deref() {
            Some(existing) if existing != policy_session_id => Err(format!(
                "journal proposal is already bound to policy_session_id `{existing}`"
            )),
            _ => {
                binding.policy_session_id = Some(policy_session_id.to_string());
                Ok(())
            }
        }
    }

    pub fn clear_policy_session(&mut self, policy_session_id: &str) -> Result<(), String> {
        let binding = self
            .active_binding
            .as_mut()
            .ok_or_else(|| "cannot clear a policy session without an active run".to_string())?;
        if binding.policy_session_id.as_deref() != Some(policy_session_id) {
            return Err("policy_session_id does not match the active journal binding".into());
        }
        binding.policy_session_id = None;
        Ok(())
    }

    pub fn clear_run_binding(&mut self, run_id: &str, client_id: &str) -> Result<(), String> {
        let binding = self
            .active_binding
            .as_ref()
            .ok_or_else(|| "journal has no active run binding".to_string())?;
        if binding.run_id != run_id || binding.client_id != client_id {
            return Err("run_id/client_id does not match the active journal binding".into());
        }
        if binding.policy_session_id.is_some() {
            return Err(
                "cannot clear an active run while a proposal policy session is bound".into(),
            );
        }
        self.active_binding = None;
        Ok(())
    }

    pub fn active_run_binding(&self) -> Option<(&str, &str, Option<&str>)> {
        self.active_binding.as_ref().map(|binding| {
            (
                binding.run_id.as_str(),
                binding.client_id.as_str(),
                binding.policy_session_id.as_deref(),
            )
        })
    }

    /// 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 binding = self.active_binding.as_ref();
            let h = event_digest(
                &prev,
                &kind,
                binding.map(|b| b.run_id.as_str()),
                binding.map(|b| b.client_id.as_str()),
                binding.and_then(|b| b.policy_session_id.as_deref()),
                action_id,
                proposal_id,
                &data,
                &timestamp,
            );
            self.last_hash = Some(h.clone());
            (Some(prev), Some(h))
        } else {
            (None, None)
        };
        let event = Event {
            kind,
            run_id: self.active_binding.as_ref().map(|b| b.run_id.clone()),
            client_id: self.active_binding.as_ref().map(|b| b.client_id.clone()),
            policy_session_id: self
                .active_binding
                .as_ref()
                .and_then(|b| b.policy_session_id.clone()),
            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()
    }

    fn prepare_critical_append(
        &mut self,
        kind: EventKind,
        action_id: Option<&str>,
        proposal_id: Option<&str>,
        data: HashMap<String, Value>,
    ) -> Result<PreparedCriticalAppend, String> {
        let binding = self.active_binding.as_ref().ok_or_else(|| {
            "critical lifecycle event requires an authenticated run binding".to_string()
        })?;
        let existing = self.events.iter().position(|event| {
            event.kind == kind
                && event.run_id.as_deref() == Some(binding.run_id.as_str())
                && event.client_id.as_deref() == Some(binding.client_id.as_str())
                && event.policy_session_id.as_deref() == binding.policy_session_id.as_deref()
                && event.action_id.as_deref() == action_id
                && event.proposal_id.as_deref() == proposal_id
                && event.data == data
        });
        if existing.is_none() && !self.critical_pending.is_empty() {
            return Err(
                "another critical lifecycle event is awaiting an exact durability retry".into(),
            );
        }

        if let Some(index) = existing {
            let line = serde_json::to_string(&self.events[index]).map_err(|e| e.to_string())?;
            return Ok(PreparedCriticalAppend {
                existing_index: Some(index),
                event: None,
                known_existing: !self.critical_pending.contains(&line),
                line,
            });
        }

        let timestamp = Utc::now();
        let (prev_hash, hash) = if self.hash_chaining {
            let prev = self.last_hash.clone().unwrap_or_default();
            let hash = event_digest(
                &prev,
                &kind,
                Some(binding.run_id.as_str()),
                Some(binding.client_id.as_str()),
                binding.policy_session_id.as_deref(),
                action_id,
                proposal_id,
                &data,
                &timestamp,
            );
            (Some(prev), Some(hash))
        } else {
            (None, None)
        };
        let event = Event {
            kind,
            run_id: Some(binding.run_id.clone()),
            client_id: Some(binding.client_id.clone()),
            policy_session_id: binding.policy_session_id.clone(),
            action_id: action_id.map(str::to_string),
            proposal_id: proposal_id.map(str::to_string),
            data,
            timestamp,
            prev_hash,
            hash,
        };
        let line = serde_json::to_string(&event).map_err(|error| error.to_string())?;
        Ok(PreparedCriticalAppend {
            existing_index: None,
            event: Some(event),
            line,
            known_existing: false,
        })
    }

    fn commit_prepared_critical(&mut self, prepared: PreparedCriticalAppend) -> (usize, String) {
        let index = match prepared.existing_index {
            Some(index) => index,
            None => {
                let event = prepared
                    .event
                    .expect("new critical append must carry its prepared event");
                if self.hash_chaining {
                    self.last_hash = event.hash.clone();
                }
                self.events.push(event);
                self.journal_lines += 1;
                self.events.len() - 1
            }
        };
        (index, prepared.line)
    }

    /// Append a lifecycle-critical event and return only after its exact JSONL
    /// row and all earlier queued rows have been flushed and fsynced. If a
    /// write/flush/fsync attempt fails, the exact event remains pending in
    /// memory so an identical retry finishes the same row rather than minting
    /// a second terminal with a new timestamp.
    ///
    /// This compatibility API performs an unbounded blocking acknowledgement
    /// wait and is intended only for genuinely synchronous callers. Async
    /// callers must use [`Self::append_critical_async`] so a stalled filesystem
    /// cannot occupy an executor worker indefinitely.
    pub fn append_critical(
        &mut self,
        kind: EventKind,
        action_id: Option<&str>,
        proposal_id: Option<&str>,
        data: HashMap<String, Value>,
    ) -> Result<&Event, String> {
        if self.journal.is_none() {
            return Err("critical lifecycle event requires an enabled journal".to_string());
        }
        let prepared = self.prepare_critical_append(kind, action_id, proposal_id, data)?;
        let acknowledgement = self
            .journal
            .as_ref()
            .expect("journal presence checked above")
            .enqueue_critical_sync(prepared.line.clone(), prepared.known_existing)
            .map_err(|error| error.to_string())?;
        let (index, line) = self.commit_prepared_critical(prepared);
        self.critical_pending.insert(line.clone());
        let result = acknowledgement
            .recv()
            .map_err(|_| {
                std::io::Error::new(
                    std::io::ErrorKind::BrokenPipe,
                    "journal writer stopped before critical acknowledgement",
                )
            })
            .and_then(|result| result);
        match result {
            Ok(()) => {
                self.critical_pending.remove(&line);
                Ok(&self.events[index])
            }
            Err(error) => {
                self.critical_pending.insert(line);
                Err(format!("critical journal append was not durable: {error}"))
            }
        }
    }

    /// Synchronous critical append with a hard acknowledgement bound.
    ///
    /// This is the startup-thread counterpart of [`Self::append_critical_async`].
    /// The exact serialized row is installed in `critical_pending` before the
    /// bounded wait begins, so timeout cannot claim success or authorize a
    /// different lifecycle event. An identical later startup replay safely
    /// reconciles the row whether or not the writer completed before timeout.
    pub fn append_critical_bounded(
        &mut self,
        kind: EventKind,
        action_id: Option<&str>,
        proposal_id: Option<&str>,
        data: HashMap<String, Value>,
        acknowledgement_timeout: Duration,
    ) -> Result<&Event, CriticalAppendError> {
        if acknowledgement_timeout.is_zero()
            || acknowledgement_timeout > MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT
        {
            return Err(CriticalAppendError::Rejected {
                reason: CriticalPreAcceptanceError::InvalidAcknowledgementTimeout {
                    requested: acknowledgement_timeout,
                }
                .to_string(),
            });
        }
        let prepared = self
            .prepare_critical_append(kind, action_id, proposal_id, data)
            .map_err(|reason| CriticalAppendError::Rejected { reason })?;
        if prepared.known_existing && self.journal.is_none() && self.journal_path.is_some() {
            let (index, _) = self.commit_prepared_critical(prepared);
            return Ok(&self.events[index]);
        }
        if self.journal.is_none() {
            let path = self
                .journal_path
                .clone()
                .ok_or_else(|| CriticalAppendError::Rejected {
                    reason: "critical lifecycle event requires an enabled journal".to_string(),
                })?;
            self.journal = Some(JournalWriter::spawn(path));
        }
        let acknowledgement = self
            .journal
            .as_ref()
            .expect("lazy journal writer installed above")
            .enqueue_critical_sync(prepared.line.clone(), prepared.known_existing)
            .map_err(|error| CriticalAppendError::Rejected {
                reason: error.to_string(),
            })?;
        let (index, line) = self.commit_prepared_critical(prepared);
        self.critical_pending.insert(line.clone());
        let result = match acknowledgement.recv_timeout(acknowledgement_timeout) {
            Ok(result) => result.map_err(|error| error.to_string()),
            Err(mpsc::RecvTimeoutError::Timeout) => Err(format!(
                "journal writer did not acknowledge within {}ms",
                acknowledgement_timeout.as_millis()
            )),
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                Err("journal writer stopped before critical acknowledgement".to_string())
            }
        };
        match result {
            Ok(()) => {
                self.critical_pending.remove(&line);
                Ok(&self.events[index])
            }
            Err(reason) => Err(CriticalAppendError::DurabilityUnknown { reason }),
        }
    }

    /// Append a lifecycle-critical event without blocking an async executor
    /// worker on filesystem acknowledgement.
    ///
    /// The acknowledgement wait is bounded by `acknowledgement_timeout`. Once
    /// the writer accepts the message, the exact serialized row is marked
    /// pending before this future can yield. Timeout, cancellation by an outer
    /// request deadline, writer failure, and acknowledgement-coordinator shutdown
    /// therefore leave an identical retry safe: it reuses the original event
    /// timestamp/hash and the writer suppresses a duplicate row. Until that
    /// exact retry reconciles the pending row, a different critical event is
    /// rejected before enqueue.
    pub async fn append_critical_async(
        &mut self,
        kind: EventKind,
        action_id: Option<&str>,
        proposal_id: Option<&str>,
        data: HashMap<String, Value>,
        acknowledgement_timeout: Duration,
    ) -> Result<&Event, CriticalAppendError> {
        let reservation = self
            .journal
            .as_ref()
            .ok_or_else(|| CriticalAppendError::Rejected {
                reason: "critical lifecycle event requires an enabled journal".to_string(),
            })?
            .reserve_async_acknowledgement(acknowledgement_timeout)
            .map_err(|error| CriticalAppendError::Rejected {
                reason: error.to_string(),
            })?;
        let prepared = self
            .prepare_critical_append(kind, action_id, proposal_id, data)
            .map_err(|reason| CriticalAppendError::Rejected { reason })?;
        let acknowledgement = self
            .journal
            .as_ref()
            .expect("journal presence checked before reservation")
            .enqueue_critical_async(prepared.line.clone(), prepared.known_existing, reservation)
            .map_err(|error| CriticalAppendError::Rejected {
                reason: error.to_string(),
            })?;
        let (index, line) = self.commit_prepared_critical(prepared);

        // This happens before the first `.await`, so dropping the future at an
        // outer Tokio timeout cannot lose the exact retry identity.
        self.critical_pending.insert(line.clone());
        match acknowledgement.await {
            Ok(()) => {
                self.critical_pending.remove(&line);
                Ok(&self.events[index])
            }
            Err(error) => Err(CriticalAppendError::DurabilityUnknown {
                reason: error.to_string(),
            }),
        }
    }

    /// 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.run_id.as_deref(),
                ev.client_id.as_deref(),
                ev.policy_session_id.as_deref(),
                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;
        };
        let compacted_lines: HashSet<String> = self
            .events
            .iter()
            .filter_map(|event| serde_json::to_string(event).ok())
            .collect();
        // Join the writer so pending lines are flushed and its handle closed.
        self.journal = None;
        let file_name = path
            .file_name()
            .and_then(|name| name.to_str())
            .unwrap_or("journal");
        let tmp = path.with_file_name(format!(".{file_name}.compact-{}.tmp", Uuid::new_v4()));
        let rewrite = (|| -> std::io::Result<()> {
            let file = create_private_file(&tmp)?;
            let mut writer = BufWriter::new(file);
            for ev in &self.events {
                let line = serde_json::to_string(ev).map_err(std::io::Error::other)?;
                writeln!(writer, "{line}")?;
            }
            writer.flush()?;
            let file = writer.into_inner().map_err(|error| error.into_error())?;
            file.sync_all()?;
            revalidate_private_file(&file)?;
            drop(file);
            atomic_replace_private_file(&tmp, &path)
        })();
        let ok = match rewrite {
            Ok(()) => {
                self.journal_lines = self.events.len();
                // Compaction fsynced these exact rows before replacing the
                // journal. Reconcile producer-side retry state before the
                // writer respawns so an identical critical retry is treated
                // as an already-written durability barrier, not appended a
                // second time by the fresh writer's empty identity cache.
                self.critical_pending
                    .retain(|line| !compacted_lines.contains(line));
                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> {
        Self::load_with_writer(path, JournalWriter::spawn(path.to_path_buf()))
    }

    /// Load a startup-reconciliation journal without attaching a writer.
    /// Existing durable rows then need no file descriptor or background
    /// thread. A bounded critical append attaches a writer lazily only when an
    /// expected lifecycle row is absent. Torn-tail repair remains enabled: a
    /// journal resumed for append must never retain a partial final record.
    pub fn load_for_reconciliation(path: &Path) -> std::io::Result<Self> {
        Self::load_from_journal(path, None, true)
    }

    /// Load and validate an event log without attaching a writer or modifying
    /// the source journal. An unterminated final row is reported as
    /// [`std::io::ErrorKind::UnexpectedEof`]; callers that own live append
    /// recovery should use [`EventLog::load`] instead.
    pub fn load_read_only(path: &Path) -> std::io::Result<Self> {
        Self::load_from_journal(path, None, false)
    }

    /// Load a journal while installing the deterministic durability-failure
    /// seam for subsequent appends.
    #[doc(hidden)]
    pub fn load_with_journal_failure_injector(
        path: &Path,
        failures: JournalFailureInjector,
    ) -> std::io::Result<Self> {
        Self::load_with_writer(
            path,
            JournalWriter::spawn_with_injector(path.to_path_buf(), failures),
        )
    }

    fn load_with_writer(path: &Path, writer: JournalWriter) -> std::io::Result<Self> {
        Self::load_from_journal(path, Some(writer), true)
    }

    fn load_from_journal(
        path: &Path,
        writer: Option<JournalWriter>,
        repair_torn_tail: bool,
    ) -> std::io::Result<Self> {
        let file = fs::File::open(path)?;
        let mut reader = BufReader::new(file);
        let mut events = Vec::new();
        let mut event_lines = Vec::new();
        let mut line_bytes = Vec::new();
        let mut line_number = 0usize;
        let mut torn_tail_line = None;

        loop {
            line_bytes.clear();
            let bytes_read = reader.read_until(b'\n', &mut line_bytes)?;
            if bytes_read == 0 {
                break;
            }
            line_number += 1;
            let terminated = line_bytes.ends_with(b"\n");
            if line_bytes.iter().all(|byte| byte.is_ascii_whitespace()) {
                if !terminated {
                    torn_tail_line = Some(line_number);
                }
                continue;
            }
            match serde_json::from_slice::<Event>(&line_bytes) {
                Ok(event) => {
                    events.push(event);
                    event_lines.push(line_number);
                    if !terminated {
                        torn_tail_line = Some(line_number);
                    }
                }
                Err(_) if !terminated => {
                    // A process can crash after writing only a prefix of its
                    // final JSONL row. Nothing follows this unterminated tail,
                    // so discard it before any resumed append. A terminated
                    // bad row is authoritative corruption and fails below.
                    torn_tail_line = Some(line_number);
                }
                Err(error) => {
                    return Err(invalid_journal_data(path, line_number, error));
                }
            }
            if !terminated {
                break;
            }
        }
        drop(reader);

        // 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 validated events. A recoverable torn
        // tail is rewritten to exactly this set before loading completes.
        let cumulative_cost_usd = events.iter().filter_map(Event::cost_usd).sum();
        let journal_lines = events.len();
        let loaded = Self {
            events,
            spans: Vec::new(),
            // Live loads journal subsequent appends back to the same file;
            // read-only loads intentionally carry no writer.
            journal: writer,
            hash_chaining,
            last_hash,
            retention: None,
            journal_path: repair_torn_tail.then(|| path.to_path_buf()),
            journal_lines,
            trimmed_events: 0,
            cumulative_cost_usd,
            active_binding: None,
            critical_pending: HashSet::new(),
        };
        if let Err(index) = loaded.verify_chain() {
            let source_line = event_lines.get(index).copied().unwrap_or(index + 1);
            return Err(invalid_journal_data(
                path,
                source_line,
                "hash chain integrity check failed",
            ));
        }
        if let Some(line_number) = torn_tail_line {
            if !repair_torn_tail {
                return Err(torn_journal_tail(path, line_number));
            }
            rewrite_loaded_journal(path, &loaded.events)?;
        }
        Ok(loaded)
    }
}

fn torn_journal_tail(path: &Path, line_number: usize) -> std::io::Error {
    std::io::Error::new(
        std::io::ErrorKind::UnexpectedEof,
        format!(
            "event journal torn tail: path={} line={} reason=unterminated final record",
            path.display(),
            line_number
        ),
    )
}

fn invalid_journal_data(
    path: &Path,
    line_number: usize,
    reason: impl std::fmt::Display,
) -> std::io::Error {
    std::io::Error::new(
        std::io::ErrorKind::InvalidData,
        format!(
            "event journal corruption: path={} line={} reason={reason}",
            path.display(),
            line_number
        ),
    )
}

fn rewrite_loaded_journal(path: &Path, events: &[Event]) -> std::io::Result<()> {
    let file_name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("journal");
    let temp = path.with_file_name(format!(".{file_name}.recover-{}.tmp", Uuid::new_v4()));
    let result = (|| {
        let file = create_private_file(&temp)?;
        let mut output = BufWriter::new(file);
        for event in events {
            let line = serde_json::to_string(event).map_err(std::io::Error::other)?;
            writeln!(output, "{line}")?;
        }
        output.flush()?;
        let file = output.into_inner().map_err(|error| error.into_error())?;
        file.sync_all()?;
        revalidate_private_file(&file)?;
        drop(file);
        atomic_replace_private_file(&temp, path)
    })();
    if result.is_err() {
        let _ = fs::remove_file(&temp);
    }
    result
}

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::*;

    struct ThreadWake(std::thread::Thread);

    impl std::task::Wake for ThreadWake {
        fn wake(self: Arc<Self>) {
            self.0.unpark();
        }

        fn wake_by_ref(self: &Arc<Self>) {
            self.0.unpark();
        }
    }

    fn test_waker() -> std::task::Waker {
        std::task::Waker::from(Arc::new(ThreadWake(std::thread::current())))
    }

    fn block_on_test_future<F: std::future::Future>(future: F) -> F::Output {
        let waker = test_waker();
        let mut context = std::task::Context::from_waker(&waker);
        let mut future = Box::pin(future);
        loop {
            match future.as_mut().poll(&mut context) {
                std::task::Poll::Ready(output) => return output,
                std::task::Poll::Pending => std::thread::park(),
            }
        }
    }

    #[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_keeps_unknown_costs_unknown_and_unattributed_apart() {
        let mut log = EventLog::new();
        let row =
            |log: &mut EventLog, agent: Option<&str>, source: Option<&str>, cost: Option<f64>| {
                let mut data: HashMap<String, Value> = HashMap::new();
                if let Some(agent) = agent {
                    data.insert("agent".into(), Value::from(agent));
                }
                if let Some(source) = source {
                    data.insert("cost_source".into(), Value::from(source));
                }
                log.append_metered(
                    EventKind::InferenceMetered,
                    None,
                    None,
                    data,
                    Metrics {
                        cost_usd: cost,
                        ..Default::default()
                    },
                );
            };
        row(&mut log, Some("agent:a"), Some("list_price"), Some(0.25));
        row(&mut log, Some("agent:a"), Some("subscription"), Some(0.0));
        row(&mut log, Some("agent:a"), Some("unknown"), None);
        row(&mut log, Some("agent:a"), None, None);
        row(&mut log, Some("agent:a"), None, Some(0.5));
        row(&mut log, None, Some("provider"), Some(1.0));
        let report = log.cost_by_agent();
        let a = report.iter().find(|r| r.agent == "agent:a").unwrap();
        assert_eq!(a.calls, 5);
        assert_eq!(a.priced_calls, 2);
        assert_eq!(a.subscription_calls, 1);
        assert_eq!(a.unpriced_calls, 2, "an unknown cost is not a $0 one");
        assert!((a.cost_usd - 0.75).abs() < 1e-9);
        let unattributed = report
            .iter()
            .find(|r| r.agent == UNATTRIBUTED_AGENT)
            .expect("its own line");
        assert_eq!(unattributed.calls, 1);
        assert!((unattributed.cost_usd - 1.0).abs() < 1e-9);
    }

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

    /// A row marked `summary_of` adds nothing to the payer's fold — not a
    /// call, not a token, not a cent — even when it carries metrics, because
    /// the calls it summarizes are rows of their own. A multi row without the
    /// mark (written by an older CAR, or by an embedder whose runner made the
    /// calls itself) folds exactly as before.
    #[test]
    fn cost_by_agent_skips_summary_rows() {
        let mut log = EventLog::new();
        // The payer's own `infer` row.
        log.append_metered(
            EventKind::InferenceMetered,
            None,
            None,
            [
                ("agent".to_string(), Value::from("agent:bot")),
                ("call_id".to_string(), Value::from("call-1")),
            ]
            .into(),
            Metrics::inference(100, 50, Some(2.0)),
        );
        // The daemon's summary of the sub-agent run that made that call.
        log.append_metered(
            EventKind::InferenceMetered,
            None,
            None,
            [
                ("agent".to_string(), Value::from("agent:bot")),
                ("sub_agent".to_string(), Value::from("Researcher")),
                (SUMMARY_OF.to_string(), Value::from("infer")),
                ("reported_cost_usd".to_string(), Value::from(2.0)),
            ]
            .into(),
            Metrics::inference(100, 50, Some(2.0)).with_duration(10.0),
        );
        // An unmarked multi row: an older CAR's, or an embedder's.
        log.append_metered(
            EventKind::InferenceMetered,
            None,
            None,
            [
                ("agent".to_string(), Value::from("agent:bot")),
                ("sub_agent".to_string(), Value::from("Verifier")),
                ("reported_cost_usd".to_string(), Value::from(0.5)),
            ]
            .into(),
            Metrics::inference(10, 5, Some(0.5)),
        );
        let report = log.cost_by_agent();
        assert_eq!(report.len(), 1, "{report:?}");
        let bot = &report[0];
        assert_eq!(bot.agent, "agent:bot");
        assert_eq!((bot.calls, bot.tokens_in, bot.tokens_out), (2, 110, 55));
        assert!((bot.cost_usd - 2.5).abs() < 1e-9, "{bot:?}");
    }

    /// Two overlapping sub-agent runs, each making one real call through
    /// `infer` with no `callback_id`: the daemon cannot tell whose row is
    /// whose, so it marks both self-reports ambiguous. The fold counts the two
    /// measured calls once, and lists the two reports apart: two calls and
    /// $2.00 measured, two ambiguous reports of $2.00 beside them, never four
    /// calls or $4.00 in the total (car#3492).
    #[test]
    fn cost_by_agent_lists_ambiguous_reports_apart_from_measured_calls() {
        let mut log = EventLog::new();
        for (call, run) in [("call-1", "Researcher"), ("call-2", "Verifier")] {
            log.append_metered(
                EventKind::InferenceMetered,
                None,
                None,
                [
                    ("agent".to_string(), Value::from("agent:bot")),
                    ("call_id".to_string(), Value::from(call)),
                ]
                .into(),
                Metrics::inference(100, 50, Some(1.0)),
            );
            log.append_metered(
                EventKind::InferenceMetered,
                None,
                None,
                [
                    ("agent".to_string(), Value::from("agent:bot")),
                    ("sub_agent".to_string(), Value::from(run)),
                    (SUMMARY_OF.to_string(), Value::from(SUMMARY_AMBIGUOUS)),
                    ("reported_cost_usd".to_string(), Value::from(1.0)),
                ]
                .into(),
                Metrics::inference(100, 50, Some(1.0)),
            );
        }
        let report = log.cost_by_agent();
        assert_eq!(report.len(), 1, "{report:?}");
        let bot = &report[0];
        assert_eq!((bot.calls, bot.priced_calls), (2, 2), "{bot:?}");
        assert_eq!((bot.tokens_in, bot.tokens_out), (200, 100), "{bot:?}");
        assert!((bot.cost_usd - 2.0).abs() < 1e-9, "{bot:?}");
        assert_eq!(bot.ambiguous_reports, 2, "{bot:?}");
        assert!(
            (bot.ambiguous_reported_cost_usd - 2.0).abs() < 1e-9,
            "{bot:?}"
        );
        let totals = log.metrics_totals();
        assert!((totals.cost_usd - 2.0).abs() < 1e-9, "{totals:?}");
        assert_eq!(totals.tokens, 300);
    }

    /// `metrics_totals` skips a row marked `summary_of` whole — tokens, cost,
    /// duration and the metered-event count — because the calls it
    /// summarizes are rows of their own in the same log. The row carries the
    /// metrics (a reader of the row sees them); the mark is what makes the
    /// fold skip, not a missing field.
    #[test]
    fn metrics_totals_skip_summary_rows() {
        let mut log = EventLog::new();
        log.append_metered(
            EventKind::InferenceMetered,
            None,
            None,
            [("call_id".to_string(), Value::from("call-1"))].into(),
            Metrics::inference(100, 50, Some(2.0)).with_duration(30.0),
        );
        log.append_metered(
            EventKind::InferenceMetered,
            None,
            None,
            [(SUMMARY_OF.to_string(), Value::from("infer"))].into(),
            Metrics::inference(100, 50, Some(2.0)).with_duration(45.0),
        );
        let t = log.metrics_totals();
        assert_eq!((t.tokens_in, t.tokens_out, t.tokens), (100, 50, 150));
        assert_eq!(t.cost_usd, 2.0);
        assert_eq!(t.duration_ms, 30.0);
        assert_eq!(t.metered_events, 1);
        // The summary's own figures are still on the row.
        assert_eq!(log.events()[1].tokens_in(), Some(100));
        assert_eq!(log.events()[1].cost_usd(), Some(2.0));
    }

    #[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 load_rejects_newline_terminated_corrupt_middle_before_later_terminal() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("corrupt-middle.jsonl");
        let started = serde_json::json!({
            "kind": "run_started",
            "run_id": "run-corrupt",
            "client_id": "client-1",
            "data": {"agent_id": "daily-continuity-newsroom"},
            "timestamp": "2026-08-30T09:30:00Z"
        });
        let completed = serde_json::json!({
            "kind": "run_completed",
            "run_id": "run-corrupt",
            "client_id": "client-1",
            "data": {
                "completion_digest": "must-not-be-trusted",
                "termination": {"kind": "outcome", "status": "success", "outcome": {}}
            },
            "timestamp": "2026-08-30T10:00:00Z"
        });
        fs::write(
            &journal,
            format!("{started}\n{{this-is-not-json}}\n{completed}\n"),
        )
        .unwrap();

        let error = match EventLog::load(&journal) {
            Ok(_) => panic!("newline-terminated middle corruption must fail closed"),
            Err(error) => error,
        };
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(error.to_string().contains("line=2"), "{error}");
    }

    #[test]
    fn load_repairs_crash_torn_final_record_before_append() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("torn-tail.jsonl");
        let started = serde_json::json!({
            "kind": "run_started",
            "run_id": "run-torn",
            "client_id": "client-1",
            "data": {"agent_id": "daily-continuity-newsroom"},
            "timestamp": "2026-08-30T09:30:00Z"
        });
        let mut journal_file = create_private_file(&journal).unwrap();
        write!(journal_file, "{started}\n{{\"kind\":\"run_completed\"").unwrap();
        drop(journal_file);

        {
            let mut loaded = EventLog::load(&journal).expect("torn final row is recoverable");
            assert_eq!(loaded.len(), 1);
            loaded.append(
                EventKind::ProposalReceived,
                None,
                Some("proposal-after-recovery"),
                HashMap::new(),
            );
        }

        let bytes = fs::read_to_string(&journal).unwrap();
        assert!(!bytes.contains("{\"kind\":\"run_completed\""), "{bytes}");
        assert!(bytes.ends_with('\n'));
        let reloaded = EventLog::load(&journal).unwrap();
        assert_eq!(reloaded.len(), 2);
        assert_eq!(reloaded.events()[0].kind, EventKind::RunStarted);
        assert_eq!(reloaded.events()[1].kind, EventKind::ProposalReceived);
    }

    #[test]
    fn load_read_only_rejects_a_torn_tail_without_modifying_the_journal() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("read-only-torn-tail.jsonl");
        let started = serde_json::json!({
            "kind": "run_started",
            "run_id": "run-torn",
            "client_id": "client-1",
            "data": {"agent_id": "daily-continuity-newsroom"},
            "timestamp": "2026-08-30T09:30:00Z"
        });
        fs::write(&journal, format!("{started}\n{{\"kind\":\"run_completed\"")).unwrap();
        let before = fs::read(&journal).unwrap();

        let error = match EventLog::load_read_only(&journal) {
            Ok(_) => panic!("read-only loading must expose a torn final row"),
            Err(error) => error,
        };

        assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof);
        assert!(error.to_string().contains("event journal torn tail"));
        assert!(error.to_string().contains("line=2"));
        assert_eq!(fs::read(&journal).unwrap(), before);
    }

    #[test]
    fn load_rejects_existing_hash_chain_tampering() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("tampered-chain.jsonl");
        {
            let mut log = EventLog::with_journal(journal.clone()).with_hash_chaining();
            log.append(
                EventKind::RunStarted,
                None,
                None,
                [(
                    "agent_id".to_string(),
                    Value::from("daily-continuity-newsroom"),
                )]
                .into(),
            );
            log.append(EventKind::RunCompleted, None, None, HashMap::new());
        }
        let mut rows: Vec<Value> = fs::read_to_string(&journal)
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect();
        rows[0]["data"]["agent_id"] = Value::from("tampered-agent");
        fs::write(
            &journal,
            rows.iter()
                .map(Value::to_string)
                .collect::<Vec<_>>()
                .join("\n")
                + "\n",
        )
        .unwrap();

        let error = match EventLog::load(&journal) {
            Ok(_) => panic!("hash-chain tampering must fail closed during load"),
            Err(error) => error,
        };
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(error.to_string().contains("hash chain"), "{error}");
        assert!(error.to_string().contains("line=1"), "{error}");
    }

    #[cfg(unix)]
    fn unix_mode(path: &Path) -> u32 {
        use std::os::unix::fs::PermissionsExt;
        fs::symlink_metadata(path).unwrap().permissions().mode() & 0o777
    }

    #[cfg(unix)]
    #[test]
    fn car_owned_journal_and_created_parents_are_private() {
        let root = tempfile::tempdir().unwrap();
        let parent = root.path().join("eventlogs").join("session");
        let journal = parent.join("events.jsonl");
        {
            let mut log = EventLog::with_journal(journal.clone());
            log.append(EventKind::StateChanged, Some("a1"), None, HashMap::new());
        }

        assert_eq!(unix_mode(&root.path().join("eventlogs")), 0o700);
        assert_eq!(unix_mode(&parent), 0o700);
        assert_eq!(unix_mode(&journal), 0o600);
    }

    #[cfg(unix)]
    #[test]
    fn append_hardens_preexisting_owned_permissive_journal() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("events.jsonl");
        fs::write(&journal, b"").unwrap();
        fs::set_permissions(&journal, fs::Permissions::from_mode(0o644)).unwrap();
        {
            let mut log = EventLog::with_journal(journal.clone());
            log.append(EventKind::StateChanged, Some("a1"), None, HashMap::new());
        }
        assert_eq!(unix_mode(&journal), 0o600);
    }

    #[cfg(unix)]
    #[test]
    fn journal_refuses_symlink_and_hardlink_destinations() {
        use std::os::unix::fs::symlink;

        let dir = tempfile::tempdir().unwrap();
        let victim = dir.path().join("victim");
        fs::write(&victim, b"unchanged").unwrap();

        for journal in [dir.path().join("symlink"), dir.path().join("hardlink")] {
            if journal.ends_with("symlink") {
                symlink(&victim, &journal).unwrap();
            } else {
                fs::hard_link(&victim, &journal).unwrap();
            }
            {
                let mut log = EventLog::with_journal(journal);
                log.append(EventKind::StateChanged, Some("a1"), None, HashMap::new());
            }
            assert_eq!(fs::read(&victim).unwrap(), b"unchanged");
        }
    }

    #[cfg(unix)]
    #[test]
    fn journal_stops_if_the_opened_path_is_substituted() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("events.jsonl");
        let moved = dir.path().join("moved.jsonl");
        let mut log = EventLog::with_journal(journal.clone());
        log.append(EventKind::StateChanged, Some("first"), None, HashMap::new());

        let deadline = std::time::Instant::now() + crate::LOADED_WAIT;
        while fs::metadata(&journal).map_or(true, |metadata| metadata.len() == 0) {
            assert!(
                std::time::Instant::now() < deadline,
                "first event was not persisted"
            );
            std::thread::yield_now();
        }
        fs::rename(&journal, &moved).unwrap();
        let _substitute = create_private_file(&journal).unwrap();
        log.append(
            EventKind::StateChanged,
            Some("second"),
            None,
            HashMap::new(),
        );
        drop(log);

        assert_eq!(EventLog::load(&moved).unwrap().len(), 1);
        assert_eq!(fs::metadata(&journal).unwrap().len(), 0);
    }

    #[cfg(unix)]
    #[test]
    fn compaction_preserves_private_mode_and_leaves_no_temp_name() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("events.jsonl");
        let mut log = EventLog::with_journal(journal.clone());
        for index in 0..4 {
            log.append(
                EventKind::StateChanged,
                Some(&format!("a{index}")),
                None,
                HashMap::new(),
            );
        }
        log.truncate_events_keep_last(2);
        assert!(log.compact_journal());
        drop(log);

        assert_eq!(unix_mode(&journal), 0o600);
        let names: Vec<_> = fs::read_dir(dir.path())
            .unwrap()
            .map(|entry| entry.unwrap().file_name())
            .collect();
        assert_eq!(names, vec![journal.file_name().unwrap()]);
    }

    #[cfg(unix)]
    #[test]
    fn historical_world_readable_journal_can_be_loaded_without_mutation() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("historical.jsonl");
        let event = Event {
            kind: EventKind::StateChanged,
            run_id: None,
            client_id: None,
            policy_session_id: None,
            action_id: Some("historical".into()),
            proposal_id: None,
            data: HashMap::new(),
            timestamp: Utc::now(),
            prev_hash: None,
            hash: None,
        };
        fs::write(
            &journal,
            format!("{}\n", serde_json::to_string(&event).unwrap()),
        )
        .unwrap();
        fs::set_permissions(&journal, fs::Permissions::from_mode(0o644)).unwrap();

        let loaded = EventLog::load(&journal).unwrap();
        assert_eq!(loaded.len(), 1);
        drop(loaded);
        assert_eq!(unix_mode(&journal), 0o644);
    }

    #[test]
    fn journal_not_created_without_appends() {
        // A session that never logs an event must leave no journal file behind.
        // Eager open created a 0-byte file per connection that accumulated
        // without bound; the writer now opens lazily on the first line.
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("no-events.jsonl");
        {
            let _log = EventLog::with_journal(journal.clone());
            // No append. Drop joins the writer thread, which never opened the
            // file because no line was ever sent.
        }
        assert!(
            !journal.exists(),
            "journal file must not be created when nothing is appended"
        );
    }

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

    #[test]
    fn active_run_binding_stamps_every_new_event_and_rejects_conflicts() {
        let mut log = EventLog::new();
        log.bind_run("run-a", "client-a")
            .expect("first active run binds");
        log.bind_policy_session("policy-session-a")
            .expect("CAR-minted policy session binds inside the run");

        log.append(
            EventKind::ProposalReceived,
            None,
            Some("same-proposal"),
            HashMap::new(),
        );
        log.append(
            EventKind::ActionSucceeded,
            Some("action-a"),
            Some("same-proposal"),
            HashMap::new(),
        );

        for event in log.events() {
            assert_eq!(event.run_id.as_deref(), Some("run-a"));
            assert_eq!(event.client_id.as_deref(), Some("client-a"));
            assert_eq!(event.policy_session_id.as_deref(), Some("policy-session-a"));
        }
        assert!(log.bind_run("run-b", "client-a").is_err());
        assert!(log.bind_run("run-a", "client-b").is_err());
        assert!(log.clear_run_binding("run-b", "client-a").is_err());
        assert_eq!(
            log.active_run_binding(),
            Some(("run-a", "client-a", Some("policy-session-a")))
        );

        log.clear_policy_session("policy-session-a")
            .expect("exact policy session clears");
        log.clear_run_binding("run-a", "client-a")
            .expect("exact active run clears");
        log.append(
            EventKind::ProposalReceived,
            None,
            Some("unbound-legacy"),
            HashMap::new(),
        );
        let legacy = log.events().last().unwrap();
        assert!(legacy.run_id.is_none());
        assert!(legacy.client_id.is_none());
        assert!(legacy.policy_session_id.is_none());
    }

    #[test]
    fn historical_event_without_binding_fields_still_deserializes() {
        let historical = r#"{"kind":"proposal_received","proposal_id":"p-old","data":{},"timestamp":"2026-01-02T03:04:05Z"}"#;
        let event: Event = serde_json::from_str(historical).expect("historical event replays");
        assert!(event.run_id.is_none());
        assert!(event.client_id.is_none());
        assert!(event.policy_session_id.is_none());
    }

    #[test]
    fn async_acknowledgement_timeout_wins_after_expiry_removal() {
        let manager = AsyncAcknowledgementManager::new(1);
        let barrier = AsyncAcknowledgementExpiryBarrier::new();
        manager.pause_next_expiry_after_removal(barrier.clone());
        let reservation = manager.reserve(Duration::from_millis(10)).unwrap();
        let sender = reservation.sender();
        let acknowledgement = reservation.into_future();

        barrier.wait_until_removed();
        sender.send(Ok(()));
        let result = block_on_test_future(acknowledgement);
        barrier.allow_timeout_completion();
        manager.shutdown();

        assert!(matches!(
            result,
            Err(CriticalPostAcceptanceError::AcknowledgementTimedOut { .. })
        ));
    }

    #[test]
    fn async_acknowledgement_capacity_is_atomic_under_concurrent_reservation() {
        let manager = Arc::new(AsyncAcknowledgementManager::new(1));
        let barrier = Arc::new(std::sync::Barrier::new(3));
        let handles: Vec<_> = (0..2)
            .map(|_| {
                let manager = manager.clone();
                let barrier = barrier.clone();
                std::thread::spawn(move || {
                    let reservation = manager.reserve(MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT);
                    barrier.wait();
                    reservation
                })
            })
            .collect();

        barrier.wait();
        let results: Vec<_> = handles
            .into_iter()
            .map(|handle| handle.join().unwrap())
            .collect();
        assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
        assert_eq!(
            results
                .iter()
                .filter(|result| {
                    matches!(
                        result,
                        Err(CriticalPreAcceptanceError::CapacityExhausted { capacity: 1 })
                    )
                })
                .count(),
            1
        );
        drop(results);
        manager.shutdown();
    }

    #[test]
    fn async_acknowledgement_completed_before_future_construction_is_observed() {
        let manager = AsyncAcknowledgementManager::new(1);
        // The longest allowed reservation: success must not depend on the clock.
        let reservation = manager
            .reserve(MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT)
            .unwrap();
        reservation.sender().send(Ok(()));
        let acknowledgement = reservation.into_future();

        assert!(block_on_test_future(acknowledgement).is_ok());
        manager.shutdown();
    }

    #[test]
    fn journal_writer_drop_completes_pending_async_acknowledgement() {
        let dir = tempfile::tempdir().unwrap();
        let failures = JournalFailureInjector::default();
        failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
        let writer = JournalWriter::spawn_with_injector(
            dir.path().join("pending-ack-shutdown.jsonl"),
            failures.clone(),
        );
        let reservation = writer
            .reserve_async_acknowledgement(MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT)
            .unwrap();
        let acknowledgement = writer
            .enqueue_critical_async("{}".to_string(), false, reservation)
            .unwrap();
        let deadline = Instant::now() + crate::LOADED_WAIT;
        while failures.held_acknowledgement_count() == 0 {
            assert!(
                Instant::now() < deadline,
                "writer never retained the pending acknowledgement"
            );
            std::thread::yield_now();
        }

        drop(writer);
        let result = block_on_test_future(acknowledgement);
        assert!(matches!(
            result,
            Err(CriticalPostAcceptanceError::CoordinatorStopped)
        ));
        failures.release_held_acknowledgements();
    }

    #[test]
    fn startup_reconciliation_opens_a_writer_only_for_a_missing_row() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("startup-lazy.jsonl");
        let data = HashMap::from([("completion_digest".to_string(), Value::from("durable"))]);
        let mut original = EventLog::with_journal(path.clone());
        original.bind_run("run", "client").unwrap();
        original
            .append_critical_bounded(
                EventKind::RunCompleted,
                None,
                None,
                data.clone(),
                MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
            )
            .unwrap();
        drop(original);

        let mut resumed = EventLog::load_for_reconciliation(&path).unwrap();
        resumed.bind_run("run", "client").unwrap();
        resumed
            .append_critical_bounded(
                EventKind::RunCompleted,
                None,
                None,
                data,
                MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
            )
            .unwrap();
        assert!(
            resumed.journal.is_none(),
            "a row that survived restart needs no append writer"
        );

        resumed
            .append_critical_bounded(
                EventKind::ProposalCompleted,
                None,
                Some("missing"),
                HashMap::new(),
                MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
            )
            .unwrap();
        assert!(
            resumed.journal.is_some(),
            "a missing row must attach a writer before acknowledging durability"
        );
    }

    #[test]
    fn prepare_failure_releases_async_acknowledgement_capacity() {
        let dir = tempfile::tempdir().unwrap();
        let mut log = EventLog::with_journal_failure_injector_and_ack_capacity(
            dir.path().join("prepare-failure-capacity.jsonl"),
            JournalFailureInjector::default(),
            1,
        );
        let data = HashMap::from([("completion_digest".to_string(), Value::from("7".repeat(64)))]);

        let error = block_on_test_future(log.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            data.clone(),
            Duration::from_secs(1),
        ))
        .expect_err("an unbound critical event must fail during preparation");
        assert!(matches!(error, CriticalAppendError::Rejected { .. }));

        log.bind_run("run-after-prepare-failure", "client-after-prepare-failure")
            .unwrap();
        block_on_test_future(log.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            data,
            MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
        ))
        .expect("the failed preparation must release the only acknowledgement slot");
    }

    #[test]
    fn async_critical_preacceptance_failures_reject_without_fabricating_events() {
        let data = HashMap::from([("completion_digest".to_string(), Value::from("f".repeat(64)))]);

        let mut no_journal = EventLog::new();
        no_journal
            .bind_run("run-no-writer", "client-no-writer")
            .unwrap();
        let error = block_on_test_future(no_journal.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            data.clone(),
            Duration::from_millis(100),
        ))
        .expect_err("a missing writer must reject before acceptance");
        assert!(matches!(error, CriticalAppendError::Rejected { .. }));
        assert!(!error.is_retry_safe());
        assert!(no_journal.events().is_empty());
        assert!(no_journal.critical_pending.is_empty());

        let dir = tempfile::tempdir().unwrap();
        let unavailable_path = dir.path().join("sender-missing.jsonl");
        let mut unavailable = EventLog::with_journal(unavailable_path.clone());
        unavailable
            .bind_run("run-sender-missing", "client-sender-missing")
            .unwrap();
        unavailable
            .journal
            .as_mut()
            .unwrap()
            .remove_sender_for_test();
        let error = block_on_test_future(unavailable.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            data.clone(),
            Duration::from_millis(100),
        ))
        .expect_err("a missing writer sender must reject before event construction");
        assert!(matches!(error, CriticalAppendError::Rejected { .. }));
        assert!(unavailable.events().is_empty());
        assert!(unavailable.critical_pending.is_empty());
        assert!(!unavailable_path.exists());

        let stopped_path = dir.path().join("receiver-stopped.jsonl");
        let mut stopped = EventLog::with_journal(stopped_path.clone());
        stopped
            .bind_run("run-receiver-stopped", "client-receiver-stopped")
            .unwrap();
        stopped.journal.as_mut().unwrap().stop_receiver_for_test();
        let error = block_on_test_future(stopped.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            data,
            Duration::from_millis(100),
        ))
        .expect_err("a stopped writer receiver must reject a failed enqueue");
        assert!(matches!(error, CriticalAppendError::Rejected { .. }));
        assert!(stopped.events().is_empty());
        assert!(stopped.critical_pending.is_empty());
        assert!(!stopped_path.exists());
    }

    #[test]
    fn async_critical_rejects_excessive_acknowledgement_duration_before_enqueue() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("excessive-ack-duration.jsonl");
        let mut log = EventLog::with_journal(path.clone());
        log.bind_run("run-excessive-ack", "client-excessive-ack")
            .unwrap();

        let error = block_on_test_future(log.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            HashMap::new(),
            // short-wait-ok: one past the maximum, to be rejected; nothing waits
            MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT + Duration::from_millis(1),
        ))
        .expect_err("an excessive acknowledgement duration must reject");

        assert!(matches!(error, CriticalAppendError::Rejected { .. }));
        assert!(log.events().is_empty());
        assert!(log.critical_pending.is_empty());
        assert!(!path.exists());
    }

    #[test]
    fn async_critical_capacity_exhaustion_rejects_before_enqueue_and_exact_retry_unblocks() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("critical-ack-capacity.jsonl");
        let failures = JournalFailureInjector::default();
        failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
        let mut log = EventLog::with_journal_failure_injector_and_ack_capacity(
            path.clone(),
            failures.clone(),
            1,
        );
        log.bind_run("run-capacity", "client-capacity").unwrap();
        let first_data =
            HashMap::from([("completion_digest".to_string(), Value::from("1".repeat(64)))]);
        let second_data =
            HashMap::from([("completion_digest".to_string(), Value::from("2".repeat(64)))]);

        let mut first = Box::pin(log.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            first_data.clone(),
            MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
        ));
        let waker = test_waker();
        let mut context = std::task::Context::from_waker(&waker);
        assert!(matches!(
            first.as_mut().poll(&mut context),
            std::task::Poll::Pending
        ));
        drop(first);
        assert_eq!(log.events().len(), 1);
        assert_eq!(log.critical_pending.len(), 1);

        for attempt in 0..100 {
            let error = block_on_test_future(log.append_critical_async(
                EventKind::ProposalCompleted,
                None,
                Some("capacity-rejected"),
                second_data.clone(),
                Duration::from_millis(100),
            ))
            .expect_err("exhausted acknowledgement capacity must reject");
            let CriticalAppendError::Rejected { reason } = error else {
                panic!("attempt {attempt} was not a pre-enqueue rejection");
            };
            assert!(
                reason.contains("acknowledgement capacity is exhausted"),
                "attempt {attempt} bypassed capacity admission: {reason}"
            );
            assert_eq!(log.events().len(), 1);
            assert_eq!(log.critical_pending.len(), 1);
        }

        let hold_deadline = std::time::Instant::now() + crate::LOADED_WAIT;
        while failures.held_acknowledgement_count() == 0 {
            assert!(
                std::time::Instant::now() < hold_deadline,
                "writer never reached the held acknowledgement"
            );
            std::thread::yield_now();
        }
        let pending_line = serde_json::to_string(&log.events()[0]).unwrap();
        assert_eq!(
            fs::read_to_string(&path).unwrap(),
            format!("{pending_line}\n")
        );
        failures.release_held_acknowledgements();

        let original_timestamp = log.events()[0].timestamp;
        let retried = block_on_test_future(log.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            first_data,
            MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
        ))
        .expect("the exact cancelled row must reconcile pending state");
        assert_eq!(retried.timestamp, original_timestamp);
        assert!(log.critical_pending.is_empty());

        block_on_test_future(log.append_critical_async(
            EventKind::ProposalCompleted,
            None,
            Some("capacity-rejected"),
            second_data,
            MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
        ))
        .expect("a distinct row is allowed after exact reconciliation");
        drop(log);

        let loaded = EventLog::load(&path).unwrap();
        assert_eq!(loaded.events().len(), 2);
        assert_eq!(loaded.events()[0].kind, EventKind::RunCompleted);
        assert_eq!(loaded.events()[1].kind, EventKind::ProposalCompleted);
    }

    #[test]
    fn async_critical_never_acknowledged_row_remains_exactly_retryable() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("critical-never-acknowledged.jsonl");
        let failures = JournalFailureInjector::default();
        failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
        let mut log = EventLog::with_journal_failure_injector(path.clone(), failures.clone());
        log.bind_run("run-never-ack", "client-never-ack").unwrap();
        let data = HashMap::from([("completion_digest".to_string(), Value::from("9".repeat(64)))]);

        let error = block_on_test_future(log.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            data.clone(),
            // short-wait-ok: the acknowledgement is held forever; the bounded wait is the outcome under test
            Duration::from_millis(20),
        ))
        .expect_err("the first acknowledgement is retained forever");
        assert!(matches!(
            error,
            CriticalAppendError::DurabilityUnknown { .. }
        ));
        let original = serde_json::to_string(&log.events()[0]).unwrap();
        assert!(log.critical_pending.contains(&original));

        let retried = block_on_test_future(log.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            data,
            MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
        ))
        .expect("the exact row must reconcile without the first acknowledgement");
        assert_eq!(serde_json::to_string(retried).unwrap(), original);
        assert!(log.critical_pending.is_empty());
        assert_eq!(
            failures.held_acknowledgement_count(),
            1,
            "the original acknowledgement must remain unsent"
        );
        drop(log);

        assert_eq!(fs::read_to_string(&path).unwrap(), format!("{original}\n"));
        assert_eq!(EventLog::load(&path).unwrap().events().len(), 1);
    }

    #[test]
    fn bounded_sync_critical_timeout_is_exactly_retryable() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("critical-bounded-sync-timeout.jsonl");
        let failures = JournalFailureInjector::default();
        failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
        let mut log = EventLog::with_journal_failure_injector(path.clone(), failures.clone());
        log.bind_run("run-bounded-sync", "client-bounded-sync")
            .unwrap();
        let data = HashMap::from([("completion_digest".to_string(), Value::from("7".repeat(64)))]);

        let error = log
            .append_critical_bounded(
                EventKind::RunCompleted,
                None,
                None,
                data.clone(),
                // short-wait-ok: the acknowledgement is held; the exact sync bound is the outcome under test
                Duration::from_millis(20),
            )
            .expect_err("the retained acknowledgement must hit the exact sync bound");
        assert_eq!(
            error,
            CriticalAppendError::DurabilityUnknown {
                reason: "journal writer did not acknowledge within 20ms".to_string()
            }
        );
        assert!(error.is_retry_safe());
        let pending = log
            .critical_pending
            .iter()
            .next()
            .expect("the exact timed-out row remains pending")
            .clone();
        assert_eq!(pending, serde_json::to_string(&log.events()[0]).unwrap());

        let retried = log
            .append_critical_bounded(
                EventKind::RunCompleted,
                None,
                None,
                data,
                MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
            )
            .expect("an exact retry must reconcile without the held acknowledgement");
        assert_eq!(serde_json::to_string(retried).unwrap(), pending);
        assert!(log.critical_pending.is_empty());
        assert_eq!(failures.held_acknowledgement_count(), 1);
        drop(log);
        assert_eq!(fs::read_to_string(&path).unwrap(), format!("{pending}\n"));
    }

    #[test]
    fn async_critical_append_bounds_unknown_ack_and_preserves_exact_retry() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("critical-unknown-ack.jsonl");
        let failures = JournalFailureInjector::default();
        failures.fail_next(JournalFailurePoint::HoldAcknowledgement);
        let mut log = EventLog::with_journal_failure_injector(path.clone(), failures.clone());
        log.bind_run("run-unknown-ack", "client-unknown-ack")
            .unwrap();
        let data = HashMap::from([("completion_digest".to_string(), Value::from("e".repeat(64)))]);

        let started = std::time::Instant::now();
        let error = block_on_test_future(log.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            data.clone(),
            // short-wait-ok: the acknowledgement is held; the 40 ms wait is the outcome under test
            std::time::Duration::from_millis(40),
        ))
        .expect_err("a retained acknowledgement must become durability-unknown");
        assert!(
            // short-wait-ok: the 40 ms acknowledgement wait is the feature; the ceiling sits well below the 30 s maximum
            started.elapsed() < std::time::Duration::from_secs(10),
            "the async acknowledgement wait exceeded its bounded allowance"
        );
        assert!(matches!(
            error,
            CriticalAppendError::DurabilityUnknown { .. }
        ));
        assert!(error.is_retry_safe());
        assert!(error
            .to_string()
            .contains("did not acknowledge within 40ms"));

        let pending = log
            .critical_pending
            .iter()
            .next()
            .expect("the exact unacknowledged row remains pending")
            .clone();
        assert_eq!(log.critical_pending.len(), 1);
        assert_eq!(pending, serde_json::to_string(&log.events()[0]).unwrap());
        let disk_row = format!("{pending}\n");
        let writer_deadline = std::time::Instant::now() + crate::LOADED_WAIT;
        while fs::read_to_string(&path).unwrap_or_default() != disk_row {
            assert!(
                std::time::Instant::now() < writer_deadline,
                "the writer never durably accepted the held-acknowledgement row"
            );
            std::thread::yield_now();
        }
        while failures.held_acknowledgement_count() == 0 {
            assert!(
                std::time::Instant::now() < writer_deadline,
                "the writer never retained the late acknowledgement"
            );
            std::thread::yield_now();
        }
        assert_eq!(failures.held_acknowledgement_count(), 1);
        failures.release_held_acknowledgements();
        let original_timestamp = log.events()[0].timestamp;

        let distinct_error = block_on_test_future(log.append_critical_async(
            EventKind::ProposalCompleted,
            None,
            Some("different-terminal"),
            HashMap::new(),
            Duration::from_millis(100),
        ))
        .expect_err("a different critical row must not bypass exact retry");
        assert!(matches!(
            distinct_error,
            CriticalAppendError::Rejected { .. }
        ));
        assert_eq!(log.events().len(), 1);

        let retried = block_on_test_future(log.append_critical_async(
            EventKind::RunCompleted,
            None,
            None,
            data,
            MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
        ))
        .expect("an identical retry must finish the retained row");
        assert_eq!(retried.timestamp, original_timestamp);
        assert!(log.critical_pending.is_empty());

        block_on_test_future(log.append_critical_async(
            EventKind::ProposalCompleted,
            None,
            Some("different-terminal"),
            HashMap::new(),
            MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT,
        ))
        .expect("a different row is allowed after exact retry reconciliation");
        drop(log);

        let loaded = EventLog::load(&path).unwrap();
        assert_eq!(loaded.events().len(), 2);
        assert_eq!(serde_json::to_string(&loaded.events()[0]).unwrap(), pending);
    }

    #[test]
    fn critical_append_failures_retry_same_row_and_fsync_prior_async_events() {
        for point in [
            JournalFailurePoint::Write,
            JournalFailurePoint::Flush,
            JournalFailurePoint::Fsync,
        ] {
            let dir = tempfile::tempdir().unwrap();
            let path = dir.path().join(format!("critical-{point:?}.jsonl"));
            let failures = JournalFailureInjector::default();
            failures.fail_next(point);
            let mut log = EventLog::with_journal_failure_injector(path.clone(), failures);
            log.bind_run("run-critical", "client-critical").unwrap();
            log.append(
                EventKind::ProposalReceived,
                None,
                Some("proposal-critical"),
                HashMap::new(),
            );
            let data =
                HashMap::from([("completion_digest".to_string(), Value::from("a".repeat(64)))]);
            assert!(
                log.append_critical(EventKind::RunCompleted, None, None, data.clone())
                    .is_err(),
                "{point:?} failure must not acknowledge"
            );
            log.append(
                EventKind::ActionSucceeded,
                Some("after-pending-terminal"),
                Some("proposal-critical"),
                HashMap::new(),
            );
            log.append_critical(EventKind::RunCompleted, None, None, data)
                .expect("retry finishes the exact critical row");
            drop(log);

            let loaded = EventLog::load(&path).unwrap();
            let events = loaded.events();
            assert_eq!(events[0].kind, EventKind::ProposalReceived);
            assert_eq!(events[1].kind, EventKind::RunCompleted);
            assert_eq!(events[2].kind, EventKind::ActionSucceeded);
            assert_eq!(
                events
                    .iter()
                    .filter(|event| event.kind == EventKind::RunCompleted)
                    .count(),
                1,
                "{point:?} retry must not duplicate a terminal"
            );
        }
    }

    #[test]
    fn a_critical_pass_moves_a_queued_failure_to_the_next_row() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("critical-pass.jsonl");
        let failures = JournalFailureInjector::default();
        failures.pass_next_critical(1);
        failures.fail_next(JournalFailurePoint::Write);
        let mut log = EventLog::with_journal_failure_injector(path.clone(), failures);
        log.bind_run("run-critical-pass", "client-critical-pass")
            .unwrap();

        log.append_critical(
            EventKind::RunCancellationRequested,
            None,
            None,
            HashMap::new(),
        )
        .expect("the passed row must be written despite the queued failure");
        assert!(
            log.append_critical(EventKind::RunCancellationResult, None, None, HashMap::new())
                .is_err(),
            "the queued failure must apply to the row after the pass"
        );
        drop(log);

        let kinds: Vec<_> = EventLog::load(&path)
            .unwrap()
            .events()
            .iter()
            .map(|event| event.kind.clone())
            .collect();
        assert_eq!(kinds, vec![EventKind::RunCancellationRequested]);
    }

    #[test]
    fn compacting_a_failed_critical_row_makes_its_retry_idempotent() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("critical-compact-retry.jsonl");
        let failures = JournalFailureInjector::default();
        failures.fail_next(JournalFailurePoint::Fsync);
        let mut log =
            EventLog::with_journal_failure_injector(path.clone(), failures).with_hash_chaining();
        log.bind_run("run-critical-compact", "client-critical-compact")
            .unwrap();
        log.append(
            EventKind::ProposalReceived,
            None,
            Some("proposal-critical-compact"),
            HashMap::new(),
        );
        let data = HashMap::from([("completion_digest".to_string(), Value::from("c".repeat(64)))]);

        assert!(
            log.append_critical(EventKind::RunCompleted, None, None, data.clone())
                .is_err(),
            "the injected fsync failure must leave the exact terminal pending"
        );
        assert!(
            log.compact_journal(),
            "compaction persists the in-memory pending terminal"
        );
        log.append_critical(EventKind::RunCompleted, None, None, data)
            .expect("identical retry recognizes the compacted terminal as durable");
        drop(log);

        let loaded = EventLog::load(&path).unwrap();
        assert_eq!(
            loaded
                .events()
                .iter()
                .filter(|event| event.kind == EventKind::RunCompleted)
                .count(),
            1,
            "writer respawn must not duplicate the compacted terminal"
        );
        assert_eq!(loaded.events()[0].kind, EventKind::ProposalReceived);
        assert_eq!(loaded.events()[1].kind, EventKind::RunCompleted);
        assert_eq!(loaded.verify_chain(), Ok(2));
    }

    #[test]
    fn critical_append_cannot_ack_until_failed_prior_async_row_is_replayed() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("prior-async-failure.jsonl");
        let failures = JournalFailureInjector::default();
        // The first failure drops the asynchronous attempt. The second makes
        // the first critical barrier prove that it cannot repair the gap yet.
        failures.fail_next(JournalFailurePoint::AsyncWrite);
        failures.fail_next(JournalFailurePoint::AsyncWrite);
        let mut log = EventLog::with_journal_failure_injector(path.clone(), failures);
        log.bind_run("run-ordered", "client-ordered").unwrap();
        log.append(
            EventKind::ActionSucceeded,
            Some("action-ordered"),
            Some("proposal-ordered"),
            HashMap::new(),
        );
        let data = HashMap::from([("completion_digest".to_string(), Value::from("b".repeat(64)))]);

        assert!(
            log.append_critical(
                EventKind::ProposalCompleted,
                None,
                Some("proposal-ordered"),
                data.clone()
            )
            .is_err(),
            "a terminal must not acknowledge while an earlier row is still missing"
        );
        log.append_critical(
            EventKind::ProposalCompleted,
            None,
            Some("proposal-ordered"),
            data,
        )
        .expect("retry repairs the prior row before acknowledging the terminal");
        drop(log);

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

    #[test]
    fn first_use_parent_sync_failure_blocks_terminal_until_prior_async_replays() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("nested").join("first-use.jsonl");
        let failures = car_secrets::PrivatePathDurabilityFailureInjector::default();
        failures.fail_next(car_secrets::PrivatePathDurabilityFailurePoint::ParentDirectorySync);
        failures.fail_next(car_secrets::PrivatePathDurabilityFailurePoint::ParentDirectorySync);
        let mut log = EventLog::with_private_path_failure_injector(path.clone(), failures);
        log.bind_run("run-first-use", "client-first-use").unwrap();
        log.append(
            EventKind::ActionSucceeded,
            Some("action-first-use"),
            Some("proposal-first-use"),
            HashMap::new(),
        );
        let data = HashMap::from([("completion_digest".to_string(), Value::from("d".repeat(64)))]);

        assert!(log
            .append_critical(
                EventKind::ProposalCompleted,
                None,
                Some("proposal-first-use"),
                data.clone(),
            )
            .is_err());
        log.append_critical(
            EventKind::ProposalCompleted,
            None,
            Some("proposal-first-use"),
            data,
        )
        .expect("retry durably replays async row then exact terminal");
        drop(log);

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