car_server_core/

run_store.rs

//! Disk-backed run-trace store (agent run tracing, U3).
//!
//! Persists the per-run trace stream — the `RunStarted` line, each
//! `RunTurn`, then the terminal `RunEnded`/`Incomplete` — as JSONL so a
//! run survives daemon restarts and stays grouped by run independent of
//! the WS connection that produced it (R4). One file per run:
//!
//! ```text
//! ~/.car/runs/{agent_id}/{run_id}.jsonl
//! ```
//!
//! ## Source-of-truth split
//!
//! This disk store is the source of truth for **replay** (U5):
//! [`RunStore::get_run_trace_page_for`] / [`RunStore::list_runs_page`] read it
//! and work after a restart when memory is empty. The in-memory
//! `RunMeta.turns` buffer (U2) stays the source for the **live** stream
//! (U4) — U3 does not remove it; it mirrors what was recorded onto disk.
//!
//! ## Layout, perms, backup exclusion (R14)
//!
//! Prompts and CLI output may carry secrets, so the `runs/` tree is
//! created `0700` and every file `0600` (Unix). The `runs/` dir is
//! marked backup-excluded — a `.nobackup` marker file plus, on macOS,
//! the `com.apple.metadata:com_apple_backup_excludeItem` xattr — so Time
//! Machine / iCloud don't silently copy plaintext traces off the box.
//!
//! ## Index
//!
//! Each agent has a fixed-record newest-first summary index plus one small
//! summary sidecar per run. `runs.list` seeks directly to its requested page;
//! it never scans every run file. The JSONL trace remains authoritative and
//! startup GC rebuilds the derived index/sidecars so interrupted updates are
//! repaired before the daemon listens. `run_id -> agent_id` resolves by
//! scanning `runs/*/` for the matching `{run_id}.jsonl` (U5's
//! `runs.get_trace` takes only a `run_id`).
//!
//! ## Retention (R6)
//!
//! GC runs on daemon boot ([`RunStore::gc`]): per agent it keeps the **50
//! most recent completed runs** and drops anything older than **30 days**,
//! whichever is more restrictive. A still-in-progress run (no terminal
//! record) is **never** evicted. Both limits are configurable via
//! `~/.car/config.toml` (`[runs] max_per_agent` / `max_age_days`) with
//! that restrictive default. **`0` disables a cap** — see
//! [`RetentionConfig`].
//!
//! Records are appended at turn granularity (not per token) — the same
//! coarse boundary the in-memory buffer uses. A corrupt/partial trailing
//! JSONL line loads the prior valid records rather than failing the whole
//! run (the error-path test).

use car_ir::{ActionProposal, ProposalLineageStatus, ProposalResult};
use car_proto::{RunRecord, RunTermination};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::{HashMap, VecDeque};
use std::fs::File;
use std::io::{BufRead, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Condvar, Mutex, Weak};

#[cfg(not(target_os = "windows"))]
fn sync_directory(path: impl AsRef<Path>) -> std::io::Result<()> {
    File::open(path)?.sync_all()
}

#[cfg(target_os = "windows")]
fn sync_directory(_path: impl AsRef<Path>) -> std::io::Result<()> {
    // Windows FlushFileBuffers rejects directory handles with
    // ERROR_ACCESS_DENIED. File contents are synchronized before each call;
    // there is no supported directory-fsync equivalent to add here.
    Ok(())
}

const COMPLETED_PROPOSAL_OWNER_INDEX_MIGRATION_VERSION: u32 = 2;
const RUN_SUMMARY_KEY_BYTES: usize = 32;
const RUN_SUMMARY_INDEX_RECORD_BYTES: usize = 8 + RUN_SUMMARY_KEY_BYTES;
const RUN_SUMMARY_INDEX_FILE: &str = ".run-summary-index-v2";
const RUN_SUMMARY_SIDECAR_DIR: &str = ".run-summaries-v2";
const RUN_TRACE_CORRUPTION_SIDECAR_DIR: &str = ".run-trace-corruption-v1";
#[cfg(not(test))]
const MAX_RESUMED_PROPOSAL_RECEIPTS_PER_RUN: usize = 1024;
#[cfg(test)]
const MAX_RESUMED_PROPOSAL_RECEIPTS_PER_RUN: usize = 4;
#[cfg(not(test))]
const MAX_RESUMED_PROPOSAL_RECEIPT_BYTES_PER_RUN: u64 = 16 * 1024 * 1024;
#[cfg(test)]
const MAX_RESUMED_PROPOSAL_RECEIPT_BYTES_PER_RUN: u64 = 64 * 1024;

#[derive(Debug, Serialize, Deserialize)]
struct CompletedProposalOwnerIndexMigration {
    version: u32,
}

/// Default per-agent cap: keep the 50 most recent completed runs.
pub const DEFAULT_MAX_RUNS_PER_AGENT: usize = 50;
/// Default age cap: drop completed runs older than 30 days.
pub const DEFAULT_MAX_AGE_DAYS: i64 = 30;

/// Terminal/in-progress status of a run, derived from its records.
///
/// Distinct from the run-level `OutcomeStatus` carried inside a terminal
/// `Outcome` — this is the coarse "what state is this run in?" the run
/// list renders. `Incomplete` is the orphan case (no terminal record + no
/// live harness, R5); `InProgress` is a run still being written.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunStatus {
    /// No terminal record yet — still being written by a live harness.
    InProgress,
    /// `runs.complete` reported a terminal `AgentOutcome`. The concrete
    /// `OutcomeStatus` lives on the `RunEnded` record itself.
    Completed,
    /// CAR confirmed controlled work stopped and committed cancellation.
    Cancelled,
    /// The harness disconnected without reporting an outcome (R5).
    Incomplete,
    /// A durable cancellation request exists without a terminal boundary.
    CancellationPending,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunTraceCorruptionKind {
    MalformedRecord,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunTraceCorruption {
    pub kind: RunTraceCorruptionKind,
    /// One-based JSONL line number.
    pub line: usize,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct RunTraceCorruptionMarker {
    run_id: String,
    agent_id: String,
    corruption: RunTraceCorruption,
}

/// One row of [`RunStore::list_runs`] — the summary U5's `runs.list`
/// returns. Persisted in a sidecar so bounded listing never loads every turn.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunSummary {
    pub run_id: String,
    pub agent_id: String,
    pub intent: String,
    pub started_at: DateTime<Utc>,
    /// When the terminal record was written, if any.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ended_at: Option<DateTime<Utc>>,
    pub status: RunStatus,
    /// Number of `RunTurn` records persisted for this run.
    pub turn_count: usize,
    /// Immutable per-agent keyset used by bounded newest-first pagination.
    /// Zero is reserved for legacy sidecars that startup has not migrated yet.
    #[serde(default)]
    pub sequence: u64,
    /// Present only when a durable newline-terminated record is malformed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub trace_corruption: Option<RunTraceCorruption>,
}

/// Durable preimage for a proposal whose actions already finished but whose
/// `proposal_completed` journal boundary has not yet been acknowledged.
///
/// The payload is deliberately server-owned instead of a `car_proto` run row:
/// it is an outbox transaction, not part of the public run trace. Keeping the
/// exact normal-serde result and its JCS digest lets a retry or daemon restart
/// finish the terminal without dispatching an action twice.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct PendingProposalFinalization {
    pub run_id: String,
    pub client_id: String,
    /// Exact caller-selected policy context. This is kept separately from
    /// `policy_session_id`: an unknown session is deliberately not trusted or
    /// journaled, but an exact retry must still present the same context.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub requested_policy_session_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub policy_session_id: Option<String>,
    pub original_proposal_id: String,
    pub final_proposal_id: String,
    /// Exact caller-supplied JSON before serde fills defaults. This recognizes
    /// an exact retry even when an omitted timestamp is regenerated on decode.
    pub original_submission: Value,
    /// Exact normal-serde proposal CAR accepted and journaled.
    pub original_proposal: ActionProposal,
    /// Exact normal-serde proposal selected by the runtime. This duplicates
    /// the active-v3 `ProposalResult.final_proposal` intentionally: validator
    /// equality makes an outbox edit or partial producer upgrade fail closed.
    pub final_proposal: ActionProposal,
    /// Generation-keyed normal-serde preimages for every accepted lineage
    /// entry, including generation zero when it was admitted.
    #[serde(default)]
    pub accepted_proposal_preimages: Vec<AcceptedProposalPreimage>,
    #[serde(deserialize_with = "deserialize_active_proposal_result")]
    pub proposal_result: ProposalResult,
    pub result_digest: String,
}

/// Durable exact response retained after `proposal_completed` is fsynced.
///
/// Unlike the finalization outbox, this receipt intentionally survives guard
/// cleanup. A same-bound-session retry can therefore recover the typed result
/// without dispatch even if the original JSON-RPC response was lost, guard
/// unlink/fsync failed, or run-trace retention later removed `RunStarted`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CompletedProposalResponse {
    pub finalization: PendingProposalFinalization,
}

/// Content-addressed ownership claim for the retry tuple used before a live
/// run is selected. The claim is deliberately independent of the run trace:
/// completed responses outlive trace retention, and an ownership lookup must
/// read at most this one exact file rather than enumerate every receipt.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct CompletedProposalOwnership {
    run_id: String,
    client_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    requested_policy_session_id: Option<String>,
    original_submission: Value,
}

impl CompletedProposalOwnership {
    fn new(
        run_id: &str,
        client_id: &str,
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> Self {
        Self {
            run_id: run_id.to_string(),
            client_id: client_id.to_string(),
            requested_policy_session_id: requested_policy_session_id.map(str::to_string),
            original_submission: original_submission.clone(),
        }
    }
}

/// Result of atomically reserving one proposal retry tuple before dispatch.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProposalRetryReservation {
    Acquired,
    Existing { run_id: String, client_id: String },
}

/// Durable rollback authority written before a retry owner is reserved.
///
/// If the daemon exits before the execution marker is durable, startup uses
/// this exact preimage to release the owner. Once a marker exists, the marker
/// remains the conservative outcome-unknown guard and startup only removes the
/// stale rollback intent.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ProposalRetryRollback {
    pub run_id: String,
    pub client_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub requested_policy_session_id: Option<String>,
    pub original_submission: Value,
}

impl ProposalRetryRollback {
    fn validate(&self) -> std::io::Result<()> {
        if self.run_id.is_empty()
            || self.client_id.is_empty()
            || self.requested_policy_session_id.as_deref() == Some("")
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "proposal retry rollback requires non-empty run/client/policy identities",
            ));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct ProposalIdClaim {
    run_id: String,
    client_id: String,
    proposal_id: String,
    original_submission: Value,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProposalIdClaimOutcome {
    Acquired,
    ExistingExact,
}

impl CompletedProposalResponse {
    pub fn proposal_result(&self) -> &ProposalResult {
        &self.finalization.proposal_result
    }

    fn validate(&self) -> std::io::Result<()> {
        self.finalization.validate()
    }

    fn execution_marker(&self) -> std::io::Result<ProposalExecutionMarker> {
        let pending = &self.finalization;
        let original_proposal = serde_json::to_value(&pending.original_proposal)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        let marker = ProposalExecutionMarker {
            run_id: pending.run_id.clone(),
            client_id: pending.client_id.clone(),
            requested_policy_session_id: pending.requested_policy_session_id.clone(),
            policy_session_id: pending.policy_session_id.clone(),
            original_proposal_id: pending.original_proposal_id.clone(),
            original_submission: pending.original_submission.clone(),
            original_proposal,
            proposal_digest: proposal_digest(&pending.original_proposal)
                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?,
        };
        marker.validate()?;
        Ok(marker)
    }
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AcceptedProposalPreimage {
    pub generation: u32,
    pub proposal: ActionProposal,
}

fn deserialize_active_proposal_result<'de, D>(deserializer: D) -> Result<ProposalResult, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let value = Value::deserialize(deserializer)?;
    let object = value
        .as_object()
        .ok_or_else(|| serde::de::Error::custom("proposal_result must be an object"))?;
    for field in [
        "proposal_id",
        "original_proposal_id",
        "final_proposal",
        "replan_lineage",
        "results",
        "cost",
    ] {
        if !object.contains_key(field) {
            return Err(serde::de::Error::custom(format!(
                "active v3 proposal_result is missing `{field}`"
            )));
        }
    }
    serde_json::from_value(value).map_err(serde::de::Error::custom)
}

fn is_lowercase_sha256(value: &str) -> bool {
    value.len() == 64
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}

fn proposal_digest(proposal: &ActionProposal) -> Result<String, String> {
    let value = serde_json::to_value(proposal).map_err(|error| error.to_string())?;
    let canonical = car_inference::catalog_identity::canonical_json(&value)?;
    Ok(format!("{:x}", Sha256::digest(canonical.as_bytes())))
}

fn validate_original_submission(
    submission: &Value,
    proposal: &ActionProposal,
) -> std::io::Result<()> {
    let invalid = |message: String| std::io::Error::new(std::io::ErrorKind::InvalidData, message);
    let object = submission.as_object().ok_or_else(|| {
        invalid("active proposal original submission must be an object".to_string())
    })?;
    if object.get("id").and_then(Value::as_str) != Some(proposal.id.as_str()) {
        return Err(invalid(
            "active proposal raw submission requires an exact string id".to_string(),
        ));
    }
    let raw_actions = object.get("actions").ok_or_else(|| {
        invalid("active proposal raw submission is missing its action array".to_string())
    })?;
    let actions: Vec<car_ir::Action> = serde_json::from_value(raw_actions.clone())
        .map_err(|error| invalid(format!("raw submission actions are invalid: {error}")))?;
    if actions != proposal.actions {
        return Err(invalid(
            "raw submission actions do not match the accepted normal-serde proposal".to_string(),
        ));
    }
    if object
        .get("source")
        .is_some_and(|source| source.as_str() != Some(proposal.source.as_str()))
    {
        return Err(invalid(
            "raw submission source does not match the accepted proposal".to_string(),
        ));
    }
    if let Some(context) = object.get("context") {
        let context: HashMap<String, Value> = serde_json::from_value(context.clone())
            .map_err(|error| invalid(format!("raw submission context is invalid: {error}")))?;
        if context != proposal.context {
            return Err(invalid(
                "raw submission context does not match the accepted proposal".to_string(),
            ));
        }
    }
    if let Some(timestamp) = object.get("timestamp") {
        let timestamp: DateTime<Utc> = serde_json::from_value(timestamp.clone())
            .map_err(|error| invalid(format!("raw submission timestamp is invalid: {error}")))?;
        if timestamp != proposal.timestamp {
            return Err(invalid(
                "raw submission timestamp does not match the accepted proposal".to_string(),
            ));
        }
    }
    Ok(())
}

/// Fsynced before runtime dispatch. If CAR restarts with this marker and no
/// exact finalization payload, the action outcome is unknown: the server must
/// quarantine the run instead of guessing that redispatch is safe.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ProposalExecutionMarker {
    pub run_id: String,
    pub client_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub requested_policy_session_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub policy_session_id: Option<String>,
    pub original_proposal_id: String,
    pub original_submission: Value,
    pub original_proposal: Value,
    pub proposal_digest: String,
}

impl ProposalExecutionMarker {
    pub fn validate(&self) -> std::io::Result<()> {
        let invalid =
            |message: String| std::io::Error::new(std::io::ErrorKind::InvalidData, message);
        if self.run_id.is_empty()
            || self.client_id.is_empty()
            || self.original_proposal_id.is_empty()
        {
            return Err(invalid(
                "proposal execution marker requires non-empty run/client/proposal identities"
                    .to_string(),
            ));
        }
        if self.policy_session_id.is_some()
            && self.policy_session_id != self.requested_policy_session_id
        {
            return Err(invalid(
                "authenticated policy session is not the caller-requested policy session"
                    .to_string(),
            ));
        }
        let proposal: ActionProposal = serde_json::from_value(self.original_proposal.clone())
            .map_err(|error| invalid(format!("execution marker proposal is invalid: {error}")))?;
        if proposal.id != self.original_proposal_id {
            return Err(invalid(
                "execution marker proposal id does not match original proposal identity"
                    .to_string(),
            ));
        }
        let digest = proposal_digest(&proposal)
            .map_err(|error| invalid(format!("execution marker proposal JCS failed: {error}")))?;
        if self.proposal_digest != digest || !is_lowercase_sha256(&self.proposal_digest) {
            return Err(invalid(
                "execution marker digest does not bind its exact proposal preimage".to_string(),
            ));
        }
        validate_original_submission(&self.original_submission, &proposal)
    }
}

impl PendingProposalFinalization {
    pub fn validate(&self) -> std::io::Result<()> {
        let invalid =
            |message: String| std::io::Error::new(std::io::ErrorKind::InvalidData, message);
        if self.run_id.is_empty()
            || self.client_id.is_empty()
            || self.original_proposal_id.is_empty()
            || self.final_proposal_id.is_empty()
        {
            return Err(invalid(
                "proposal finalization requires non-empty run/client/proposal identities"
                    .to_string(),
            ));
        }
        if self.policy_session_id.is_some()
            && self.policy_session_id != self.requested_policy_session_id
        {
            return Err(invalid(
                "authenticated pending policy session is not the requested policy session"
                    .to_string(),
            ));
        }
        if self.original_proposal.id != self.original_proposal_id
            || self.proposal_result.original_proposal_id != self.original_proposal_id
        {
            return Err(invalid(
                "proposal finalization original proposal identities do not match".to_string(),
            ));
        }
        if self.final_proposal.id != self.final_proposal_id
            || self.proposal_result.proposal_id != self.final_proposal_id
            || self.proposal_result.final_proposal.as_ref() != Some(&self.final_proposal)
        {
            return Err(invalid(
                "active v3 proposal finalization final proposal preimage/id do not match"
                    .to_string(),
            ));
        }
        validate_original_submission(&self.original_submission, &self.original_proposal)?;

        let result_value = serde_json::to_value(&self.proposal_result)
            .map_err(|error| invalid(format!("proposal result serialization failed: {error}")))?;
        let canonical = car_inference::catalog_identity::canonical_json(&result_value)
            .map_err(|error| invalid(format!("proposal result JCS failed: {error}")))?;
        let result_digest = format!("{:x}", Sha256::digest(canonical.as_bytes()));
        if self.result_digest != result_digest || !is_lowercase_sha256(&self.result_digest) {
            return Err(invalid(
                "proposal finalization result digest does not match typed result".to_string(),
            ));
        }

        let lineage = &self.proposal_result.replan_lineage;
        if lineage.is_empty() {
            return Err(invalid(
                "active v3 proposal result is missing generation-zero lineage".to_string(),
            ));
        }
        let original_digest = proposal_digest(&self.original_proposal)
            .map_err(|error| invalid(format!("original proposal JCS failed: {error}")))?;
        if lineage[0].generation != 0
            || lineage[0].proposal_id != self.original_proposal_id
            || lineage[0].proposal_digest.as_deref() != Some(original_digest.as_str())
        {
            return Err(invalid(
                "generation-zero lineage does not bind the exact original proposal preimage"
                    .to_string(),
            ));
        }
        if lineage[0].status == ProposalLineageStatus::Rejected && lineage.len() != 1 {
            return Err(invalid(
                "a generation-zero rejection cannot be followed by another generation".to_string(),
            ));
        }

        let mut accepted_by_generation = HashMap::new();
        for accepted in &self.accepted_proposal_preimages {
            if accepted_by_generation
                .insert(accepted.generation, &accepted.proposal)
                .is_some()
            {
                return Err(invalid(format!(
                    "duplicate accepted proposal preimage for generation {}",
                    accepted.generation
                )));
            }
        }
        for (index, entry) in lineage.iter().enumerate() {
            if entry.generation != u32::try_from(index).unwrap_or(u32::MAX) {
                return Err(invalid(
                    "proposal lineage generations are not contiguous from zero".to_string(),
                ));
            }
            match entry.status {
                ProposalLineageStatus::Accepted => {
                    if entry.rejection_reason.is_some() {
                        return Err(invalid(format!(
                            "accepted lineage generation {} cannot carry a rejection reason",
                            entry.generation
                        )));
                    }
                    let proposal =
                        accepted_by_generation
                            .get(&entry.generation)
                            .ok_or_else(|| {
                                invalid(format!(
                            "accepted lineage generation {} is missing its exact proposal preimage",
                            entry.generation
                        ))
                            })?;
                    let digest = proposal_digest(proposal).map_err(|error| {
                        invalid(format!(
                            "accepted proposal generation {} JCS failed: {error}",
                            entry.generation
                        ))
                    })?;
                    if proposal.id != entry.proposal_id
                        || entry.proposal_digest.as_deref() != Some(digest.as_str())
                        || !is_lowercase_sha256(&digest)
                    {
                        return Err(invalid(format!(
                            "accepted lineage generation {} does not bind its exact lowercase-JCS preimage",
                            entry.generation
                        )));
                    }
                }
                ProposalLineageStatus::Rejected => {
                    if entry
                        .rejection_reason
                        .as_deref()
                        .is_none_or(|reason| reason.trim().is_empty())
                    {
                        return Err(invalid(format!(
                            "rejected lineage generation {} is missing its exact rejection reason",
                            entry.generation
                        )));
                    }
                    if entry
                        .proposal_digest
                        .as_deref()
                        .is_some_and(|digest| !is_lowercase_sha256(digest))
                    {
                        return Err(invalid(format!(
                            "rejected lineage generation {} has an invalid proposal digest",
                            entry.generation
                        )));
                    }
                    if accepted_by_generation.contains_key(&entry.generation) {
                        return Err(invalid(format!(
                            "rejected lineage generation {} cannot carry an accepted preimage",
                            entry.generation
                        )));
                    }
                }
            }
        }
        if accepted_by_generation.len()
            != lineage
                .iter()
                .filter(|entry| entry.status == ProposalLineageStatus::Accepted)
                .count()
        {
            return Err(invalid(
                "proposal finalization contains an unbound accepted proposal preimage".to_string(),
            ));
        }

        if let Some(last_accepted) = lineage
            .iter()
            .rev()
            .find(|entry| entry.status == ProposalLineageStatus::Accepted)
        {
            let final_preimage = accepted_by_generation
                .get(&last_accepted.generation)
                .expect("accepted lineage was validated above");
            if *final_preimage != &self.final_proposal {
                return Err(invalid(
                    "most recent accepted proposal is not the active v3 final proposal".to_string(),
                ));
            }
        } else if lineage.len() != 1
            || lineage[0].status != ProposalLineageStatus::Rejected
            || self.final_proposal != self.original_proposal
        {
            return Err(invalid(
                "a result without an accepted generation must be a generation-zero rejection"
                    .to_string(),
            ));
        }

        let final_action_ids: std::collections::HashSet<_> = self
            .final_proposal
            .actions
            .iter()
            .map(|action| action.id.as_str())
            .collect();
        let result_action_ids: std::collections::HashSet<_> = self
            .proposal_result
            .results
            .iter()
            .map(|result| result.action_id.as_str())
            .collect();
        if final_action_ids.len() != self.final_proposal.actions.len()
            || result_action_ids.len() != self.proposal_result.results.len()
            || final_action_ids != result_action_ids
        {
            return Err(invalid(
                "typed proposal results do not match the final proposal action identities"
                    .to_string(),
            ));
        }
        Ok(())
    }

    fn validate_provenance(
        &self,
        started: &car_proto::RunStarted,
        marker: &ProposalExecutionMarker,
    ) -> std::io::Result<()> {
        let invalid =
            |message: String| std::io::Error::new(std::io::ErrorKind::InvalidData, message);
        let started_client = started.client_id.as_deref().ok_or_else(|| {
            invalid("durable RunStarted is missing its client identity".to_string())
        })?;
        if started.run_id != self.run_id || started_client != self.client_id {
            return Err(invalid(
                "pending run/client identity does not match durable RunStarted".to_string(),
            ));
        }
        let original_proposal = serde_json::to_value(&self.original_proposal).map_err(|error| {
            invalid(format!(
                "pending original proposal serialization failed: {error}"
            ))
        })?;
        let original_digest = proposal_digest(&self.original_proposal)
            .map_err(|error| invalid(format!("pending original proposal JCS failed: {error}")))?;
        if marker.run_id != self.run_id
            || marker.client_id != self.client_id
            || marker.requested_policy_session_id != self.requested_policy_session_id
            || marker.policy_session_id != self.policy_session_id
            || marker.original_proposal_id != self.original_proposal_id
            || marker.original_submission != self.original_submission
            || marker.original_proposal != original_proposal
            || marker.proposal_digest != original_digest
        {
            return Err(invalid(
                "pending finalization identity does not match durable execution marker".to_string(),
            ));
        }
        Ok(())
    }

    pub fn event_data(&self) -> HashMap<String, Value> {
        HashMap::from([
            (
                "original_submission_id".to_string(),
                Value::from(self.original_proposal_id.clone()),
            ),
            (
                "final_proposal_id".to_string(),
                Value::from(self.final_proposal_id.clone()),
            ),
            (
                "original_proposal".to_string(),
                serde_json::to_value(&self.original_proposal).unwrap_or(Value::Null),
            ),
            (
                "final_proposal".to_string(),
                serde_json::to_value(&self.final_proposal).unwrap_or(Value::Null),
            ),
            (
                "original_submission".to_string(),
                self.original_submission.clone(),
            ),
            (
                "replan_lineage".to_string(),
                serde_json::to_value(&self.proposal_result.replan_lineage).unwrap_or(Value::Null),
            ),
            (
                "all_succeeded".to_string(),
                Value::from(self.proposal_result.all_succeeded()),
            ),
            (
                "action_count".to_string(),
                Value::from(self.proposal_result.results.len()),
            ),
            (
                "proposal_result".to_string(),
                serde_json::to_value(&self.proposal_result).unwrap_or(Value::Null),
            ),
            (
                "result_digest".to_string(),
                Value::from(self.result_digest.clone()),
            ),
        ])
    }
}

/// Retention policy for the run store (R6). Defaults are the restrictive
/// 50-per-agent / 30-day caps; `~/.car/config.toml [runs]` overrides.
///
/// **A non-positive cap is DISABLED, not zero-tolerance** (car#1338). This used
/// to be read literally, and the literal reading is destructive in the one
/// direction nobody intends: `max_per_agent = 0` made `completed_rank >= 0`
/// always true and `max_age_days = 0` put the cutoff at now, so either value
/// deleted every collectable run trace on the next boot, unrecoverably.
///
/// An operator writing `0` means "no cap" — that is what it means for
/// `max_session_wall_secs`, and for the coder's own `max_sessions` /
/// `max_session_age_days` in the neighbouring `~/.car/coder.toml`. Two
/// similarly-named retention knobs in one daemon whose zero values inverted was
/// a hazard whose failure mode was silent data loss, so the two now agree.
///
/// This is a choice, not an unqualified convention: `CoderConfig`'s
/// `max_check_timeout_secs` and `approval_patch_bytes` read `0` as UNSET and
/// fall back to their defaults, because zero would floor a check at one second
/// or blind the approval surface. That treatment is the other candidate fix for
/// this bug and it was rejected on the merits — "unset" restores 50/30, which
/// still deletes, so it does not close the data loss. `0` is a legal value with
/// an ambiguous meaning and this defines it; a value that could not have been
/// intended (a negative or unrepresentable age) is a different case and is
/// warned about at load.
/// The fields are PRIVATE and the caps are read as `Option`s, so "disabled" is
/// a state this type carries rather than a sentinel every call site has to
/// remember. That is not decoration: the bug was a bare `rank >= cap`
/// comparison, and leaving the raw values public would let the next one be
/// written with no compiler complaint.
#[derive(Debug, Clone, Copy)]
pub struct RetentionConfig {
    max_per_agent: usize,
    max_age_days: i64,
}

impl RetentionConfig {
    /// Build a policy. `0` (or a negative age) disables that cap — see the
    /// type's note.
    pub fn new(max_per_agent: usize, max_age_days: i64) -> Self {
        Self {
            max_per_agent,
            max_age_days,
        }
    }

    /// The count cap, or `None` when disabled.
    pub fn count_cap(&self) -> Option<usize> {
        (self.max_per_agent > 0).then_some(self.max_per_agent)
    }

    /// The age cap in days, or `None` when disabled — including when the
    /// configured value cannot be turned into a cutoff at all.
    pub fn age_cap_days(&self) -> Option<i64> {
        self.age_cutoff(Utc::now()).map(|_| self.max_age_days)
    }

    /// The age cutoff at `now`, or `None` when disabled.
    ///
    /// `checked_sub_signed` rather than plain subtraction: `max_age_days` is an
    /// operator-supplied `i64`, and `chrono` panics rather than saturating on a
    /// value that overflows. A config typo must not take down the daemon at
    /// boot, so an unrepresentable cutoff reads as no cap — the safe direction,
    /// since the alternative is deleting on a value nobody could have meant.
    fn age_cutoff(&self, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
        if self.max_age_days <= 0 {
            return None;
        }
        chrono::Duration::try_days(self.max_age_days).and_then(|d| now.checked_sub_signed(d))
    }
}

impl Default for RetentionConfig {
    fn default() -> Self {
        Self::new(DEFAULT_MAX_RUNS_PER_AGENT, DEFAULT_MAX_AGE_DAYS)
    }
}

/// `[runs]` section of `~/.car/config.toml`. Both keys optional; an
/// absent key keeps the restrictive default.
#[derive(Debug, Clone, Default, Deserialize)]
struct RunsConfigFile {
    #[serde(default)]
    runs: RunsSection,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct RunsSection {
    #[serde(default)]
    max_per_agent: Option<usize>,
    #[serde(default)]
    max_age_days: Option<i64>,
}

impl RetentionConfig {
    /// Load the retention policy from `<car_dir>/config.toml`'s `[runs]`
    /// section, falling back to the restrictive default for any missing
    /// key or an unreadable/malformed file (config errors must never make
    /// the daemon refuse to start — same posture as the rest of `.car`).
    ///
    /// A present key is taken as written, `0` included — see the type's note on
    /// what `0` means. Filtering it here would silently restore the default
    /// instead of honoring "keep everything".
    pub fn from_car_dir(car_dir: &Path) -> Self {
        let mut cfg = Self::default();
        let path = car_dir.join("config.toml");
        let Ok(text) = std::fs::read_to_string(&path) else {
            return cfg;
        };
        let Ok(parsed) = toml::from_str::<RunsConfigFile>(&text) else {
            return cfg;
        };
        if let Some(n) = parsed.runs.max_per_agent {
            cfg.max_per_agent = n;
        }
        if let Some(d) = parsed.runs.max_age_days {
            cfg.max_age_days = d;
        }
        // `0` is a choice and is documented; a NEGATIVE or unrepresentable age
        // is a typo, and absorbing one without saying so is the only genuinely
        // silent behaviour here. It reads as no cap — the safe direction, since
        // the alternative is deleting on a number nobody meant — but an operator
        // who wrote it has to be able to find out.
        if cfg.max_age_days < 0 || (cfg.max_age_days > 0 && cfg.age_cutoff(Utc::now()).is_none()) {
            tracing::warn!(
                max_age_days = cfg.max_age_days,
                path = %path.display(),
                "[runs] max_age_days is not a usable number of days; treating the age cap as disabled"
            );
        }
        cfg
    }
}

/// JSONL run-trace store rooted at `<car_dir>/runs/`.
///
/// Stateless across calls — each append opens, writes, and closes the
/// run's file. Flush points are sparse (turn granularity), so there is no
/// long-lived file handle to manage, and a concurrently-restarting daemon
/// always sees a consistent on-disk tail.
#[derive(Debug, Clone)]
pub struct RunStore {
    /// `<car_dir>/runs` — the tree root, created `0700`.
    root: PathBuf,
    retention: RetentionConfig,
    failures: RunStoreFailureInjector,
    summary_read_gate: Option<RunStoreSummaryReadGate>,
    summary_write_gate: Option<RunStoreSummaryWriteGate>,
    append_gate: Option<RunStoreAppendGate>,
    lookup_gate: Option<RunStoreLookupGate>,
    private_path_failures: Option<car_secrets::PrivatePathDurabilityFailureInjector>,
    append_locks: Arc<Mutex<HashMap<PathBuf, Weak<Mutex<()>>>>>,
    trace_corruptions: Arc<Mutex<HashMap<(String, String), RunTraceCorruption>>>,
}

#[derive(Debug, Default)]
struct SummaryReadGateState {
    armed: bool,
    entered: bool,
}

/// Deterministic test/embedder seam for proving summary reads do not hold
/// async runtime locks. It is inert unless explicitly armed.
#[derive(Debug, Clone, Default)]
pub struct RunStoreSummaryReadGate {
    state: Arc<(Mutex<SummaryReadGateState>, Condvar)>,
}

pub type RunStoreAppendGate = RunStoreSummaryReadGate;
pub type RunStoreLookupGate = RunStoreSummaryReadGate;
#[doc(hidden)]
pub type RunStoreSummaryWriteGate = RunStoreSummaryReadGate;

impl RunStoreSummaryReadGate {
    pub fn block_next(&self) {
        let (lock, _) = &*self.state;
        let mut state = lock.lock().expect("run-store gate mutex poisoned");
        state.armed = true;
        state.entered = false;
    }

    pub fn wait_until_entered(&self, timeout: std::time::Duration) -> bool {
        let (lock, ready) = &*self.state;
        let state = lock.lock().expect("run-store gate mutex poisoned");
        let (state, _) = ready
            .wait_timeout_while(state, timeout, |state| !state.entered)
            .expect("run-store gate mutex poisoned while waiting");
        state.entered
    }

    pub fn release(&self) {
        let (lock, ready) = &*self.state;
        let mut state = lock.lock().expect("run-store gate mutex poisoned");
        state.armed = false;
        ready.notify_all();
    }

    fn wait_if_armed(&self) {
        let (lock, ready) = &*self.state;
        let mut state = lock.lock().expect("run-store gate mutex poisoned");
        if !state.armed || state.entered {
            return;
        }
        state.entered = true;
        ready.notify_all();
        while state.armed {
            state = ready
                .wait(state)
                .expect("run-store gate mutex poisoned while blocked");
        }
    }
}

#[derive(Debug, Clone)]
pub struct ProposalTraceEnsure {
    pub records: Vec<RunRecord>,
    pub appended: bool,
}

/// Exact durable append boundary used by deterministic fault-injection tests.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RunStoreFailurePoint {
    MarkerWrite,
    Write,
    Flush,
    Fsync,
    PendingUnlink,
    MarkerUnlink,
    DirectoryFsync,
    ResponseSerialization,
    SummaryWrite,
    CorruptionMarkerWrite,
    CorruptionSummaryInvalidate,
}

#[derive(Debug, Clone, Default)]
pub struct RunStoreFailureInjector {
    failures: Arc<Mutex<VecDeque<RunStoreFailurePoint>>>,
}

fn injected_storage_full() -> std::io::Error {
    std::io::Error::new(
        std::io::ErrorKind::StorageFull,
        "injected run-store storage full",
    )
}

impl RunStoreFailureInjector {
    pub fn fail_next(&self, point: RunStoreFailurePoint) {
        self.failures
            .lock()
            .expect("run-store failure injector mutex poisoned")
            .push_back(point);
    }

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

impl RunStore {
    /// Construct a store rooted at `runs_root` (the `runs/` dir itself).
    /// Use [`RunStore::from_journal_dir`] from the daemon, which derives
    /// the root from the configured journal dir; this constructor is the
    /// test/embedder seam.
    /// The retention policy in force, for a caller reporting what GC did.
    pub fn retention(&self) -> RetentionConfig {
        self.retention
    }

    pub fn new(runs_root: PathBuf, retention: RetentionConfig) -> Self {
        Self {
            root: runs_root,
            retention,
            failures: RunStoreFailureInjector::default(),
            summary_read_gate: None,
            summary_write_gate: None,
            append_gate: None,
            lookup_gate: None,
            private_path_failures: None,
            append_locks: Arc::new(Mutex::new(HashMap::new())),
            trace_corruptions: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    pub fn with_failure_injector(mut self, failures: RunStoreFailureInjector) -> Self {
        self.failures = failures;
        self
    }

    pub fn with_summary_read_gate(mut self, gate: RunStoreSummaryReadGate) -> Self {
        self.summary_read_gate = Some(gate);
        self
    }

    #[doc(hidden)]
    pub fn with_summary_write_gate(mut self, gate: RunStoreSummaryWriteGate) -> Self {
        self.summary_write_gate = Some(gate);
        self
    }

    pub fn with_append_gate(mut self, gate: RunStoreAppendGate) -> Self {
        self.append_gate = Some(gate);
        self
    }

    pub fn with_lookup_gate(mut self, gate: RunStoreLookupGate) -> Self {
        self.lookup_gate = Some(gate);
        self
    }

    /// Test/embedder seam for deterministic first-use directory-entry faults.
    pub fn with_private_path_failure_injector(
        mut self,
        failures: car_secrets::PrivatePathDurabilityFailureInjector,
    ) -> Self {
        self.private_path_failures = Some(failures);
        self
    }

    fn ensure_private_dir(&self, path: &Path) -> std::io::Result<()> {
        match self.private_path_failures.as_ref() {
            Some(failures) => car_secrets::ensure_private_dir_with_failure_injector(path, failures),
            None => car_secrets::ensure_private_dir(path),
        }
    }

    fn create_private_file(&self, path: &Path) -> std::io::Result<File> {
        match self.private_path_failures.as_ref() {
            Some(failures) => {
                car_secrets::create_private_file_with_failure_injector(path, failures)
            }
            None => car_secrets::create_private_file(path),
        }
    }

    fn open_private_append(&self, path: &Path) -> std::io::Result<File> {
        match self.private_path_failures.as_ref() {
            Some(failures) => {
                car_secrets::open_private_append_with_failure_injector(path, failures)
            }
            None => car_secrets::open_private_append(path),
        }
    }

    /// Derive the store from the daemon's journal dir. The journal lives
    /// at `~/.car/journals`, so the run store is its sibling
    /// `~/.car/runs`; retention is read from `~/.car/config.toml`. When
    /// the journal dir has no parent (a bare relative path), the store
    /// falls back to `journal_dir/../runs` resolved lexically.
    pub fn from_journal_dir(journal_dir: &Path) -> Self {
        let car_dir = journal_dir
            .parent()
            .map(Path::to_path_buf)
            .unwrap_or_else(|| PathBuf::from("."));
        let root = car_dir.join("runs");
        let retention = RetentionConfig::from_car_dir(&car_dir);
        Self::new(root, retention)
    }

    /// The `runs/` tree root.
    pub fn root(&self) -> &Path {
        &self.root
    }

    fn proposal_outbox_root(&self) -> PathBuf {
        self.root
            .parent()
            .unwrap_or_else(|| Path::new("."))
            .join("proposal-finalization")
    }

    fn proposal_outbox_path(&self, run_id: &str) -> PathBuf {
        let key = format!("{:x}", Sha256::digest(run_id.as_bytes()));
        self.proposal_outbox_root().join(format!("{key}.json"))
    }

    fn proposal_execution_root(&self) -> PathBuf {
        self.root
            .parent()
            .unwrap_or_else(|| Path::new("."))
            .join("proposal-execution")
    }

    fn proposal_execution_path(&self, run_id: &str) -> PathBuf {
        let key = format!("{:x}", Sha256::digest(run_id.as_bytes()));
        self.proposal_execution_root().join(format!("{key}.json"))
    }

    fn proposal_retry_rollback_root(&self) -> PathBuf {
        self.root
            .parent()
            .unwrap_or_else(|| Path::new("."))
            .join("proposal-retry-rollbacks")
    }

    fn proposal_retry_rollback_path(&self, run_id: &str) -> PathBuf {
        let key = format!("{:x}", Sha256::digest(run_id.as_bytes()));
        self.proposal_retry_rollback_root()
            .join(format!("{key}.json"))
    }

    fn proposal_id_claim_root(&self, run_id: &str) -> PathBuf {
        let run_key = format!("{:x}", Sha256::digest(run_id.as_bytes()));
        self.root
            .parent()
            .unwrap_or_else(|| Path::new("."))
            .join("proposal-id-claims")
            .join(run_key)
    }

    fn proposal_id_claim_path(&self, run_id: &str, proposal_id: &str) -> PathBuf {
        let proposal_key = format!("{:x}", Sha256::digest(proposal_id.as_bytes()));
        self.proposal_id_claim_root(run_id)
            .join(format!("{proposal_key}.json"))
    }

    fn completed_response_root(&self) -> PathBuf {
        self.root
            .parent()
            .unwrap_or_else(|| Path::new("."))
            .join("proposal-completed")
    }

    fn completed_response_run_root(&self, run_id: &str) -> PathBuf {
        let run_key = format!("{:x}", Sha256::digest(run_id.as_bytes()));
        self.completed_response_root().join(run_key)
    }

    fn completed_response_index_root(&self) -> PathBuf {
        self.root
            .parent()
            .unwrap_or_else(|| Path::new("."))
            .join("proposal-completed-index")
    }

    fn completed_response_index_migration_path(&self) -> PathBuf {
        self.completed_response_index_root()
            .join("owner-index-migration.json")
    }

    fn completed_response_owner_key(
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> std::io::Result<String> {
        let key_preimage = serde_json::to_vec(&(requested_policy_session_id, original_submission))
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        Ok(format!("{:x}", Sha256::digest(key_preimage)))
    }

    fn completed_response_owner_path(
        &self,
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> std::io::Result<PathBuf> {
        Ok(self.completed_response_index_root().join(format!(
            "{}.json",
            Self::completed_response_owner_key(requested_policy_session_id, original_submission,)?
        )))
    }

    fn completed_response_key(
        client_id: &str,
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> std::io::Result<String> {
        let key_preimage =
            serde_json::to_vec(&(client_id, requested_policy_session_id, original_submission))
                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        Ok(format!("{:x}", Sha256::digest(key_preimage)))
    }

    fn completed_response_path(
        &self,
        run_id: &str,
        client_id: &str,
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> std::io::Result<PathBuf> {
        Ok(self.completed_response_run_root(run_id).join(format!(
            "{}.json",
            Self::completed_response_key(
                client_id,
                requested_policy_session_id,
                original_submission,
            )?
        )))
    }

    /// Read the exact durable `RunStarted` boundary for `run_id` without
    /// collapsing absence into corruption. A zero-byte file left by an open
    /// that failed before its first write is still absent; any non-empty file
    /// without one valid, unambiguous `RunStarted` is invalid durable state.
    pub fn run_started(&self, run_id: &str) -> std::io::Result<Option<car_proto::RunStarted>> {
        if let Some(gate) = &self.lookup_gate {
            gate.wait_if_armed();
        }
        match self.open_root_for_read() {
            Ok(()) => {}
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        }
        let file_name = format!("{}.jsonl", sanitize(run_id));
        let mut found = None;
        for agent in std::fs::read_dir(&self.root)? {
            let agent = agent?;
            let file_type = agent.file_type()?;
            if !file_type.is_dir() || file_type.is_symlink() {
                continue;
            }
            car_secrets::ensure_private_dir(&agent.path())?;
            let path = agent.path().join(&file_name);
            match car_secrets::open_private_read(&path) {
                Ok(file) if found.is_none() => found = Some((path, file)),
                Ok(_) => {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        format!("durable run `{run_id}` exists under multiple agents"),
                    ))
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(error) => return Err(error),
            }
        }
        let Some((path, file)) = found else {
            return Ok(None);
        };
        let file_len = file.metadata()?.len();
        let records = load_private_records(&path, &file)?;
        if records.is_empty() && file_len == 0 {
            return Ok(None);
        }
        let mut started_rows = records.iter().filter_map(|record| match record {
            RunRecord::Started(started) => Some(started),
            _ => None,
        });
        let started = started_rows.next().ok_or_else(|| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("durable run record for `{run_id}` has no RunStarted"),
            )
        })?;
        if started_rows.next().is_some()
            || started.run_id != run_id
            || started.client_id.as_deref().is_none_or(str::is_empty)
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("durable RunStarted identity for `{run_id}` is ambiguous or invalid"),
            ));
        }
        Ok(Some(started.clone()))
    }

    fn durable_run_started(&self, run_id: &str) -> std::io::Result<car_proto::RunStarted> {
        self.run_started(run_id)?.ok_or_else(|| {
            std::io::Error::new(
                std::io::ErrorKind::NotFound,
                format!("durable RunStarted for run `{run_id}` is absent"),
            )
        })
    }

    /// Remove only the zero-byte trace created when `runs.start` opened its
    /// destination but failed before writing `RunStarted`. The full expected
    /// identity is accepted so callers cannot use this as a broad delete
    /// primitive. Any bytes or parsed rows make rollback fail closed.
    pub fn rollback_empty_run_start(&self, started: &car_proto::RunStarted) -> std::io::Result<()> {
        let dir = match self.open_agent_dir_for_read(&started.agent_id) {
            Ok(dir) => dir,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
            Err(error) => return Err(error),
        };
        let path = self.run_path(&started.agent_id, &started.run_id);
        let append_lock = self.append_lock(&path);
        let _guard = append_lock
            .lock()
            .map_err(|_| std::io::Error::other("run append lock poisoned"))?;
        let file = match car_secrets::open_private_read(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
            Err(error) => return Err(error),
        };
        car_secrets::revalidate_private_path(&path, &file)?;
        if file.metadata()?.len() != 0 {
            return Err(std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                format!(
                    "run `{}` acquired durable bytes while empty-start rollback was pending",
                    started.run_id
                ),
            ));
        }
        drop(file);
        std::fs::remove_file(&path)?;
        sync_directory(dir)
    }

    pub fn pending_provenance(
        &self,
        pending: &PendingProposalFinalization,
    ) -> std::io::Result<(car_proto::RunStarted, ProposalExecutionMarker)> {
        pending.validate()?;
        let started = self.durable_run_started(&pending.run_id)?;
        let marker = self.execution_marker(&pending.run_id)?.ok_or_else(|| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "pending finalization is missing its durable execution marker",
            )
        })?;
        pending.validate_provenance(&started, &marker)?;
        Ok((started, marker))
    }

    /// Write the durable `execution_in_progress` marker before tool dispatch.
    pub fn write_execution_marker(&self, marker: &ProposalExecutionMarker) -> std::io::Result<()> {
        marker.validate()?;
        if self.failures.take(RunStoreFailurePoint::MarkerWrite) {
            return Err(std::io::Error::other(
                "injected proposal execution marker write failure",
            ));
        }
        let root = self.proposal_execution_root();
        self.ensure_private_dir(&root)?;
        ensure_backup_excluded(&root)?;
        let path = self.proposal_execution_path(&marker.run_id);
        match car_secrets::open_private_read(&path) {
            Ok(file) => {
                let existing: ProposalExecutionMarker = serde_json::from_reader(file)
                    .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
                if existing == *marker {
                    return Ok(());
                }
                return Err(std::io::Error::new(
                    std::io::ErrorKind::AlreadyExists,
                    "run already has a different execution-in-progress marker",
                ));
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }
        let temp = root.join(format!(
            ".{}.{}.tmp",
            path.file_stem()
                .and_then(|value| value.to_str())
                .unwrap_or("proposal"),
            uuid::Uuid::new_v4().simple()
        ));
        let write_result = (|| {
            let mut file = self.create_private_file(&temp)?;
            serde_json::to_writer(&mut file, marker)
                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
            file.flush()?;
            file.sync_all()?;
            car_secrets::revalidate_private_path(&temp, &file)?;
            drop(file);
            car_secrets::atomic_replace_private_file(&temp, &path)?;
            sync_directory(&root)
        })();
        if write_result.is_err() {
            let _ = std::fs::remove_file(&temp);
        }
        write_result
    }

    pub fn execution_marker(
        &self,
        run_id: &str,
    ) -> std::io::Result<Option<ProposalExecutionMarker>> {
        let path = self.proposal_execution_path(run_id);
        let file = match car_secrets::open_private_read(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let marker: ProposalExecutionMarker = serde_json::from_reader(file)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        if marker.run_id != run_id {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!(
                    "proposal execution marker run_id `{}` does not match requested run `{run_id}`",
                    marker.run_id
                ),
            ));
        }
        marker.validate()?;
        Ok(Some(marker))
    }

    /// Persist the pre-execution rollback authority before acquiring the
    /// content-addressed retry owner. Exact rewrites are idempotent; a
    /// conflicting preimage for the same run fails closed.
    pub fn write_proposal_retry_rollback(
        &self,
        rollback: &ProposalRetryRollback,
    ) -> std::io::Result<()> {
        rollback.validate()?;
        let root = self.proposal_retry_rollback_root();
        self.ensure_private_dir(&root)?;
        ensure_backup_excluded(&root)?;
        let path = self.proposal_retry_rollback_path(&rollback.run_id);
        match car_secrets::open_private_read(&path) {
            Ok(file) => {
                let existing: ProposalRetryRollback = serde_json::from_reader(file)
                    .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
                existing.validate()?;
                if existing == *rollback {
                    return Ok(());
                }
                return Err(std::io::Error::new(
                    std::io::ErrorKind::AlreadyExists,
                    "run already has a different proposal retry rollback intent",
                ));
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }
        self.write_private_json_atomic(&root, &path, rollback)
    }

    pub fn proposal_retry_rollback(
        &self,
        run_id: &str,
    ) -> std::io::Result<Option<ProposalRetryRollback>> {
        let path = self.proposal_retry_rollback_path(run_id);
        let file = match car_secrets::open_private_read(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let rollback: ProposalRetryRollback = serde_json::from_reader(file)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        rollback.validate()?;
        if rollback.run_id != run_id || path != self.proposal_retry_rollback_path(&rollback.run_id)
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "proposal retry rollback path does not match its run identity",
            ));
        }
        Ok(Some(rollback))
    }

    pub fn clear_proposal_retry_rollback(
        &self,
        expected: &ProposalRetryRollback,
    ) -> std::io::Result<()> {
        let Some(existing) = self.proposal_retry_rollback(&expected.run_id)? else {
            return Ok(());
        };
        if existing != *expected {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "proposal retry rollback does not match the exact authority",
            ));
        }
        let root = self.proposal_retry_rollback_root();
        std::fs::remove_file(self.proposal_retry_rollback_path(&expected.run_id))?;
        sync_directory(root)
    }

    /// Resolve every crash-left rollback intent. The intent is removed last:
    /// a second crash at any earlier boundary leaves enough authority for the
    /// next startup to repeat the exact cleanup.
    pub fn reconcile_proposal_retry_rollbacks(&self) -> std::io::Result<usize> {
        let root = self.proposal_retry_rollback_root();
        let entries = match std::fs::read_dir(&root) {
            Ok(entries) => entries,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0),
            Err(error) => return Err(error),
        };
        let mut reconciled = 0usize;
        for entry in entries {
            let entry = entry?;
            if entry.path().extension().and_then(|value| value.to_str()) != Some("json") {
                continue;
            }
            let file = car_secrets::open_private_read(&entry.path())?;
            let rollback: ProposalRetryRollback = serde_json::from_reader(file)
                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
            rollback.validate()?;
            if entry.path() != self.proposal_retry_rollback_path(&rollback.run_id) {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "proposal retry rollback filename does not match its run identity",
                ));
            }
            if self.execution_marker(&rollback.run_id)?.is_none()
                && self.completed_proposal_retry_owner(
                    rollback.requested_policy_session_id.as_deref(),
                    &rollback.original_submission,
                )? == Some((rollback.run_id.clone(), rollback.client_id.clone()))
            {
                self.release_proposal_retry_owner(
                    &rollback.run_id,
                    &rollback.client_id,
                    rollback.requested_policy_session_id.as_deref(),
                    &rollback.original_submission,
                )?;
            }
            self.clear_proposal_retry_rollback(&rollback)?;
            reconciled = reconciled.saturating_add(1);
        }
        Ok(reconciled)
    }

    /// Claim a proposal identifier inside one authenticated run before any
    /// dispatch. Exact-submission retries are idempotent; reusing the same id
    /// for changed bytes is rejected durably, including after restart.
    pub fn claim_proposal_id(
        &self,
        run_id: &str,
        client_id: &str,
        proposal_id: &str,
        original_submission: &Value,
    ) -> std::io::Result<ProposalIdClaimOutcome> {
        if run_id.is_empty() || client_id.is_empty() || proposal_id.is_empty() {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "proposal id claim requires non-empty run/client/proposal identities",
            ));
        }
        let expected = ProposalIdClaim {
            run_id: run_id.to_string(),
            client_id: client_id.to_string(),
            proposal_id: proposal_id.to_string(),
            original_submission: original_submission.clone(),
        };
        let root = self.proposal_id_claim_root(run_id);
        self.ensure_private_dir(&root)?;
        ensure_backup_excluded(&root)?;
        let path = self.proposal_id_claim_path(run_id, proposal_id);
        let lock = self.append_lock(&path);
        let _guard = lock
            .lock()
            .map_err(|_| std::io::Error::other("proposal id claim lock poisoned"))?;
        match car_secrets::open_private_read(&path) {
            Ok(file) => {
                let existing: ProposalIdClaim = serde_json::from_reader(file)
                    .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
                if existing == expected {
                    return Ok(ProposalIdClaimOutcome::ExistingExact);
                }
                return Err(std::io::Error::new(
                    std::io::ErrorKind::AlreadyExists,
                    "proposal id is already bound to a different submission in this run",
                ));
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }
        self.write_private_json_atomic(&root, &path, &expected)?;
        Ok(ProposalIdClaimOutcome::Acquired)
    }

    fn write_private_json_atomic<T: Serialize>(
        &self,
        root: &Path,
        path: &Path,
        value: &T,
    ) -> std::io::Result<()> {
        let temp = root.join(format!(
            ".{}.{}.tmp",
            path.file_stem()
                .and_then(|value| value.to_str())
                .unwrap_or("record"),
            uuid::Uuid::new_v4().simple()
        ));
        let write_result = (|| {
            let mut file = self.create_private_file(&temp)?;
            serde_json::to_writer(&mut file, value)
                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
            file.flush()?;
            file.sync_all()?;
            car_secrets::revalidate_private_path(&temp, &file)?;
            drop(file);
            car_secrets::atomic_replace_private_file(&temp, path)?;
            sync_directory(root)
        })();
        if write_result.is_err() {
            let _ = std::fs::remove_file(&temp);
        }
        write_result
    }

    pub fn clear_execution_marker(
        &self,
        expected: &ProposalExecutionMarker,
    ) -> std::io::Result<()> {
        let Some(existing) = self.execution_marker(&expected.run_id)? else {
            return Ok(());
        };
        if existing != *expected {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "proposal execution marker does not match the full expected identity tuple",
            ));
        }
        let root = self.proposal_execution_root();
        std::fs::remove_file(self.proposal_execution_path(&expected.run_id))?;
        sync_directory(root)
    }

    /// Persist one exact proposal-finalization transaction with private
    /// permissions and an fsynced atomic replacement. Exact retries are
    /// idempotent; a conflicting preimage for the same run is rejected.
    pub fn write_pending_proposal(
        &self,
        pending: &PendingProposalFinalization,
    ) -> std::io::Result<()> {
        self.pending_provenance(pending)?;
        if pending.run_id.is_empty()
            || pending.client_id.is_empty()
            || pending.original_proposal_id.is_empty()
            || pending.final_proposal_id.is_empty()
            || pending.result_digest.len() != 64
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "proposal finalization requires run/client/proposal identities and a SHA-256 digest",
            ));
        }
        let root = self.proposal_outbox_root();
        self.ensure_private_dir(&root)?;
        ensure_backup_excluded(&root)?;
        let path = self.proposal_outbox_path(&pending.run_id);
        match car_secrets::open_private_read(&path) {
            Ok(file) => {
                let existing: PendingProposalFinalization = serde_json::from_reader(file)
                    .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
                if existing == *pending {
                    return Ok(());
                }
                return Err(std::io::Error::new(
                    std::io::ErrorKind::AlreadyExists,
                    "run already has a different pending proposal finalization",
                ));
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }

        let temp = root.join(format!(
            ".{}.{}.tmp",
            path.file_stem()
                .and_then(|value| value.to_str())
                .unwrap_or("proposal"),
            uuid::Uuid::new_v4().simple()
        ));
        let write_result = (|| {
            let mut file = self.create_private_file(&temp)?;
            serde_json::to_writer(&mut file, pending)
                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
            file.flush()?;
            file.sync_all()?;
            car_secrets::revalidate_private_path(&temp, &file)?;
            drop(file);
            car_secrets::atomic_replace_private_file(&temp, &path)?;
            sync_directory(&root)
        })();
        if write_result.is_err() {
            let _ = std::fs::remove_file(&temp);
        }
        write_result
    }

    pub fn pending_proposal(
        &self,
        run_id: &str,
    ) -> std::io::Result<Option<PendingProposalFinalization>> {
        let path = self.proposal_outbox_path(run_id);
        let file = match car_secrets::open_private_read(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let pending: PendingProposalFinalization = serde_json::from_reader(file)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        if pending.run_id != run_id {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!(
                    "proposal finalization run_id `{}` does not match requested run `{run_id}`",
                    pending.run_id
                ),
            ));
        }
        self.pending_provenance(&pending)?;
        Ok(Some(pending))
    }

    pub fn all_pending_proposals(&self) -> std::io::Result<Vec<PendingProposalFinalization>> {
        let root = self.proposal_outbox_root();
        let entries = match std::fs::read_dir(root) {
            Ok(entries) => entries,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(error) => return Err(error),
        };
        let mut pending = Vec::new();
        for entry in entries {
            let entry = entry?;
            if entry.path().extension().and_then(|ext| ext.to_str()) != Some("json") {
                continue;
            }
            let file = car_secrets::open_private_read(&entry.path())?;
            let row: PendingProposalFinalization = serde_json::from_reader(file)
                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
            if entry.path() != self.proposal_outbox_path(&row.run_id) {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!(
                        "proposal finalization filename does not match internal run_id `{}`",
                        row.run_id
                    ),
                ));
            }
            self.pending_provenance(&row)?;
            pending.push(row);
        }
        Ok(pending)
    }

    /// Remove only the exact, provenance-validated transaction whose terminal
    /// was acknowledged. The caller clears this before its matching execution
    /// marker so provenance remains available for the check.
    pub fn clear_pending_proposal(
        &self,
        expected: &PendingProposalFinalization,
    ) -> std::io::Result<()> {
        self.pending_provenance(expected)?;
        let existing = self.pending_proposal(&expected.run_id)?.ok_or_else(|| {
            std::io::Error::new(
                std::io::ErrorKind::NotFound,
                "proposal finalization outbox row is absent",
            )
        })?;
        if existing != *expected {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "proposal finalization outbox does not match the full expected transaction",
            ));
        }
        let root = self.proposal_outbox_root();
        std::fs::remove_file(self.proposal_outbox_path(&expected.run_id))?;
        sync_directory(root)
    }

    /// Validate only the receipt's self-contained typed preimages. The sole
    /// creation path separately requires [`Self::pending_provenance`] before
    /// publishing, while retained reads cannot depend on a GC-eligible trace.
    fn validate_completed_response(
        &self,
        receipt: &CompletedProposalResponse,
    ) -> std::io::Result<()> {
        receipt.validate()?;
        Ok(())
    }

    fn read_completed_proposal_owner(
        &self,
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> std::io::Result<Option<CompletedProposalOwnership>> {
        let path =
            self.completed_response_owner_path(requested_policy_session_id, original_submission)?;
        let file = match car_secrets::open_private_read(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let owner: CompletedProposalOwnership = serde_json::from_reader(file)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        if owner.requested_policy_session_id.as_deref() != requested_policy_session_id
            || owner.original_submission != *original_submission
            || path
                != self.completed_response_owner_path(
                    owner.requested_policy_session_id.as_deref(),
                    &owner.original_submission,
                )?
            || owner.run_id.is_empty()
            || owner.client_id.is_empty()
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "completed proposal ownership index does not match its retry tuple",
            ));
        }
        Ok(Some(owner))
    }

    /// Atomically reserve the process-wide retry tuple before any execution
    /// marker or runtime invocation. Direct exclusive creation is the linear
    /// point: one run/client wins, every concurrent or restarted contender
    /// observes that durable owner, and a partial claim fails closed.
    pub fn reserve_proposal_retry_owner(
        &self,
        run_id: &str,
        client_id: &str,
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> std::io::Result<ProposalRetryReservation> {
        if run_id.is_empty() || client_id.is_empty() || requested_policy_session_id == Some("") {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "proposal retry reservation requires non-empty run/client/policy identities",
            ));
        }
        let expected = CompletedProposalOwnership::new(
            run_id,
            client_id,
            requested_policy_session_id,
            original_submission,
        );
        let root = self.completed_response_index_root();
        self.ensure_private_dir(&root)?;
        ensure_backup_excluded(&root)?;
        sync_directory(root.parent().unwrap_or_else(|| Path::new(".")))?;
        let path =
            self.completed_response_owner_path(requested_policy_session_id, original_submission)?;
        let lock = self.append_lock(&path);
        let _guard = lock
            .lock()
            .map_err(|_| std::io::Error::other("proposal retry reservation lock poisoned"))?;
        if let Some(existing) =
            self.read_completed_proposal_owner(requested_policy_session_id, original_submission)?
        {
            return Ok(ProposalRetryReservation::Existing {
                run_id: existing.run_id,
                client_id: existing.client_id,
            });
        }

        let mut file = match self.create_private_file(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                let existing = self
                    .read_completed_proposal_owner(
                        requested_policy_session_id,
                        original_submission,
                    )?
                    .ok_or_else(|| {
                        std::io::Error::new(
                            std::io::ErrorKind::InvalidData,
                            "proposal retry reservation appeared without a readable owner",
                        )
                    })?;
                return Ok(ProposalRetryReservation::Existing {
                    run_id: existing.run_id,
                    client_id: existing.client_id,
                });
            }
            Err(error) => return Err(error),
        };
        serde_json::to_writer(&mut file, &expected)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        file.flush()?;
        file.sync_all()?;
        car_secrets::revalidate_private_path(&path, &file)?;
        sync_directory(&root)?;
        Ok(ProposalRetryReservation::Acquired)
    }

    /// Release only the exact pre-dispatch retry owner. This is used when
    /// execution never became admissible (marker durability or policy bind
    /// failed), so a later scheduled run may safely submit the same proposal.
    /// The content-addressed owner is validated before unlink and the owner
    /// directory is fsynced before success is reported.
    pub fn release_proposal_retry_owner(
        &self,
        run_id: &str,
        client_id: &str,
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> std::io::Result<()> {
        let path =
            self.completed_response_owner_path(requested_policy_session_id, original_submission)?;
        let lock = self.append_lock(&path);
        let _guard = lock
            .lock()
            .map_err(|_| std::io::Error::other("proposal retry release lock poisoned"))?;
        let Some(existing) =
            self.read_completed_proposal_owner(requested_policy_session_id, original_submission)?
        else {
            return Ok(());
        };
        if existing.run_id != run_id || existing.client_id != client_id {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "proposal retry owner does not match the exact rollback authority",
            ));
        }
        let root = self.completed_response_index_root();
        std::fs::remove_file(path)?;
        sync_directory(root)
    }

    /// Verify or backfill the pre-execution owner before publishing a
    /// completed response. Startup uses the same operation to migrate legacy
    /// receipts that predate reservations.
    fn claim_completed_proposal_owner(
        &self,
        pending: &PendingProposalFinalization,
    ) -> std::io::Result<()> {
        match self.reserve_proposal_retry_owner(
            &pending.run_id,
            &pending.client_id,
            pending.requested_policy_session_id.as_deref(),
            &pending.original_submission,
        )? {
            ProposalRetryReservation::Acquired => Ok(()),
            ProposalRetryReservation::Existing { run_id, client_id }
                if run_id == pending.run_id && client_id == pending.client_id =>
            {
                Ok(())
            }
            ProposalRetryReservation::Existing { .. } => Err(std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                "proposal retry tuple already belongs to a different completed response",
            )),
        }
    }

    /// Persist an exact typed response only after the matching critical
    /// `proposal_completed` append acknowledged. The finalization outbox and
    /// execution marker must still be present and provenance-valid here, so a
    /// forged receipt cannot become a cleanup authority.
    pub fn write_completed_proposal(
        &self,
        pending: &PendingProposalFinalization,
    ) -> std::io::Result<CompletedProposalResponse> {
        self.pending_provenance(pending)?;
        let receipt = CompletedProposalResponse {
            finalization: pending.clone(),
        };
        self.validate_completed_response(&receipt)?;
        self.claim_completed_proposal_owner(pending)?;
        let root = self.completed_response_root();
        self.ensure_private_dir(&root)?;
        ensure_backup_excluded(&root)?;
        sync_directory(root.parent().unwrap_or_else(|| Path::new(".")))?;
        let run_root = self.completed_response_run_root(&pending.run_id);
        self.ensure_private_dir(&run_root)?;
        sync_directory(&root)?;
        let path = self.completed_response_path(
            &pending.run_id,
            &pending.client_id,
            pending.requested_policy_session_id.as_deref(),
            &pending.original_submission,
        )?;
        match car_secrets::open_private_read(&path) {
            Ok(file) => {
                let existing: CompletedProposalResponse = serde_json::from_reader(file)
                    .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
                self.validate_completed_response(&existing)?;
                if existing == receipt {
                    return Ok(existing);
                }
                return Err(std::io::Error::new(
                    std::io::ErrorKind::AlreadyExists,
                    "exact proposal retry key already has a different completed response",
                ));
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }

        let temp = run_root.join(format!(".response.{}.tmp", uuid::Uuid::new_v4().simple()));
        let write_result = (|| {
            let mut file = self.create_private_file(&temp)?;
            serde_json::to_writer(&mut file, &receipt)
                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
            file.flush()?;
            file.sync_all()?;
            car_secrets::revalidate_private_path(&temp, &file)?;
            drop(file);
            car_secrets::atomic_replace_private_file(&temp, &path)?;
            sync_directory(&run_root)
        })();
        if write_result.is_err() {
            let _ = std::fs::remove_file(&temp);
        }
        write_result.map(|()| receipt)
    }

    /// Load only the receipt whose immutable retry tuple exactly matches this
    /// bound run/client/policy/raw submission. Absence permits a new proposal;
    /// a present corrupt or self-inconsistent receipt fails closed.
    pub fn completed_proposal(
        &self,
        run_id: &str,
        client_id: &str,
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> std::io::Result<Option<CompletedProposalResponse>> {
        let path = self.completed_response_path(
            run_id,
            client_id,
            requested_policy_session_id,
            original_submission,
        )?;
        let file = match car_secrets::open_private_read(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let receipt: CompletedProposalResponse = serde_json::from_reader(file)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        self.validate_completed_response(&receipt)?;
        let pending = &receipt.finalization;
        if pending.run_id != run_id
            || pending.client_id != client_id
            || pending.requested_policy_session_id.as_deref() != requested_policy_session_id
            || pending.original_submission != *original_submission
            || path
                != self.completed_response_path(
                    &pending.run_id,
                    &pending.client_id,
                    pending.requested_policy_session_id.as_deref(),
                    &pending.original_submission,
                )?
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "completed proposal response does not match its exact retry key",
            ));
        }
        Ok(Some(receipt))
    }

    /// Recover the sole retained response for an exact proposal after an
    /// authenticated run owner reconnects with a newly minted policy session.
    ///
    /// The lookup is deliberately bounded to one run and one durable client.
    /// It never treats the replacement policy-session id as durable identity,
    /// and ambiguity fails closed instead of selecting an arbitrary receipt.
    pub fn completed_proposal_for_resumed_owner(
        &self,
        run_id: &str,
        durable_client_id: &str,
        original_submission: &Value,
    ) -> std::io::Result<Option<CompletedProposalResponse>> {
        let run_root = self.completed_response_run_root(run_id);
        let entries = match std::fs::read_dir(&run_root) {
            Ok(entries) => entries,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        car_secrets::ensure_private_dir(&run_root)?;

        let mut receipt_paths = Vec::new();
        for entry in entries {
            let entry = entry?;
            if entry.path().extension().and_then(|ext| ext.to_str()) != Some("json") {
                continue;
            }
            if receipt_paths.len() >= MAX_RESUMED_PROPOSAL_RECEIPTS_PER_RUN {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "resumed proposal recovery exceeded the per-run receipt count limit",
                ));
            }
            receipt_paths.push(entry.path());
        }

        let mut matching = None;
        let mut scanned_bytes = 0_u64;
        for path in receipt_paths {
            let file = car_secrets::open_private_read(&path)?;
            scanned_bytes = scanned_bytes
                .checked_add(file.metadata()?.len())
                .filter(|total| *total <= MAX_RESUMED_PROPOSAL_RECEIPT_BYTES_PER_RUN)
                .ok_or_else(|| {
                    std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "resumed proposal recovery exceeded the per-run receipt byte limit",
                    )
                })?;
            let receipt: CompletedProposalResponse = serde_json::from_reader(file)
                .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
            self.validate_completed_response(&receipt)?;
            let pending = &receipt.finalization;
            if pending.run_id != run_id
                || path
                    != self.completed_response_path(
                        &pending.run_id,
                        &pending.client_id,
                        pending.requested_policy_session_id.as_deref(),
                        &pending.original_submission,
                    )?
            {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "completed proposal response path does not match its typed identity",
                ));
            }
            if pending.client_id != durable_client_id
                || pending.original_submission != *original_submission
            {
                continue;
            }
            if matching.replace(receipt).is_some() {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "resumed proposal recovery is ambiguous for the exact durable owner and submission",
                ));
            }
        }
        Ok(matching)
    }

    pub fn all_completed_proposals(&self) -> std::io::Result<Vec<CompletedProposalResponse>> {
        let root = self.completed_response_root();
        let run_dirs = match std::fs::read_dir(&root) {
            Ok(entries) => entries,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(error) => return Err(error),
        };
        let mut receipts = Vec::new();
        for run_dir in run_dirs {
            let run_dir = run_dir?;
            if !run_dir.file_type()?.is_dir() || run_dir.file_type()?.is_symlink() {
                continue;
            }
            car_secrets::ensure_private_dir(&run_dir.path())?;
            for entry in std::fs::read_dir(run_dir.path())? {
                let entry = entry?;
                if entry.path().extension().and_then(|ext| ext.to_str()) != Some("json") {
                    continue;
                }
                let file = car_secrets::open_private_read(&entry.path())?;
                let receipt: CompletedProposalResponse = serde_json::from_reader(file)
                    .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
                self.validate_completed_response(&receipt)?;
                let pending = &receipt.finalization;
                if run_dir.path() != self.completed_response_run_root(&pending.run_id)
                    || entry.path()
                        != self.completed_response_path(
                            &pending.run_id,
                            &pending.client_id,
                            pending.requested_policy_session_id.as_deref(),
                            &pending.original_submission,
                        )?
                {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "completed proposal response path does not match its typed identity",
                    ));
                }
                self.claim_completed_proposal_owner(pending)?;
                receipts.push(receipt);
            }
        }
        Ok(receipts)
    }

    /// Backfill the content-addressed retry-owner index exactly once for
    /// receipts written before that index existed. Permanent receipts grow
    /// without bound by design, so normal daemon startup must never enumerate
    /// them. After this versioned checkpoint is durable, crash cleanup is
    /// driven by the bounded pending-finalization outbox instead.
    pub fn reconcile_completed_proposal_migration(&self) -> std::io::Result<()> {
        let checkpoint_path = self.completed_response_index_migration_path();
        match car_secrets::open_private_read(&checkpoint_path) {
            Ok(file) => {
                let checkpoint: CompletedProposalOwnerIndexMigration =
                    serde_json::from_reader(file).map_err(|error| {
                        std::io::Error::new(std::io::ErrorKind::InvalidData, error)
                    })?;
                if checkpoint.version > COMPLETED_PROPOSAL_OWNER_INDEX_MIGRATION_VERSION {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        format!(
                            "unsupported completed-proposal owner-index migration version {}",
                            checkpoint.version
                        ),
                    ));
                }
                if checkpoint.version == COMPLETED_PROPOSAL_OWNER_INDEX_MIGRATION_VERSION {
                    return Ok(());
                }
                // An older checkpoint indexed owners but did not guarantee
                // trace backfill plus authorized guard cleanup. Re-run the
                // streamed migration and publish v2 only after both finish.
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }

        // Stream permanent receipts one at a time. A corrupt or unexpectedly
        // large legacy row fails closed before the checkpoint, making the
        // migration crash-resumable without an unbounded all-receipts Vec.
        const MAX_LEGACY_RECEIPT_BYTES: u64 = 16 * 1024 * 1024;
        let receipts_root = self.completed_response_root();
        match std::fs::read_dir(&receipts_root) {
            Ok(run_dirs) => {
                for run_dir in run_dirs {
                    let run_dir = run_dir?;
                    if !run_dir.file_type()?.is_dir() || run_dir.file_type()?.is_symlink() {
                        continue;
                    }
                    car_secrets::ensure_private_dir(&run_dir.path())?;
                    for entry in std::fs::read_dir(run_dir.path())? {
                        let entry = entry?;
                        if entry.path().extension().and_then(|ext| ext.to_str()) != Some("json") {
                            continue;
                        }
                        let file = car_secrets::open_private_read(&entry.path())?;
                        if file.metadata()?.len() > MAX_LEGACY_RECEIPT_BYTES {
                            return Err(std::io::Error::new(
                                std::io::ErrorKind::InvalidData,
                                "completed proposal receipt exceeds migration byte limit",
                            ));
                        }
                        let receipt: CompletedProposalResponse = serde_json::from_reader(file)
                            .map_err(|error| {
                                std::io::Error::new(std::io::ErrorKind::InvalidData, error)
                            })?;
                        self.validate_completed_response(&receipt)?;
                        let pending = &receipt.finalization;
                        if run_dir.path() != self.completed_response_run_root(&pending.run_id)
                            || entry.path()
                                != self.completed_response_path(
                                    &pending.run_id,
                                    &pending.client_id,
                                    pending.requested_policy_session_id.as_deref(),
                                    &pending.original_submission,
                                )?
                        {
                            return Err(std::io::Error::new(
                                std::io::ErrorKind::InvalidData,
                                "completed proposal response path does not match its typed identity",
                            ));
                        }
                        self.claim_completed_proposal_owner(pending)?;
                        if self.run_started(&pending.run_id)?.is_some() {
                            self.ensure_proposal_turns(pending)?;
                        }
                        // Cleanup is part of the authorized migration. The
                        // checkpoint is deliberately published only after
                        // every receipt's guards have been durably cleared.
                        self.cleanup_completed_proposal_guards(&receipt)?;
                    }
                }
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }
        let root = self.completed_response_index_root();
        self.ensure_private_dir(&root)?;
        ensure_backup_excluded(&root)?;
        sync_directory(root.parent().unwrap_or_else(|| Path::new(".")))?;
        let temp = root.join(format!(
            ".owner-index-migration.{}.tmp",
            uuid::Uuid::new_v4().simple()
        ));
        let write_result = (|| {
            let mut file = self.create_private_file(&temp)?;
            serde_json::to_writer(
                &mut file,
                &CompletedProposalOwnerIndexMigration {
                    version: COMPLETED_PROPOSAL_OWNER_INDEX_MIGRATION_VERSION,
                },
            )
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
            file.flush()?;
            file.sync_all()?;
            car_secrets::revalidate_private_path(&temp, &file)?;
            drop(file);
            car_secrets::atomic_replace_private_file(&temp, &checkpoint_path)?;
            sync_directory(&root)
        })();
        if write_result.is_err() {
            let _ = std::fs::remove_file(&temp);
        }
        write_result
    }

    pub fn completed_proposal_retry_owner(
        &self,
        requested_policy_session_id: Option<&str>,
        original_submission: &Value,
    ) -> std::io::Result<Option<(String, String)>> {
        Ok(self
            .read_completed_proposal_owner(requested_policy_session_id, original_submission)?
            .map(|owner| (owner.run_id, owner.client_id)))
    }

    /// Remove leftover pre-response guards only under a completed receipt's
    /// exact typed authority. Each deletion is independently durable; a fault
    /// leaves the receipt and any remaining guard for safe retry/restart.
    pub fn cleanup_completed_proposal_guards(
        &self,
        receipt: &CompletedProposalResponse,
    ) -> std::io::Result<()> {
        self.validate_completed_response(receipt)?;
        let pending = &receipt.finalization;
        // Remove the execution marker first. If either unlink boundary is
        // interrupted, the pending-finalization outbox remains as the bounded
        // startup cleanup index. Removing pending first could leave a lone
        // marker that was discoverable only by rescanning every permanent
        // completed receipt.
        let expected_marker = receipt.execution_marker()?;
        if let Some(existing) = self.execution_marker(&pending.run_id)? {
            if existing != expected_marker {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "completed response does not authorize this execution marker cleanup",
                ));
            }
            if self.failures.take(RunStoreFailurePoint::MarkerUnlink) {
                return Err(std::io::Error::other(
                    "injected execution marker unlink failure",
                ));
            }
            std::fs::remove_file(self.proposal_execution_path(&pending.run_id))?;
            self.sync_cleanup_directory(&self.proposal_execution_root())?;
        }

        let pending_path = self.proposal_outbox_path(&pending.run_id);
        match car_secrets::open_private_read(&pending_path) {
            Ok(file) => {
                let existing: PendingProposalFinalization = serde_json::from_reader(file)
                    .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
                existing.validate()?;
                if existing != *pending {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "completed response does not authorize this pending finalization cleanup",
                    ));
                }
                if self.failures.take(RunStoreFailurePoint::PendingUnlink) {
                    return Err(std::io::Error::other(
                        "injected pending proposal unlink failure",
                    ));
                }
                std::fs::remove_file(&pending_path)?;
                self.sync_cleanup_directory(&self.proposal_outbox_root())?;
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }
        Ok(())
    }

    fn sync_cleanup_directory(&self, root: &Path) -> std::io::Result<()> {
        if self.failures.take(RunStoreFailurePoint::DirectoryFsync) {
            return Err(std::io::Error::other(
                "injected proposal cleanup directory fsync failure",
            ));
        }
        sync_directory(root)
    }

    pub fn completed_proposal_response_value(
        &self,
        receipt: &CompletedProposalResponse,
    ) -> std::io::Result<Value> {
        self.validate_completed_response(receipt)?;
        if self
            .failures
            .take(RunStoreFailurePoint::ResponseSerialization)
        {
            return Err(std::io::Error::other(
                "injected completed proposal response serialization failure",
            ));
        }
        let result = receipt.proposal_result();
        let mut value = serde_json::to_value(result)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;

        // Keep rollback provenance explicit at the durable-response boundary.
        // `ActionResult`'s serde shape already carries this field, but this is
        // the second serialization after the engine result crossed the pending
        // finalization outbox. Project true markers from the typed receipt so a
        // response-shaping refactor cannot silently reduce rolled-back success
        // to an ordinary committed success. False remains omitted for backward
        // compatibility.
        let wire_results = value
            .get_mut("results")
            .and_then(Value::as_array_mut)
            .ok_or_else(|| {
                std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "completed proposal response is missing its results array",
                )
            })?;
        if wire_results.len() != result.results.len() {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "completed proposal response result count changed during serialization",
            ));
        }
        for (wire, typed) in wire_results.iter_mut().zip(&result.results) {
            if typed.rolled_back {
                wire.as_object_mut()
                    .ok_or_else(|| {
                        std::io::Error::new(
                            std::io::ErrorKind::InvalidData,
                            "completed proposal response contains a non-object action result",
                        )
                    })?
                    .insert("rolled_back".to_string(), Value::Bool(true));
            }
        }
        Ok(value)
    }

    /// Prepare the CAR-owned run tree for daemon startup. This is intentionally
    /// fallible: the daemon must refuse adoption/listening when the root or
    /// backup marker cannot be proven owner-private.
    pub fn prepare_storage(&self) -> std::io::Result<()> {
        self.ensure_root()
    }

    /// Path to a run's JSONL file: `runs/{agent_id}/{run_id}.jsonl`.
    fn run_path(&self, agent_id: &str, run_id: &str) -> PathBuf {
        self.root
            .join(sanitize(agent_id))
            .join(format!("{}.jsonl", sanitize(run_id)))
    }

    fn run_summary_index_path(agent_path: &Path) -> PathBuf {
        agent_path.join(RUN_SUMMARY_INDEX_FILE)
    }

    fn run_summary_sidecar_root(agent_path: &Path) -> PathBuf {
        agent_path.join(RUN_SUMMARY_SIDECAR_DIR)
    }

    fn run_summary_key(run_id: &str) -> [u8; RUN_SUMMARY_KEY_BYTES] {
        Sha256::digest(run_id.as_bytes()).into()
    }

    fn run_summary_sidecar_path(agent_path: &Path, key: &[u8; 32]) -> PathBuf {
        let name = key
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect::<String>();
        Self::run_summary_sidecar_root(agent_path).join(format!("{name}.json"))
    }

    fn run_trace_corruption_sidecar_root(agent_path: &Path) -> PathBuf {
        agent_path.join(RUN_TRACE_CORRUPTION_SIDECAR_DIR)
    }

    fn run_trace_corruption_sidecar_path(agent_path: &Path, key: &[u8; 32]) -> PathBuf {
        let name = key
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect::<String>();
        Self::run_trace_corruption_sidecar_root(agent_path).join(format!("{name}.json"))
    }

    fn read_run_trace_corruption_marker(
        &self,
        agent_path: &Path,
        agent_id: &str,
        run_id: &str,
    ) -> std::io::Result<Option<RunTraceCorruption>> {
        let key = Self::run_summary_key(run_id);
        let path = Self::run_trace_corruption_sidecar_path(agent_path, &key);
        let file = match car_secrets::open_private_read(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let marker: RunTraceCorruptionMarker = serde_json::from_reader(file)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        if marker.run_id != run_id || marker.agent_id != agent_id {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "run trace corruption marker identity does not match its path",
            ));
        }
        Ok(Some(marker.corruption))
    }

    fn write_run_trace_corruption_marker(
        &self,
        agent_path: &Path,
        agent_id: &str,
        run_id: &str,
        corruption: &RunTraceCorruption,
    ) -> std::io::Result<()> {
        if let Some(existing) =
            self.read_run_trace_corruption_marker(agent_path, agent_id, run_id)?
        {
            if existing == *corruption {
                return Ok(());
            }
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "run trace corruption marker conflicts with an existing marker",
            ));
        }
        if self
            .failures
            .take(RunStoreFailurePoint::CorruptionMarkerWrite)
        {
            return Err(injected_storage_full());
        }
        let root = Self::run_trace_corruption_sidecar_root(agent_path);
        self.ensure_private_dir(&root)?;
        let path =
            Self::run_trace_corruption_sidecar_path(agent_path, &Self::run_summary_key(run_id));
        self.write_private_json_atomic(
            &root,
            &path,
            &RunTraceCorruptionMarker {
                run_id: run_id.to_string(),
                agent_id: agent_id.to_string(),
                corruption: corruption.clone(),
            },
        )
    }

    fn persist_run_trace_corruption_marker(
        &self,
        agent_path: &Path,
        agent_id: &str,
        run_id: &str,
        corruption: &RunTraceCorruption,
    ) -> std::io::Result<()> {
        let error = match self
            .write_run_trace_corruption_marker(agent_path, agent_id, run_id, corruption)
        {
            Ok(()) => return Ok(()),
            Err(error) => error,
        };
        if self
            .read_run_trace_corruption_marker(agent_path, agent_id, run_id)
            .ok()
            .flatten()
            .is_some()
        {
            return Err(error);
        }

        // If the independent marker cannot be made durable, invalidate the
        // derived healthy summary. The bounded list path then fails closed
        // instead of trusting stale state; the authoritative JSONL remains.
        let sidecar_root = Self::run_summary_sidecar_root(agent_path);
        let sidecar = Self::run_summary_sidecar_path(agent_path, &Self::run_summary_key(run_id));
        if self
            .failures
            .take(RunStoreFailurePoint::CorruptionSummaryInvalidate)
        {
            return Err(std::io::Error::new(
                error.kind(),
                format!("{error}; stale run summary invalidation also failed: permission denied"),
            ));
        }
        match std::fs::remove_file(&sidecar) {
            Ok(()) => sync_directory(&sidecar_root)?,
            Err(remove_error) if remove_error.kind() == std::io::ErrorKind::NotFound => {}
            Err(remove_error) => {
                return Err(std::io::Error::new(
                    error.kind(),
                    format!("{error}; stale run summary invalidation also failed: {remove_error}"),
                ));
            }
        }
        Err(error)
    }

    fn remember_run_trace_corruption(
        &self,
        agent_id: &str,
        run_id: &str,
        corruption: &RunTraceCorruption,
    ) {
        self.trace_corruptions
            .lock()
            .expect("run trace corruption registry poisoned")
            .insert(
                (agent_id.to_string(), run_id.to_string()),
                corruption.clone(),
            );
    }

    fn known_run_trace_corruption(
        &self,
        agent_id: &str,
        run_id: &str,
    ) -> Option<RunTraceCorruption> {
        self.trace_corruptions
            .lock()
            .expect("run trace corruption registry poisoned")
            .get(&(agent_id.to_string(), run_id.to_string()))
            .cloned()
    }

    /// Publish corruption discovered by a trust-bearing read through the same
    /// marker/summary fail-closed boundary used by writers. The in-process
    /// registry is updated first so even marker ENOSPC cannot make a later
    /// read or write treat the run as healthy in this process.
    fn publish_strict_read_corruption(
        &self,
        agent_path: &Path,
        agent_id: &str,
        run_id: &str,
        error: std::io::Error,
    ) -> std::io::Error {
        let Some(corruption) = trace_corruption_from_error(&error) else {
            return error;
        };
        self.remember_run_trace_corruption(agent_id, run_id, &corruption);
        match self.persist_run_trace_corruption_marker(agent_path, agent_id, run_id, &corruption) {
            Ok(()) => error,
            Err(source) => summary_refresh_error(source, corruption),
        }
    }

    fn apply_run_trace_corruption_marker(
        &self,
        agent_path: &Path,
        summary: &mut RunSummary,
    ) -> std::io::Result<()> {
        let corruption = match self.known_run_trace_corruption(&summary.agent_id, &summary.run_id) {
            Some(corruption) => Some(corruption),
            None => self.read_run_trace_corruption_marker(
                agent_path,
                &summary.agent_id,
                &summary.run_id,
            )?,
        };
        if let Some(corruption) = corruption {
            summary.status = RunStatus::Incomplete;
            summary.trace_corruption = Some(corruption);
        }
        Ok(())
    }

    fn read_summary_sidecar(
        &self,
        agent_path: &Path,
        key: &[u8; 32],
    ) -> std::io::Result<RunSummary> {
        let path = Self::run_summary_sidecar_path(agent_path, key);
        let file = car_secrets::open_private_read(&path)?;
        let summary: RunSummary = serde_json::from_reader(file)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        if Self::run_summary_key(&summary.run_id) != *key {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "run summary sidecar identity does not match its index key",
            ));
        }
        Ok(summary)
    }

    fn write_summary_sidecar(
        &self,
        agent_path: &Path,
        summary: &RunSummary,
    ) -> std::io::Result<()> {
        if let Some(gate) = &self.summary_write_gate {
            gate.wait_if_armed();
        }
        if self.failures.take(RunStoreFailurePoint::SummaryWrite) {
            return Err(injected_storage_full());
        }
        let root = Self::run_summary_sidecar_root(agent_path);
        self.ensure_private_dir(&root)?;
        let path =
            Self::run_summary_sidecar_path(agent_path, &Self::run_summary_key(&summary.run_id));
        self.write_private_json_atomic(&root, &path, summary)
    }

    fn write_summary_index(
        &self,
        agent_path: &Path,
        summaries: &[RunSummary],
    ) -> std::io::Result<()> {
        let mut ordered = summaries.to_vec();
        let mut next_sequence = ordered.iter().map(|row| row.sequence).max().unwrap_or(0);
        let mut missing = ordered
            .iter_mut()
            .filter(|row| row.sequence == 0)
            .collect::<Vec<_>>();
        missing.sort_by(|left, right| {
            left.started_at
                .cmp(&right.started_at)
                .then_with(|| right.run_id.cmp(&left.run_id))
        });
        for summary in missing {
            next_sequence = next_sequence.checked_add(1).ok_or_else(|| {
                std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "run summary sequence exhausted",
                )
            })?;
            summary.sequence = next_sequence;
        }
        ordered.sort_by(|left, right| right.sequence.cmp(&left.sequence));
        if ordered
            .windows(2)
            .any(|pair| pair[0].sequence == pair[1].sequence)
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "run summary index contains duplicate sequences",
            ));
        }
        let sidecar_root = Self::run_summary_sidecar_root(agent_path);
        self.ensure_private_dir(&sidecar_root)?;
        let mut retained = std::collections::HashSet::new();
        let mut bytes = Vec::with_capacity(ordered.len() * RUN_SUMMARY_INDEX_RECORD_BYTES);
        for summary in &ordered {
            let key = Self::run_summary_key(&summary.run_id);
            retained.insert(key);
            self.write_summary_sidecar(agent_path, summary)?;
            bytes.extend_from_slice(&summary.sequence.to_be_bytes());
            bytes.extend_from_slice(&key);
        }
        if let Ok(entries) = std::fs::read_dir(&sidecar_root) {
            for entry in entries.flatten() {
                let path = entry.path();
                let Some(stem) = path.file_stem().and_then(|value| value.to_str()) else {
                    continue;
                };
                let is_retained = retained.iter().any(|key| {
                    key.iter()
                        .map(|byte| format!("{byte:02x}"))
                        .collect::<String>()
                        == stem
                });
                if !is_retained && path.extension().and_then(|value| value.to_str()) == Some("json")
                {
                    let _ = std::fs::remove_file(path);
                }
            }
        }
        let path = Self::run_summary_index_path(agent_path);
        let temp = agent_path.join(format!(
            ".run-summary-index-v2.{}.tmp",
            uuid::Uuid::new_v4().simple()
        ));
        let result = (|| {
            let mut file = self.create_private_file(&temp)?;
            file.write_all(&bytes)?;
            file.flush()?;
            file.sync_all()?;
            car_secrets::revalidate_private_path(&temp, &file)?;
            drop(file);
            car_secrets::atomic_replace_private_file(&temp, &path)?;
            sync_directory(agent_path)
        })();
        if result.is_err() {
            let _ = std::fs::remove_file(temp);
        }
        result
    }

    fn refresh_run_summary(
        &self,
        agent_id: &str,
        run_id: &str,
        add_to_order: bool,
    ) -> std::io::Result<()> {
        let agent_path = self.open_agent_dir_for_read(agent_id)?;
        // Every summary refresh for an agent shares the index path as its
        // serialization key. Terminal refreshes update one sidecar; new-run
        // refreshes rewrite every retained sidecar plus the index. The lock
        // starts before the scan so a slower, older refresh cannot publish a
        // stale summary after a newer lifecycle boundary.
        let index_path = Self::run_summary_index_path(&agent_path);
        let index_lock = self.append_lock(&index_path);
        let _guard = index_lock
            .lock()
            .map_err(|_| std::io::Error::other("run summary index lock poisoned"))?;
        let path = self.run_path(agent_id, run_id);
        let scan = match summarize_file_checked(&path) {
            Ok(scan) => scan,
            Err(error) => {
                let Some(corruption) = trace_corruption_from_error(&error) else {
                    return Err(error);
                };
                self.remember_run_trace_corruption(agent_id, run_id, &corruption);
                self.persist_run_trace_corruption_marker(
                    &agent_path,
                    agent_id,
                    run_id,
                    &corruption,
                )
                .map_err(|source| summary_refresh_error(source, corruption))?;
                return Err(error);
            }
        };
        if let Some(corruption) = &scan.corruption {
            self.remember_run_trace_corruption(agent_id, run_id, corruption);
            self.persist_run_trace_corruption_marker(&agent_path, agent_id, run_id, corruption)
                .map_err(|source| summary_refresh_error(source, corruption.clone()))?;
        }
        let mut summary = scan.summary.ok_or_else(|| {
            let error = std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "durable trace cannot be summarized for the run index",
            );
            match scan.corruption.clone() {
                Some(corruption) => summary_refresh_error(error, corruption),
                None => error,
            }
        })?;
        self.apply_run_trace_corruption_marker(&agent_path, &mut summary)?;
        let corruption = summary.trace_corruption.clone();
        let update = (|| {
            if !add_to_order {
                let key = Self::run_summary_key(run_id);
                summary.sequence = self.read_summary_sidecar(&agent_path, &key)?.sequence;
                self.write_summary_sidecar(&agent_path, &summary)?;
                return Ok(());
            }
            let mut summaries = Vec::new();
            if let Ok(mut index) = car_secrets::open_private_read(&index_path) {
                let len = index.metadata()?.len() as usize;
                if !len.is_multiple_of(RUN_SUMMARY_INDEX_RECORD_BYTES) {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "run summary index has a partial record",
                    ));
                }
                let mut record = [0u8; RUN_SUMMARY_INDEX_RECORD_BYTES];
                while index.read_exact(&mut record).is_ok() {
                    let sequence = u64::from_be_bytes(record[..8].try_into().expect("eight bytes"));
                    let key: [u8; RUN_SUMMARY_KEY_BYTES] =
                        record[8..].try_into().expect("summary key bytes");
                    let existing = self.read_summary_sidecar(&agent_path, &key)?;
                    if existing.sequence != sequence {
                        return Err(std::io::Error::new(
                            std::io::ErrorKind::InvalidData,
                            "run summary sidecar sequence does not match its index record",
                        ));
                    }
                    if existing.run_id != run_id {
                        summaries.push(existing);
                    } else {
                        summary.sequence = sequence;
                    }
                }
            }
            summaries.push(summary.clone());
            self.write_summary_index(&agent_path, &summaries)
        })();
        preserve_summary_corruption(update, &corruption)?;
        reject_corrupt_summary(&summary)
    }

    /// Ensure the `runs/` root exists with `0700` perms and is marked
    /// backup-excluded. Idempotent. Called lazily on the first append so
    /// constructing a `RunStore` is free (no disk touch until a run is
    /// actually recorded).
    fn ensure_root(&self) -> std::io::Result<()> {
        self.ensure_private_dir(&self.root)?;
        ensure_backup_excluded(&self.root)?;
        Ok(())
    }

    /// Validate and harden an existing runs root without creating a missing
    /// store as a side effect of a replay/list request.
    fn open_root_for_read(&self) -> std::io::Result<()> {
        std::fs::symlink_metadata(&self.root)?;
        self.ensure_root()
    }

    /// Ensure an agent's run dir exists `0700`.
    fn ensure_agent_dir(&self, agent_id: &str) -> std::io::Result<PathBuf> {
        self.ensure_root()?;
        let dir = self.root.join(sanitize(agent_id));
        self.ensure_private_dir(&dir)?;
        Ok(dir)
    }

    /// Validate and harden an existing agent directory without creating an
    /// empty directory for an unknown replay/list key.
    fn open_agent_dir_for_read(&self, agent_id: &str) -> std::io::Result<PathBuf> {
        self.open_root_for_read()?;
        let dir = self.root.join(sanitize(agent_id));
        std::fs::symlink_metadata(&dir)?;
        self.ensure_private_dir(&dir)?;
        Ok(dir)
    }

    /// Append one or more `RunRecord`s to a run's file, creating it `0600`
    /// on first write. Records are written one JSONL line each, in order.
    ///
    /// This is the single low-level flush primitive the wiring calls at
    /// each boundary: `RunStarted` on `runs.start`, `RunTurn`s as the
    /// recorder produces them, and the terminal `RunEnded`/`Incomplete` on
    /// `runs.complete`/disconnect.
    pub fn append_records(
        &self,
        agent_id: &str,
        run_id: &str,
        records: &[RunRecord],
    ) -> std::io::Result<()> {
        if records.is_empty() {
            return Ok(());
        }
        self.ensure_agent_dir(agent_id)?;
        let path = self.run_path(agent_id, run_id);
        let append_lock = self.append_lock(&path);
        let _guard = append_lock
            .lock()
            .map_err(|_| std::io::Error::other("run append lock poisoned"))?;
        let existed_before_open = path.exists();
        let mut file = self.open_private_append(&path)?;
        if self.failures.take(RunStoreFailurePoint::Write) {
            return Err(injected_storage_full());
        }
        append_jsonl_batch_to_path(&path, &mut file, records)?;
        self.durable_receipt(&path, &mut file, existed_before_open)
    }

    /// Append the `RunStarted` line + create the run file (`runs.start`).
    pub fn write_started(&self, started: &car_proto::RunStarted) -> std::io::Result<()> {
        if started.run_id.is_empty()
            || started.agent_id.is_empty()
            || started.client_id.as_deref().is_none_or(str::is_empty)
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "active RunStarted requires client_id",
            ));
        }
        self.ensure_boundary_record(
            &started.agent_id,
            &started.run_id,
            RunRecord::Started(started.clone()),
        )?;
        self.refresh_run_summary(&started.agent_id, &started.run_id, true)
    }

    /// Append `RunTurn` records (the recorder's per-proposal output).
    pub fn append_turns(
        &self,
        agent_id: &str,
        run_id: &str,
        turns: &[RunRecord],
    ) -> std::io::Result<()> {
        if let Some(gate) = &self.append_gate {
            gate.wait_if_armed();
        }
        self.append_records(agent_id, run_id, turns)?;
        if let Err(error) = self.refresh_run_summary(agent_id, run_id, false) {
            if is_trace_corruption_error(&error) {
                return Err(error);
            }
            tracing::error!(%agent_id, %run_id, %error, "run summary index update deferred until startup repair");
        }
        Ok(())
    }

    /// Read the durable summary's corruption marker without materializing the
    /// trace. Active subscribe reads this off the async lock path, then
    /// revalidates live state before it trusts a snapshot or registers.
    pub fn run_trace_corruption_for(
        &self,
        agent_id: &str,
        run_id: &str,
    ) -> std::io::Result<Option<RunTraceCorruption>> {
        if let Some(gate) = &self.summary_read_gate {
            gate.wait_if_armed();
        }
        if let Some(corruption) = self.known_run_trace_corruption(agent_id, run_id) {
            return Ok(Some(corruption));
        }
        let agent_path = match self.open_agent_dir_for_read(agent_id) {
            Ok(path) => path,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        if let Some(corruption) =
            self.read_run_trace_corruption_marker(&agent_path, agent_id, run_id)?
        {
            return Ok(Some(corruption));
        }
        let key = Self::run_summary_key(run_id);
        match self.read_summary_sidecar(&agent_path, &key) {
            Ok(summary) => {
                if summary.run_id != run_id || summary.agent_id != agent_id {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "run summary sidecar owner does not match requested run",
                    ));
                }
                Ok(summary.trace_corruption)
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(error) => Err(error),
        }
    }

    /// Ensure the exact action trace authenticated by a durable proposal
    /// finalization exists once before the terminal journal/receipt can
    /// advance. Existing exact rows make retries idempotent; partial or
    /// conflicting rows fail closed.
    pub fn ensure_proposal_turns(
        &self,
        pending: &PendingProposalFinalization,
    ) -> std::io::Result<ProposalTraceEnsure> {
        if let Some(gate) = &self.append_gate {
            gate.wait_if_armed();
        }
        pending.validate()?;
        let started = self.durable_run_started(&pending.run_id)?;
        match self.execution_marker(&pending.run_id)? {
            Some(marker) => pending.validate_provenance(&started, &marker)?,
            None => {
                let completed = self.completed_proposal(
                    &pending.run_id,
                    &pending.client_id,
                    pending.requested_policy_session_id.as_deref(),
                    &pending.original_submission,
                )?;
                if completed.as_ref().map(|receipt| &receipt.finalization) != Some(pending) {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "proposal trace has neither its execution marker nor an exact completed receipt",
                    ));
                }
            }
        }
        let mut expected = crate::run_trace::record_turns(
            &pending.final_proposal,
            &pending.proposal_result.results,
            0,
        );
        for record in &mut expected {
            let RunRecord::Turn(turn) = record else {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "proposal trace generator returned a non-turn record",
                ));
            };
            if !crate::handler::enforce_proposal_turn_byte_cap(turn) {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "proposal turn exceeds the durable byte limit without changing authenticated parameters",
                ));
            }
        }
        if expected.is_empty() {
            return Ok(ProposalTraceEnsure {
                records: expected,
                appended: false,
            });
        }

        let path = self.run_path(&started.agent_id, &pending.run_id);
        let append_lock = self.append_lock(&path);
        let _guard = append_lock
            .lock()
            .map_err(|_| std::io::Error::other("proposal trace append lock poisoned"))?;
        let file = car_secrets::open_private_read(&path)?;
        let mut turn_count = 0usize;
        let mut matching = Vec::new();
        let scan = scan_records(&file, |record| {
            if let RunRecord::Turn(turn) = record {
                turn_count = turn_count.saturating_add(1);
                if turn.proposal_id.as_deref() == Some(pending.final_proposal_id.as_str()) {
                    matching.push(RunRecord::Turn(turn));
                }
            }
        });
        if let Err(error) = scan {
            let error = scan_error_to_io(error);
            let _ = self.refresh_run_summary(&started.agent_id, &pending.run_id, false);
            return Err(error);
        }
        car_secrets::revalidate_private_path(&path, &file)?;

        let to_append = if !matching.is_empty() {
            let start = matching
                .first()
                .and_then(|record| match record {
                    RunRecord::Turn(turn) => Some(turn.index),
                    _ => None,
                })
                .unwrap_or(0);
            for (offset, record) in expected.iter_mut().enumerate() {
                if let RunRecord::Turn(turn) = record {
                    turn.index = start + offset;
                }
            }
            let exact_prefix = matching.len() <= expected.len()
                && matching.iter().zip(&expected).all(|(left, right)| {
                    serde_json::to_value(left).ok() == serde_json::to_value(right).ok()
                });
            if !exact_prefix || turn_count != start.saturating_add(matching.len()) {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "durable proposal trace conflicts with final proposal identity or has later rows",
                ));
            }
            if matching.len() == expected.len() {
                return Ok(ProposalTraceEnsure {
                    records: matching,
                    appended: false,
                });
            }
            expected[matching.len()..].to_vec()
        } else {
            for (offset, record) in expected.iter_mut().enumerate() {
                if let RunRecord::Turn(turn) = record {
                    turn.index = turn_count + offset;
                }
            }
            expected.clone()
        };

        if turn_count.saturating_add(to_append.len()) > crate::session::RECORD_TURNS_RUN_CEILING {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "proposal trace would exceed the durable run turn ceiling",
            ));
        }

        let existed_before_open = path.exists();
        let mut append = self.open_private_append(&path)?;
        if self.failures.take(RunStoreFailurePoint::Write) {
            return Err(injected_storage_full());
        }
        append_jsonl_batch_to_path(&path, &mut append, &to_append)?;
        self.durable_receipt(&path, &mut append, existed_before_open)?;
        if let Err(error) = self.refresh_run_summary(&started.agent_id, &pending.run_id, false) {
            if is_trace_corruption_error(&error) {
                return Err(error);
            }
            tracing::error!(run_id = %pending.run_id, %error, "proposal trace summary update deferred until startup repair");
        }
        Ok(ProposalTraceEnsure {
            records: expected,
            appended: true,
        })
    }

    /// Append the terminal `RunEnded` line (`runs.complete` or the
    /// disconnect-`Incomplete` path).
    pub fn write_ended(&self, ended: &car_proto::RunEnded) -> std::io::Result<()> {
        if ended.run_id.is_empty()
            || ended.agent_id.is_empty()
            || ended.client_id.as_deref().is_none_or(str::is_empty)
            || ended.completion_digest.as_deref().is_none_or(str::is_empty)
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "active RunEnded requires client_id and completion_digest",
            ));
        }
        let expected_digest = crate::session::run_completion_digest(&ended.termination)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        if ended.completion_digest.as_deref() != Some(expected_digest.as_str()) {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "RunEnded completion_digest does not match JCS RunTermination",
            ));
        }
        self.ensure_boundary_record(
            &ended.agent_id,
            &ended.run_id,
            RunRecord::Ended(ended.clone()),
        )?;
        self.refresh_run_summary(&ended.agent_id, &ended.run_id, false)
    }

    /// Durably append the body-free cancellation request before attempting
    /// control. The exact key/preimage is idempotent; another key cannot race
    /// the active request.
    pub fn write_cancellation_requested(
        &self,
        agent_id: &str,
        requested: &car_proto::RunCancellationRequested,
    ) -> std::io::Result<()> {
        self.ensure_boundary_record(
            agent_id,
            &requested.run_id,
            RunRecord::CancellationRequested(requested.clone()),
        )?;
        self.refresh_run_summary(agent_id, &requested.run_id, false)
    }

    /// Persist an unconfirmed cancellation receipt. Confirmed cancellation is
    /// represented by the terminal `RunEnded::Cancelled` row instead.
    pub fn write_cancellation_result(
        &self,
        agent_id: &str,
        result: &car_proto::RunCancelResponse,
    ) -> std::io::Result<()> {
        self.ensure_boundary_record(
            agent_id,
            &result.run_id,
            RunRecord::CancellationResult(result.clone()),
        )?;
        self.refresh_run_summary(agent_id, &result.run_id, false)
    }

    fn append_lock(&self, path: &Path) -> Arc<Mutex<()>> {
        let mut locks = self
            .append_locks
            .lock()
            .expect("run append-lock registry poisoned");
        if let Some(lock) = locks.get(path).and_then(Weak::upgrade) {
            return lock;
        }
        locks.retain(|_, lock| lock.strong_count() > 0);
        let lock = Arc::new(Mutex::new(()));
        locks.insert(path.to_path_buf(), Arc::downgrade(&lock));
        lock
    }

    fn ensure_boundary_record(
        &self,
        agent_id: &str,
        run_id: &str,
        record: RunRecord,
    ) -> std::io::Result<()> {
        self.ensure_agent_dir(agent_id)?;
        let path = self.run_path(agent_id, run_id);
        let append_lock = self.append_lock(&path);
        let _guard = append_lock
            .lock()
            .map_err(|_| std::io::Error::other("run append lock poisoned"))?;
        let existed_before_open = path.exists();
        let mut file = self.open_private_append(&path)?;
        car_secrets::revalidate_private_path(&path, &file)?;
        let existing = match load_records(&file) {
            Ok(records) => records,
            Err(error) => {
                // Publish the fail-closed summary marker before returning the
                // corruption error to the attempted boundary writer.
                let _ = self.refresh_run_summary(agent_id, run_id, false);
                return Err(error);
            }
        };

        let exact_exists = match &record {
            RunRecord::Started(wanted) => {
                if let Some(found) = existing.iter().find_map(|row| match row {
                    RunRecord::Started(started) => Some(started),
                    _ => None,
                }) {
                    if found != wanted {
                        return Err(std::io::Error::new(
                            std::io::ErrorKind::AlreadyExists,
                            "run_id is already reserved by a different RunStarted owner/preimage",
                        ));
                    }
                    true
                } else {
                    if !existing.is_empty() {
                        return Err(std::io::Error::new(
                            std::io::ErrorKind::InvalidData,
                            "non-empty run trace is missing its RunStarted boundary",
                        ));
                    }
                    false
                }
            }
            RunRecord::Ended(wanted) => {
                let started = existing.iter().find_map(|row| match row {
                    RunRecord::Started(started) => Some(started),
                    _ => None,
                });
                let Some(started) = started else {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "RunEnded cannot precede RunStarted",
                    ));
                };
                if started.run_id != wanted.run_id
                    || started.agent_id != wanted.agent_id
                    || started.client_id != wanted.client_id
                {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::PermissionDenied,
                        "RunEnded owner does not match durable RunStarted",
                    ));
                }
                if let Some(found) = existing.iter().find_map(|row| match row {
                    RunRecord::Ended(ended) => Some(ended),
                    _ => None,
                }) {
                    if !same_run_ended(found, wanted) {
                        return Err(std::io::Error::new(
                            std::io::ErrorKind::AlreadyExists,
                            "run already has a different durable terminal",
                        ));
                    }
                    true
                } else {
                    false
                }
            }
            RunRecord::Turn(_) => false,
            RunRecord::CancellationRequested(wanted) => {
                let started = existing.iter().find_map(|row| match row {
                    RunRecord::Started(started) => Some(started),
                    _ => None,
                });
                let Some(started) = started else {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "cancellation request cannot precede RunStarted",
                    ));
                };
                if started.run_id != wanted.run_id || started.agent_id != agent_id {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::PermissionDenied,
                        "cancellation request owner does not match durable RunStarted",
                    ));
                }
                if existing
                    .iter()
                    .any(|row| matches!(row, RunRecord::Ended(_)))
                {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::AlreadyExists,
                        "run is already terminal",
                    ));
                }
                if let Some(found) = existing.iter().find_map(|row| match row {
                    RunRecord::CancellationRequested(row) => Some(row),
                    _ => None,
                }) {
                    if found != wanted {
                        return Err(std::io::Error::new(
                            std::io::ErrorKind::AlreadyExists,
                            "run already has a different cancellation request",
                        ));
                    }
                    true
                } else {
                    false
                }
            }
            RunRecord::CancellationResult(wanted) => {
                if wanted.status != car_proto::RunCancellationStatus::TerminationUnconfirmed
                    || wanted.terminal_digest.is_some()
                {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidInput,
                        "only body-free termination_unconfirmed receipts are nonterminal records",
                    ));
                }
                let request = existing.iter().find_map(|row| match row {
                    RunRecord::CancellationRequested(row) => Some(row),
                    _ => None,
                });
                let Some(request) = request else {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "cancellation result cannot precede its request",
                    ));
                };
                if request.run_id != wanted.run_id
                    || request.idempotency_key != wanted.idempotency_key
                    || request.reason_digest != wanted.reason_digest
                    || request.principal != wanted.principal
                    || request.action_id != wanted.action_id
                    || request.request_id != wanted.request_id
                {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "cancellation result conflicts with its durable request",
                    ));
                }
                if existing
                    .iter()
                    .any(|row| matches!(row, RunRecord::Ended(_)))
                {
                    return Err(std::io::Error::new(
                        std::io::ErrorKind::AlreadyExists,
                        "run is already terminal",
                    ));
                }
                if let Some(found) = existing.iter().find_map(|row| match row {
                    RunRecord::CancellationResult(row) => Some(row),
                    _ => None,
                }) {
                    if found != wanted {
                        return Err(std::io::Error::new(
                            std::io::ErrorKind::AlreadyExists,
                            "run already has a different cancellation result",
                        ));
                    }
                    true
                } else {
                    false
                }
            }
        };

        if !exact_exists {
            if self.failures.take(RunStoreFailurePoint::Write) {
                return Err(injected_storage_full());
            }
            append_jsonl_batch_to_path(&path, &mut file, &[record])?;
        }
        self.durable_receipt(&path, &mut file, existed_before_open)
    }

    fn durable_receipt(
        &self,
        path: &Path,
        file: &mut File,
        existed_before_open: bool,
    ) -> std::io::Result<()> {
        if self.failures.take(RunStoreFailurePoint::Flush) {
            return Err(std::io::Error::other("injected run-store flush failure"));
        }
        file.flush()?;
        if self.failures.take(RunStoreFailurePoint::Fsync) {
            return Err(std::io::Error::other("injected run-store fsync failure"));
        }
        file.sync_all()?;
        if !existed_before_open {
            if let Some(parent) = path.parent() {
                sync_directory(parent)?;
            }
        }
        car_secrets::revalidate_private_path(path, file)
    }

    /// Load a run's full ordered trace from disk by `run_id` — the U5
    /// `runs.get_trace` read path. Works after a restart when memory is
    /// empty. Resolves `run_id -> agent_id` by scanning the tree, then
    /// reads the JSONL. A malformed unterminated final crash tail is ignored;
    /// committed corruption makes this compatibility wrapper return `None`.
    /// Returns `None` when no file exists for the `run_id`.
    pub fn get_run_trace(&self, run_id: &str) -> Option<Vec<RunRecord>> {
        self.get_run_trace_checked(run_id).ok().flatten()
    }

    /// Strict replay read used by trust-bearing RPC surfaces. A malformed
    /// newline-terminated row rejects the trace; only a torn final row is
    /// ignored as an uncommitted crash tail.
    pub fn get_run_trace_checked(&self, run_id: &str) -> std::io::Result<Option<Vec<RunRecord>>> {
        let Some((path, file)) = self.resolve_run_file(run_id) else {
            return Ok(None);
        };
        let Some(agent_path) = path.parent() else {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "run trace path has no owning agent directory",
            ));
        };
        let Some(agent_id) = agent_path.file_name().and_then(|value| value.to_str()) else {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "run trace path has no valid owning agent id",
            ));
        };
        load_private_records(&path, &file)
            .map(Some)
            .map_err(|error| {
                self.publish_strict_read_corruption(agent_path, agent_id, run_id, error)
            })
    }

    /// Load a run's trace given both keys (cheaper — no tree scan). Used
    /// by `list_runs` internally and available to callers that already
    /// know the owning agent.
    pub fn get_run_trace_for(&self, agent_id: &str, run_id: &str) -> Option<Vec<RunRecord>> {
        self.get_run_trace_for_checked(agent_id, run_id)
            .ok()
            .flatten()
    }

    pub fn get_run_trace_for_checked(
        &self,
        agent_id: &str,
        run_id: &str,
    ) -> std::io::Result<Option<Vec<RunRecord>>> {
        let agent_path = match self.open_agent_dir_for_read(agent_id) {
            Ok(path) => path,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let path = self.run_path(agent_id, run_id);
        let file = match car_secrets::open_private_read(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        load_private_records(&path, &file)
            .map(Some)
            .map_err(|error| {
                self.publish_strict_read_corruption(&agent_path, agent_id, run_id, error)
            })
    }

    /// Stream one bounded page from a run's JSONL trace. `limit + 1`
    /// records are retained only to determine whether a continuation exists;
    /// accumulated history is never materialized by the dashboard read path.
    pub fn get_run_trace_page_for(
        &self,
        agent_id: &str,
        run_id: &str,
        cursor: usize,
        limit: usize,
    ) -> std::io::Result<Option<(Vec<RunRecord>, Option<usize>)>> {
        let agent_path = match self.open_agent_dir_for_read(agent_id) {
            Ok(path) => path,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let path = self.run_path(agent_id, run_id);
        let file = match car_secrets::open_private_read(&path) {
            Ok(file) => file,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let mut records = Vec::with_capacity(limit.saturating_add(1));
        let mut valid_index = 0usize;
        if let Err(error) = scan_records(&file, |record| {
            if valid_index >= cursor && records.len() <= limit {
                records.push(record);
            }
            valid_index = valid_index.saturating_add(1);
        }) {
            let error = scan_error_to_io(error);
            return Err(self.publish_strict_read_corruption(&agent_path, agent_id, run_id, error));
        }
        car_secrets::revalidate_private_path(&path, &file)?;
        let has_more = records.len() > limit;
        records.truncate(limit);
        let next_cursor = has_more.then_some(cursor.saturating_add(records.len()));
        Ok(Some((records, next_cursor)))
    }

    /// Stream a bounded turns-only page for `runs.subscribe`. The exact total
    /// comes from the durable summary sidecar, so returning `live_cursor` never
    /// requires materializing the trace.
    pub fn get_run_turn_page_for(
        &self,
        agent_id: &str,
        run_id: &str,
        cursor: usize,
        limit: usize,
    ) -> std::io::Result<Option<(Vec<RunRecord>, Option<usize>, usize, RunStatus)>> {
        let agent_path = match self.open_agent_dir_for_read(agent_id) {
            Ok(path) => path,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        let key = Self::run_summary_key(run_id);
        let mut summary = match self.read_summary_sidecar(&agent_path, &key) {
            Ok(summary) if summary.run_id == run_id && summary.agent_id == agent_id => summary,
            Ok(_) => {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "run summary identity does not match the requested trace",
                ))
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error),
        };
        self.apply_run_trace_corruption_marker(&agent_path, &mut summary)?;
        reject_corrupt_summary(&summary)?;
        if cursor > summary.turn_count {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                format!(
                    "runs.subscribe cursor {cursor} exceeds live_cursor {}",
                    summary.turn_count
                ),
            ));
        }
        let path = self.run_path(agent_id, run_id);
        let file = car_secrets::open_private_read(&path)?;
        let mut turns = Vec::with_capacity(limit.saturating_add(1));
        let mut turn_index = 0usize;
        if let Err(error) = scan_records(&file, |record| {
            if !matches!(record, RunRecord::Turn(_)) {
                return;
            }
            if turn_index >= cursor && turns.len() <= limit {
                turns.push(record);
            }
            turn_index = turn_index.saturating_add(1);
        }) {
            let error = scan_error_to_io(error);
            return Err(self.publish_strict_read_corruption(&agent_path, agent_id, run_id, error));
        }
        car_secrets::revalidate_private_path(&path, &file)?;
        if turn_index < summary.turn_count && turns.len() <= limit {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "run trace ended before its durable summary turn count",
            ));
        }
        let has_more = turns.len() > limit;
        turns.truncate(limit);
        let next_cursor = has_more.then_some(cursor.saturating_add(turns.len()));
        Ok(Some((
            turns,
            next_cursor,
            summary.turn_count,
            summary.status,
        )))
    }

    /// Read a newest-first run-summary page through the durable fixed-record
    /// order index. The page performs at most `limit + 1` sidecar reads and no
    /// run-directory enumeration or trace scan.
    pub fn list_runs_page(
        &self,
        agent_id: &str,
        cursor: usize,
        limit: usize,
    ) -> std::io::Result<(Vec<RunSummary>, Option<usize>)> {
        let agent_path = match self.open_agent_dir_for_read(agent_id) {
            Ok(path) => path,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                return Ok((Vec::new(), None))
            }
            Err(error) => return Err(error),
        };
        let index_path = Self::run_summary_index_path(&agent_path);
        let mut index = car_secrets::open_private_read(&index_path)?;
        let byte_len = index.metadata()?.len() as usize;
        if !byte_len.is_multiple_of(RUN_SUMMARY_INDEX_RECORD_BYTES) {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "run summary index has a partial record",
            ));
        }
        let total = byte_len / RUN_SUMMARY_INDEX_RECORD_BYTES;
        if total == 0 {
            return Ok((Vec::new(), None));
        }

        // `0` means the current head. Every non-zero cursor is an exclusive
        // immutable sequence bound, so inserts at the head cannot shift a
        // continuation and duplicate/skip rows. Locate the first record with
        // sequence < cursor using O(log n) fixed-record seeks.
        let mut start = 0usize;
        if cursor != 0 {
            let cursor = u64::try_from(cursor).map_err(|_| {
                std::io::Error::new(std::io::ErrorKind::InvalidInput, "cursor overflow")
            })?;
            let mut low = 0usize;
            let mut high = total;
            while low < high {
                let mid = low + (high - low) / 2;
                index.seek(SeekFrom::Start(
                    (mid * RUN_SUMMARY_INDEX_RECORD_BYTES) as u64,
                ))?;
                let mut sequence = [0u8; 8];
                index.read_exact(&mut sequence)?;
                if u64::from_be_bytes(sequence) >= cursor {
                    low = mid + 1;
                } else {
                    high = mid;
                }
            }
            start = low;
        }
        if start >= total {
            return Ok((Vec::new(), None));
        }
        index.seek(SeekFrom::Start(
            (start * RUN_SUMMARY_INDEX_RECORD_BYTES) as u64,
        ))?;
        let count = limit.saturating_add(1).min(total - start);
        let mut summaries = Vec::with_capacity(count);
        for _ in 0..count {
            let mut record = [0u8; RUN_SUMMARY_INDEX_RECORD_BYTES];
            index.read_exact(&mut record)?;
            let sequence = u64::from_be_bytes(record[..8].try_into().expect("eight bytes"));
            let key: [u8; RUN_SUMMARY_KEY_BYTES] =
                record[8..].try_into().expect("summary key bytes");
            let mut summary = self.read_summary_sidecar(&agent_path, &key)?;
            if summary.agent_id != agent_id || summary.sequence != sequence {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "run summary index does not match its durable sidecar",
                ));
            }
            self.apply_run_trace_corruption_marker(&agent_path, &mut summary)?;
            summaries.push(summary);
        }
        let has_more = summaries.len() > limit;
        summaries.truncate(limit);
        let next_cursor = if has_more {
            summaries
                .last()
                .map(|summary| {
                    usize::try_from(summary.sequence).map_err(|_| {
                        std::io::Error::new(
                            std::io::ErrorKind::InvalidData,
                            "run summary sequence cannot be represented by the wire cursor",
                        )
                    })
                })
                .transpose()?
        } else {
            None
        };
        Ok((summaries, next_cursor))
    }

    /// List an agent's runs newest-first — the U5 `runs.list` read path.
    /// Each summary is built from the run file's records (head for
    /// `RunStarted`, tail for the terminal record, count of `Turn`s).
    /// Returns an empty Vec for an agent with no runs (the empty-state).
    pub fn list_runs(&self, agent_id: &str) -> Vec<RunSummary> {
        let Ok(dir) = self.open_agent_dir_for_read(agent_id) else {
            return Vec::new();
        };
        let mut out = Vec::new();
        let Ok(entries) = std::fs::read_dir(&dir) else {
            return out;
        };
        for entry in entries {
            let Ok(entry) = entry else {
                return Vec::new();
            };
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
                continue;
            }
            if let Some(mut summary) = summarize_file(&path) {
                if self
                    .apply_run_trace_corruption_marker(&dir, &mut summary)
                    .is_err()
                {
                    return Vec::new();
                }
                out.push(summary);
            }
        }
        // Newest first by start time.
        out.sort_by(|a, b| {
            b.started_at
                .cmp(&a.started_at)
                .then_with(|| a.run_id.cmp(&b.run_id))
        });
        out
    }

    /// Visit only the durable lifecycle boundaries for one trace at a time.
    /// Startup reconciliation needs neither turns nor a process-wide snapshot;
    /// streaming each file prevents accumulated newsroom history from being
    /// materialized before the daemon can listen and schedule.
    pub fn visit_run_boundaries<F>(&self, mut visitor: F)
    where
        F: FnMut(
            car_proto::RunStarted,
            Option<car_proto::RunEnded>,
            Option<car_proto::RunCancellationRequested>,
            Option<car_proto::RunCancelResponse>,
        ),
    {
        if self.open_root_for_read().is_err() {
            return;
        }
        let Ok(agent_dirs) = std::fs::read_dir(&self.root) else {
            return;
        };
        for agent in agent_dirs.flatten() {
            let Ok(file_type) = agent.file_type() else {
                continue;
            };
            if !file_type.is_dir() || file_type.is_symlink() {
                continue;
            }
            let Ok(entries) = std::fs::read_dir(agent.path()) else {
                continue;
            };
            for entry in entries.flatten() {
                let path = entry.path();
                if path.extension().and_then(|ext| ext.to_str()) != Some("jsonl") {
                    continue;
                }
                let Ok(file) = car_secrets::open_private_read(&path) else {
                    continue;
                };
                if let Some((started, ended, requested, result)) =
                    load_private_run_boundaries(&path, &file)
                {
                    visitor(started, ended, requested, result);
                }
            }
        }
    }

    /// Resolve `run_id -> path` by scanning `runs/*/{run_id}.jsonl`. U5's
    /// `runs.get_trace` takes only a `run_id`, so the owning agent must be
    /// discovered. Returns the first match (run ids are uuids — unique).
    fn resolve_run_file(&self, run_id: &str) -> Option<(PathBuf, File)> {
        self.open_root_for_read().ok()?;
        let file_name = format!("{}.jsonl", sanitize(run_id));
        let agent_dirs = std::fs::read_dir(&self.root).ok()?;
        for agent in agent_dirs {
            let agent = agent.ok()?;
            let file_type = agent.file_type().ok()?;
            if !file_type.is_dir() || file_type.is_symlink() {
                continue;
            }
            car_secrets::ensure_private_dir(&agent.path()).ok()?;
            let candidate = agent.path().join(&file_name);
            match car_secrets::open_private_read(&candidate) {
                Ok(file) => return Some((candidate, file)),
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
                Err(_) => return None,
            }
        }
        None
    }

    /// Resolve the owning `agent_id` for a `run_id` from disk. Mirrors
    /// [`Self::resolve_run_file`] but returns the agent dir name — U5's
    /// authorization check (KTD10) needs `run_id -> agent_id` to verify
    /// ownership before serving a trace.
    pub fn agent_for_run(&self, run_id: &str) -> Option<String> {
        let (path, _file) = self.resolve_run_file(run_id)?;
        path.parent()
            .and_then(Path::file_name)
            .and_then(|s| s.to_str())
            .map(str::to_string)
    }

    /// Retention GC (R6) — call on daemon boot. Per agent: keep the most
    /// recent `max_per_agent` **completed** runs and drop any completed
    /// run older than `max_age_days`, whichever is more restrictive. An
    /// in-progress run (no terminal record) is NEVER evicted. Returns the
    /// number of run files removed.
    pub fn gc(&self) -> usize {
        let mut removed = 0;
        if self.open_root_for_read().is_err() {
            return 0;
        }
        let Ok(agent_dirs) = std::fs::read_dir(&self.root) else {
            return 0;
        };
        let cutoff = self.retention.age_cutoff(Utc::now());
        for agent in agent_dirs {
            let Ok(agent) = agent else {
                return removed;
            };
            let Ok(file_type) = agent.file_type() else {
                return removed;
            };
            if !file_type.is_dir() || file_type.is_symlink() {
                continue;
            }
            let agent_path = agent.path();
            if car_secrets::ensure_private_dir(&agent_path).is_err() {
                return removed;
            }
            removed += self.gc_agent_dir(&agent_path, cutoff);
        }
        removed
    }

    /// Adopt crash-orphaned runs at boot (FIX 4). A daemon crash mid-run
    /// leaves an on-disk run with `RunStarted` (+ `Turn`s) but no terminal
    /// `RunEnded`, so it reads `InProgress` forever and the GC — which never
    /// evicts an in-progress run — can never reclaim it. The file leaks
    /// across every crash.
    ///
    /// This runs at store construction/boot, BEFORE `gc()`. At that moment
    /// the in-memory `runs` map is always empty, so any run that is
    /// `InProgress` on disk cannot have a live harness writing to it — it is
    /// necessarily a crashed prior process. We append an `Incomplete`
    /// terminal `RunEnded` marker to adopt it, making it terminal and thus
    /// age-GC-eligible (so a later `gc()` in the same boot can reclaim it).
    ///
    /// Returns the number of runs adopted. Best-effort: an unwritable file is
    /// skipped rather than failing startup.
    pub fn adopt_orphans(&self) -> usize {
        let mut adopted = 0;
        if self.open_root_for_read().is_err() {
            return 0;
        }
        let Ok(agent_dirs) = std::fs::read_dir(&self.root) else {
            return 0;
        };
        let now = Utc::now();
        for agent in agent_dirs {
            let Ok(agent) = agent else {
                return adopted;
            };
            let Ok(file_type) = agent.file_type() else {
                return adopted;
            };
            if !file_type.is_dir() || file_type.is_symlink() {
                continue;
            }
            let agent_path = agent.path();
            if car_secrets::ensure_private_dir(&agent_path).is_err() {
                return adopted;
            }
            let Ok(entries) = std::fs::read_dir(&agent_path) else {
                continue;
            };
            for entry in entries {
                let Ok(entry) = entry else {
                    return adopted;
                };
                let path = entry.path();
                if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
                    continue;
                }
                let Some(summary) = summarize_file(&path) else {
                    continue;
                };
                if summary.status != RunStatus::InProgress {
                    continue;
                }
                // A pre-dispatch marker with no exact result means the daemon
                // may have died after an irreversible effect. Preserve the
                // open run as quarantined/outcome-unknown; an automatic
                // Incomplete terminal would hide the unresolved proposal and
                // invite an unsafe retry.
                let pending_state = self.pending_proposal(&summary.run_id);
                let execution_state = self.execution_marker(&summary.run_id);
                if !matches!(pending_state, Ok(None)) || !matches!(execution_state, Ok(None)) {
                    continue;
                }
                // No terminal record + memory empty at boot => crash orphan.
                let client_id = self
                    .get_run_trace_for(&summary.agent_id, &summary.run_id)
                    .and_then(|records| {
                        records.into_iter().find_map(|record| match record {
                            RunRecord::Started(started) => started.client_id,
                            _ => None,
                        })
                    });
                let termination = RunTermination::Incomplete;
                let completion_digest = crate::session::run_completion_digest(&termination).ok();
                let incomplete = RunRecord::Ended(car_proto::RunEnded {
                    run_id: summary.run_id.clone(),
                    client_id,
                    agent_id: summary.agent_id.clone(),
                    termination,
                    completion_digest,
                    ended_at: now,
                });
                if self
                    .append_records(&summary.agent_id, &summary.run_id, &[incomplete])
                    .is_ok()
                {
                    adopted += 1;
                }
            }
        }
        adopted
    }

    /// GC one agent's run dir against the retention caps.
    fn gc_agent_dir(&self, agent_path: &Path, age_cutoff: Option<DateTime<Utc>>) -> usize {
        // Collect (path, summary) for every run file, ignoring unreadable
        // ones (a malformed file with no RunStarted can't be summarized;
        // leave it rather than risk evicting something we can't classify).
        let mut runs: Vec<(PathBuf, RunSummary)> = Vec::new();
        let Ok(entries) = std::fs::read_dir(agent_path) else {
            return 0;
        };
        for entry in entries {
            let Ok(entry) = entry else {
                return 0;
            };
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
                continue;
            }
            let file_run_id = path
                .file_stem()
                .and_then(|value| value.to_str())
                .unwrap_or_default();
            let key = Self::run_summary_key(file_run_id);
            let existing = self.read_summary_sidecar(agent_path, &key).ok();
            let scan = summarize_file_checked(&path);
            let mut summary = match scan {
                Ok(scan) => match scan.summary {
                    Some(summary) => {
                        if let Some(corruption) = &scan.corruption {
                            self.remember_run_trace_corruption(
                                &summary.agent_id,
                                file_run_id,
                                corruption,
                            );
                        }
                        summary
                    }
                    None => {
                        let Some(mut existing) = existing.clone() else {
                            continue;
                        };
                        let corruption = scan
                            .corruption
                            .or_else(|| {
                                self.known_run_trace_corruption(&existing.agent_id, file_run_id)
                            })
                            .or_else(|| {
                                self.read_run_trace_corruption_marker(
                                    agent_path,
                                    &existing.agent_id,
                                    file_run_id,
                                )
                                .ok()
                                .flatten()
                            });
                        let Some(corruption) = corruption else {
                            continue;
                        };
                        self.remember_run_trace_corruption(
                            &existing.agent_id,
                            file_run_id,
                            &corruption,
                        );
                        existing.status = RunStatus::Incomplete;
                        existing.trace_corruption = Some(corruption);
                        existing
                    }
                },
                Err(error) => {
                    let Some(corruption) = trace_corruption_from_error(&error) else {
                        continue;
                    };
                    let Some(mut existing) = existing.clone() else {
                        continue;
                    };
                    self.remember_run_trace_corruption(
                        &existing.agent_id,
                        file_run_id,
                        &corruption,
                    );
                    existing.status = RunStatus::Incomplete;
                    existing.trace_corruption = Some(corruption);
                    existing
                }
            };
            if let Some(existing) = existing {
                summary.sequence = existing.sequence;
            }
            if self
                .apply_run_trace_corruption_marker(agent_path, &mut summary)
                .is_err()
            {
                continue;
            }
            runs.push((path, summary));
        }
        // Newest-first so the most recent completed runs rank first.
        runs.sort_by(|a, b| b.1.started_at.cmp(&a.1.started_at));

        let mut removed = 0;
        // Rank among COMPLETED/Incomplete runs only — an in-progress run
        // must not consume a keeper slot, so the count cap is measured by
        // completed-run rank, not the combined sorted index (R6).
        let mut completed_rank = 0usize;
        for (path, summary) in runs.iter() {
            // Corruption is durable audit evidence, not an ordinary incomplete
            // run. Never age/count-evict it until an operator repairs it.
            if summary.trace_corruption.is_some() {
                continue;
            }
            // NEVER evict an in-progress run — it has no terminal record
            // and a live harness may still be writing to it (R6).
            if matches!(
                summary.status,
                RunStatus::InProgress | RunStatus::CancellationPending
            ) {
                continue;
            }
            let over_count = self
                .retention
                .count_cap()
                .is_some_and(|cap| completed_rank >= cap);
            completed_rank += 1;
            // Age the run by its TERMINAL time, not its start. A long run
            // that started >max_age_days ago but completed recently is still
            // a recent result and must not be evicted (FIX 2). Completed and
            // Incomplete runs always have an `ended_at`; fall back to
            // `started_at` only if a terminal record somehow lacks one.
            let term_time = summary.ended_at.unwrap_or(summary.started_at);
            let too_old = age_cutoff.is_some_and(|cut| term_time < cut);
            if (over_count || too_old) && std::fs::remove_file(path).is_ok() {
                removed += 1;
            }
        }
        let retained: Vec<RunSummary> = runs
            .into_iter()
            .filter_map(|(path, summary)| path.exists().then_some(summary))
            .collect();
        if let Err(error) = self.write_summary_index(agent_path, &retained) {
            tracing::error!(path = %agent_path.display(), %error, "run summary index startup repair failed");
        }
        removed
    }
}

fn same_run_ended(left: &car_proto::RunEnded, right: &car_proto::RunEnded) -> bool {
    serde_json::to_value(left).ok() == serde_json::to_value(right).ok()
}

/// Serialize and append a complete batch through one already-validated file
/// descriptor. Tail inspection and repair deliberately use the same
/// descriptor as the `O_APPEND` write: swapping the path after open cannot
/// redirect any bytes into a symlink target or replacement inode.
fn append_jsonl_batch_on_descriptor(file: &mut File, records: &[RunRecord]) -> std::io::Result<()> {
    car_secrets::revalidate_private_file(file)?;
    // FIX 6/7: build the WHOLE batch into one buffer and write it with a
    // single `write_all`. With `O_APPEND` a single write is positioned and
    // appended atomically relative to other appenders, so two concurrent
    // batches for the same file can't interleave mid-record. A leading '\n'
    // repairs a torn tail as part of the same atomic write.
    let mut buf = Vec::new();
    if last_byte_is_not_newline(file)? {
        buf.push(b'\n');
    }
    for rec in records {
        let line = serde_json::to_string(rec)
            .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
        buf.extend_from_slice(line.as_bytes());
        buf.push(b'\n');
    }
    file.write_all(&buf)?;
    car_secrets::revalidate_private_file(file)
}

/// Bind the descriptor-safe append to the canonical path before and after the
/// write. The second check converts a concurrent rename/substitution into a
/// failed append receipt while guaranteeing the replacement path was never
/// opened for writing.
fn append_jsonl_batch_to_path(
    path: &Path,
    file: &mut File,
    records: &[RunRecord],
) -> std::io::Result<()> {
    car_secrets::revalidate_private_path(path, file)?;
    append_jsonl_batch_on_descriptor(file, records)?;
    car_secrets::revalidate_private_path(path, file)
}

/// Return `true` when the opened file's last byte is not `\n` — i.e. the
/// previous append was torn. An empty file returns `false`. The caller passes
/// the already-validated read+append descriptor so no path lookup occurs
/// between inspection and repair.
fn last_byte_is_not_newline(file: &mut File) -> std::io::Result<bool> {
    let len = file.seek(SeekFrom::End(0))?;
    if len == 0 {
        return Ok(false);
    }
    file.seek(SeekFrom::End(-1))?;
    let mut buf = [0u8; 1];
    file.read_exact(&mut buf)?;
    Ok(buf[0] != b'\n')
}

#[derive(Debug)]
enum RecordScanError {
    Io(std::io::Error),
    Malformed { line: usize, detail: String },
}

#[derive(Debug)]
struct StrictRunTraceCorruptionError {
    corruption: RunTraceCorruption,
    detail: String,
}

impl std::fmt::Display for StrictRunTraceCorruptionError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            formatter,
            "malformed run trace record at line {}: {}",
            self.corruption.line, self.detail
        )
    }
}

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

fn scan_error_to_io(error: RecordScanError) -> std::io::Error {
    match error {
        RecordScanError::Io(error) => error,
        RecordScanError::Malformed { line, detail } => std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            StrictRunTraceCorruptionError {
                corruption: RunTraceCorruption {
                    kind: RunTraceCorruptionKind::MalformedRecord,
                    line,
                },
                detail,
            },
        ),
    }
}

/// Visit every committed JSONL row. A malformed newline-terminated row is
/// durable corruption and rejects the trace. A malformed final row without a
/// newline is an uncommitted crash tail and is ignored.
fn scan_records<F>(file: &File, mut visit: F) -> Result<(), RecordScanError>
where
    F: FnMut(RunRecord),
{
    let cloned = file.try_clone().map_err(RecordScanError::Io)?;
    let mut reader = std::io::BufReader::new(cloned);
    let mut bytes = Vec::new();
    let mut line = 0usize;
    loop {
        bytes.clear();
        let read = reader
            .read_until(b'\n', &mut bytes)
            .map_err(RecordScanError::Io)?;
        if read == 0 {
            break;
        }
        line = line.saturating_add(1);
        let terminated = bytes.last() == Some(&b'\n');
        if terminated {
            bytes.pop();
            if bytes.last() == Some(&b'\r') {
                bytes.pop();
            }
        }
        if bytes.iter().all(u8::is_ascii_whitespace) {
            continue;
        }
        match serde_json::from_slice::<RunRecord>(&bytes) {
            Ok(record) => visit(record),
            Err(_) if !terminated => break,
            Err(error) => {
                return Err(RecordScanError::Malformed {
                    line,
                    detail: error.to_string(),
                })
            }
        }
    }
    Ok(())
}

fn load_records(file: &File) -> std::io::Result<Vec<RunRecord>> {
    let mut records = Vec::new();
    scan_records(file, |record| records.push(record)).map_err(scan_error_to_io)?;
    Ok(records)
}

fn load_private_records(path: &Path, file: &File) -> std::io::Result<Vec<RunRecord>> {
    let records = load_records(file)?;
    car_secrets::revalidate_private_path(path, file)?;
    Ok(records)
}

/// Stream a trace and retain only its lifecycle boundary rows. This preserves
/// the JSONL reader's torn-line tolerance without allocating every recorded
/// turn during daemon startup.
fn load_private_run_boundaries(
    path: &Path,
    file: &File,
) -> Option<(
    car_proto::RunStarted,
    Option<car_proto::RunEnded>,
    Option<car_proto::RunCancellationRequested>,
    Option<car_proto::RunCancelResponse>,
)> {
    let mut started = None;
    let mut ended = None;
    let mut requested = None;
    let mut result = None;
    scan_records(file, |record| match record {
        RunRecord::Started(row) if started.is_none() => started = Some(row),
        RunRecord::Ended(row) if ended.is_none() => ended = Some(row),
        RunRecord::CancellationRequested(row) if requested.is_none() => requested = Some(row),
        RunRecord::CancellationResult(row) if result.is_none() => result = Some(row),
        _ => {}
    })
    .ok()?;
    car_secrets::revalidate_private_path(path, file).ok()?;
    started.map(|started| (started, ended, requested, result))
}

fn reject_corrupt_summary(summary: &RunSummary) -> std::io::Result<()> {
    if let Some(corruption) = &summary.trace_corruption {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("malformed run trace record at line {}", corruption.line),
        ));
    }
    Ok(())
}

#[derive(Debug)]
struct RunSummaryRefreshError {
    source: std::io::Error,
    corruption: RunTraceCorruption,
}

impl std::fmt::Display for RunSummaryRefreshError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            formatter,
            "malformed run trace record at line {}; summary persistence also failed: {}",
            self.corruption.line, self.source
        )
    }
}

impl std::error::Error for RunSummaryRefreshError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(&self.source)
    }
}

fn summary_refresh_error(source: std::io::Error, corruption: RunTraceCorruption) -> std::io::Error {
    std::io::Error::new(source.kind(), RunSummaryRefreshError { source, corruption })
}

fn preserve_summary_corruption<T>(
    result: std::io::Result<T>,
    corruption: &Option<RunTraceCorruption>,
) -> std::io::Result<T> {
    match corruption {
        Some(corruption) => {
            result.map_err(|source| summary_refresh_error(source, corruption.clone()))
        }
        None => result,
    }
}

pub(crate) fn trace_corruption_from_error(error: &std::io::Error) -> Option<RunTraceCorruption> {
    let source = error.get_ref()?;
    if let Some(failure) = source.downcast_ref::<RunSummaryRefreshError>() {
        return Some(failure.corruption.clone());
    }
    source
        .downcast_ref::<StrictRunTraceCorruptionError>()
        .map(|failure| failure.corruption.clone())
}

pub(crate) fn is_trace_corruption_error(error: &std::io::Error) -> bool {
    trace_corruption_from_error(error).is_some()
        || (error.kind() == std::io::ErrorKind::InvalidData
            && error.to_string().contains("malformed run trace record"))
}

fn summary_from_records(
    started: Option<car_proto::RunStarted>,
    ended: Option<car_proto::RunEnded>,
    turn_count: usize,
    cancellation_pending: bool,
    trace_corruption: Option<RunTraceCorruption>,
) -> Option<RunSummary> {
    let started = started?;
    let (status, ended_at) = if trace_corruption.is_some() {
        (RunStatus::Incomplete, None)
    } else {
        match &ended {
            Some(e) => {
                let status = match &e.termination {
                    RunTermination::Outcome { .. } => RunStatus::Completed,
                    RunTermination::Incomplete => RunStatus::Incomplete,
                    RunTermination::Cancelled { .. } => RunStatus::Cancelled,
                };
                (status, Some(e.ended_at))
            }
            None if cancellation_pending => (RunStatus::CancellationPending, None),
            None => (RunStatus::InProgress, None),
        }
    };
    Some(RunSummary {
        run_id: started.run_id,
        agent_id: started.agent_id,
        intent: started.intent,
        started_at: started.started_at,
        ended_at,
        status,
        turn_count,
        sequence: 0,
        trace_corruption,
    })
}

/// Build a [`RunSummary`] from a run's JSONL file: `RunStarted` from the
/// first valid record, terminal status from the last, and `turn_count`
/// from the `Turn`s in between. Returns `None` when the file has no
/// `RunStarted` (it can't be keyed/summarized).
struct RunSummaryScan {
    summary: Option<RunSummary>,
    corruption: Option<RunTraceCorruption>,
}

fn summarize_file_checked(path: &Path) -> std::io::Result<RunSummaryScan> {
    let file = car_secrets::open_private_read(path)?;
    let mut started: Option<car_proto::RunStarted> = None;
    let mut ended: Option<car_proto::RunEnded> = None;
    let mut turn_count = 0usize;
    let mut cancellation_pending = false;
    let scan = scan_records(&file, |record| match record {
        RunRecord::Started(s) => started = Some(s),
        RunRecord::Ended(e) => ended = Some(e),
        RunRecord::Turn(_) => turn_count += 1,
        RunRecord::CancellationRequested(_) | RunRecord::CancellationResult(_) => {
            cancellation_pending = true
        }
    });
    let trace_corruption = match scan {
        Ok(()) => None,
        Err(RecordScanError::Malformed { line, .. }) => Some(RunTraceCorruption {
            kind: RunTraceCorruptionKind::MalformedRecord,
            line,
        }),
        Err(RecordScanError::Io(error)) => return Err(error),
    };
    preserve_summary_corruption(
        car_secrets::revalidate_private_path(path, &file),
        &trace_corruption,
    )?;
    let summary = summary_from_records(
        started,
        ended,
        turn_count,
        cancellation_pending,
        trace_corruption.clone(),
    );
    Ok(RunSummaryScan {
        summary,
        corruption: trace_corruption,
    })
}

fn summarize_file(path: &Path) -> Option<RunSummary> {
    summarize_file_checked(path).ok()?.summary
}

/// Sanitize an id for use as a portable path segment — replace path
/// separators, Windows-reserved punctuation, and control characters, then
/// strip `..` so a hostile `agent_id`/`run_id` can't escape the `runs/` tree
/// or make an otherwise valid trace unpublishable on Windows. Ids are uuids /
/// slugs in practice; this is defense in depth.
fn sanitize(id: &str) -> String {
    let cleaned: String = id
        .chars()
        .map(|c| {
            if c.is_control() || matches!(c, '<' | '>' | ':' | '"' | '/' | '\\' | '|' | '?' | '*') {
                '_'
            } else {
                c
            }
        })
        .collect();
    let trimmed = cleaned.trim_matches(['.', ' ']);
    if trimmed.is_empty() {
        "_".to_string()
    } else {
        trimmed.to_string()
    }
}

/// Mark a directory backup-excluded so Time Machine / iCloud don't copy
/// plaintext traces off the machine (R14). The portable `.nobackup` marker is
/// fail-closed CAR-owned state at `0600`; only the supplementary macOS xattr
/// remains best-effort.
fn ensure_backup_excluded(dir: &Path) -> std::io::Result<()> {
    let marker = dir.join(".nobackup");
    match car_secrets::create_private_file(&marker) {
        Ok(mut file) => {
            file.write_all(b"car run traces - excluded from backup\n")?;
            car_secrets::revalidate_private_file(&file)?;
            car_secrets::revalidate_private_path(&marker, &file)?;
        }
        Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
            let file = car_secrets::open_private_read(&marker)?;
            car_secrets::revalidate_private_path(&marker, &file)?;
        }
        Err(error) => return Err(error),
    }
    #[cfg(target_os = "macos")]
    set_macos_backup_excluded(dir);
    car_secrets::ensure_private_dir(dir)?;
    Ok(())
}

/// Set the macOS Time Machine exclusion xattr on `dir`. Uses the `xattr`
/// CLI (always present on macOS) rather than linking a native crate. Time
/// Machine treats any non-empty value on this attr as "exclude". Best-
/// effort — a missing `xattr` or a failure is ignored.
#[cfg(target_os = "macos")]
fn set_macos_backup_excluded(dir: &Path) {
    // `xattr -w` writes a string value; Time Machine only checks for the
    // attr's presence/non-emptiness, not its exact bytes.
    let _ = std::process::Command::new("xattr")
        .args(["-w", "com.apple.metadata:com_apple_backup_excludeItem", "1"])
        .arg(dir)
        .output();
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_ir::{AgentOutcome, CostSummary, OutcomeMetrics, OutcomeStatus, ProposalLineageEntry};
    use car_proto::{RunEnded, RunStarted, RunTurn, VerifierVerdict};
    use serde_json::json;

    #[test]
    fn run_path_segments_replace_windows_reserved_characters() {
        assert_eq!(sanitize("name:bulldozer-agent"), "name_bulldozer-agent");
        assert_eq!(sanitize("<>:\"/\\|?*\0"), "__________");
        assert_eq!(sanitize(".. "), "_");
    }

    fn store(root: PathBuf) -> RunStore {
        RunStore::new(root, RetentionConfig::default())
    }

    fn started(run_id: &str, agent_id: &str, when: DateTime<Utc>) -> RunStarted {
        RunStarted {
            run_id: run_id.to_string(),
            client_id: Some("test-client".to_string()),
            agent_id: agent_id.to_string(),
            intent: "do the thing".to_string(),
            outcome_description: None,
            started_at: when,
        }
    }

    fn turn(index: usize, prompt: &str) -> RunRecord {
        RunRecord::Turn(RunTurn {
            index,
            proposal_id: None,
            action_id: None,
            action_status: None,
            action_duration_ms: None,
            action_completed_at: None,
            depends_on: None,
            state_dependencies: None,
            prompt: Some(prompt.to_string()),
            tool: Some("drive_cli".to_string()),
            parameters: json!({ "prompt": prompt }),
            output: Some(json!({ "exit_code": 0 })),
            cli_outcome: None,
            verifier_verdict: VerifierVerdict::NotRun,
            policy_rejected: None,
        })
    }

    fn ended(run_id: &str, agent_id: &str, status: OutcomeStatus) -> RunRecord {
        let outcome = AgentOutcome {
            status,
            summary: "done".to_string(),
            evidence: vec![],
            metrics: OutcomeMetrics::default(),
            timestamp: Utc::now(),
        };
        let termination = RunTermination::Outcome { status, outcome };
        let completion_digest = crate::session::run_completion_digest(&termination).unwrap();
        RunRecord::Ended(RunEnded {
            run_id: run_id.to_string(),
            client_id: Some("test-client".to_string()),
            agent_id: agent_id.to_string(),
            termination,
            completion_digest: Some(completion_digest),
            ended_at: Utc::now(),
        })
    }

    fn valid_pending(run_id: &str) -> PendingProposalFinalization {
        let proposal = ActionProposal {
            id: format!("proposal-{run_id}"),
            source: "run-store-test".to_string(),
            actions: Vec::new(),
            timestamp: Utc::now(),
            context: HashMap::new(),
        };
        let digest = proposal_digest(&proposal).unwrap();
        let proposal_result = ProposalResult {
            proposal_id: proposal.id.clone(),
            original_proposal_id: proposal.id.clone(),
            final_proposal: Some(proposal.clone()),
            replan_lineage: vec![ProposalLineageEntry {
                generation: 0,
                proposal_id: proposal.id.clone(),
                proposal_digest: Some(digest.clone()),
                status: ProposalLineageStatus::Accepted,
                rejection_reason: None,
            }],
            accepted_proposal_preimages: vec![car_ir::AcceptedProposalPreimage {
                generation: 0,
                proposal_digest: digest,
                proposal: proposal.clone(),
            }],
            results: Vec::new(),
            cost: CostSummary::default(),
        };
        let result_value = serde_json::to_value(&proposal_result).unwrap();
        let canonical = car_inference::catalog_identity::canonical_json(&result_value).unwrap();
        PendingProposalFinalization {
            run_id: run_id.to_string(),
            client_id: "test-client".to_string(),
            requested_policy_session_id: None,
            policy_session_id: None,
            original_proposal_id: proposal.id.clone(),
            final_proposal_id: proposal.id.clone(),
            original_submission: json!({
                "id": proposal.id.clone(),
                "source": "run-store-test",
                "actions": []
            }),
            original_proposal: proposal.clone(),
            final_proposal: proposal.clone(),
            accepted_proposal_preimages: vec![AcceptedProposalPreimage {
                generation: 0,
                proposal,
            }],
            proposal_result,
            result_digest: format!("{:x}", Sha256::digest(canonical.as_bytes())),
        }
    }

    fn refresh_pending_result_digest(pending: &mut PendingProposalFinalization) {
        let value = serde_json::to_value(&pending.proposal_result).unwrap();
        let canonical = car_inference::catalog_identity::canonical_json(&value).unwrap();
        pending.result_digest = format!("{:x}", Sha256::digest(canonical.as_bytes()));
    }

    fn marker_for_pending(pending: &PendingProposalFinalization) -> ProposalExecutionMarker {
        ProposalExecutionMarker {
            run_id: pending.run_id.clone(),
            client_id: pending.client_id.clone(),
            requested_policy_session_id: pending.requested_policy_session_id.clone(),
            policy_session_id: pending.policy_session_id.clone(),
            original_proposal_id: pending.original_proposal_id.clone(),
            original_submission: pending.original_submission.clone(),
            original_proposal: serde_json::to_value(&pending.original_proposal).unwrap(),
            proposal_digest: proposal_digest(&pending.original_proposal).unwrap(),
        }
    }

    fn rebind_pending_original(pending: &mut PendingProposalFinalization) {
        pending.original_submission["source"] = json!("pending-self-claim");
        pending.original_proposal.source = "pending-self-claim".to_string();
        pending.final_proposal = pending.original_proposal.clone();
        pending.accepted_proposal_preimages[0].proposal = pending.original_proposal.clone();
        pending.proposal_result.final_proposal = Some(pending.original_proposal.clone());
        pending.proposal_result.replan_lineage[0].proposal_digest =
            Some(proposal_digest(&pending.original_proposal).unwrap());
        refresh_pending_result_digest(pending);
    }

    fn write_pending_provenance(store: &RunStore, pending: &PendingProposalFinalization) {
        store
            .write_started(&started(&pending.run_id, "agent-a", Utc::now()))
            .unwrap();
        store
            .write_execution_marker(&marker_for_pending(pending))
            .unwrap();
    }

    fn write_completed_receipt_fixture(
        store: &RunStore,
        pending: PendingProposalFinalization,
        exact_bytes: Option<usize>,
        invalid_tail: bool,
    ) -> CompletedProposalResponse {
        let receipt = CompletedProposalResponse {
            finalization: pending,
        };
        receipt.validate().unwrap();
        let pending = &receipt.finalization;
        let run_root = store.completed_response_run_root(&pending.run_id);
        store.ensure_private_dir(&run_root).unwrap();
        let path = store
            .completed_response_path(
                &pending.run_id,
                &pending.client_id,
                pending.requested_policy_session_id.as_deref(),
                &pending.original_submission,
            )
            .unwrap();
        let mut bytes = serde_json::to_vec(&receipt).unwrap();
        if let Some(exact_bytes) = exact_bytes {
            assert!(bytes.len() <= exact_bytes);
            bytes.resize(exact_bytes, b' ');
        }
        if invalid_tail {
            bytes.push(b'!');
        }
        let mut file = store.create_private_file(&path).unwrap();
        file.write_all(&bytes).unwrap();
        file.sync_all().unwrap();
        receipt
    }

    #[test]
    fn first_use_agent_receipt_parent_sync_failure_is_not_acknowledged_and_retry_is_exact() {
        let tmp = tempfile::TempDir::new().unwrap();
        let failures = car_secrets::PrivatePathDurabilityFailureInjector::default();
        let store =
            store(tmp.path().join("runs")).with_private_path_failure_injector(failures.clone());
        store.prepare_storage().unwrap();
        failures.fail_next(car_secrets::PrivatePathDurabilityFailurePoint::ParentDirectorySync);
        let boundary = started("private-first-use", "new-agent", Utc::now());

        let error = store.write_started(&boundary).unwrap_err();
        assert!(error
            .to_string()
            .contains("injected private-path parent directory sync failure"));
        assert!(
            store.get_run_trace(&boundary.run_id).is_none(),
            "an unacknowledged agent-directory entry must not contain a run boundary"
        );

        store.write_started(&boundary).unwrap();
        let trace = store.get_run_trace(&boundary.run_id).unwrap();
        assert_eq!(
            trace
                .iter()
                .filter(|row| matches!(row, RunRecord::Started(_)))
                .count(),
            1,
            "retry must persist exactly one RunStarted boundary"
        );
    }

    #[test]
    fn proposal_sidecar_parent_sync_failures_are_not_acknowledged_and_retry_exactly() {
        let tmp = tempfile::TempDir::new().unwrap();
        let failures = car_secrets::PrivatePathDurabilityFailureInjector::default();
        let store =
            store(tmp.path().join("runs")).with_private_path_failure_injector(failures.clone());
        let pending = valid_pending("private-sidecar-first-use");
        store
            .write_started(&started(&pending.run_id, "agent-a", Utc::now()))
            .unwrap();
        let marker = marker_for_pending(&pending);

        failures.fail_next(car_secrets::PrivatePathDurabilityFailurePoint::ParentDirectorySync);
        assert!(store.write_execution_marker(&marker).is_err());
        assert_eq!(store.execution_marker(&pending.run_id).unwrap(), None);
        store.write_execution_marker(&marker).unwrap();
        assert_eq!(
            store.execution_marker(&pending.run_id).unwrap(),
            Some(marker)
        );

        failures.fail_next(car_secrets::PrivatePathDurabilityFailurePoint::ParentDirectorySync);
        assert!(store.write_pending_proposal(&pending).is_err());
        assert_eq!(store.pending_proposal(&pending.run_id).unwrap(), None);
        store.write_pending_proposal(&pending).unwrap();
        assert_eq!(
            store.pending_proposal(&pending.run_id).unwrap(),
            Some(pending)
        );
    }

    /// A completed run is readable from a brand-new store instance — the
    /// "simulated daemon restart" (new store, empty memory): the trace
    /// must come back from disk (R4).
    #[test]
    fn completed_run_readable_after_restart() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s1 = store(root.clone());
        s1.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();
        s1.append_turns("agent-a", "run-1", &[turn(0, "first")])
            .unwrap();
        s1.append_records(
            "agent-a",
            "run-1",
            &[ended("run-1", "agent-a", OutcomeStatus::Success)],
        )
        .unwrap();

        // Brand-new store over the same root = simulated restart with empty memory.
        let s2 = store(root);
        let trace = s2
            .get_run_trace("run-1")
            .expect("trace readable after restart");
        assert!(matches!(trace.first(), Some(RunRecord::Started(_))));
        assert!(matches!(trace.last(), Some(RunRecord::Ended(_))));
        let turns = trace
            .iter()
            .filter(|r| matches!(r, RunRecord::Turn(_)))
            .count();
        assert_eq!(turns, 1);
    }

    /// Each run persists to its own `(agent_id, run_id)` file — no
    /// cross-contamination between runs or agents (R1).
    #[test]
    fn runs_isolated_per_agent_and_run() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        s.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();
        s.append_turns("agent-a", "run-1", &[turn(0, "a-first")])
            .unwrap();
        s.write_started(&started("run-2", "agent-a", Utc::now()))
            .unwrap();
        s.append_turns("agent-a", "run-2", &[turn(0, "a-second")])
            .unwrap();
        s.write_started(&started("run-3", "agent-b", Utc::now()))
            .unwrap();
        s.append_turns("agent-b", "run-3", &[turn(0, "b-first")])
            .unwrap();

        // Distinct files; each holds only its own turn.
        let t1 = s.get_run_trace("run-1").unwrap();
        let t2 = s.get_run_trace("run-2").unwrap();
        let t3 = s.get_run_trace("run-3").unwrap();
        assert_eq!(turn_prompt(&t1), "a-first");
        assert_eq!(turn_prompt(&t2), "a-second");
        assert_eq!(turn_prompt(&t3), "b-first");
        // run_id -> agent_id resolution works for replay authz.
        assert_eq!(s.agent_for_run("run-1").as_deref(), Some("agent-a"));
        assert_eq!(s.agent_for_run("run-3").as_deref(), Some("agent-b"));
        // agent-a lists 2 runs, agent-b lists 1.
        assert_eq!(s.list_runs("agent-a").len(), 2);
        assert_eq!(s.list_runs("agent-b").len(), 1);
    }

    #[test]
    fn summary_index_pages_in_stable_newest_first_order() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        let now = Utc::now();
        s.write_started(&started(
            "run-old",
            "agent-page",
            now - chrono::Duration::seconds(2),
        ))
        .unwrap();
        s.append_turns("agent-page", "run-old", &[turn(0, "old")])
            .unwrap();
        s.write_started(&started("run-new", "agent-page", now))
            .unwrap();
        s.append_turns(
            "agent-page",
            "run-new",
            &[turn(0, "new-0"), turn(1, "new-1")],
        )
        .unwrap();

        let (first, next) = s.list_runs_page("agent-page", 0, 1).unwrap();
        assert_eq!(first.len(), 1);
        assert_eq!(first[0].run_id, "run-new");
        assert_eq!(first[0].turn_count, 2);
        let next = next.expect("older run remains");

        // A new run arriving at the head must not shift an offset cursor and
        // cause the previous first row to be returned twice.
        s.write_started(&started(
            "run-newest",
            "agent-page",
            now + chrono::Duration::seconds(1),
        ))
        .unwrap();

        let (second, next) = s.list_runs_page("agent-page", next, 1).unwrap();
        assert_eq!(second.len(), 1);
        assert_eq!(second[0].run_id, "run-old");
        assert_eq!(second[0].turn_count, 1);
        assert_eq!(next, None);

        let (past_end, next) = s.list_runs_page("agent-page", 1, 1).unwrap();
        assert!(past_end.is_empty());
        assert_eq!(next, None);
    }

    #[test]
    fn newline_terminated_malformed_record_quarantines_completed_trace() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root.clone());
        s.write_started(&started("corrupt", "agent-a", Utc::now()))
            .unwrap();
        s.append_turns("agent-a", "corrupt", &[turn(0, "before")])
            .unwrap();

        let path = root.join("agent-a").join("corrupt.jsonl");
        let mut file = car_secrets::open_private_append(&path).unwrap();
        file.write_all(b"{malformed-middle}\n").unwrap();
        file.flush().unwrap();
        file.sync_all().unwrap();
        drop(file);

        let RunRecord::Ended(terminal) = ended("corrupt", "agent-a", OutcomeStatus::Success) else {
            unreachable!()
        };
        let error = s
            .write_ended(&terminal)
            .expect_err("a durable malformed row must prevent completion");
        assert!(error.to_string().contains("line 3"), "{error}");

        let (summaries, _) = s.list_runs_page("agent-a", 0, 1).unwrap();
        assert_eq!(summaries[0].status, RunStatus::Incomplete);
        assert_eq!(
            serde_json::to_value(&summaries[0]).unwrap()["trace_corruption"],
            json!({"kind":"malformed_record","line":3})
        );
        assert!(s
            .get_run_turn_page_for("agent-a", "corrupt", 0, 10)
            .unwrap_err()
            .to_string()
            .contains("line 3"));
    }

    fn turn_prompt(trace: &[RunRecord]) -> String {
        trace
            .iter()
            .find_map(|r| match r {
                RunRecord::Turn(t) => t.prompt.clone(),
                _ => None,
            })
            .unwrap_or_default()
    }

    /// Files are `0600` and dirs are `0700` (R14 — assert the modes).
    #[cfg(unix)]
    #[test]
    fn perms_are_0600_files_0700_dirs() {
        use std::os::unix::fs::PermissionsExt;
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root.clone());
        s.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();

        let file = root.join("agent-a").join("run-1.jsonl");
        let fmode = std::fs::metadata(&file).unwrap().permissions().mode() & 0o777;
        assert_eq!(fmode, 0o600, "run file must be 0600, got {:o}", fmode);

        let root_mode = std::fs::metadata(&root).unwrap().permissions().mode() & 0o777;
        assert_eq!(
            root_mode, 0o700,
            "runs/ dir must be 0700, got {:o}",
            root_mode
        );
        let agent_mode = std::fs::metadata(root.join("agent-a"))
            .unwrap()
            .permissions()
            .mode()
            & 0o777;
        assert_eq!(
            agent_mode, 0o700,
            "agent dir must be 0700, got {:o}",
            agent_mode
        );

        // The backup-exclusion marker is present.
        let marker = root.join(".nobackup");
        assert!(marker.exists(), ".nobackup marker written");
        let marker_mode = std::fs::metadata(marker).unwrap().permissions().mode() & 0o777;
        assert_eq!(
            marker_mode, 0o600,
            ".nobackup marker must be 0600, got {:o}",
            marker_mode
        );
    }

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

        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let agent = root.join("agent-a");
        std::fs::create_dir_all(&agent).unwrap();
        let path = agent.join("run-1.jsonl");
        let line =
            serde_json::to_string(&RunRecord::Started(started("run-1", "agent-a", Utc::now())))
                .unwrap();
        let original = format!("{line}\n");
        std::fs::write(&path, original.as_bytes()).unwrap();
        std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o755)).unwrap();
        std::fs::set_permissions(&agent, std::fs::Permissions::from_mode(0o755)).unwrap();
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();

        let trace = store(root.clone())
            .get_run_trace_for("agent-a", "run-1")
            .expect("legacy run remains replayable");
        assert_eq!(trace.len(), 1);
        assert_eq!(std::fs::read_to_string(&path).unwrap(), original);
        assert_eq!(
            std::fs::metadata(root).unwrap().permissions().mode() & 0o777,
            0o700
        );
        assert_eq!(
            std::fs::metadata(agent).unwrap().permissions().mode() & 0o777,
            0o700
        );
        assert_eq!(
            std::fs::metadata(path).unwrap().permissions().mode() & 0o777,
            0o600
        );
    }

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

        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let agent = root.join("agent-a");
        std::fs::create_dir_all(&agent).unwrap();
        let victim = tmp.path().join("victim.jsonl");
        std::fs::write(&victim, b"victim\n").unwrap();

        let symlink_path = agent.join("symlink.jsonl");
        symlink(&victim, &symlink_path).unwrap();
        let s = store(root.clone());
        assert!(s.get_run_trace_for("agent-a", "symlink").is_none());
        assert!(s
            .append_records(
                "agent-a",
                "symlink",
                &[RunRecord::Started(started(
                    "symlink",
                    "agent-a",
                    Utc::now()
                ))],
            )
            .is_err());
        assert_eq!(std::fs::read(&victim).unwrap(), b"victim\n");

        std::fs::remove_file(&symlink_path).unwrap();
        let hardlink_path = agent.join("hardlink.jsonl");
        std::fs::hard_link(&victim, &hardlink_path).unwrap();
        assert!(s.get_run_trace_for("agent-a", "hardlink").is_none());
        assert!(s
            .append_records(
                "agent-a",
                "hardlink",
                &[RunRecord::Started(started(
                    "hardlink",
                    "agent-a",
                    Utc::now()
                ))],
            )
            .is_err());
        assert_eq!(std::fs::read(&victim).unwrap(), b"victim\n");
    }

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

        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        std::fs::create_dir(&root).unwrap();
        let victim_dir = tmp.path().join("victim-agent");
        std::fs::create_dir(&victim_dir).unwrap();
        symlink(&victim_dir, root.join("agent-link")).unwrap();

        let s = store(root.clone());
        assert!(s.list_runs("agent-link").is_empty());
        assert!(s
            .append_records(
                "agent-link",
                "run-1",
                &[RunRecord::Started(started(
                    "run-1",
                    "agent-link",
                    Utc::now()
                ))],
            )
            .is_err());
        assert!(std::fs::read_dir(&victim_dir).unwrap().next().is_none());

        let agent = root.join("agent-a");
        std::fs::create_dir(&agent).unwrap();
        let socket_path = agent.join("socket.jsonl");
        let _listener = UnixListener::bind(&socket_path).unwrap();
        assert!(s.get_run_trace_for("agent-a", "socket").is_none());
        assert!(s
            .append_records(
                "agent-a",
                "socket",
                &[RunRecord::Started(started("socket", "agent-a", Utc::now()))],
            )
            .is_err());
    }

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

        let tmp = tempfile::TempDir::new().unwrap();
        let path = tmp.path().join("run.jsonl");
        std::fs::write(&path, b"torn").unwrap();
        let mut file = car_secrets::open_private_append(&path).unwrap();
        let victim = tmp.path().join("victim");
        std::fs::write(&victim, b"victim\n").unwrap();

        std::fs::remove_file(&path).unwrap();
        symlink(&victim, &path).unwrap();
        assert!(append_jsonl_batch_to_path(&path, &mut file, &[turn(0, "safe")]).is_err());

        assert_eq!(std::fs::read(&victim).unwrap(), b"victim\n");
    }

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

        let tmp = tempfile::TempDir::new().unwrap();
        let path = tmp.path().join("run.jsonl");
        let line =
            serde_json::to_string(&RunRecord::Started(started("run-1", "agent-a", Utc::now())))
                .unwrap();
        std::fs::write(&path, format!("{line}\n")).unwrap();
        let file = car_secrets::open_private_read(&path).unwrap();
        let victim = tmp.path().join("victim");
        std::fs::write(&victim, b"victim\n").unwrap();

        std::fs::rename(&path, tmp.path().join("moved.jsonl")).unwrap();
        symlink(&victim, &path).unwrap();

        assert!(load_private_records(&path, &file).is_err());
        assert_eq!(std::fs::read(&victim).unwrap(), b"victim\n");
    }

    /// A run with no `RunEnded` reads `InProgress`; once the disconnect
    /// path writes the `Incomplete` terminal it reads `Incomplete` — never
    /// silently `Completed`. This is the R5 distinguishability the
    /// dashboard renders.
    #[test]
    fn orphan_run_status_distinguishes_inprogress_from_incomplete() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        // Stale start time (no harness writing — an orphan).
        let stale = Utc::now() - chrono::Duration::hours(6);
        s.write_started(&started("run-1", "agent-a", stale))
            .unwrap();
        s.append_turns("agent-a", "run-1", &[turn(0, "first")])
            .unwrap();

        // Still open: no terminal record yet.
        let open = &s.list_runs("agent-a")[0];
        assert_eq!(open.status, RunStatus::InProgress);

        // Disconnect path writes the Incomplete terminal.
        let incomplete = RunRecord::Ended(RunEnded {
            run_id: "run-1".to_string(),
            client_id: Some("test-client".to_string()),
            agent_id: "agent-a".to_string(),
            termination: RunTermination::Incomplete,
            completion_digest: Some("test-digest".to_string()),
            ended_at: Utc::now(),
        });
        s.append_records("agent-a", "run-1", &[incomplete]).unwrap();
        let closed = &s.list_runs("agent-a")[0];
        assert_eq!(closed.status, RunStatus::Incomplete);
    }

    /// Retention evicts beyond the per-agent cap, never an in-progress
    /// run (R6).
    #[test]
    fn gc_evicts_beyond_per_agent_cap_but_never_in_progress() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = RunStore::new(root, RetentionConfig::new(3, 30));
        // 5 completed runs with increasing start times.
        let base = Utc::now() - chrono::Duration::days(1);
        for i in 0..5 {
            let id = format!("c{i}");
            let when = base + chrono::Duration::minutes(i);
            s.write_started(&started(&id, "agent-a", when)).unwrap();
            s.append_records(
                "agent-a",
                &id,
                &[ended(&id, "agent-a", OutcomeStatus::Success)],
            )
            .unwrap();
        }
        // 1 in-progress run (no terminal) — must survive GC.
        s.write_started(&started("live", "agent-a", Utc::now()))
            .unwrap();

        let removed = s.gc();
        // Keep 3 most-recent completed; evict the 2 oldest completed. The
        // in-progress run is never counted/evicted.
        assert_eq!(removed, 2, "should evict the 2 oldest completed runs");
        let remaining = s.list_runs("agent-a");
        // 3 kept completed + the live one = 4.
        assert_eq!(remaining.len(), 4);
        assert!(
            remaining.iter().any(|r| r.run_id == "live"),
            "in-progress run must never be evicted"
        );
        // The two oldest (c0, c1) are gone.
        assert!(!remaining.iter().any(|r| r.run_id == "c0"));
        assert!(!remaining.iter().any(|r| r.run_id == "c1"));
    }

    /// `0` disables a cap. It used to delete everything (car#1338).
    ///
    /// Read literally, `max_per_agent = 0` makes `completed_rank >= 0` true for
    /// every run and `max_age_days = 0` puts the cutoff at now — so either value
    /// wiped every collectable trace on the next boot, unrecoverably. Nobody
    /// types `0` meaning "keep nothing"; they mean "no cap", which is what it
    /// means for `max_session_wall_secs` and for the coder's own retention in
    /// the neighbouring config file.
    #[test]
    fn a_zero_cap_keeps_everything_instead_of_deleting_it() {
        // Each case disables ONE cap and keeps the run set clear of the other,
        // so a pass means the disabled cap stood down rather than the other one
        // simply having nothing to take.
        let recent = Utc::now();
        let ancient = Utc::now() - chrono::Duration::days(400);
        for (retention, count, when) in [
            // No count cap: 60 runs, well past the default 50, all recent.
            (RetentionConfig::new(0, 30), 60, recent),
            // No age cap: ancient runs, but fewer than the count cap.
            (RetentionConfig::new(50, 0), 5, ancient),
            // A negative age is not a cap either; it would put the cutoff in the
            // future and evict runs that have not happened yet.
            (RetentionConfig::new(50, -1), 5, recent),
            // Neither cap: both dimensions blown well past the defaults.
            (RetentionConfig::new(0, 0), 60, ancient),
        ] {
            let tmp = tempfile::TempDir::new().unwrap();
            let s = RunStore::new(tmp.path().join("runs"), retention);
            for i in 0..count {
                let id = format!("r{i}");
                s.write_started(&started(&id, "agent-a", when)).unwrap();
                s.append_records(
                    "agent-a",
                    &id,
                    &[ended_at(&id, "agent-a", OutcomeStatus::Success, when)],
                )
                .unwrap();
            }

            assert_eq!(s.gc(), 0, "{retention:?} must evict nothing");
            assert_eq!(s.list_runs("agent-a").len(), count, "{retention:?}");
        }
    }

    /// An operator typo must not take the daemon down at boot.
    ///
    /// `chrono::Duration::days` PANICS rather than saturating on a value that
    /// overflows, and `max_age_days` is an unvalidated `i64` straight from
    /// `config.toml`. An unrepresentable cutoff reads as no cap, which is the
    /// safe direction — the alternative is deleting on a number nobody meant.
    #[test]
    fn an_absurd_age_cap_disables_itself_rather_than_panicking() {
        // Asserted on the cap, not only on `gc() == 0`: a cutoff at year
        // -292_277_022_365 would also evict nothing, so the weaker assertion
        // would pass without the cap actually being disabled.
        assert_eq!(RetentionConfig::new(50, i64::MAX).age_cap_days(), None);
        assert_eq!(RetentionConfig::new(50, -1).age_cap_days(), None);
        assert_eq!(RetentionConfig::new(50, 30).age_cap_days(), Some(30));

        let tmp = tempfile::TempDir::new().unwrap();
        let s = RunStore::new(tmp.path().join("runs"), RetentionConfig::new(50, i64::MAX));
        let old = Utc::now() - chrono::Duration::days(400);
        s.write_started(&started("r0", "agent-a", old)).unwrap();
        s.append_records(
            "agent-a",
            "r0",
            &[ended_at("r0", "agent-a", OutcomeStatus::Success, old)],
        )
        .unwrap();

        assert_eq!(s.gc(), 0);
        assert_eq!(s.list_runs("agent-a").len(), 1);
    }

    /// A present `0` is honored, not silently replaced by the default. Filtering
    /// it at load would turn "keep everything" back into the 50/30 caps.
    #[test]
    fn a_configured_zero_survives_loading() {
        let dir = tempfile::TempDir::new().unwrap();
        std::fs::write(
            dir.path().join("config.toml"),
            "[runs]\nmax_per_agent = 0\nmax_age_days = 0\n",
        )
        .unwrap();
        let cfg = RetentionConfig::from_car_dir(dir.path());
        assert_eq!(cfg.count_cap(), None, "a configured 0 must not become 50");
        assert_eq!(
            cfg.age_cap_days(),
            None,
            "a configured 0 must not become 30"
        );
    }

    /// Retention evicts completed runs older than the age cap (R6).
    #[test]
    fn gc_evicts_runs_older_than_age_cap() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = RunStore::new(tmp.path().join("runs"), RetentionConfig::new(50, 30));
        // One old completed run — started AND ended 40 days ago (its
        // terminal time is what the age cap measures, FIX 2) — and one
        // fresh run.
        let old = Utc::now() - chrono::Duration::days(40);
        s.write_started(&started("old", "agent-a", old)).unwrap();
        s.append_records(
            "agent-a",
            "old",
            &[ended_at("old", "agent-a", OutcomeStatus::Success, old)],
        )
        .unwrap();
        s.write_started(&started("fresh", "agent-a", Utc::now()))
            .unwrap();
        s.append_records(
            "agent-a",
            "fresh",
            &[ended("fresh", "agent-a", OutcomeStatus::Success)],
        )
        .unwrap();

        let removed = s.gc();
        assert_eq!(removed, 1, "the 40-day-old run should be evicted");
        let remaining = s.list_runs("agent-a");
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].run_id, "fresh");
    }

    /// An old run that is still in progress is NOT evicted by the age cap
    /// (R6 — never evict an open run, even a stale one).
    #[test]
    fn gc_never_evicts_stale_in_progress_run() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = RunStore::new(tmp.path().join("runs"), RetentionConfig::new(1, 1));
        let old = Utc::now() - chrono::Duration::days(40);
        // Old + in-progress (no terminal).
        s.write_started(&started("stale-live", "agent-a", old))
            .unwrap();
        let removed = s.gc();
        assert_eq!(removed, 0);
        assert!(s
            .list_runs("agent-a")
            .iter()
            .any(|r| r.run_id == "stale-live"));
    }

    /// A corrupt/partial trailing JSONL line loads the prior valid
    /// records rather than failing the whole run (error path).
    #[test]
    fn corrupt_trailing_line_loads_prior_records() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root.clone());
        s.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();
        s.append_turns("agent-a", "run-1", &[turn(0, "first"), turn(1, "second")])
            .unwrap();

        // Append a partial/garbage line directly (simulating a crash
        // mid-append).
        let path = root.join("agent-a").join("run-1.jsonl");
        let mut f = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap();
        f.write_all(b"{\"record\":\"turn\",\"index\":2,\"prom")
            .unwrap();

        let trace = s.get_run_trace("run-1").expect("trace still loads");
        // RunStarted + 2 valid turns; the garbage trailing line is dropped.
        let turns = trace
            .iter()
            .filter(|r| matches!(r, RunRecord::Turn(_)))
            .count();
        assert_eq!(turns, 2, "prior valid turns load; corrupt line skipped");
        assert!(matches!(trace.first(), Some(RunRecord::Started(_))));
    }

    /// `list_runs` for an agent with no runs returns empty (empty-state).
    #[test]
    fn list_runs_empty_for_unknown_agent() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        assert!(s.list_runs("nobody").is_empty());
        assert!(s.get_run_trace("nope").is_none());
        assert!(s.agent_for_run("nope").is_none());
    }

    /// `from_journal_dir` roots the store at `<car_dir>/runs` as a sibling
    /// of the journal dir.
    #[test]
    fn from_journal_dir_roots_at_car_runs() {
        let s = RunStore::from_journal_dir(Path::new("/home/u/.car/journals"));
        assert_eq!(s.root(), Path::new("/home/u/.car/runs"));
    }

    /// Retention config loads `[runs]` overrides from config.toml; missing
    /// keys keep the restrictive default.
    #[test]
    fn retention_config_reads_overrides() {
        let tmp = tempfile::TempDir::new().unwrap();
        std::fs::write(
            tmp.path().join("config.toml"),
            "[runs]\nmax_per_agent = 10\n",
        )
        .unwrap();
        let cfg = RetentionConfig::from_car_dir(tmp.path());
        assert_eq!(cfg.count_cap(), Some(10));
        // Missing key keeps the default.
        assert_eq!(cfg.age_cap_days(), Some(DEFAULT_MAX_AGE_DAYS));
    }

    /// A missing/malformed config.toml falls back to the restrictive
    /// default — never refuses.
    #[test]
    fn retention_config_defaults_on_missing_file() {
        let tmp = tempfile::TempDir::new().unwrap();
        let cfg = RetentionConfig::from_car_dir(tmp.path());
        assert_eq!(cfg.count_cap(), Some(DEFAULT_MAX_RUNS_PER_AGENT));
        assert_eq!(cfg.age_cap_days(), Some(DEFAULT_MAX_AGE_DAYS));
    }

    /// Append the terminal `RunEnded` with an explicit `ended_at` (the
    /// disconnect/complete path uses `Utc::now()`, but GC ages by terminal
    /// time, so tests need to control it).
    fn ended_at(
        run_id: &str,
        agent_id: &str,
        status: OutcomeStatus,
        when: DateTime<Utc>,
    ) -> RunRecord {
        let outcome = AgentOutcome {
            status,
            summary: "done".to_string(),
            evidence: vec![],
            metrics: OutcomeMetrics::default(),
            timestamp: when,
        };
        RunRecord::Ended(RunEnded {
            run_id: run_id.to_string(),
            client_id: Some("test-client".to_string()),
            agent_id: agent_id.to_string(),
            termination: RunTermination::Outcome { status, outcome },
            completion_digest: Some("test-digest".to_string()),
            ended_at: when,
        })
    }

    /// FIX 2: a run that STARTED >max_age_days ago but COMPLETED recently is
    /// a fresh result and must NOT be evicted by the age cap. GC must age by
    /// the terminal time, not the start time.
    #[test]
    fn gc_age_cap_uses_terminal_time_not_start() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = RunStore::new(tmp.path().join("runs"), RetentionConfig::new(50, 30));
        // Long-running run: started 40 days ago, completed 1 day ago.
        let started_40d = Utc::now() - chrono::Duration::days(40);
        let ended_1d = Utc::now() - chrono::Duration::days(1);
        s.write_started(&started("long", "agent-a", started_40d))
            .unwrap();
        s.append_records(
            "agent-a",
            "long",
            &[ended_at(
                "long",
                "agent-a",
                OutcomeStatus::Success,
                ended_1d,
            )],
        )
        .unwrap();

        let removed = s.gc();
        assert_eq!(
            removed, 0,
            "a run completed 1 day ago must survive the 30-day age cap, \
             even if it started 40 days ago"
        );
        let remaining = s.list_runs("agent-a");
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].run_id, "long");
    }

    /// FIX 4: a crash-orphaned run (RunStarted + Turn, no RunEnded) on disk
    /// reads `InProgress`. A fresh store's `adopt_orphans()` at boot — when
    /// the in-memory map is empty — appends an `Incomplete` terminal so the
    /// run becomes terminal (and thus age-GC-eligible), no longer leaking.
    #[test]
    fn adopt_orphans_marks_crashed_inprogress_runs_incomplete() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        // Prior process wrote a start + a turn, then crashed (no terminal).
        let s1 = store(root.clone());
        s1.write_started(&started("orphan", "agent-a", Utc::now()))
            .unwrap();
        s1.append_turns("agent-a", "orphan", &[turn(0, "first")])
            .unwrap();
        assert_eq!(
            s1.list_runs("agent-a")[0].status,
            RunStatus::InProgress,
            "precondition: orphan reads InProgress before adoption"
        );

        // Fresh store = new process boot, empty in-memory map.
        let s2 = store(root);
        let adopted = s2.adopt_orphans();
        assert_eq!(adopted, 1, "the crash orphan should be adopted");

        let after = &s2.list_runs("agent-a")[0];
        assert_eq!(
            after.status,
            RunStatus::Incomplete,
            "adopted orphan now reads Incomplete (terminal)"
        );
        assert!(after.ended_at.is_some(), "terminal record has an ended_at");

        // Idempotent: a second boot finds no orphans (already terminal).
        assert_eq!(s2.adopt_orphans(), 0);
    }

    fn write_truncated_private_file(path: &Path) {
        let parent = path.parent().unwrap();
        car_secrets::ensure_private_dir(parent).unwrap();
        let mut file = car_secrets::create_private_file(path).unwrap();
        file.write_all(b"{").unwrap();
        file.sync_all().unwrap();
    }

    #[test]
    fn corrupt_execution_marker_quarantines_orphan_adoption() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root);
        s.write_started(&started("marker-corrupt", "agent-a", Utc::now()))
            .unwrap();
        write_truncated_private_file(&s.proposal_execution_path("marker-corrupt"));

        assert!(s.execution_marker("absent-marker").unwrap().is_none());
        assert!(s.execution_marker("marker-corrupt").is_err());

        assert_eq!(
            s.adopt_orphans(),
            0,
            "present-but-invalid marker is outcome-unknown, never an adoptable absence"
        );
        assert_eq!(s.list_runs("agent-a")[0].status, RunStatus::InProgress);
    }

    #[test]
    fn corrupt_finalization_outbox_quarantines_orphan_adoption() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root);
        s.write_started(&started("outbox-corrupt", "agent-a", Utc::now()))
            .unwrap();
        write_truncated_private_file(&s.proposal_outbox_path("outbox-corrupt"));

        assert!(s.pending_proposal("absent-outbox").unwrap().is_none());
        assert!(s.pending_proposal("outbox-corrupt").is_err());

        assert_eq!(
            s.adopt_orphans(),
            0,
            "present-but-invalid outbox is outcome-unknown, never an adoptable absence"
        );
        assert_eq!(s.list_runs("agent-a")[0].status, RunStatus::InProgress);
    }

    #[test]
    fn proposal_sidecars_reject_wrong_run_identity() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        let marker = ProposalExecutionMarker {
            run_id: "other-run".to_string(),
            client_id: "client".to_string(),
            requested_policy_session_id: None,
            policy_session_id: None,
            original_proposal_id: "proposal".to_string(),
            original_submission: json!({"id":"proposal","source":"test","actions":[]}),
            original_proposal: json!({"id":"proposal","source":"test","actions":[]}),
            proposal_digest: "a".repeat(64),
        };
        let path = s.proposal_execution_path("requested-run");
        car_secrets::ensure_private_dir(path.parent().unwrap()).unwrap();
        let mut file = car_secrets::create_private_file(&path).unwrap();
        serde_json::to_writer(&mut file, &marker).unwrap();
        file.sync_all().unwrap();

        let error = s.execution_marker("requested-run").unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(error.to_string().contains("does not match requested run"));

        let mut pending = valid_pending("other-outbox-run");
        pending.client_id = "other-client".to_string();
        let path = s.proposal_outbox_path("requested-outbox-run");
        car_secrets::ensure_private_dir(path.parent().unwrap()).unwrap();
        let mut file = car_secrets::create_private_file(&path).unwrap();
        serde_json::to_writer(&mut file, &pending).unwrap();
        file.sync_all().unwrap();
        let error = s.pending_proposal("requested-outbox-run").unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(error.to_string().contains("does not match requested run"));
    }

    #[test]
    fn execution_marker_reader_rejects_semantically_invalid_preimage() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        let pending = valid_pending("invalid-marker");
        let marker = ProposalExecutionMarker {
            run_id: pending.run_id.clone(),
            client_id: pending.client_id.clone(),
            requested_policy_session_id: None,
            policy_session_id: None,
            original_proposal_id: pending.original_proposal_id.clone(),
            original_submission: pending.original_submission.clone(),
            original_proposal: serde_json::to_value(&pending.original_proposal).unwrap(),
            proposal_digest: "A".repeat(64),
        };
        let path = s.proposal_execution_path(&pending.run_id);
        car_secrets::ensure_private_dir(path.parent().unwrap()).unwrap();
        let mut file = car_secrets::create_private_file(&path).unwrap();
        serde_json::to_writer(&mut file, &marker).unwrap();
        file.sync_all().unwrap();

        let error = s.execution_marker(&pending.run_id).unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(error.to_string().contains("does not bind"));
    }

    #[test]
    fn execution_marker_cleanup_requires_the_full_identity_tuple() {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = store(tmp.path().join("runs"));
        let pending = valid_pending("marker-cleanup-tuple");
        let marker = marker_for_pending(&pending);
        store.write_execution_marker(&marker).unwrap();

        let mut mismatched = marker.clone();
        mismatched.requested_policy_session_id = Some("self-claimed-policy".to_string());
        assert!(store.clear_execution_marker(&mismatched).is_err());
        assert_eq!(
            store.execution_marker(&marker.run_id).unwrap(),
            Some(marker.clone()),
            "tuple mismatch must preserve the exact durable marker"
        );

        store.clear_execution_marker(&marker).unwrap();
        assert!(store.execution_marker(&marker.run_id).unwrap().is_none());
    }

    #[test]
    fn typed_finalization_rejects_invalid_result_and_lineage() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        let valid = valid_pending("typed-valid");
        valid.validate().unwrap();
        write_pending_provenance(&s, &valid);
        s.write_pending_proposal(&valid).unwrap();
        assert_eq!(s.pending_proposal("typed-valid").unwrap(), Some(valid));

        let mut missing_final = valid_pending("typed-missing-final");
        missing_final.proposal_result.final_proposal = None;
        assert!(missing_final.validate().is_err());

        let mut wrong_digest = valid_pending("typed-wrong-digest");
        wrong_digest.proposal_result.replan_lineage[0].proposal_digest = Some("A".repeat(64));
        assert!(wrong_digest.validate().is_err());

        let mut wrong_final = valid_pending("typed-wrong-final");
        wrong_final.final_proposal_id = "another-final".to_string();
        assert!(wrong_final.validate().is_err());

        let mut rejected_tail = valid_pending("typed-rejected-tail");
        rejected_tail
            .proposal_result
            .replan_lineage
            .push(ProposalLineageEntry {
                generation: 1,
                proposal_id: "rejected-candidate".to_string(),
                proposal_digest: Some("d".repeat(64)),
                status: ProposalLineageStatus::Rejected,
                rejection_reason: Some("candidate failed quality gate".to_string()),
            });
        let value = serde_json::to_value(&rejected_tail.proposal_result).unwrap();
        let canonical = car_inference::catalog_identity::canonical_json(&value).unwrap();
        rejected_tail.result_digest = format!("{:x}", Sha256::digest(canonical.as_bytes()));
        rejected_tail.validate().unwrap();

        let mut rejected_without_reason = valid_pending("typed-rejected-no-reason");
        rejected_without_reason
            .proposal_result
            .replan_lineage
            .push(ProposalLineageEntry {
                generation: 1,
                proposal_id: "rejected-candidate".to_string(),
                proposal_digest: None,
                status: ProposalLineageStatus::Rejected,
                rejection_reason: None,
            });
        assert!(rejected_without_reason.validate().is_err());

        let mut rejected_zero_then_accepted = valid_pending("typed-rejected-zero");
        rejected_zero_then_accepted.proposal_result.replan_lineage[0].status =
            ProposalLineageStatus::Rejected;
        rejected_zero_then_accepted.proposal_result.replan_lineage[0].rejection_reason =
            Some("original rejected".to_string());
        rejected_zero_then_accepted
            .proposal_result
            .replan_lineage
            .push(ProposalLineageEntry {
                generation: 1,
                proposal_id: rejected_zero_then_accepted.final_proposal_id.clone(),
                proposal_digest: Some(
                    proposal_digest(&rejected_zero_then_accepted.final_proposal).unwrap(),
                ),
                status: ProposalLineageStatus::Accepted,
                rejection_reason: None,
            });
        rejected_zero_then_accepted
            .accepted_proposal_preimages
            .push(AcceptedProposalPreimage {
                generation: 1,
                proposal: rejected_zero_then_accepted.final_proposal.clone(),
            });
        refresh_pending_result_digest(&mut rejected_zero_then_accepted);
        assert!(rejected_zero_then_accepted.validate().is_err());
    }

    #[test]
    fn active_pending_requires_exact_raw_submission_shape() {
        let valid = valid_pending("raw-valid");
        valid.validate().unwrap();

        for raw in [
            Value::Null,
            json!([]),
            json!({"id": valid.original_proposal_id, "source": "run-store-test"}),
            json!({
                "id": valid.original_proposal_id,
                "source": "run-store-test",
                "actions": [{"id":"not-the-typed-action","type":"state_read"}]
            }),
        ] {
            let mut invalid = valid.clone();
            invalid.original_submission = raw;
            assert!(
                invalid.validate().is_err(),
                "active pending accepted invalid raw submission: {}",
                invalid.original_submission
            );
        }

        let mut with_extra = valid;
        with_extra.original_submission["caller_extension"] = json!({"kept": true});
        with_extra.validate().unwrap();
        assert_eq!(
            with_extra.event_data()["original_submission"]["caller_extension"],
            json!({"kept": true})
        );
    }

    #[test]
    fn fresh_pending_write_requires_exact_run_and_marker_provenance() {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = store(tmp.path().join("runs"));
        let pending = valid_pending("fresh-provenance");

        assert!(
            store.write_pending_proposal(&pending).is_err(),
            "pending without durable RunStarted and execution marker must be rejected"
        );
        write_pending_provenance(&store, &pending);
        store.write_pending_proposal(&pending).unwrap();
        assert_eq!(
            store.pending_proposal(&pending.run_id).unwrap(),
            Some(pending.clone())
        );

        store.clear_pending_proposal(&pending).unwrap();
        let mut client_mismatch = pending.clone();
        client_mismatch.client_id = "self-claimed-client".to_string();
        let mut requested_policy_mismatch = pending.clone();
        requested_policy_mismatch.requested_policy_session_id =
            Some("self-claimed-policy".to_string());
        let mut run_mismatch = pending.clone();
        run_mismatch.run_id = "self-claimed-run".to_string();
        let mut original_mismatch = pending;
        rebind_pending_original(&mut original_mismatch);
        for (name, mismatched) in [
            ("client", client_mismatch),
            ("requested policy", requested_policy_mismatch),
            ("run", run_mismatch),
            ("original proposal", original_mismatch),
        ] {
            assert!(
                store.write_pending_proposal(&mismatched).is_err(),
                "pending {name} self-claim must not override durable run/marker provenance"
            );
        }

        let mut authenticated = valid_pending("fresh-auth-provenance");
        authenticated.requested_policy_session_id = Some("live-policy".to_string());
        authenticated.policy_session_id = Some("live-policy".to_string());
        write_pending_provenance(&store, &authenticated);
        authenticated.policy_session_id = None;
        assert!(
            store.write_pending_proposal(&authenticated).is_err(),
            "pending authenticated policy self-claim must match the durable marker"
        );
    }

    #[test]
    fn outbox_reader_requires_active_v3_typed_result_fields() {
        let tmp = tempfile::TempDir::new().unwrap();
        let s = store(tmp.path().join("runs"));
        let pending = valid_pending("strict-result");
        let mut value = serde_json::to_value(&pending).unwrap();
        value["proposal_result"]
            .as_object_mut()
            .unwrap()
            .remove("cost");
        let path = s.proposal_outbox_path("strict-result");
        car_secrets::ensure_private_dir(path.parent().unwrap()).unwrap();
        let mut file = car_secrets::create_private_file(&path).unwrap();
        serde_json::to_writer(&mut file, &value).unwrap();
        file.sync_all().unwrap();

        let error = s.pending_proposal("strict-result").unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(error.to_string().contains("missing `cost`"));
    }

    /// FIX 4 (corollary): a completed run is NOT adopted (it already has a
    /// terminal record).
    #[test]
    fn adopt_orphans_leaves_completed_runs_alone() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root);
        s.write_started(&started("done", "agent-a", Utc::now()))
            .unwrap();
        s.append_records(
            "agent-a",
            "done",
            &[ended("done", "agent-a", OutcomeStatus::Success)],
        )
        .unwrap();
        assert_eq!(s.adopt_orphans(), 0);
        assert_eq!(s.list_runs("agent-a")[0].status, RunStatus::Completed);
    }

    /// A torn final tail is tolerated only until another committed record
    /// follows it. The separator makes the prior fragment a durable malformed
    /// middle row, so the trace must then be quarantined rather than filtered.
    #[test]
    fn appending_after_torn_tail_quarantines_trace() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let s = store(root.clone());
        s.write_started(&started("run-1", "agent-a", Utc::now()))
            .unwrap();
        s.append_turns("agent-a", "run-1", &[turn(0, "first")])
            .unwrap();

        // Simulate a torn append: a partial line with NO trailing newline.
        let path = root.join("agent-a").join("run-1.jsonl");
        {
            let mut f = std::fs::OpenOptions::new()
                .append(true)
                .open(&path)
                .unwrap();
            // No writeln! — deliberately leaves the file's last byte != '\n'.
            f.write_all(b"{\"record\":\"turn\",\"index\":1,\"prom")
                .unwrap();
        }
        let mut opened = car_secrets::open_private_append(&path).unwrap();
        assert!(
            last_byte_is_not_newline(&mut opened).unwrap(),
            "precondition: tail is torn (no trailing newline)"
        );
        drop(opened);

        // A following durable record turns the torn fragment into a committed
        // middle row. The append itself is durable but the trace is no longer
        // eligible for trusted replay/completion.
        let append_error = s
            .append_turns("agent-a", "run-1", &[turn(2, "third")])
            .expect_err("committed middle corruption must fail the append receipt");
        assert!(
            append_error.to_string().contains("line 3"),
            "{append_error}"
        );

        let error = s.get_run_trace_checked("run-1").unwrap_err();
        assert!(error.to_string().contains("line 3"), "{error}");
        let summary = &s.list_runs("agent-a")[0];
        assert_eq!(summary.status, RunStatus::Incomplete);
        assert_eq!(summary.trace_corruption.as_ref().unwrap().line, 3);
    }

    #[test]
    fn durable_boundaries_fail_closed_and_retry_without_duplicate_rows() {
        for point in [
            RunStoreFailurePoint::Write,
            RunStoreFailurePoint::Flush,
            RunStoreFailurePoint::Fsync,
        ] {
            let tmp = tempfile::TempDir::new().unwrap();
            let root = tmp.path().join("runs");
            let failures = RunStoreFailureInjector::default();
            let store = RunStore::new(root.clone(), RetentionConfig::default())
                .with_failure_injector(failures.clone());
            let started = started("durable", "agent-a", Utc::now());

            failures.fail_next(point);
            assert!(
                store.write_started(&started).is_err(),
                "{point:?} start must not acknowledge"
            );
            store
                .write_started(&started)
                .expect("exact start retry reaches durability");

            let ended = match ended("durable", "agent-a", OutcomeStatus::Success) {
                RunRecord::Ended(ended) => ended,
                _ => unreachable!(),
            };
            failures.fail_next(point);
            assert!(
                store.write_ended(&ended).is_err(),
                "{point:?} terminal must not acknowledge"
            );
            store
                .write_ended(&ended)
                .expect("exact terminal retry reaches durability");

            let restarted = RunStore::new(root, RetentionConfig::default());
            let trace = restarted.get_run_trace("durable").unwrap();
            assert_eq!(
                trace
                    .iter()
                    .filter(|row| matches!(row, RunRecord::Started(_)))
                    .count(),
                1,
                "{point:?} retry duplicated RunStarted"
            );
            assert_eq!(
                trace
                    .iter()
                    .filter(|row| matches!(row, RunRecord::Ended(_)))
                    .count(),
                1,
                "{point:?} retry duplicated RunEnded"
            );
        }
    }

    #[test]
    fn retained_completed_proposal_survives_run_gc_and_restart() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        // An ordinary age cap, with the run stamped old enough to trip it.
        // This used `max_per_agent: 0` to evict everything — which worked only
        // because a zero cap was read as zero tolerance, the destructive
        // reading car#1338 removed. A test leaning on that is part of why it
        // stayed invisible: the footgun looked like a feature from in here.
        let retention = RetentionConfig::new(DEFAULT_MAX_RUNS_PER_AGENT, DEFAULT_MAX_AGE_DAYS);
        let long_ago = Utc::now() - chrono::Duration::days(DEFAULT_MAX_AGE_DAYS + 10);
        let store = RunStore::new(root.clone(), retention);
        let pending = valid_pending("completed-after-gc");
        write_pending_provenance(&store, &pending);
        store.write_pending_proposal(&pending).unwrap();
        let receipt = store.write_completed_proposal(&pending).unwrap();
        store.cleanup_completed_proposal_guards(&receipt).unwrap();
        let mut ended = match ended(&pending.run_id, "agent-a", OutcomeStatus::Success) {
            RunRecord::Ended(ended) => ended,
            _ => unreachable!(),
        };
        // Backdate the terminal so the age cap takes it. `write_ended` verifies
        // the digest, which covers the `RunTermination` and not this stamp, so
        // it stays valid — and this field is exactly what the age cap reads.
        ended.ended_at = long_ago;
        store.write_ended(&ended).unwrap();

        assert_eq!(store.gc(), 1, "the completed run trace must be evicted");
        assert!(store.get_run_trace(&pending.run_id).is_none());
        assert_eq!(
            store
                .completed_proposal(
                    &pending.run_id,
                    &pending.client_id,
                    pending.requested_policy_session_id.as_deref(),
                    &pending.original_submission,
                )
                .unwrap(),
            Some(receipt.clone()),
            "receipt validity must not depend on a GC-eligible RunStarted"
        );

        let restarted = RunStore::new(root, retention);
        assert_eq!(
            restarted
                .completed_proposal_retry_owner(
                    pending.requested_policy_session_id.as_deref(),
                    &pending.original_submission,
                )
                .unwrap(),
            Some((pending.run_id.clone(), pending.client_id.clone())),
            "the retry tuple owner must remain recoverable after restart"
        );
        assert_eq!(restarted.all_completed_proposals().unwrap(), vec![receipt]);
    }

    #[test]
    fn resumed_completed_proposal_lookup_requires_one_exact_receipt() {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = store(tmp.path().join("runs"));
        let mut first = valid_pending("resumed-policy-rotation");
        first.requested_policy_session_id = Some("original-policy-one".to_string());
        first.policy_session_id = first.requested_policy_session_id.clone();
        write_pending_provenance(&store, &first);
        store.write_pending_proposal(&first).unwrap();
        let first_receipt = store.write_completed_proposal(&first).unwrap();
        store
            .cleanup_completed_proposal_guards(&first_receipt)
            .unwrap();

        assert_eq!(
            store
                .completed_proposal_for_resumed_owner(
                    &first.run_id,
                    &first.client_id,
                    &first.original_submission,
                )
                .unwrap(),
            Some(first_receipt),
            "a single exact receipt is recoverable without the closed policy-session id"
        );
        assert!(
            store
                .completed_proposal_for_resumed_owner(
                    &first.run_id,
                    &first.client_id,
                    &json!({"id":"different"}),
                )
                .unwrap()
                .is_none(),
            "a different proposal must not match"
        );

        let mut second = first.clone();
        second.requested_policy_session_id = Some("original-policy-two".to_string());
        second.policy_session_id = second.requested_policy_session_id.clone();
        store
            .write_execution_marker(&marker_for_pending(&second))
            .unwrap();
        store.write_pending_proposal(&second).unwrap();
        let second_receipt = store.write_completed_proposal(&second).unwrap();
        store
            .cleanup_completed_proposal_guards(&second_receipt)
            .unwrap();

        let error = store
            .completed_proposal_for_resumed_owner(
                &first.run_id,
                &first.client_id,
                &first.original_submission,
            )
            .unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(error.to_string().contains("ambiguous"));
    }

    #[test]
    fn resumed_completed_proposal_lookup_enforces_receipt_count_before_deserialization() {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = store(tmp.path().join("runs"));
        let target = valid_pending("resumed-receipt-count-limit");
        let target_receipt = write_completed_receipt_fixture(&store, target.clone(), None, false);
        for index in 1..MAX_RESUMED_PROPOSAL_RECEIPTS_PER_RUN {
            let mut unrelated = target.clone();
            unrelated.client_id = format!("unrelated-client-{index}");
            write_completed_receipt_fixture(&store, unrelated, None, false);
        }

        assert_eq!(
            store
                .completed_proposal_for_resumed_owner(
                    &target.run_id,
                    &target.client_id,
                    &target.original_submission,
                )
                .unwrap(),
            Some(target_receipt),
            "the exact receipt-count limit remains recoverable"
        );

        let corrupt_path = store
            .completed_response_run_root(&target.run_id)
            .join("count-plus-one-is-never-deserialized.json");
        let mut corrupt = store.create_private_file(&corrupt_path).unwrap();
        corrupt.write_all(b"not-json").unwrap();
        corrupt.sync_all().unwrap();
        let error = store
            .completed_proposal_for_resumed_owner(
                &target.run_id,
                &target.client_id,
                &target.original_submission,
            )
            .unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(error.to_string().contains("receipt count limit"));
    }

    #[test]
    fn resumed_completed_proposal_lookup_enforces_byte_limit_before_deserialization() {
        let at_limit = tempfile::TempDir::new().unwrap();
        let at_limit_store = store(at_limit.path().join("runs"));
        let target = valid_pending("resumed-receipt-byte-limit");
        let first_receipt_bytes = MAX_RESUMED_PROPOSAL_RECEIPT_BYTES_PER_RUN as usize / 2;
        let target_receipt = write_completed_receipt_fixture(
            &at_limit_store,
            target.clone(),
            Some(first_receipt_bytes),
            false,
        );
        let mut unrelated = target.clone();
        unrelated.client_id = "unrelated-byte-limit-client".to_string();
        write_completed_receipt_fixture(
            &at_limit_store,
            unrelated,
            Some(MAX_RESUMED_PROPOSAL_RECEIPT_BYTES_PER_RUN as usize - first_receipt_bytes),
            false,
        );
        assert_eq!(
            at_limit_store
                .completed_proposal_for_resumed_owner(
                    &target.run_id,
                    &target.client_id,
                    &target.original_submission,
                )
                .unwrap(),
            Some(target_receipt),
            "the exact cumulative-byte limit remains recoverable"
        );

        let cumulative_over_limit = tempfile::TempDir::new().unwrap();
        let cumulative_over_limit_store = store(cumulative_over_limit.path().join("runs"));
        write_completed_receipt_fixture(
            &cumulative_over_limit_store,
            target.clone(),
            Some(first_receipt_bytes),
            false,
        );
        let mut unrelated = target.clone();
        unrelated.client_id = "unrelated-byte-over-limit-client".to_string();
        write_completed_receipt_fixture(
            &cumulative_over_limit_store,
            unrelated,
            Some(MAX_RESUMED_PROPOSAL_RECEIPT_BYTES_PER_RUN as usize - first_receipt_bytes + 1),
            false,
        );
        let error = cumulative_over_limit_store
            .completed_proposal_for_resumed_owner(
                &target.run_id,
                &target.client_id,
                &target.original_submission,
            )
            .unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(error.to_string().contains("receipt byte limit"));

        let invalid_over_limit = tempfile::TempDir::new().unwrap();
        let invalid_over_limit_store = store(invalid_over_limit.path().join("runs"));
        write_completed_receipt_fixture(
            &invalid_over_limit_store,
            target.clone(),
            Some(MAX_RESUMED_PROPOSAL_RECEIPT_BYTES_PER_RUN as usize),
            true,
        );
        let error = invalid_over_limit_store
            .completed_proposal_for_resumed_owner(
                &target.run_id,
                &target.client_id,
                &target.original_submission,
            )
            .unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(
            error.to_string().contains("receipt byte limit"),
            "the byte ceiling must reject an oversized corrupt receipt before deserialization"
        );
    }

    #[test]
    fn resumed_completed_proposal_lookup_rejects_corrupt_receipt_under_limits() {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = store(tmp.path().join("runs"));
        let target = valid_pending("resumed-corrupt-receipt");
        let run_root = store.completed_response_run_root(&target.run_id);
        store.ensure_private_dir(&run_root).unwrap();
        let path = store
            .completed_response_path(
                &target.run_id,
                &target.client_id,
                target.requested_policy_session_id.as_deref(),
                &target.original_submission,
            )
            .unwrap();
        let mut corrupt = store.create_private_file(&path).unwrap();
        corrupt.write_all(b"not-json").unwrap();
        corrupt.sync_all().unwrap();

        let error = store
            .completed_proposal_for_resumed_owner(
                &target.run_id,
                &target.client_id,
                &target.original_submission,
            )
            .unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert!(!error.to_string().contains("receipt count limit"));
        assert!(!error.to_string().contains("receipt byte limit"));
    }

    #[test]
    fn completed_proposal_owner_lookup_does_not_scan_unrelated_receipts() {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = store(tmp.path().join("runs"));
        let unrelated_dir = store.completed_response_root().join("unrelated-run");
        car_secrets::ensure_private_dir(&unrelated_dir).unwrap();
        let path = unrelated_dir.join("corrupt.json");
        let mut file = car_secrets::create_private_file(&path).unwrap();
        file.write_all(b"not-json").unwrap();
        file.sync_all().unwrap();

        assert_eq!(
            store
                .completed_proposal_retry_owner(None, &json!({"id": "absent"}))
                .unwrap(),
            None,
            "a bounded content-addressed miss must not enumerate unrelated receipts"
        );
    }

    #[test]
    fn proposal_retry_reservation_is_first_writer_wins_and_exact_owner_idempotent() {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = store(tmp.path().join("runs"));
        let submission = json!({"id":"reserved","source":"test","actions":[]});
        assert_eq!(
            store
                .reserve_proposal_retry_owner("first-run", "first-client", None, &submission)
                .unwrap(),
            ProposalRetryReservation::Acquired
        );
        for (run_id, client_id) in [
            ("first-run", "first-client"),
            ("second-run", "second-client"),
        ] {
            assert_eq!(
                store
                    .reserve_proposal_retry_owner(run_id, client_id, None, &submission)
                    .unwrap(),
                ProposalRetryReservation::Existing {
                    run_id: "first-run".to_string(),
                    client_id: "first-client".to_string(),
                }
            );
        }
        assert_eq!(
            store
                .completed_proposal_retry_owner(None, &submission)
                .unwrap(),
            Some(("first-run".to_string(), "first-client".to_string()))
        );
    }

    #[test]
    fn partial_proposal_retry_reservation_fails_closed_without_replacement() {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = store(tmp.path().join("runs"));
        let submission = json!({"id":"partial","source":"test","actions":[]});
        let root = store.completed_response_index_root();
        car_secrets::ensure_private_dir(&root).unwrap();
        let path = store
            .completed_response_owner_path(None, &submission)
            .unwrap();
        let mut file = car_secrets::create_private_file(&path).unwrap();
        file.write_all(b"{").unwrap();
        file.sync_all().unwrap();

        let error = store
            .reserve_proposal_retry_owner("partial-run", "partial-client", None, &submission)
            .unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert_eq!(std::fs::read(&path).unwrap(), b"{");
    }

    #[test]
    fn completed_proposal_owner_claim_rejects_a_different_run_for_the_same_retry_tuple() {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = store(tmp.path().join("runs"));
        let first = valid_pending("owner-first-run");
        write_pending_provenance(&store, &first);
        store.write_pending_proposal(&first).unwrap();
        store.write_completed_proposal(&first).unwrap();

        let mut conflicting = first.clone();
        conflicting.run_id = "owner-conflicting-run".to_string();
        write_pending_provenance(&store, &conflicting);
        store.write_pending_proposal(&conflicting).unwrap();
        let error = store.write_completed_proposal(&conflicting).unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists);
        assert!(error.to_string().contains("different completed response"));
        assert_eq!(
            store
                .completed_proposal_retry_owner(
                    first.requested_policy_session_id.as_deref(),
                    &first.original_submission,
                )
                .unwrap(),
            Some((first.run_id.clone(), first.client_id.clone())),
            "a conflicting writer must not replace the first durable owner"
        );
        assert!(
            store
                .completed_proposal(
                    &conflicting.run_id,
                    &conflicting.client_id,
                    conflicting.requested_policy_session_id.as_deref(),
                    &conflicting.original_submission,
                )
                .unwrap()
                .is_none(),
            "the rejected owner must not publish a receipt"
        );
    }

    #[test]
    fn startup_enumeration_backfills_a_legacy_completed_proposal_owner() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("runs");
        let initial = store(root.clone());
        let pending = valid_pending("legacy-owner-backfill");
        write_pending_provenance(&initial, &pending);
        initial.write_pending_proposal(&pending).unwrap();
        let receipt = initial.write_completed_proposal(&pending).unwrap();
        let owner_path = initial
            .completed_response_owner_path(
                pending.requested_policy_session_id.as_deref(),
                &pending.original_submission,
            )
            .unwrap();
        std::fs::remove_file(&owner_path).unwrap();
        sync_directory(owner_path.parent().unwrap()).unwrap();

        let restarted = store(root);
        assert!(
            restarted
                .completed_proposal_retry_owner(
                    pending.requested_policy_session_id.as_deref(),
                    &pending.original_submission,
                )
                .unwrap()
                .is_none(),
            "pre-index receipts begin without an ownership claim"
        );
        assert_eq!(restarted.all_completed_proposals().unwrap(), vec![receipt]);
        assert_eq!(
            restarted
                .completed_proposal_retry_owner(
                    pending.requested_policy_session_id.as_deref(),
                    &pending.original_submission,
                )
                .unwrap(),
            Some((pending.run_id, pending.client_id)),
            "startup receipt enumeration must durably backfill the owner claim"
        );
    }
}