car_server_core/coder/

session.rs

//! Coder session state machine, event stream, and persistence.
//!
//! A session moves `Created → ContractProposed → ContractConfirmed → Running →
//! NeedsApproval → Merged`, with `Failed`/`Abandoned` as the other terminal
//! states. Every transition is validated, emitted as a [`CoderEvent`], audited
//! to the event log, and snapshotted as JSON under the state dir so a daemon
//! restart can at least report orphaned sessions (full resume is out of scope).

use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use car_eventlog::{EventKind, EventLog};
use car_multi::{AgentWorkspace, WorkspaceConfig};

use super::contract::{CheckResult, OutcomeContract};
use super::router::EngineChoice;

/// Cooperative cancellation flag, checked between turns and checks.
pub type CancelFlag = Arc<AtomicBool>;

/// Callback receiving every [`CoderEvent`] (WS fanout, CLI rendering, tests).
pub type EventEmitter = Arc<dyn Fn(CoderEvent) + Send + Sync>;

/// Mid-session user-input rendezvous.
///
/// When a loop wants to ask the user a question, it parks a oneshot sender here
/// and awaits the receiver; `coder.respond` takes the sender and fulfills it.
/// At most one request is pending at a time — a loop runs single-threaded, so
/// it cannot have two questions in flight, and `coder.respond` errors cleanly
/// when nothing is parked. The sender is dropped (which surfaces as a closed
/// channel to the waiter) if the session is cancelled or torn down before the
/// user answers.
#[derive(Default)]
pub struct UserInputGate {
    pending: Mutex<Option<tokio::sync::oneshot::Sender<String>>>,
    /// The prompt of the currently-parked question, so a board can render
    /// *what* is being asked from a session summary without replaying the
    /// event stream. Cleared whenever the gate is.
    prompt: Mutex<Option<String>>,
}

impl UserInputGate {
    pub fn new() -> Self {
        Self::default()
    }

    /// Park a fresh oneshot for a new question, returning the receiver the
    /// caller awaits. Any previously-parked (unanswered) sender is dropped,
    /// which closes its receiver — the prior waiter, if somehow still alive,
    /// then unblocks with an error rather than hanging forever.
    pub fn park(&self, prompt: &str) -> tokio::sync::oneshot::Receiver<String> {
        let (tx, rx) = tokio::sync::oneshot::channel();
        *self.pending.lock().expect("user-input gate poisoned") = Some(tx);
        *self.prompt.lock().expect("user-input gate poisoned") = Some(prompt.to_string());
        rx
    }

    /// The prompt of the currently-parked question, if any.
    pub fn pending_prompt(&self) -> Option<String> {
        self.prompt
            .lock()
            .expect("user-input gate poisoned")
            .clone()
    }

    /// Fulfill the parked request with `answer`. Returns `Err` when nothing is
    /// pending (so `coder.respond` can report "no pending request") or when the
    /// waiter has already gone away (cancelled/timed-out).
    pub fn fulfill(&self, answer: String) -> Result<(), String> {
        let tx = self
            .pending
            .lock()
            .expect("user-input gate poisoned")
            .take()
            .ok_or("no pending user-input request for this session")?;
        *self.prompt.lock().expect("user-input gate poisoned") = None;
        tx.send(answer)
            .map_err(|_| "the session is no longer waiting for input".to_string())
    }

    /// Drop any parked sender (cancellation/teardown): unblocks a waiter with a
    /// closed channel.
    pub fn clear(&self) {
        *self.pending.lock().expect("user-input gate poisoned") = None;
        *self.prompt.lock().expect("user-input gate poisoned") = None;
    }

    /// Whether a request is currently parked.
    pub fn is_pending(&self) -> bool {
        self.pending
            .lock()
            .expect("user-input gate poisoned")
            .is_some()
    }
}

/// `coder` under the CAR state root (`~/.car/coder` unless `CAR_HOME` moves the
/// root) — session snapshots, event journals, and worktrees. This is only the
/// default; `CAR_CODER_STATE_DIR` still overrides it outright in
/// `coder_state_dir`.
pub fn default_state_dir() -> Result<PathBuf, String> {
    let root = car_home::root()
        .ok_or("cannot resolve home directory (CAR_HOME/HOME/USERPROFILE unset)")?;
    Ok(root.join("coder"))
}

fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

fn proactive_maintenance_event_data(
    report: &car_memgine::ProactiveMaintenanceReport,
) -> HashMap<String, Value> {
    let mut data = proactive_trigger_event_data(&report.trigger);
    data.insert(
        "saved_count".to_string(),
        Value::from(report.saved.len() as u64),
    );
    data.insert(
        "skipped_existing".to_string(),
        Value::from(report.skipped_existing as u64),
    );
    data.insert(
        "status_updated".to_string(),
        Value::from(report.status.is_some()),
    );
    data
}

fn proactive_intervention_event_data(
    decision: &car_memgine::ProactiveMemoryDecision,
) -> HashMap<String, Value> {
    let mut data = HashMap::new();
    match decision {
        car_memgine::ProactiveMemoryDecision::Inject {
            selected,
            candidates,
            bank,
            ..
        } => {
            data.insert("decision".to_string(), Value::from("inject"));
            data.insert("selected_id".to_string(), Value::from(selected.id.clone()));
            data.insert(
                "selected_kind".to_string(),
                Value::from(format!("{:?}", selected.kind).to_ascii_lowercase()),
            );
            data.insert(
                "candidate_count".to_string(),
                Value::from(candidates.len() as u64),
            );
            data.insert(
                "bank_knowledge".to_string(),
                Value::from(bank.knowledge as u64),
            );
            data.insert(
                "bank_procedural".to_string(),
                Value::from(bank.procedural as u64),
            );
            data.insert(
                "bank_open_subgoals".to_string(),
                Value::from(bank.open_subgoals as u64),
            );
        }
        car_memgine::ProactiveMemoryDecision::Silent {
            reason,
            candidates,
            bank,
        } => {
            data.insert("decision".to_string(), Value::from("silent"));
            data.insert("reason".to_string(), Value::from(reason.clone()));
            data.insert(
                "candidate_count".to_string(),
                Value::from(candidates.len() as u64),
            );
            data.insert(
                "bank_knowledge".to_string(),
                Value::from(bank.knowledge as u64),
            );
            data.insert(
                "bank_procedural".to_string(),
                Value::from(bank.procedural as u64),
            );
            data.insert(
                "bank_open_subgoals".to_string(),
                Value::from(bank.open_subgoals as u64),
            );
        }
    }
    data
}

fn proactive_trigger_event_data(
    trigger: &car_memgine::ProactiveMemoryTrigger,
) -> HashMap<String, Value> {
    HashMap::from([
        (
            "repeated_failures".to_string(),
            Value::from(trigger.repeated_failures as u64),
        ),
        ("tool_error".to_string(), Value::from(trigger.tool_error)),
        (
            "explicit_uncertainty".to_string(),
            Value::from(trigger.explicit_uncertainty),
        ),
        (
            "high_risk_action".to_string(),
            Value::from(trigger.high_risk_action),
        ),
        (
            "context_shift".to_string(),
            Value::from(trigger.context_shift),
        ),
    ])
}

/// Session lifecycle states.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CoderState {
    Created,
    ContractProposed,
    ContractConfirmed,
    Running,
    NeedsApproval,
    Merged,
    /// The session finished correctly **without a diff**: it investigated and
    /// established that no code should change. A non-failure terminal, and the
    /// only one that is not reached through a green contract over a diff.
    ///
    /// Reachable two ways, both gated outside inference — see
    /// [`NoChangeVerification`]. A model can *nominate* this outcome; it can
    /// never transition into it.
    Reported,
    Failed,
    Abandoned,
}

impl CoderState {
    pub fn is_terminal(&self) -> bool {
        matches!(
            self,
            Self::Merged | Self::Reported | Self::Failed | Self::Abandoned
        )
    }

    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Created => "created",
            Self::ContractProposed => "contract_proposed",
            Self::ContractConfirmed => "contract_confirmed",
            Self::Running => "running",
            Self::NeedsApproval => "needs_approval",
            Self::Merged => "merged",
            Self::Reported => "reported",
            Self::Failed => "failed",
            Self::Abandoned => "abandoned",
        }
    }
}

/// What a session is waiting on a *human* for, right now.
///
/// Computed server-side and shipped on every session summary so every client
/// (the `car board` TUI, CarHost, milo) says the same words about the same
/// state — the same precedent as `DiffReady::overlap_disclosure`, where
/// hand-rolling the sentence per renderer had already produced two divergent
/// copies of one sentence.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NeedsYou {
    /// A drafted outcome contract is waiting for confirm/reject/revise.
    Contract,
    /// The loop asked a mid-session question and is parked on the answer.
    Question,
    /// The work is done and the diff is waiting for merge approval.
    Approval,
    /// A no-change finding is waiting to be accepted or rejected. Distinct from
    /// [`NeedsYou::Approval`] on purpose: there is no diff, and a board that
    /// says "diff ready for approval" over an empty worktree is lying to the
    /// operator about what they are being asked to look at.
    Finding,
    /// The run is blocked on sign-in and is waiting for a credential.
    Auth,
}

impl NeedsYou {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Contract => "contract",
            Self::Question => "question",
            Self::Approval => "approval",
            Self::Finding => "finding",
            Self::Auth => "auth",
        }
    }

    /// The fixed operator-facing wording. The daemon owns it so two boards
    /// never disagree about what the same session needs.
    pub fn label(&self) -> &'static str {
        match self {
            Self::Contract => "contract awaiting confirmation",
            Self::Question => "question waiting",
            Self::Approval => "diff ready for approval",
            Self::Finding => "finding ready for review",
            Self::Auth => "sign-in needed",
        }
    }

    /// Parse the wire form back (used when reading a persisted snapshot's
    /// last-known value for a non-live session).
    pub fn parse(s: &str) -> Option<Self> {
        match s {
            "contract" => Some(Self::Contract),
            "question" => Some(Self::Question),
            "approval" => Some(Self::Approval),
            "finding" => Some(Self::Finding),
            "auth" => Some(Self::Auth),
            _ => None,
        }
    }
}

/// Which gate a `NeedsApproval` session is sitting on.
///
/// `NeedsApproval` is deliberately reused for findings rather than growing a
/// second pending state — it already *is* the pending state, and a parallel one
/// would duplicate the gate and complicate every FFI consumer. This field is
/// how a client tells the two apart, so nothing has to infer the gate type from
/// whether a diff happens to exist.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ApprovalKind {
    /// A diff is waiting to be merged.
    Merge,
    /// A no-change finding is waiting to be accepted.
    Finding,
}

impl ApprovalKind {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Merge => "merge",
            Self::Finding => "finding",
        }
    }

    pub fn parse(s: &str) -> Option<Self> {
        match s {
            "merge" => Some(Self::Merge),
            "finding" => Some(Self::Finding),
            _ => None,
        }
    }
}

/// Why a session concluded no code should change.
///
/// The runtime can only independently verify the first of these, and only in a
/// narrow operational sense — see [`NoChangeVerification::RuntimeBaselineGreen`].
/// The other two rest on judgement a green test run cannot supply, so they
/// always reach a human.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NoChangeKind {
    /// The reported problem does not exist — the code already handles the case.
    PremiseWrong,
    /// The behaviour is intentional. Passing tests establish what the code does
    /// now, never that maintainers intended it or still want it, so this is not
    /// runtime-verifiable at all.
    DeliberateBehavior,
    /// The fix is real but is not a code change — a migration, an operator
    /// decision, a rollout. No diff substitutes for it.
    NonCodeDecision,
}

impl NoChangeKind {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::PremiseWrong => "premise_wrong",
            Self::DeliberateBehavior => "deliberate_behavior",
            Self::NonCodeDecision => "non_code_decision",
        }
    }

    pub fn parse(s: &str) -> Option<Self> {
        match s {
            "premise_wrong" => Some(Self::PremiseWrong),
            "deliberate_behavior" => Some(Self::DeliberateBehavior),
            "non_code_decision" => Some(Self::NonCodeDecision),
            _ => None,
        }
    }
}

/// What makes a [`NoChangeKind`] verdict trustworthy.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NoChangeVerification {
    /// Every check of a **trusted** contract passed against the unmodified
    /// baseline. The claim this licenses is narrow and should be read literally:
    /// *the contract was already satisfied before any work*. It is NOT a claim
    /// that the model's explanation of why is true.
    ///
    /// "Trusted" is the load-bearing word and it is defined in exactly one
    /// place, [`ContractProvenance::is_trusted`]. Widening it is how this gate
    /// stops working.
    RuntimeBaselineGreen,
    /// A human read the finding and accepted it.
    HumanApproved,
}

impl NoChangeVerification {
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::RuntimeBaselineGreen => "runtime_baseline_green",
            Self::HumanApproved => "human_approved",
        }
    }
}

/// Where an outcome contract's bytes came from.
///
/// This exists because the autonomous no-change path is gated on it, which
/// makes it a security boundary rather than bookkeeping. A model that can
/// author its own contract and then cite that contract passing as proof it need
/// not work has an escape hatch from every hard task.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ContractProvenance {
    /// Supplied as bytes by the operator (`--contract-file`).
    OperatorSupplied,
    /// A human confirmed it through the contract gate before work began.
    HumanConfirmed,
    /// Generated by the runtime from telemetry it collected itself, before this
    /// session existed — e.g. a reproduction attached to a report CAR filed from
    /// a `DurableFixProposal`. External to the session in the sense that matters:
    /// neither this session's model nor a member of the public wrote it.
    RuntimeGenerated,
    /// Derived by this session's model, typically from an issue body. **Never
    /// trusted**, at any tier. An issue body is untrusted input — on a public
    /// tracker it is attacker-controlled — so a contract derived from one is a
    /// stranger's definition of done.
    ModelDerived,
}

impl ContractProvenance {
    /// The single definition of "trusted". Every caller must route through this
    /// rather than matching the variants itself, so widening it is one visible
    /// edit rather than a drift across call sites.
    pub fn is_trusted(&self) -> bool {
        match self {
            Self::OperatorSupplied | Self::HumanConfirmed | Self::RuntimeGenerated => true,
            Self::ModelDerived => false,
        }
    }

    pub fn as_str(&self) -> &'static str {
        match self {
            Self::OperatorSupplied => "operator_supplied",
            Self::HumanConfirmed => "human_confirmed",
            Self::RuntimeGenerated => "runtime_generated",
            Self::ModelDerived => "model_derived",
        }
    }
}

/// What the model actually said when it called `report_no_change`.
///
/// Raw, unjudged, and carried out of the loop untouched. The loop does not have
/// the baseline results, the contract's provenance, or the mutation ledger, and
/// giving it those so it could self-adjudicate is precisely the design this
/// avoids: nomination and adjudication live in different layers on purpose.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct NoChangeNomination {
    pub kind: NoChangeKind,
    pub summary: String,
    pub evidence: String,
}

/// A nominated (or accepted) "no code should change" conclusion.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct NoChangeFinding {
    pub kind: NoChangeKind,
    /// One-line conclusion.
    pub summary: String,
    /// What was examined to reach it.
    pub evidence: String,
    /// `None` while the finding is still nominated and awaiting a human.
    pub verification: Option<NoChangeVerification>,
    pub proposed_at: u64,
    pub resolved_at: Option<u64>,
    pub resolver_comment: Option<String>,
    /// The baseline check results as they stood when the finding was nominated,
    /// captured so the verdict can be audited later without re-running anything.
    pub baseline_checks: Vec<CheckResult>,
}

/// Derive [`NeedsYou`] from the three facts that decide it. Split out from the
/// live registry so the table in `docs/proposals/coder-board-wire-contract.md`
/// §1 is directly testable without a daemon.
///
/// `auth_outstanding` means "an `auth_required` event is the latest unresolved
/// auth event" — the caller clears it on any subsequent non-auth event or state
/// change (see `coder::rpc::AttentionState`).
pub fn needs_you_from(
    state: CoderState,
    question_pending: bool,
    auth_outstanding: bool,
    approval_kind: Option<ApprovalKind>,
) -> Option<NeedsYou> {
    match state {
        CoderState::ContractProposed => Some(NeedsYou::Contract),
        // Which gate is decided by `approval_kind`, never by guessing from the
        // presence of a diff. Absent (an older snapshot) reads as a merge gate,
        // which is what every pre-finding session was.
        CoderState::NeedsApproval => match approval_kind {
            Some(ApprovalKind::Finding) => Some(NeedsYou::Finding),
            Some(ApprovalKind::Merge) | None => Some(NeedsYou::Approval),
        },
        // Question wins over auth: a parked question is a literal prompt on
        // screen with a waiter behind it, while an outstanding auth event only
        // means the loop is polling for a credential.
        CoderState::Running if question_pending => Some(NeedsYou::Question),
        CoderState::Running if auth_outstanding => Some(NeedsYou::Auth),
        _ => None,
    }
}

/// Whether `from → to` is a legal transition. Any non-terminal state may move
/// to `Failed` (errors happen anywhere) or `Abandoned` (user cancel); terminal
/// states never move.
pub fn can_transition(from: CoderState, to: CoderState) -> bool {
    use CoderState::*;
    if from.is_terminal() {
        return false;
    }
    matches!(to, Failed | Abandoned)
        || matches!(
            (from, to),
            (Created, ContractProposed)
                | (ContractProposed, ContractProposed) // re-propose after edit
                | (ContractProposed, ContractConfirmed)
                | (ContractConfirmed, Running)
                | (Running, NeedsApproval)
                | (NeedsApproval, Merged)
                // The autonomous no-change path, gated on a trusted contract
                // being green against an unmodified baseline.
                | (Running, Reported)
                // A human accepted a nominated finding.
                | (NeedsApproval, Reported)
                // A human REJECTED a nominated finding. The first backward edge
                // in this table, and deliberate: a rejected nomination must not
                // become a scored loss. The session goes back to work and can
                // still reach a diff.
                | (NeedsApproval, Running)
        )
}

/// One event in a session's stream. `seq` is monotonically increasing per
/// session so clients can resume from a cursor after reconnect.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CoderEvent {
    pub session_id: String,
    pub seq: u64,
    pub ts: u64,
    #[serde(flatten)]
    pub kind: CoderEventKind,
}

/// What happened. Serialized with `"type": "snake_case_name"` for WS clients.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum CoderEventKind {
    StateChanged {
        from: String,
        to: String,
    },
    ContractProposed {
        contract: OutcomeContract,
    },
    EngineSelected {
        engine: String,
        reason: String,
    },
    EngineFallback {
        from: String,
        to: String,
        reason: String,
    },
    /// The preferred inference lane was skipped and a different model served
    /// the call. Distinct from `EngineFallback`, which is about the coder
    /// ENGINE (native vs an external CLI), not the model behind it.
    ///
    /// Emitted at most once per PHASE — contract derivation, each contract
    /// revision, and the run loop announce independently, and a session that
    /// degrades in more than one of them emits more than one. The guard is
    /// against narrating every routing decision inside a phase, not against a
    /// second phase reporting a degrade the operator has not seen resolved.
    ModelFallback {
        /// The lane that was skipped.
        from: String,
        /// The model that actually served the call.
        to: String,
        /// Why, in words an operator can act on.
        reason: String,
    },
    IterationStarted {
        n: u32,
        max: u32,
    },
    /// The run is blocked on **sign-in** and is waiting for the human, rather
    /// than failing. Not terminal: if a credential appears within `wait_secs`
    /// the session resumes from where it stopped, worktree intact.
    ///
    /// Distinct from `Error` on purpose. A client should surface this as an
    /// action the user can take ("sign in to continue"), because it is the one
    /// failure mode a person standing at the machine can clear in seconds — and
    /// previously it read as `no inference backend is available`, which points
    /// at models and accounts instead of at the sign-in it actually needs.
    AuthRequired {
        /// The underlying auth error, for diagnosis.
        message: String,
        /// How long the session will wait before giving up.
        wait_secs: u64,
    },
    /// The session hit its wall-clock ceiling and the next iteration was not
    /// admitted. Terminal, and the session ends `Failed` — it never reaches the
    /// merge gate, which requires green checks. The worktree IS retained for
    /// postmortem (the budget path forces `keep_workspace_on_failure`), so the
    /// partial work survives on disk at `workspace_path`.
    BudgetExhausted {
        /// Human-readable, naming both the elapsed time and the ceiling.
        reason: String,
        elapsed_secs: u64,
        /// Iterations completed before the ceiling was reached.
        iterations: u32,
    },
    /// A worker invocation died mid-run (timeout / I/O) with the contract still
    /// red, and the same hypothesis is being re-invoked.
    ///
    /// Distinct from `IterationStarted` on purpose: a retry costs no hypothesis,
    /// so folding it in would make that event's `n`/`max` misreport the budget.
    /// A chronically flaky CLI is otherwise indistinguishable from a fast clean
    /// one in the A/B's wall-clock.
    InvocationRetried {
        /// The hypothesis being retried (`IterationStarted.n`).
        hypothesis: u32,
        /// Transport error that ended the invocation.
        reason: String,
        /// Transient retries left for this session.
        retries_remaining: u32,
    },
    PlanText {
        text: String,
    },
    ToolCall {
        tool: String,
        params_preview: String,
    },
    ToolResult {
        tool: String,
        ok: bool,
        preview: String,
    },
    CheckStarted {
        name: String,
    },
    CheckCompleted {
        result: CheckResult,
    },
    /// The contract evaluated against the **unmodified** worktree at session
    /// start (car#707). Distinct from `CheckStarted`/`CheckCompleted`, which
    /// mean "the contract is being evaluated on the work" — replaying those for
    /// a baseline would show checks going green before a line was written.
    /// `gates_nothing` is true when every check already passed, i.e. the
    /// contract verifies nothing for this task.
    ContractBaseline {
        results: Vec<CheckResult>,
        gates_nothing: bool,
    },
    ExternalEvent {
        raw: Value,
    },
    /// The loop nominated a "no code should change" conclusion. A nomination,
    /// not a verdict: the runtime decides what happens next.
    FindingProposed {
        finding: NoChangeFinding,
    },
    /// A nominated finding was accepted or rejected.
    FindingResolved {
        accepted: bool,
        comment: Option<String>,
    },
    DiffReady {
        stat: String,
        /// The patch body, tail-capped to the configured budget. Named for what
        /// it is; `patch_truncated` (the bool) says whether it is partial.
        patch: String,
        /// True when `patch` is a tail. A UI must be able to say "you are
        /// approving against a partial diff" without string-matching the
        /// `…[truncated]…` marker (car#706).
        patch_truncated: bool,
        /// Size of the untruncated patch.
        patch_full_bytes: usize,
        /// How many distinct paths the diff touches. Named `paths`, not
        /// `files`, because a rename contributes BOTH of its endpoints — one
        /// file moved is two paths touched, and for a reviewer asking "what did
        /// this session reach into" that is the honest number. `stat` carries it
        /// too, but only as prose a client must parse; scope explosion is what a
        /// reviewer most needs stated plainly before deciding whether to read
        /// the patch at all.
        changed_paths: usize,
        /// Contract checks whose commands execute a path this diff modified.
        /// Disclosure, never denial: editing tests is frequently the task, and
        /// the human at the gate is who should judge which case this is.
        contract_overlap: Vec<super::overlap::CheckOverlap>,
        /// The rendered disclosure sentence, or `None` when nothing overlaps.
        ///
        /// On the wire so every surface prints the SAME words. Hand-rolling it
        /// per renderer had already lost "that is often legitimate" from both
        /// the CLI and the host app while the log kept it — dropping the
        /// non-accusatory half of a sentence whose entire design posture is
        /// disclosure rather than accusation, and leaving the only tested copy
        /// the one no human reads. `contract_overlap` stays alongside it for
        /// machine consumers that want the structure.
        overlap_disclosure: Option<String>,
    },
    UserInputRequested {
        prompt: String,
    },
    /// The mid-session question's answer window closed server-side without an
    /// answer. The loop carried on without one; the prompt is DEAD.
    ///
    /// Its own event because a client has no other way to learn: the gate
    /// simply stops being pending, which is a state a board can only discover
    /// by asking again. Without this, a board kept rendering the question as
    /// live — and counting it under "needs you" — until the operator happened
    /// to refresh. It is also what drives the `coder.session_changed` fanout
    /// that drops `needs_you` back to null.
    UserInputExpired {
        /// The question that went unanswered, so a client can match it to the
        /// prompt it is showing.
        prompt: String,
        /// How long the daemon waited.
        waited_secs: u64,
    },
    /// A `coder.revise_contract` request could NOT be honored: the redraft did
    /// not validate, or the request was not expressible as checks. The session
    /// stays at the gate with the PREVIOUS contract intact.
    ///
    /// Its own event rather than a generic `Error` because the operator needs
    /// to know the contract they are still looking at is the old one — a
    /// revision that silently passes as applied is the failure mode this
    /// exists to make impossible.
    ContractRevisionRejected {
        /// The operator's plain-English request, verbatim.
        request: String,
        /// Why it could not be honored (the derivation/validation failure).
        reason: String,
    },
    MergeCompleted {
        branch: String,
    },
    Error {
        message: String,
    },
}

/// Per-session event fanout + audit. Emits to the registered emitter (WS
/// subscribers) and journals the audit-relevant subset to a JSONL event log.
pub struct EventSink {
    session_id: String,
    seq: AtomicU64,
    emitter: Option<EventEmitter>,
    journal: Option<Mutex<EventLog>>,
}

impl EventSink {
    pub fn new(
        session_id: impl Into<String>,
        emitter: Option<EventEmitter>,
        journal_path: Option<PathBuf>,
    ) -> Self {
        Self {
            session_id: session_id.into(),
            seq: AtomicU64::new(0),
            emitter,
            journal: journal_path.map(|p| Mutex::new(EventLog::with_journal(p))),
        }
    }

    /// A sink that drops everything — unit tests that don't assert on events.
    pub fn test_sink() -> Self {
        Self::new("coder-test", None, None)
    }

    /// Collect events into a shared Vec — tests that DO assert on events.
    pub fn collecting(session_id: &str) -> (Self, Arc<Mutex<Vec<CoderEvent>>>) {
        let collected: Arc<Mutex<Vec<CoderEvent>>> = Arc::new(Mutex::new(Vec::new()));
        let sink_copy = collected.clone();
        let emitter: EventEmitter = Arc::new(move |e| {
            sink_copy.lock().expect("collector poisoned").push(e);
        });
        (Self::new(session_id, Some(emitter), None), collected)
    }

    pub fn emit(&self, kind: CoderEventKind) -> CoderEvent {
        let event = CoderEvent {
            session_id: self.session_id.clone(),
            seq: self.seq.fetch_add(1, Ordering::SeqCst),
            ts: now_secs(),
            kind,
        };
        self.audit(&event);
        if let Some(emitter) = &self.emitter {
            emitter(event.clone());
        }
        event
    }

    /// Append a durable `TurnCompleted` audit record for a coder-loop terminal.
    ///
    /// The coder loop has no `Runtime` in scope (only this sink), so this mirrors
    /// [`car_engine::Runtime::record_turn_completed`] directly onto the coder
    /// session journal — the same `EventKind::TurnCompleted` + data shape the
    /// assistant path emits (via the shared `car_engine::goal::turn_completed_data`),
    /// so the coder-path false-success / truncation / turn-budget-burn signal is
    /// captured in the exact form the harness miners already understand.
    ///
    /// Consumption is a separate follow-up, NOT done here: these events land in
    /// the coder session journal (`<state_dir>/<session_id>.events.jsonl`), a
    /// durable record read offline / via the FFI `diagnose_from_jsonl`. The
    /// in-process daemon miners (`harness_adapt::diagnose`,
    /// `evolution::failed_trace_events`) run over `session.runtime.log` (the
    /// assistant path), so they do NOT yet consume this coder journal — wiring it
    /// into the daemon evolution path is tracked separately. Journal-only: not a
    /// WS-streamed `CoderEvent`, matching how P0b kept `TurnCompleted` off the
    /// live `AssistantEvent` stream (no WS/FFI surface change).
    pub fn record_turn_completed(
        &self,
        decision: &str,
        stop_reason: Option<&str>,
        was_truncated: bool,
        turns: u32,
        model: &str,
        // Every distinct model that served a turn in this iteration, first-seen
        // order. `model` is whichever one reached the terminal; these are all of
        // them, which is a different question when the chain routes per request.
        models_served: &[String],
    ) {
        let Some(journal) = &self.journal else { return };
        let mut data = car_engine::goal::turn_completed_data(
            decision,
            stop_reason,
            was_truncated,
            turns,
            model,
        );
        // Added here rather than in `turn_completed_data`, which the assistant
        // path shares: this is a coder-loop fact, and widening the shared
        // helper's signature would make every caller answer a question only
        // this one has. Additive on the journal record — a reader that does not
        // know the key sees exactly what it saw before.
        if !models_served.is_empty() {
            data.insert(
                "models_served".to_string(),
                Value::Array(
                    models_served
                        .iter()
                        .map(|m| Value::String(m.clone()))
                        .collect(),
                ),
            );
        }
        if let Ok(mut log) = journal.lock() {
            log.append(EventKind::TurnCompleted, Some(&self.session_id), None, data);
        }
    }

    /// Journal one mid-run backbone change.
    ///
    /// **Written directly rather than recognized in [`EventSink::audit`], once
    /// per distinct hop rather than once per session.** The live
    /// `ModelFallback` event is behind a once-per-phase latch so the stream
    /// does not narrate every routing decision — right for a stream, wrong for
    /// a record. A run that degraded twice has two facts, and journaling off
    /// the latched emit would keep only the first.
    ///
    /// The caller supplies the deduplication, because the fact worth recording
    /// is a distinct transition and a persistent condition re-reports the same
    /// one on every turn. This method itself always appends: it is the record,
    /// not the policy.
    ///
    /// car#1333 recorded WHO wrote each turn (`models_served` on the turn
    /// terminal). This records WHY the backbone moved, which is a genuinely
    /// different fact rather than the same one twice: without it, a run that
    /// silently degraded to another model looks — to `harness_adapt`-style
    /// mining and to a human reading the journal — like the code under test
    /// behaving badly (car#1351).
    ///
    /// Journal-only, no WS/FFI surface, same as the two above.
    pub fn record_model_fallback(&self, from: &str, to: &str, reason: &str) {
        let Some(journal) = &self.journal else { return };
        let mut data = std::collections::HashMap::new();
        data.insert("from".to_string(), Value::String(from.to_string()));
        data.insert("to".to_string(), Value::String(to.to_string()));
        data.insert("reason".to_string(), Value::String(reason.to_string()));
        if let Ok(mut log) = journal.lock() {
            log.append(EventKind::ModelFallback, Some(&self.session_id), None, data);
        }
    }

    /// Journal one merge-gate verdict.
    ///
    /// **Written here rather than derived from the event stream, deliberately.**
    /// The obvious shape is to bridge the verdict out as a `foreman: "gate"`
    /// `CoderEvent` and have [`EventSink::audit`] recognize it — one path, live
    /// and durable together. That is forgeable. `process_stream` fires the
    /// emitter on every line the supervised CLI prints, and `StreamEvent`'s
    /// `#[serde(flatten)] extra` absorbs arbitrary top-level keys and re-emits
    /// them at top level — so a single line of stdout from the model being
    /// supervised satisfies any predicate `audit` could key on, and writes "the
    /// gate accepted this patch" into the audit record with the gate never
    /// having run. It would not even be scoped to foreman sessions: `audit` runs
    /// for every coder session with a journal.
    ///
    /// car#1243 is what makes that fatal rather than untidy. The patches this
    /// gate rules on are authored on peers this host does not control, which is
    /// an argument for a record the audited party CANNOT write to. So the live
    /// `foreman: "gate"` event stays narration — spoofable, and only ever read
    /// as narration — and the durable record is written straight to the journal
    /// from the foreman loop, on the same shape as
    /// [`EventSink::record_turn_completed`] above and for the same reason.
    ///
    /// `kind` is the gate's own [`EventKind::GateAccepted`] / [`EventKind::GateRejected`],
    /// and `data` is its payload verbatim. Note the gate treats `Inconclusive`
    /// (verify timed out, or not configured) as not-accepted, so "we don't know"
    /// journals as `GateRejected` carrying `outcome: "inconclusive"` — the kind
    /// is binary, the payload is not.
    ///
    /// Journal-only, no WS/FFI surface: same as `record_turn_completed`.
    pub fn record_gate_verdict(
        &self,
        kind: EventKind,
        data: std::collections::HashMap<String, Value>,
    ) {
        let Some(journal) = &self.journal else { return };
        if let Ok(mut log) = journal.lock() {
            log.append(kind, Some(&self.session_id), None, data);
        }
    }

    /// Every model that completed a turn in this session, from the journal,
    /// distinct and in first-seen order.
    ///
    /// `record_turn_completed` already writes `model_id` on every terminal
    /// native path, so this reads what is there rather than adding a second
    /// record that could disagree with it.
    ///
    /// **A set, not the last one.** `TurnCompleted` is a PER-ITERATION
    /// terminal, and an unpinned session is free to route each iteration
    /// differently — `strict_model` is `cfg.model.is_some()`, so an unpinned
    /// chain may also degrade to another model on an outage. A reviewer judges
    /// the accumulated worktree diff, not the last iteration, so taking the
    /// last name would let the model that wrote most of the change review it
    /// as long as something else finished the run.
    ///
    /// **Mid-iteration models count too** (car#1333). A terminal names only the
    /// model that reached it, so a model that wrote turns 1-3 of an iteration
    /// another model finished used to leave no record at all — and could then
    /// sit on the panel reviewing what it had written. `models_served` carries
    /// the rest, and both are folded here.
    ///
    /// A record written before that field existed simply has no `models_served`,
    /// and folds to its `model_id` alone — the same answer it gave before.
    ///
    /// Empty for a run with no native turns — a foreman or external session
    /// farms to a coding CLI whose backbone CAR never resolved, so there is no
    /// honest answer. Empty means "nobody asked CAR's own loop to write this",
    /// NOT "nobody wrote it".
    pub fn authoring_models(&self) -> Vec<String> {
        let Some(journal) = &self.journal else {
            return Vec::new();
        };
        // Under the journal lock rather than `events().to_vec()`: a session
        // that made hundreds of tool calls has hundreds of records, and cloning
        // all of them to read one field each runs while the session mutex is
        // also held by `finalize_outcome`.
        let Ok(log) = journal.lock() else {
            return Vec::new();
        };
        let mut models: Vec<String> = Vec::new();
        for event in log.events() {
            if event.kind != EventKind::TurnCompleted {
                continue;
            }
            let served = event
                .data
                .get("models_served")
                .and_then(|v| v.as_array())
                .map(|a| a.iter().filter_map(|v| v.as_str()).collect::<Vec<_>>())
                .unwrap_or_default();
            let terminal = event.data.get("model_id").and_then(|v| v.as_str());
            for model in served.into_iter().chain(terminal) {
                let model = model.trim();
                if model.is_empty() {
                    continue;
                }
                if !models.iter().any(|m| m == model) {
                    models.push(model.to_string());
                }
            }
        }
        models
    }

    pub fn events(&self) -> Vec<car_eventlog::Event> {
        let Some(journal) = &self.journal else {
            return Vec::new();
        };
        journal
            .lock()
            .map(|log| log.events().to_vec())
            .unwrap_or_default()
    }

    pub fn record_proactive_memory(
        &self,
        maintenance: &car_memgine::ProactiveMaintenanceReport,
        decision: &car_memgine::ProactiveMemoryDecision,
    ) {
        let Some(journal) = &self.journal else {
            return;
        };
        if let Ok(mut log) = journal.lock() {
            log.append(
                EventKind::ProactiveMemoryMaintained,
                Some(&self.session_id),
                None,
                proactive_maintenance_event_data(maintenance),
            );
            log.append(
                EventKind::ProactiveMemoryIntervention,
                Some(&self.session_id),
                None,
                proactive_intervention_event_data(decision),
            );
        }
    }

    /// Journal the audit-relevant subset (transitions, tool calls, checks,
    /// errors). Narration-only events (plan text, iteration markers, diffs)
    /// live in the WS stream and the session snapshot instead.
    fn audit(&self, event: &CoderEvent) {
        let Some(journal) = &self.journal else { return };
        let (kind, mut data): (EventKind, HashMap<String, Value>) = match &event.kind {
            CoderEventKind::StateChanged { from, to } => (
                EventKind::StateChanged,
                HashMap::from([
                    ("from".to_string(), Value::String(from.clone())),
                    ("to".to_string(), Value::String(to.clone())),
                ]),
            ),
            CoderEventKind::ToolCall {
                tool,
                params_preview,
            } => (
                EventKind::ActionExecuting,
                HashMap::from([
                    ("tool".to_string(), Value::String(tool.clone())),
                    ("params".to_string(), Value::String(params_preview.clone())),
                ]),
            ),
            CoderEventKind::ToolResult { tool, ok, preview } => (
                if *ok {
                    EventKind::ActionSucceeded
                } else {
                    EventKind::ActionFailed
                },
                HashMap::from([
                    ("tool".to_string(), Value::String(tool.clone())),
                    ("result".to_string(), Value::String(preview.clone())),
                ]),
            ),
            CoderEventKind::CheckCompleted { result } => (
                if result.passed {
                    EventKind::ActionSucceeded
                } else {
                    EventKind::ActionFailed
                },
                HashMap::from([
                    ("check".to_string(), Value::String(result.name.clone())),
                    (
                        "exit_code".to_string(),
                        result.exit_code.map(Value::from).unwrap_or(Value::Null),
                    ),
                ]),
            ),
            // Journalled as an observation, never as a failure: an all-green
            // baseline is a fact about the contract, not a failed action, and
            // recording it as `ActionFailed` would poison `harness_adapt`'s
            // failure-mechanism diagnosis with a non-failure.
            CoderEventKind::ContractBaseline {
                results,
                gates_nothing,
            } => (
                EventKind::ActionSucceeded,
                HashMap::from([
                    ("baseline_checks".to_string(), Value::from(results.len())),
                    (
                        "baseline_passed".to_string(),
                        Value::from(results.iter().filter(|r| r.passed).count()),
                    ),
                    (
                        "contract_gates_nothing".to_string(),
                        Value::Bool(*gates_nothing),
                    ),
                ]),
            ),
            CoderEventKind::Error { message } => (
                EventKind::ActionFailed,
                HashMap::from([("error".to_string(), Value::String(message.clone()))]),
            ),
            CoderEventKind::MergeCompleted { branch } => (
                EventKind::ActionSucceeded,
                HashMap::from([("branch".to_string(), Value::String(branch.clone()))]),
            ),
            // `ActionSkipped`, which is literally what happened: the next
            // iteration was not admitted. Deliberately NOT `ActionFailed` — a
            // session that ran out of clock did not fail an action, and filing
            // it as one would poison `harness_adapt`'s failure-mechanism
            // diagnosis with a non-failure, the same trap `ContractBaseline`
            // avoids.
            CoderEventKind::BudgetExhausted {
                reason,
                elapsed_secs,
                iterations,
            } => (
                EventKind::ActionSkipped,
                HashMap::from([
                    ("reason".to_string(), Value::String(reason.clone())),
                    ("elapsed_secs".to_string(), Value::from(*elapsed_secs)),
                    ("iterations".to_string(), Value::from(*iterations)),
                ]),
            ),
            // Journaled so `harness_adapt::diagnose` can see a CLI that keeps
            // dying under us. Without this arm a chronically flaky engine is
            // indistinguishable from a fast clean one in the run record.
            CoderEventKind::InvocationRetried {
                hypothesis,
                reason,
                retries_remaining,
            } => (
                // `ActionRetrying`, not `ActionFailed`: `harness_adapt` tallies
                // the two separately, and a retried invocation is not a failed
                // action — filing it as one would inflate the failure tally
                // that drives intervention thresholds.
                EventKind::ActionRetrying,
                HashMap::from([
                    ("hypothesis".to_string(), Value::from(*hypothesis)),
                    ("reason".to_string(), Value::String(reason.clone())),
                    (
                        "retries_remaining".to_string(),
                        Value::from(*retries_remaining),
                    ),
                ]),
            ),
            _ => return,
        };
        data.insert(
            "coder_event".to_string(),
            Value::String(coder_event_name(&event.kind).to_string()),
        );
        data.insert("seq".to_string(), Value::from(event.seq));
        // The event's `action_id` identifies WHICH action, keyed by the tool or
        // check name so `harness_adapt::diagnose` can tally failures per-tool
        // (`run_command` failing 4× → a targeted intervention) instead of lumping
        // every failure under the session id. The session is already the journal
        // file's identity; other events fall back to it.
        let action_id: String = match &event.kind {
            CoderEventKind::ToolCall { tool, .. } | CoderEventKind::ToolResult { tool, .. } => {
                tool.clone()
            }
            CoderEventKind::CheckCompleted { result } => format!("check:{}", result.name),
            // Its own bucket, for the reason stated above: falling through to
            // the session id would pool transport retries with every other
            // coder error against one `min_occurrences` threshold, so neither
            // signal would mean what `diagnose` reads it as.
            CoderEventKind::InvocationRetried { .. } => "invocation_retry".to_string(),
            CoderEventKind::BudgetExhausted { .. } => "session_budget".to_string(),
            _ => self.session_id.clone(),
        };
        if let Ok(mut log) = journal.lock() {
            log.append(kind, Some(&action_id), None, data);
        }
    }
}

fn coder_event_name(kind: &CoderEventKind) -> &'static str {
    match kind {
        CoderEventKind::StateChanged { .. } => "coder.state_changed",
        CoderEventKind::ContractProposed { .. } => "coder.contract_proposed",
        CoderEventKind::EngineSelected { .. } => "coder.engine_selected",
        CoderEventKind::EngineFallback { .. } => "coder.engine_fallback",
        CoderEventKind::ModelFallback { .. } => "coder.model_fallback",
        CoderEventKind::IterationStarted { .. } => "coder.iteration_started",
        CoderEventKind::AuthRequired { .. } => "coder.auth_required",
        CoderEventKind::BudgetExhausted { .. } => "coder.budget_exhausted",
        CoderEventKind::InvocationRetried { .. } => "coder.invocation_retried",
        CoderEventKind::PlanText { .. } => "coder.plan_text",
        CoderEventKind::ToolCall { .. } => "coder.tool_call",
        CoderEventKind::ToolResult { .. } => "coder.tool_result",
        CoderEventKind::CheckStarted { .. } => "coder.check_started",
        CoderEventKind::CheckCompleted { .. } => "coder.check_completed",
        CoderEventKind::ContractBaseline { .. } => "coder.contract_baseline",
        CoderEventKind::ExternalEvent { .. } => "coder.external_event",
        CoderEventKind::FindingProposed { .. } => "coder.finding_proposed",
        CoderEventKind::FindingResolved { .. } => "coder.finding_resolved",
        CoderEventKind::DiffReady { .. } => "coder.diff_ready",
        CoderEventKind::UserInputRequested { .. } => "coder.user_input_requested",
        CoderEventKind::UserInputExpired { .. } => "coder.user_input_expired",
        CoderEventKind::ContractRevisionRejected { .. } => "coder.contract_revision_rejected",
        CoderEventKind::MergeCompleted { .. } => "coder.merge_completed",
        CoderEventKind::Error { .. } => "coder.error",
    }
}

/// The phase an Agent-project build is currently executing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AgentBuildPhase {
    GeneratingSpec,
    Repairing,
    RunningScenario,
}

/// Live progress for an Agent-project build.
///
/// Persisted with the session so `coder.get` does not lose the last known phase
/// across a daemon restart. While the session is live, the RPC refreshes
/// `elapsed_secs` from `started_at` on every poll.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AgentBuildProgress {
    pub phase: AgentBuildPhase,
    pub attempt: u32,
    pub max_attempts: u32,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub scenario: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub scenarios_total: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub model: Option<String>,
    pub started_at: u64,
    pub elapsed_secs: u64,
}

impl AgentBuildProgress {
    pub fn refresh_elapsed(&mut self) {
        self.elapsed_secs = now_secs().saturating_sub(self.started_at);
    }
}

/// A coding session. Serializes to the JSON snapshot persisted on every
/// transition; the live worktree handle is process-only (`#[serde(skip)]`).
#[derive(Debug, Serialize, Deserialize)]
pub struct CoderSession {
    pub id: String,
    /// The user's repository root (never written to directly).
    pub repo: PathBuf,
    pub intent: String,
    pub engine: EngineChoice,
    pub state: CoderState,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub contract: Option<OutcomeContract>,
    /// Where the throwaway worktree lives (kept in the snapshot so orphaned
    /// sessions after a daemon restart can still report it).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub workspace_path: Option<PathBuf>,
    /// When this session works on a CAR-managed project (vs. a raw repo path),
    /// the project slug + kind. Drives delivery (commit straight to the
    /// project's `main` instead of publishing a `car/coder/<id>` branch) and,
    /// for `Agent` projects, the scenario-based contract + agent registration
    /// on approve. `None` = raw-repo session (the original behavior).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub project: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub project_kind: Option<super::project::ProjectKind>,
    /// For an `Agent` project: the declarative agent spec the build loop
    /// produced, stashed so `approve_merge` can register it. Persisted so
    /// `coder.get` can show what was built.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub built_agent: Option<car_registry::declarative::DeclarativeAgentSpec>,
    /// Present only for Agent-project builds. Additive on `coder.get`; older
    /// hosts ignore it and older snapshots deserialize with `None`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub agent_build_progress: Option<AgentBuildProgress>,
    pub iterations: u32,
    pub max_iterations: u32,
    /// Metered inference spend, when anything reported it. `None` is
    /// **unknown**, not free — the native loop does not meter.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cost_usd: Option<f64>,
    /// Per-session external-engine hypothesis budget. `None` = engine default.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub repair_invokes: Option<u32>,
    /// Per-session external-engine availability budget. `None` = engine default.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub transient_retries: Option<u32>,
    /// When a session ends `Failed`, keep the throwaway worktree on disk (and
    /// its handle in-process) so the operator can inspect it for a postmortem
    /// instead of having it reaped on the terminal transition. Sourced from
    /// `~/.car/coder.toml` (`keep_workspace_on_failure`); default `false`.
    #[serde(default)]
    pub keep_workspace_on_failure: bool,
    /// Pin the native loop's inference model (e.g. `"parslee/reasoning"`).
    /// `None` = adaptive routing. Sourced from `~/.car/coder.toml` (`model`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub model: Option<String>,
    #[serde(default)]
    pub last_check_results: Vec<CheckResult>,
    /// The contract's red-green baseline — how each check fared against the
    /// **unmodified** worktree (car#707). Stored rather than recomputed because
    /// it is part of how the current draft READS: a board renders the contract
    /// with its baseline beside it, so a `coder.revise_contract` that could not
    /// be honored has to hand back both, or the contract it swore was unchanged
    /// visibly changes anyway when the baseline blanks out.
    #[serde(default)]
    pub baseline: Vec<CheckResult>,
    /// Whether every baseline check already passed — i.e. the contract gates
    /// nothing for this task. Travels with `baseline` for the same reason.
    #[serde(default)]
    pub baseline_gates_nothing: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub result_branch: Option<String>,
    // NOTE: there is deliberately no persisted `needs_you` here. It only ever
    // stored `"contract"` / `"approval"` — exactly what `needs_you_from` already
    // derives from `state` — and a non-live session is now reported as not
    // actionable regardless, so the field earned nothing and is gone.
    /// Why a `failed` session failed, as a machine-readable kind:
    /// `"budget_exhausted"` | `"auth_required"` | `"configuration"` |
    /// `"infrastructure"` | `"error"`. Persisted so a summary read from disk
    /// after a daemon restart still distinguishes "ran out of clock" from
    /// "nobody signed in" from "routing excludes every model" from "the
    /// machinery broke" from "the work was judged and rejected" — a live-only
    /// derivation would go blank exactly when the operator comes back to look.
    ///
    /// `"infrastructure"` is the one a scorer must act on: nothing was judged,
    /// so the session is not a task loss. See `rpc::failure_kind_for` for how it
    /// is chosen and why it is not folded into `"error"`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub failure_kind: Option<String>,
    /// The `coder.discuss` conversation this run was distilled from, when the
    /// operator went through a discussion. Provenance only — the run itself is
    /// independent of the discussion's lifetime.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub discussion_id: Option<String>,
    /// Whether this session explicitly opted into the assistant's browser tool
    /// surface. False by default and persisted so contract review cannot change
    /// which tools the later confirmed run receives.
    #[serde(default)]
    pub browser: bool,
    /// Farm subtasks across reachable CAR instances rather than this machine
    /// alone. Only the `foreman` engine reads it.
    ///
    /// Persisted, and reported on the `coder.list` row, so a finished run can
    /// say which way it ran. That is the whole reason — a session does NOT
    /// resume across a daemon restart (`adopt_orphaned_sessions` rewrites every
    /// non-terminal orphan to `Failed`), so this is not protecting a resumed
    /// run from silently becoming local.
    #[serde(default)]
    pub distributed: bool,
    /// Instances a distributed run is restricted to. Empty = every instance
    /// that can serve the repository.
    ///
    /// The operator's FILTER, not the resolved pool — see [`Self::pool_workers`]
    /// for what the run actually got.
    #[serde(default)]
    pub workers: Vec<String>,
    /// The peers the pool actually resolved to, recorded when it is built.
    ///
    /// `workers` above is what the operator asked to restrict to; this is what
    /// answering that restriction against reachability, enrollment and repo
    /// eligibility produced. It was emitted as a `foreman: "pool"` event and
    /// then dropped, so a subscriber that was not attached saw nothing and the
    /// snapshot carried no record at all.
    ///
    /// Written BEFORE any subtask runs, which is what makes it the durable
    /// answer to "which machines was this farmed to?". [`Self::placements`]
    /// cannot answer it on its own: a placement is recorded when a worker
    /// RETURNS, and foreman runs a level under `join_all` rather than spawning,
    /// so `coder.cancel`'s abort drops every in-flight future before it records
    /// — the subtasks running at the moment an operator gives up are exactly
    /// the ones the ledger omits (car#1346). Empty on a local run.
    #[serde(default)]
    pub pool_workers: Vec<String>,
    /// Where each of a distributed run's subtasks actually ran, including the
    /// workers that failed it first.
    ///
    /// Empty for a local run, and for a distributed one that fell off the
    /// foreman rung before farming anything out.
    ///
    /// **Not a proxy for "a distributed run completed."** A cancelled run
    /// carries this populated with [`Self::integrated_subtasks`] empty, and so
    /// does a run whose foreman union the gate rejected — three different
    /// states, separated by `state` and by nothing else. And it is a floor, not
    /// a census: subtasks still in flight at a cancel never reach it, which is
    /// what [`Self::pool_workers`] is for.
    ///
    /// `foreman.run` has reported this all along; the pipeline that DELIVERS had
    /// strictly less provenance than the one that only reports, because
    /// `fleet_pool_for` erased `FleetPool` to `Arc<dyn WorktreeAgent>` and
    /// `placements()` is on the concrete type (car#1322). CAR's position is that
    /// receipts decide completion, and a delivered commit whose hunks were
    /// authored on unnamed machines is the wrong artifact for that claim — the
    /// question arrives the first time a distributed run produces something
    /// surprising.
    #[serde(default)]
    pub placements: Vec<car_multi::Placement>,
    /// Which of those subtasks actually LANDED in the worktree, and what each
    /// wrote. The diagnostic record above says what happened; this says what is
    /// in the tree, and only this may back a claim in the delivered commit.
    #[serde(default)]
    pub integrated_subtasks: Vec<IntegratedSubtask>,
    /// Foreman's union was integrated and then the native loop repaired on top
    /// of it, so some delivered hunks were written locally by no listed worker.
    ///
    /// Without this the commit body would credit the fleet for a diff it only
    /// partly wrote — the same false attribution as crediting it for one it did
    /// not write at all, in a milder form.
    #[serde(default)]
    pub repaired_locally: bool,
    /// Every model that AUTHORED part of this session's work, as opposed to
    /// [`Self::model`], which is the pin the caller asked for.
    ///
    /// They differ exactly when it matters: unpinned, the router chooses, and
    /// on a machine with one reachable credential that choice can also be a
    /// review-panel seat — the self-review the gate refuses when a coder is
    /// pinned, permitted by default because nothing knew who wrote the change
    /// (car#1299).
    ///
    /// A projection of the journal ([`EventSink::authoring_models`]), not a
    /// second record of the same fact — two records can disagree and then the
    /// question is which one the gate believes. Empty for a session with no
    /// native turns (foreman/external), which is the DEFAULT engine: read it as
    /// "CAR's own loop did not write this", not "nobody did".
    #[serde(default)]
    pub authored_by: Vec<String>,
    pub created_at: u64,
    pub updated_at: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    /// RAII worktree handle. Dropping it removes the worktree, so terminal
    /// transitions release it explicitly.
    #[serde(skip)]
    pub workspace: Option<AgentWorkspace>,
    /// Where snapshots/journals/worktrees go; `None` disables persistence.
    #[serde(skip)]
    pub state_dir: Option<PathBuf>,
}

impl CoderSession {
    pub fn new(
        repo: impl Into<PathBuf>,
        intent: impl Into<String>,
        engine: EngineChoice,
        max_iterations: u32,
        state_dir: Option<PathBuf>,
    ) -> Self {
        let now = now_secs();
        Self {
            id: format!("coder-{}", uuid::Uuid::new_v4().simple()),
            repo: repo.into(),
            intent: intent.into(),
            engine,
            state: CoderState::Created,
            contract: None,
            cost_usd: None,
            repair_invokes: None,
            transient_retries: None,
            workspace_path: None,
            project: None,
            project_kind: None,
            built_agent: None,
            agent_build_progress: None,
            iterations: 0,
            max_iterations: max_iterations.max(1),
            keep_workspace_on_failure: false,
            model: None,
            last_check_results: Vec::new(),
            baseline: Vec::new(),
            baseline_gates_nothing: false,
            browser: false,
            distributed: false,
            workers: Vec::new(),
            pool_workers: Vec::new(),
            authored_by: Vec::new(),
            placements: Vec::new(),
            integrated_subtasks: Vec::new(),
            repaired_locally: false,
            result_branch: None,
            failure_kind: None,
            discussion_id: None,
            created_at: now,
            updated_at: now,
            error: None,
            workspace: None,
            state_dir,
        }
    }

    /// Mark this session as working on a managed project (builder so existing
    /// call sites and tests stay green).
    pub fn with_project(
        mut self,
        slug: impl Into<String>,
        kind: super::project::ProjectKind,
    ) -> Self {
        self.project = Some(slug.into());
        self.project_kind = Some(kind);
        self
    }

    /// Short suffix for branch names and worktree dirs.
    pub fn short_id(&self) -> &str {
        // "coder-<32 hex>" → last 8 chars are plenty unique per repo.
        &self.id[self.id.len().saturating_sub(8)..]
    }

    /// Provision the throwaway git worktree under the state dir (NOT inside
    /// the user's repo, so their `git status` stays clean).
    pub fn provision_workspace(&mut self) -> Result<PathBuf, String> {
        let state_dir = self
            .state_dir
            .clone()
            .ok_or("session has no state dir; cannot provision a worktree")?;
        let config = WorkspaceConfig::git_worktree_at(&self.repo, state_dir.join("worktrees"));
        let workspace = AgentWorkspace::provision(&config, &self.id)?;
        let path = workspace.path().to_path_buf();
        self.workspace_path = Some(path.clone());
        self.workspace = Some(workspace);
        Ok(path)
    }

    /// Validated state transition: updates timestamps, emits `StateChanged`,
    /// persists the snapshot, and releases the worktree on terminal states.
    pub fn transition(&mut self, to: CoderState, sink: &EventSink) -> Result<(), String> {
        if !can_transition(self.state, to) {
            return Err(format!(
                "illegal coder transition {} → {}",
                self.state.as_str(),
                to.as_str()
            ));
        }
        let from = self.state;
        self.state = to;
        self.updated_at = now_secs();
        sink.emit(CoderEventKind::StateChanged {
            from: from.as_str().to_string(),
            to: to.as_str().to_string(),
        });
        if to.is_terminal() {
            // Drop the RAII handle → worktree removed. The one exception:
            // when `keep_workspace_on_failure` is set (operator config) and the
            // terminal state is `Failed`, we `leak()` the handle so the worktree
            // survives on disk for a postmortem. `workspace_path` is always kept
            // in the snapshot regardless, so a dropped tree still reports where
            // it *was*; with the flag set the tree is actually still there.
            if to == CoderState::Failed && self.keep_workspace_on_failure {
                // Suppress the RAII `Drop` so the git worktree survives on disk
                // for a postmortem. The cost is a leaked `git worktree`
                // registration in the user's repo; it's reaped on next
                // provision (AgentWorkspace::provision self-heals stale entries)
                // or by `git worktree prune`. `workspace_path` stays in the
                // snapshot so the operator knows exactly where to look.
                if let Some(ws) = self.workspace.take() {
                    std::mem::forget(ws);
                }
            } else {
                self.workspace = None;
            }
        }
        if let Err(e) = self.persist() {
            tracing::warn!(session = %self.id, "coder snapshot persist failed: {e}");
        }
        Ok(())
    }

    /// Write the JSON snapshot to `<state_dir>/<id>.json` (no-op without a
    /// state dir, e.g. in unit tests).
    pub fn persist(&self) -> Result<(), String> {
        let Some(dir) = &self.state_dir else {
            return Ok(());
        };
        std::fs::create_dir_all(dir).map_err(|e| format!("create {}: {e}", dir.display()))?;
        let path = dir.join(format!("{}.json", self.id));
        let json = serde_json::to_string_pretty(self).map_err(|e| e.to_string())?;
        std::fs::write(&path, json).map_err(|e| format!("write {}: {e}", path.display()))
    }

    /// Load a snapshot from disk. The worktree handle is NOT restored — a
    /// loaded session is read-only history unless re-provisioned.
    pub fn load(path: &Path) -> Result<Self, String> {
        let text =
            std::fs::read_to_string(path).map_err(|e| format!("read {}: {e}", path.display()))?;
        serde_json::from_str(&text).map_err(|e| format!("parse {}: {e}", path.display()))
    }

    /// All persisted sessions under `state_dir`, newest first.
    pub fn list(state_dir: &Path) -> Vec<CoderSession> {
        let Ok(entries) = std::fs::read_dir(state_dir) else {
            return Vec::new();
        };
        let mut sessions: Vec<CoderSession> = entries
            .flatten()
            .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
            .filter_map(|e| Self::load(&e.path()).ok())
            .collect();
        sessions.sort_by(|a, b| b.updated_at.cmp(&a.updated_at));
        sessions
    }
}

/// Best-effort unlink. `true` when the file is gone because this call removed
/// it — a missing file is not a collection, so it does not count as one.
fn unlink(path: &Path) -> bool {
    match std::fs::remove_file(path) {
        Ok(()) => true,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
        Err(e) => {
            tracing::warn!(path = %path.display(), "coder retention unlink failed: {e}");
            false
        }
    }
}

/// Remove a stranded git worktree, mirroring `AgentWorkspace`'s own `Drop`.
///
/// `git worktree remove --force` first so the registration in the user's
/// repository goes with the directory — removing the directory alone leaves a
/// dangling entry that only a later `provision` or a manual
/// `git worktree prune` clears. Best-effort throughout: this runs at boot, and
/// a repository that has since moved or been deleted must not stop the daemon
/// from starting.
fn reap_worktree(repo: &Path, path: &Path) {
    if !path.is_dir() {
        return;
    }
    let _ = std::process::Command::new("git")
        .arg("-C")
        .arg(repo)
        .args(["worktree", "remove", "--force"])
        .arg(path)
        .output();
    if let Err(e) = std::fs::remove_dir_all(path) {
        if e.kind() != std::io::ErrorKind::NotFound {
            tracing::warn!(path = %path.display(), "could not reap a stranded coder worktree: {e}");
        }
    }
}

/// One subtask whose patch reached the session worktree, and the files it wrote.
///
/// Distinct from a [`car_multi::Placement`], which records that a worker RAN a
/// subtask — recorded when the worker returns, before the per-patch gate rules
/// on what it produced. Only this says what is in the delivered tree, which is
/// the claim a commit body makes (car#1322).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct IntegratedSubtask {
    pub subtask_id: String,
    /// Repo-relative paths the subtask's patch touched, from the gate's own
    /// parser. A subtask id is opaque model output; these are what make the row
    /// reviewable.
    #[serde(default)]
    pub files: Vec<String>,
}

/// Retention policy for the coder state dir, from `~/.car/coder.toml`.
///
/// Mirrors [`RunStore`](crate::run_store::RunStore)'s `[runs]` caps, which
/// solve this exact shape for run traces: a count cap, an age cap, and an
/// exemption for work that is not finished.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SessionRetention {
    /// Keep at most this many collectable snapshots. `0` = unlimited.
    pub max_sessions: usize,
    /// Drop a collectable snapshot older than this. `0` = unlimited.
    pub max_age_days: u64,
}

/// Delete session snapshots (and their journals) beyond the retention caps.
///
/// Nothing pruned this directory before car#1310, so `<id>.json` and the
/// larger `<id>.events.jsonl` beside it accumulated for the life of the
/// installation — and `coder.list` pays a read, a `serde_json` parse and a
/// `stat` for every one of them on every call. car#1262 bounded the in-memory
/// registry and explicitly left this alone; this is the disk arm.
///
/// **Two exemptions, both about not destroying the only record of something
/// that still exists:**
///
/// - A session that is not [terminal](CoderState::is_terminal) is never
///   collected. Same rule as `RunStore`'s in-progress exemption. Note this
///   covers `NeedsApproval`, which is deliberately non-terminal — a snapshot
///   waiting on a human is not garbage however old it is.
/// - A session whose `workspace_path` is still a directory is never collected.
///   The snapshot is the only thing that names that worktree; deleting it
///   turns a directory an operator kept (`keep_workspace_on_failure`) or a
///   preserved `needs_approval` orphan into an unattributable leak.
///
/// **The count cap ranks only COLLECTABLE sessions**, matching `RunStore`'s
/// `completed_rank`: an exempt session neither dies nor consumes a keeper slot.
/// So `max_sessions` is a bound on what retention manages, NOT on the size of
/// the directory — an install that sets `keep_workspace_on_failure` holds every
/// failed session's snapshot, journal AND worktree on top of the cap, by the
/// operator's own request. [`adopt_orphaned_sessions`] reaps a CRASH orphan's
/// worktree precisely so that population does not join the exempt set.
///
/// The age cap reads `updated_at`, which is stamped on every transition, so it
/// measures time since the session last did anything rather than since it
/// started.
///
/// The journal is unlinked BEFORE the snapshot, and the collection is counted
/// on the snapshot. Only `*.json` is enumerated, so a journal whose snapshot is
/// already gone is invisible to the candidate pass — removing the snapshot
/// first would strand the larger file permanently on any unlink error. (The
/// pass at the end sweeps journals already stranded that way, including by a
/// `coder.start` that created the sink and died before its first `persist`.)
///
/// Best-effort: an unreadable snapshot is skipped, and a failed unlink is
/// logged rather than propagated — this runs at boot and must not block it.
/// Returns the number of sessions collected.
/// Which process state a [`gc_sessions`] sweep is running in.
///
/// Not an `Option<&HashSet>`: that carries two orthogonal bits in one type and
/// only one of them is about the set. `None` would have to mean BOTH "filter
/// nothing" AND "sweep orphan journals" — so a caller with no live set to hand
/// over, the natural reading of `None`, would silently re-enable a deletion
/// pass that is safe only where no live sink can own a journal.
pub enum SweepScope<'a> {
    /// Daemon construction, where the session registry is provably empty. Also
    /// sweeps orphan journals, which is safe only here.
    Boot,
    /// Mid-lifetime, carrying the ids the in-process registry still holds.
    Live(&'a std::collections::HashSet<String>),
}

pub fn gc_sessions(state_dir: &Path, retention: &SessionRetention, scope: SweepScope<'_>) -> usize {
    let Ok(entries) = std::fs::read_dir(state_dir) else {
        return 0;
    };
    // The DirEntry's own path, not one rebuilt from the parsed id: those can
    // disagree (a `foo.bak.json` copy names the id inside it), and a deleter
    // must remove what it classified.
    let mut candidates: Vec<(u64, PathBuf)> = Vec::new();
    let mut snapshots: Vec<PathBuf> = Vec::new();
    let mut journals: Vec<PathBuf> = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        if path.to_string_lossy().ends_with(".events.jsonl") {
            journals.push(path);
            continue;
        }
        if path.extension().is_none_or(|x| x != "json") {
            continue;
        }
        snapshots.push(path.clone());
        let Ok(session) = CoderSession::load(&path) else {
            continue;
        };
        if !session.state.is_terminal() {
            continue;
        }
        // `is_dir`, the same predicate `adopt_orphaned_sessions` asks this
        // question with. A stray FILE at a worktree path is not a worktree, and
        // exempting a session forever over one would be the leak this guards
        // against, arrived at backwards.
        if session.workspace_path.as_ref().is_some_and(|p| p.is_dir()) {
            continue;
        }
        // Never collect a snapshot the in-memory registry still holds an entry
        // for. `prune_finished_sessions` treats "snapshot missing on disk" as
        // "keep the entry rather than lose the session", so deleting one out
        // from under a live entry converts it into a permanent memory pin —
        // re-opening the leak car#1262 closed, through the door that was added
        // to bound the disk (car#1339). Terminal-and-registered is the ordinary
        // window between a loop finishing and the next `coder.start` pruning it.
        if matches!(&scope, SweepScope::Live(ids) if ids.contains(&session.id)) {
            continue;
        }
        candidates.push((session.updated_at, path));
    }

    // Newest first, so the count cap keeps the head and drops the tail.
    candidates.sort_by(|a, b| b.0.cmp(&a.0));

    let age_cutoff = (retention.max_age_days > 0)
        .then(|| now_secs().saturating_sub(retention.max_age_days.saturating_mul(24 * 60 * 60)));

    let mut collected = 0;
    for (rank, (updated_at, snapshot)) in candidates.iter().enumerate() {
        let over_count = retention.max_sessions > 0 && rank >= retention.max_sessions;
        let too_old = age_cutoff.is_some_and(|cut| *updated_at < cut);
        if !over_count && !too_old {
            continue;
        }
        // Journal first. It is the larger artifact and it is only reachable
        // through its snapshot, so removing the snapshot first and then failing
        // here would strand it for good.
        let journal = snapshot.with_extension("events.jsonl");
        unlink(&journal);
        if unlink(snapshot) {
            snapshots.retain(|p| p != snapshot);
            collected += 1;
        }
    }

    // Journals with no snapshot beside them. A `coder.start` creates the
    // `EventSink`'s journal before anything calls `persist()`, so a start that
    // dies in between leaves one that no candidate pass can ever reach. Safe
    // ONLY at boot, where the session registry is empty and no live sink can
    // own one — the same invariant `adopt_orphaned_sessions` relies on.
    //
    // Mid-lifetime the pass is unsafe for a reason the live set cannot fix, and
    // it is NOT the ordering inside one `coder.start`: that registers its entry
    // before it emits anything, so its own journal does not exist yet. The race
    // is with a CONCURRENT start, which registers after this sweep's caller
    // snapshotted the live set, then emits and so opens its journal. Its
    // journal exists, its snapshot does not, and its id is in no set we hold.
    // A snapshot caught that way is saved by carrying a non-terminal state; a
    // journal carries no state at all, so nothing can exempt it. Skip the pass
    // rather than filter it — which is why the scope is an enum, not the set.
    if matches!(scope, SweepScope::Live(_)) {
        return collected;
    }
    for journal in &journals {
        let snapshot = journal
            .to_string_lossy()
            .strip_suffix(".events.jsonl")
            .map(|stem| PathBuf::from(format!("{stem}.json")));
        if snapshot.is_some_and(|p| snapshots.contains(&p)) {
            continue;
        }
        unlink(journal);
    }
    collected
}

/// What [`adopt_orphaned_sessions`] decided about one on-disk snapshot.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AdoptionOutcome {
    /// A non-terminal orphan was rewritten to `Failed` ("daemon restarted
    /// mid-session"), so `coder.list`/`coder.get` stop reporting it as live.
    Failed,
    /// A `needs_approval` orphan whose worktree still exists — left untouched
    /// so the user can inspect the diff and approve-by-hand. NOT auto-published.
    Preserved,
}

/// Adopt crash/restart-orphaned coder sessions at daemon boot.
///
/// A daemon restart drops the in-memory `CoderSessionEntry` registry; only the
/// JSON snapshot under `state_dir` survives (the worktree under
/// `state_dir/worktrees` survives too). Any snapshot left in a **non-terminal**
/// state (`created`/`contract_proposed`/`contract_confirmed`/`running`/
/// `needs_approval`) therefore has no live loop driving it and would otherwise
/// report its stale state — "running" forever — to `coder.list`/`coder.get`.
///
/// This runs once at [`ServerState`](crate::session::ServerState) construction,
/// where the in-memory registry is always empty, so every non-terminal on-disk
/// snapshot is necessarily a prior process's orphan (no live writer can race).
///
/// Policy:
/// - A `needs_approval` orphan whose worktree directory **still exists** is
///   PRESERVED untouched: the diff is real and the snapshot stays inspectable
///   on disk, with the worktree path recorded so the user can review and merge
///   it by hand (`git -C <worktree> diff` / `git branch`). It is NOT approvable
///   through `coder.approve_merge` after a restart — that handler requires a
///   live `CoderSessionEntry`, which adoption deliberately does not rehydrate
///   (re-establishing a live entry without the running loop would bypass the
///   invariant the merge gate relies on). We never auto-publish.
/// - Every other non-terminal orphan — including `needs_approval` whose
///   worktree is gone — is rewritten to `Failed` with
///   `error = "daemon restarted mid-session"` and re-persisted.
///
/// Full live re-attach (resuming the loop where it left off) is explicitly OUT
/// OF SCOPE: the generator, sink, cancel flag, and RAII worktree handle are all
/// process-local and cannot be reconstructed from the snapshot. This only stops
/// the snapshots from lying about their state.
///
/// Best-effort: an unreadable or unwritable snapshot is skipped rather than
/// failing startup. Returns one [`AdoptionOutcome`] per snapshot it acted on.
pub fn adopt_orphaned_sessions(state_dir: &Path) -> Vec<(String, AdoptionOutcome)> {
    let Ok(entries) = std::fs::read_dir(state_dir) else {
        return Vec::new();
    };
    let mut outcomes = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        if path.extension().is_none_or(|x| x != "json") {
            continue;
        }
        let Ok(mut session) = CoderSession::load(&path) else {
            continue;
        };
        if session.state.is_terminal() {
            continue;
        }
        // A live worktree keeps a needs_approval orphan inspectable/approvable.
        let worktree_alive = session.state == CoderState::NeedsApproval
            && session.workspace_path.as_ref().is_some_and(|p| p.is_dir());
        if worktree_alive {
            outcomes.push((session.id.clone(), AdoptionOutcome::Preserved));
            continue;
        }
        // Reap the worktree the same way a live terminal transition would.
        //
        // A snapshot loaded from disk has no `AgentWorkspace` (it is
        // `#[serde(skip)]`), so nothing drops and nothing removes the
        // directory — and this assigns `state` directly rather than going
        // through `transition`, so even the handle logic there is bypassed.
        // Before car#1310 that was invisible: the leaked worktree simply sat
        // there. Now that retention exempts a session whose worktree still
        // exists — because the snapshot is the only thing naming it — leaving
        // it would make every crash orphan PERMANENTLY uncollectable, and a
        // git checkout is orders of magnitude larger than the snapshot and
        // journal retention was written to reclaim. Converting a bounded leak
        // into an unbounded one is not an improvement.
        //
        // `keep_workspace_on_failure` is honored exactly as `transition` honors
        // it: the operator asked to keep it, so it stays (and stays exempt,
        // which is the point of asking).
        if !session.keep_workspace_on_failure {
            if let Some(path) = session.workspace_path.clone() {
                reap_worktree(&session.repo, &path);
            }
        }
        session.state = CoderState::Failed;
        session.error = Some("daemon restarted mid-session".to_string());
        // The machinery died, nothing was judged — without this the summary
        // defaults to `"error"` and a board renders a restart as work that came
        // back red.
        session.failure_kind = Some("infrastructure".to_string());
        session.updated_at = now_secs();
        // load() drops the (serde-skipped) state_dir; restore it so persist()
        // writes back to the same snapshot instead of no-op'ing.
        session.state_dir = Some(state_dir.to_path_buf());
        if session.persist().is_ok() {
            outcomes.push((session.id.clone(), AdoptionOutcome::Failed));
        }
    }
    outcomes
}

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

    /// A run that degrades TWICE has two facts, and the journal must hold
    /// both.
    ///
    /// This is the half car#1333 did not cover. `models_served` answers WHO
    /// wrote each turn; this answers WHY the backbone moved, and the two are
    /// different facts rather than one recorded twice. Without it, mining and
    /// humans alike read a degraded run as the code under test behaving badly.
    ///
    /// Every transition, deliberately. The live `ModelFallback` event is behind
    /// a once-per-phase latch so the stream does not narrate every routing
    /// decision — right for a stream, wrong for a record. Journaling off the
    /// latched emit would keep only the first.
    #[test]
    fn every_backbone_change_is_journaled_not_just_the_first() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("coder-mf.events.jsonl");
        let sink = EventSink::new("coder-mf", None, Some(journal.clone()));

        sink.record_model_fallback("parslee/reasoning", "openai/gpt-5.6", "rate_limited");
        sink.record_model_fallback("openai/gpt-5.6", "anthropic/claude-opus-5", "timed_out");
        // JournalWriter hands lines to a background thread; Drop flushes and
        // joins, and durability is the whole claim here.
        drop(sink);

        let body = std::fs::read_to_string(&journal).expect("the journal exists");
        let lines: Vec<&str> = body.lines().filter(|l| !l.trim().is_empty()).collect();
        assert_eq!(
            lines.len(),
            2,
            "both transitions, not just the first: {body}"
        );

        // The reason is the point — a degrade with no cause recorded is the
        // gap this closes, and the two causes need different responses.
        assert!(body.contains("rate_limited"), "{body}");
        assert!(body.contains("timed_out"), "{body}");
        // And the chain is readable end to end: the second hop starts where
        // the first landed.
        assert!(body.contains("parslee/reasoning"), "{body}");
        assert!(body.contains("anthropic/claude-opus-5"), "{body}");
    }

    #[test]
    fn coder_journal_is_diagnosable_per_tool() {
        // Proves the coder-A/B loop's premise: the coder's action journal is a
        // real car_eventlog log whose failures `harness_adapt::diagnose` can
        // attribute PER TOOL (not lumped under the session id).
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("coder-x.events.jsonl");
        let sink = EventSink::new("coder-x", None, Some(journal.clone()));
        // The same tool fails twice → a diagnosable pattern at min=2.
        for _ in 0..2 {
            sink.emit(CoderEventKind::ToolResult {
                tool: "run_command".into(),
                ok: false,
                preview: "exit 1 at runtime".into(),
            });
        }
        // A successful tool must not create a failure pattern.
        sink.emit(CoderEventKind::ToolResult {
            tool: "edit_file".into(),
            ok: true,
            preview: "ok".into(),
        });
        // The journal writer runs on its own thread and flushes on drain/drop, so
        // drop the sink before reading — otherwise the read races the async write.
        drop(sink);
        let jsonl = std::fs::read_to_string(&journal).unwrap();
        let report = car_eventlog::harness_adapt::diagnose_from_jsonl(&jsonl, 2);
        assert!(
            report
                .interventions
                .iter()
                .any(|i| i.target == "run_command"),
            "diagnose must tally run_command failures per-tool: {:?}",
            report.interventions
        );
        assert!(
            !report.interventions.iter().any(|i| i.target == "edit_file"),
            "a succeeding tool must not be flagged"
        );
    }

    fn session() -> (CoderSession, EventSink) {
        (
            CoderSession::new("/tmp/repo", "do it", EngineChoice::Native, 8, None),
            EventSink::test_sink(),
        )
    }

    fn init_repo(dir: &Path) {
        for args in [
            vec!["init", "-q", "-b", "main"],
            vec![
                "-c",
                "user.name=t",
                "-c",
                "user.email=t@t",
                "commit",
                "-q",
                "--allow-empty",
                "-m",
                "init",
            ],
        ] {
            let out = std::process::Command::new("git")
                .arg("-C")
                .arg(dir)
                .args(&args)
                .output()
                .unwrap();
            assert!(
                out.status.success(),
                "{}",
                String::from_utf8_lossy(&out.stderr)
            );
        }
    }

    /// `keep_workspace_on_failure = true`: a Failed terminal transition leaves
    /// the git worktree on disk (RAII drop suppressed) for a postmortem; the
    /// snapshot still records its path.
    #[test]
    fn failed_with_keep_flag_retains_worktree() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let state_dir = tempfile::tempdir().unwrap();

        let mut s = CoderSession::new(
            repo.path(),
            "x",
            EngineChoice::Native,
            2,
            Some(state_dir.path().to_path_buf()),
        );
        s.keep_workspace_on_failure = true;
        let worktree = s.provision_workspace().unwrap();
        assert!(worktree.is_dir());

        let sink = EventSink::test_sink();
        s.transition(CoderState::Failed, &sink).unwrap();
        // Handle taken out of the session, but Drop suppressed → tree survives.
        assert!(s.workspace.is_none());
        assert!(
            worktree.is_dir(),
            "worktree should be retained for postmortem"
        );
        assert_eq!(s.workspace_path.as_deref(), Some(worktree.as_path()));

        // Clean up the leaked worktree registration so the temp repo can drop.
        let _ = std::process::Command::new("git")
            .arg("-C")
            .arg(repo.path())
            .args(["worktree", "remove", "--force"])
            .arg(&worktree)
            .output();
    }

    /// Default (`keep_workspace_on_failure = false`): a Failed transition reaps
    /// the worktree, same as every other terminal state.
    #[test]
    fn failed_without_keep_flag_reaps_worktree() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let state_dir = tempfile::tempdir().unwrap();

        let mut s = CoderSession::new(
            repo.path(),
            "x",
            EngineChoice::Native,
            2,
            Some(state_dir.path().to_path_buf()),
        );
        // keep_workspace_on_failure defaults to false.
        let worktree = s.provision_workspace().unwrap();
        assert!(worktree.is_dir());

        let sink = EventSink::test_sink();
        s.transition(CoderState::Failed, &sink).unwrap();
        assert!(s.workspace.is_none());
        assert!(
            !worktree.exists(),
            "worktree should be reaped on failure by default"
        );
    }

    #[test]
    fn happy_path_transitions_are_legal() {
        let (mut s, sink) = session();
        for to in [
            CoderState::ContractProposed,
            CoderState::ContractProposed, // re-propose
            CoderState::ContractConfirmed,
            CoderState::Running,
            CoderState::NeedsApproval,
            CoderState::Merged,
        ] {
            s.transition(to, &sink).unwrap();
        }
        assert!(s.state.is_terminal());
    }

    #[test]
    fn illegal_jumps_are_rejected() {
        let (mut s, sink) = session();
        assert!(s.transition(CoderState::Running, &sink).is_err());
        assert!(s.transition(CoderState::Merged, &sink).is_err());
        assert!(s.transition(CoderState::NeedsApproval, &sink).is_err());
        // State unchanged after rejections.
        assert_eq!(s.state, CoderState::Created);
    }

    #[test]
    fn any_non_terminal_state_can_fail_or_abandon() {
        for terminal in [CoderState::Failed, CoderState::Abandoned] {
            let (mut s, sink) = session();
            s.transition(CoderState::ContractProposed, &sink).unwrap();
            s.transition(terminal, &sink).unwrap();
            // Terminal is sticky.
            assert!(s.transition(CoderState::Running, &sink).is_err());
            assert!(s.transition(CoderState::Failed, &sink).is_err());
        }
    }

    #[test]
    fn event_seq_is_monotonic_and_session_tagged() {
        let (sink, collected) = EventSink::collecting("coder-seq");
        for _ in 0..5 {
            sink.emit(CoderEventKind::PlanText { text: "x".into() });
        }
        let events = collected.lock().unwrap();
        assert_eq!(events.len(), 5);
        for (i, e) in events.iter().enumerate() {
            assert_eq!(e.seq, i as u64);
            assert_eq!(e.session_id, "coder-seq");
        }
    }

    /// Persist one snapshot (and a journal beside it) in a given terminal
    /// state, stamped `age_days` old.
    fn retained(
        state_dir: &Path,
        state: CoderState,
        age_days: u64,
        workspace: Option<PathBuf>,
    ) -> String {
        let mut s = CoderSession::new(
            "/tmp/repo",
            "intent",
            EngineChoice::Auto,
            4,
            Some(state_dir.to_path_buf()),
        );
        s.state = state;
        s.workspace_path = workspace;
        s.updated_at = now_secs().saturating_sub(age_days * 24 * 60 * 60);
        s.persist().unwrap();
        std::fs::write(state_dir.join(format!("{}.events.jsonl", s.id)), "{}\n").unwrap();
        s.id
    }

    fn on_disk(state_dir: &Path, id: &str) -> bool {
        state_dir.join(format!("{id}.json")).exists()
    }

    const KEEP_ALL: SessionRetention = SessionRetention {
        max_sessions: 0,
        max_age_days: 0,
    };

    #[test]
    fn gc_evicts_beyond_the_count_cap_oldest_first() {
        let dir = tempfile::tempdir().unwrap();
        // Ages 0..5 days; the cap keeps the three most recently updated.
        let ids: Vec<String> = (0..5)
            .map(|age| retained(dir.path(), CoderState::Merged, age, None))
            .collect();

        let collected = gc_sessions(
            dir.path(),
            &SessionRetention {
                max_sessions: 3,
                ..KEEP_ALL
            },
            SweepScope::Boot,
        );

        assert_eq!(collected, 2);
        for id in &ids[..3] {
            assert!(on_disk(dir.path(), id), "newest three must survive");
        }
        for id in &ids[3..] {
            assert!(!on_disk(dir.path(), id), "oldest two must be collected");
        }
    }

    #[test]
    fn gc_evicts_snapshots_past_the_age_cap() {
        let dir = tempfile::tempdir().unwrap();
        let fresh = retained(dir.path(), CoderState::Reported, 3, None);
        let stale = retained(dir.path(), CoderState::Merged, 40, None);

        let collected = gc_sessions(
            dir.path(),
            &SessionRetention {
                max_age_days: 30,
                ..KEEP_ALL
            },
            SweepScope::Boot,
        );

        assert_eq!(collected, 1);
        assert!(on_disk(dir.path(), &fresh));
        assert!(!on_disk(dir.path(), &stale));
    }

    /// The two exemptions, and the reason each exists: collecting either would
    /// destroy the only record of something that still exists.
    #[test]
    fn gc_never_collects_an_unfinished_session() {
        let dir = tempfile::tempdir().unwrap();
        // NeedsApproval is deliberately NOT terminal — a session waiting on a
        // human is not garbage however old it is.
        for state in [
            CoderState::Created,
            CoderState::ContractProposed,
            CoderState::ContractConfirmed,
            CoderState::Running,
            CoderState::NeedsApproval,
        ] {
            // TWO of them, and `max_sessions: 1`. With one the count arm is
            // inert (`rank 0 >= 1` is false) and only the age arm is ever
            // tested — the exemption could be missing from the count path and
            // this would stay green.
            let a = retained(dir.path(), state, 9999, None);
            let b = retained(dir.path(), state, 9998, None);
            let collected = gc_sessions(
                dir.path(),
                &SessionRetention {
                    max_sessions: 1,
                    max_age_days: 1,
                },
                SweepScope::Boot,
            );
            assert_eq!(collected, 0, "{state:?} must be exempt");
            for id in [&a, &b] {
                assert!(on_disk(dir.path(), id), "{state:?} must survive");
                std::fs::remove_file(dir.path().join(format!("{id}.json"))).unwrap();
                std::fs::remove_file(dir.path().join(format!("{id}.events.jsonl"))).unwrap();
            }
        }
    }

    /// An exempt session neither dies nor consumes a keeper slot — the
    /// `completed_rank` semantics `RunStore` uses, and the reason `max_sessions`
    /// is NOT a bound on the size of the directory.
    #[test]
    fn exempt_sessions_do_not_spend_the_count_budget() {
        let dir = tempfile::tempdir().unwrap();
        let live = tempfile::tempdir().unwrap();
        let waiting = retained(dir.path(), CoderState::NeedsApproval, 500, None);
        let kept = retained(
            dir.path(),
            CoderState::Failed,
            500,
            Some(live.path().to_path_buf()),
        );
        let newest = retained(dir.path(), CoderState::Merged, 1, None);
        let older = retained(dir.path(), CoderState::Merged, 2, None);

        let collected = gc_sessions(
            dir.path(),
            &SessionRetention {
                max_sessions: 1,
                max_age_days: 0,
            },
            SweepScope::Boot,
        );

        // One collectable session over the cap of one — the exempt pair did not
        // fill it.
        assert_eq!(collected, 1);
        assert!(on_disk(dir.path(), &waiting));
        assert!(on_disk(dir.path(), &kept));
        assert!(on_disk(dir.path(), &newest));
        assert!(!on_disk(dir.path(), &older));
        // And the directory holds three sessions under a cap of one, which is
        // the documented behaviour rather than an accident.
        assert_eq!(CoderSession::list(dir.path()).len(), 3);
    }

    #[test]
    fn a_journal_with_no_snapshot_is_swept() {
        let dir = tempfile::tempdir().unwrap();
        // `coder.start` opens the sink's journal before anything persists a
        // snapshot, so a start that dies in between leaves exactly this — and
        // the candidate pass enumerates only `*.json`, so nothing else can ever
        // reach it.
        let orphan = dir.path().join("coder-died-before-persist.events.jsonl");
        std::fs::write(&orphan, "{}\n").unwrap();
        let live = retained(dir.path(), CoderState::Merged, 0, None);
        let live_journal = dir.path().join(format!("{live}.events.jsonl"));

        gc_sessions(dir.path(), &KEEP_ALL, SweepScope::Boot);

        assert!(!orphan.exists(), "an unreachable journal must be swept");
        assert!(
            live_journal.exists(),
            "a journal whose snapshot is retained must be left alone"
        );
    }

    /// The boot sequence, end to end: a crash orphan's worktree is reaped by
    /// adoption, which is what makes the snapshot collectable at all.
    ///
    /// Without the reap the directory survives, the worktree exemption fires
    /// forever, and every daemon crash permanently adds a snapshot, a journal
    /// and a full git checkout — a bounded leak turned into an unbounded one by
    /// the change that was supposed to close it.
    #[test]
    fn adoption_reaps_a_crash_orphans_worktree_so_retention_can_collect_it() {
        let dir = tempfile::tempdir().unwrap();
        let worktrees = dir.path().join("worktrees");
        let stranded = worktrees.join("coder-crashed");
        std::fs::create_dir_all(&stranded).unwrap();

        let id = retained(
            dir.path(),
            CoderState::Running,
            9999,
            Some(stranded.clone()),
        );

        let outcomes = adopt_orphaned_sessions(dir.path());
        assert_eq!(outcomes.len(), 1);
        assert_eq!(outcomes[0].1, AdoptionOutcome::Failed);
        assert!(
            !stranded.exists(),
            "a stranded worktree must be reaped, not left to exempt its own snapshot"
        );

        // Adoption stamps a fresh `updated_at`, so the age cap cannot take it on
        // this boot — the count cap can, and that is what proves it is no longer
        // exempt.
        let collected = gc_sessions(
            dir.path(),
            &SessionRetention {
                max_sessions: 0,
                max_age_days: 1,
            },
            SweepScope::Boot,
        );
        assert_eq!(
            collected, 0,
            "adoption stamped it now; the age cap must not fire"
        );

        let collected = gc_sessions(
            dir.path(),
            &SessionRetention {
                max_sessions: 0,
                max_age_days: 0,
            },
            SweepScope::Boot,
        );
        assert_eq!(collected, 0, "both caps disabled");

        let mut session = CoderSession::load(&dir.path().join(format!("{id}.json"))).unwrap();
        session.updated_at = now_secs().saturating_sub(9999 * 24 * 60 * 60);
        session.state_dir = Some(dir.path().to_path_buf());
        session.persist().unwrap();
        let collected = gc_sessions(
            dir.path(),
            &SessionRetention {
                max_sessions: 0,
                max_age_days: 30,
            },
            SweepScope::Boot,
        );
        assert_eq!(collected, 1, "collectable once the worktree is gone");
        assert!(!on_disk(dir.path(), &id));
    }

    /// A mid-lifetime sweep must never collect a snapshot the in-memory
    /// registry still holds an entry for.
    ///
    /// `prune_finished_sessions` reads "snapshot missing on disk" as "keep the
    /// entry rather than lose the session". So deleting one out from under a
    /// registered entry does not free anything — it converts that entry into a
    /// permanent memory pin, reopening the leak car#1262 closed through the
    /// door opened to bound the disk (car#1339). Terminal-and-still-registered
    /// is the ordinary window between a loop finishing and the next
    /// `coder.start` pruning it, not a rare race.
    #[test]
    fn a_mid_lifetime_sweep_never_collects_a_session_the_registry_still_holds() {
        let dir = tempfile::tempdir().unwrap();
        let mut ids = Vec::new();
        for _ in 0..3 {
            let mut s = CoderSession::new(
                Path::new("/tmp/repo"),
                "x",
                EngineChoice::Native,
                2,
                Some(dir.path().to_path_buf()),
            );
            s.state = CoderState::Merged;
            s.updated_at = now_secs().saturating_sub(9999 * 24 * 60 * 60);
            s.persist().unwrap();
            ids.push(s.id.clone());
        }

        // Boot form: nothing is live, every one of them is over the age cap.
        let retention = SessionRetention {
            max_sessions: 0,
            max_age_days: 30,
        };
        let live: std::collections::HashSet<String> = ids.iter().cloned().collect();

        let collected = gc_sessions(dir.path(), &retention, SweepScope::Live(&live));
        assert_eq!(
            collected, 0,
            "every id is registered; a mid-lifetime sweep must collect none"
        );
        for id in &ids {
            assert!(
                on_disk(dir.path(), id),
                "{id} was collected out from under a live entry"
            );
        }

        // Drop one from the registry and it becomes collectable — proving the
        // filter is what held it, not some other exemption.
        let mut partial = live.clone();
        partial.remove(&ids[0]);
        let collected = gc_sessions(dir.path(), &retention, SweepScope::Live(&partial));
        assert_eq!(collected, 1, "the unregistered one is collectable");
        assert!(!on_disk(dir.path(), &ids[0]));
        assert!(on_disk(dir.path(), &ids[1]));
    }

    /// The orphan-journal sweep is boot-only, and the live set cannot make it
    /// safe: `coder.start` opens the journal BEFORE inserting the entry, so a
    /// start racing a mid-lifetime sweep is journal-without-snapshot AND absent
    /// from any set the sweep could be handed.
    #[test]
    fn a_mid_lifetime_sweep_leaves_orphan_journals_alone() {
        let dir = tempfile::tempdir().unwrap();
        let journal = dir.path().join("just-started.events.jsonl");
        std::fs::write(&journal, "{}\n").unwrap();

        let retention = SessionRetention {
            max_sessions: 0,
            max_age_days: 30,
        };
        gc_sessions(
            dir.path(),
            &retention,
            SweepScope::Live(&std::collections::HashSet::new()),
        );
        assert!(
            journal.exists(),
            "a start that has not persisted yet must keep its journal"
        );

        // At boot the same file IS garbage — no live sink can own it there.
        gc_sessions(dir.path(), &retention, SweepScope::Boot);
        assert!(
            !journal.exists(),
            "boot must still sweep a stranded journal"
        );
    }

    #[test]
    fn adoption_keeps_the_worktree_the_operator_asked_to_keep() {
        let dir = tempfile::tempdir().unwrap();
        let kept = dir.path().join("worktrees").join("coder-postmortem");
        std::fs::create_dir_all(&kept).unwrap();

        let mut s = CoderSession::new(
            "/tmp/repo",
            "intent",
            EngineChoice::Auto,
            4,
            Some(dir.path().to_path_buf()),
        );
        s.state = CoderState::Running;
        s.workspace_path = Some(kept.clone());
        s.keep_workspace_on_failure = true;
        s.persist().unwrap();

        adopt_orphaned_sessions(dir.path());

        assert!(
            kept.is_dir(),
            "`keep_workspace_on_failure` is the operator asking for exactly this"
        );
        // And it stays exempt, which is the point of asking.
        assert_eq!(
            gc_sessions(
                dir.path(),
                &SessionRetention {
                    max_sessions: 0,
                    max_age_days: 1,
                },
                SweepScope::Boot,
            ),
            0
        );
    }

    #[test]
    fn gc_never_collects_a_session_whose_worktree_still_exists() {
        let dir = tempfile::tempdir().unwrap();
        let live = tempfile::tempdir().unwrap();
        // `keep_workspace_on_failure` left this directory behind; the snapshot
        // is the only thing that names it.
        let kept = retained(
            dir.path(),
            CoderState::Failed,
            9999,
            Some(live.path().to_path_buf()),
        );
        let reaped = retained(
            dir.path(),
            CoderState::Failed,
            9999,
            Some(dir.path().join("worktrees").join("gone")),
        );
        // A stray FILE at a worktree path is not a worktree. `exists()` would
        // exempt this session forever — the leak this guard exists to prevent,
        // reached backwards.
        let stray = dir.path().join("not-a-worktree");
        std::fs::write(&stray, "").unwrap();
        let not_a_worktree = retained(dir.path(), CoderState::Failed, 9999, Some(stray));

        let collected = gc_sessions(
            dir.path(),
            &SessionRetention {
                max_sessions: 0,
                max_age_days: 1,
            },
            SweepScope::Boot,
        );

        assert_eq!(collected, 2);
        assert!(on_disk(dir.path(), &kept), "a live worktree is not garbage");
        assert!(!on_disk(dir.path(), &reaped));
        assert!(!on_disk(dir.path(), &not_a_worktree));
    }

    #[test]
    fn gc_takes_the_journal_with_the_snapshot() {
        let dir = tempfile::tempdir().unwrap();
        let id = retained(dir.path(), CoderState::Merged, 99, None);
        let journal = dir.path().join(format!("{id}.events.jsonl"));
        assert!(journal.exists());

        gc_sessions(
            dir.path(),
            &SessionRetention {
                max_age_days: 30,
                ..KEEP_ALL
            },
            SweepScope::Boot,
        );

        assert!(!on_disk(dir.path(), &id));
        assert!(
            !journal.exists(),
            "the journal is the larger artifact; retaining it alone keeps the \
             bytes and drops the index that explains them"
        );
    }

    #[test]
    fn gc_with_both_caps_disabled_keeps_everything() {
        let dir = tempfile::tempdir().unwrap();
        let ids: Vec<String> = (0..4)
            .map(|i| retained(dir.path(), CoderState::Merged, i * 1000, None))
            .collect();

        assert_eq!(gc_sessions(dir.path(), &KEEP_ALL, SweepScope::Boot), 0);
        for id in &ids {
            assert!(on_disk(dir.path(), id));
        }
    }

    #[test]
    fn gc_on_a_missing_state_dir_is_not_an_error() {
        let dir = tempfile::tempdir().unwrap();
        let missing = dir.path().join("never-created");
        assert_eq!(
            gc_sessions(
                &missing,
                &SessionRetention {
                    max_sessions: 1,
                    max_age_days: 1
                },
                SweepScope::Boot,
            ),
            0
        );
    }

    #[test]
    fn authoring_models_are_every_distinct_model_that_completed_a_turn() {
        let dir = tempfile::tempdir().unwrap();
        let sink = EventSink::new(
            "coder-authors",
            None,
            Some(dir.path().join("coder-authors.events.jsonl")),
        );

        // `TurnCompleted` is a PER-ITERATION terminal and an unpinned session
        // routes per request, so a reviewer judging the accumulated diff has
        // to be checked against all of them. Taking the last would clear the
        // model that wrote most of this change.
        sink.record_turn_completed("empty_tool_calls", None, false, 3, "model-a", &[]);
        sink.record_turn_completed("empty_tool_calls", None, false, 5, "model-b", &[]);
        sink.record_turn_completed("max_turns", None, false, 9, "model-a", &[]);
        // Neither blank nor whitespace is an attribution.
        sink.record_turn_completed("empty_tool_calls", None, false, 1, "   ", &[]);

        assert_eq!(sink.authoring_models(), vec!["model-a", "model-b"]);
    }

    /// A model that wrote turns inside an iteration another model FINISHED.
    ///
    /// The terminal names only whoever reached it, so before car#1333 this
    /// model left no record anywhere — and could then sit on the review panel
    /// judging a change it had largely written, which is exactly the collision
    /// the gate exists to refuse.
    #[test]
    fn a_model_replaced_before_the_terminal_is_still_an_author() {
        let dir = tempfile::tempdir().unwrap();
        let sink = EventSink::new(
            "coder-midturn",
            None,
            Some(dir.path().join("coder-midturn.events.jsonl")),
        );

        // `writer` served turns 1-3; the chain degraded and `finisher` declared
        // done. Only `finisher` reaches the terminal.
        sink.record_turn_completed(
            "empty_tool_calls",
            None,
            false,
            4,
            "finisher",
            &["writer".to_string(), "finisher".to_string()],
        );

        let authors = sink.authoring_models();
        assert!(
            authors.contains(&"writer".to_string()),
            "the model that wrote most of the change must be named: {authors:?}"
        );
        assert!(authors.contains(&"finisher".to_string()), "{authors:?}");
    }

    /// A journal written before `models_served` existed folds to exactly the
    /// answer it gave before — the field is additive, not a new requirement.
    #[test]
    fn a_record_without_models_served_still_attributes_its_terminal() {
        let dir = tempfile::tempdir().unwrap();
        let sink = EventSink::new(
            "coder-legacy",
            None,
            Some(dir.path().join("coder-legacy.events.jsonl")),
        );
        sink.record_turn_completed("empty_tool_calls", None, false, 2, "only-model", &[]);
        assert_eq!(sink.authoring_models(), vec!["only-model"]);
    }

    #[test]
    fn a_session_with_no_journal_attributes_nothing() {
        // Foreman and external runs farm to a CLI whose backbone CAR never
        // resolved. Empty means "CAR's own loop did not write this", and the
        // gate reads it that way rather than as a fault.
        let sink = EventSink::test_sink();
        assert!(sink.authoring_models().is_empty());
    }

    #[test]
    fn snapshot_round_trips_without_workspace_handle() {
        let dir = tempfile::tempdir().unwrap();
        let mut s = CoderSession::new(
            "/tmp/repo",
            "intent",
            EngineChoice::Auto,
            4,
            Some(dir.path().to_path_buf()),
        );
        s.contract = Some(OutcomeContract {
            description: "d".into(),
            checks: vec![],
        });
        s.persist().unwrap();
        let loaded = CoderSession::load(&dir.path().join(format!("{}.json", s.id))).unwrap();
        assert_eq!(loaded.id, s.id);
        assert_eq!(loaded.state, CoderState::Created);
        assert!(loaded.workspace.is_none());
        assert!(loaded.contract.is_some());

        let listed = CoderSession::list(dir.path());
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].id, s.id);
    }

    #[test]
    fn event_json_shape_is_ws_friendly() {
        let e = CoderEvent {
            session_id: "coder-x".into(),
            seq: 3,
            ts: 1,
            kind: CoderEventKind::CheckStarted {
                name: "tests".into(),
            },
        };
        let v = serde_json::to_value(&e).unwrap();
        assert_eq!(v["type"], "check_started");
        assert_eq!(v["name"], "tests");
        assert_eq!(v["seq"], 3);
    }

    // --- daemon-restart orphan adoption -----------------------------------

    /// Write a snapshot directly in `state` (bypassing the transition guard,
    /// which is exactly the situation a daemon crash leaves on disk).
    fn write_snapshot(dir: &Path, state: CoderState, workspace_path: Option<PathBuf>) -> String {
        let mut s = CoderSession::new(
            "/tmp/repo",
            "intent",
            EngineChoice::Native,
            4,
            Some(dir.to_path_buf()),
        );
        s.state = state;
        s.workspace_path = workspace_path;
        s.persist().unwrap();
        s.id
    }

    fn reload(dir: &Path, id: &str) -> CoderSession {
        CoderSession::load(&dir.join(format!("{id}.json"))).unwrap()
    }

    #[test]
    fn adoption_fails_running_and_confirmed_orphans() {
        let dir = tempfile::tempdir().unwrap();
        let running = write_snapshot(dir.path(), CoderState::Running, None);
        let confirmed = write_snapshot(dir.path(), CoderState::ContractConfirmed, None);
        let created = write_snapshot(dir.path(), CoderState::Created, None);
        let proposed = write_snapshot(dir.path(), CoderState::ContractProposed, None);

        let outcomes = adopt_orphaned_sessions(dir.path());
        assert_eq!(outcomes.len(), 4);
        assert!(outcomes.iter().all(|(_, o)| *o == AdoptionOutcome::Failed));

        for id in [&running, &confirmed, &created, &proposed] {
            let s = reload(dir.path(), id);
            assert_eq!(s.state, CoderState::Failed, "{id} should be failed");
            assert_eq!(s.error.as_deref(), Some("daemon restarted mid-session"));
        }
    }

    #[test]
    fn adoption_preserves_needs_approval_with_live_worktree() {
        let dir = tempfile::tempdir().unwrap();
        // A real directory standing in for the surviving worktree.
        let worktree = dir.path().join("worktrees").join("wt-1");
        std::fs::create_dir_all(&worktree).unwrap();
        let id = write_snapshot(
            dir.path(),
            CoderState::NeedsApproval,
            Some(worktree.clone()),
        );

        let outcomes = adopt_orphaned_sessions(dir.path());
        assert_eq!(outcomes, vec![(id.clone(), AdoptionOutcome::Preserved)]);

        let s = reload(dir.path(), &id);
        // Untouched: still inspectable/approvable-by-hand, worktree path intact.
        assert_eq!(s.state, CoderState::NeedsApproval);
        assert!(s.error.is_none());
        assert_eq!(s.workspace_path.as_deref(), Some(worktree.as_path()));
    }

    #[test]
    fn adoption_fails_needs_approval_when_worktree_gone() {
        let dir = tempfile::tempdir().unwrap();
        // Worktree path recorded but never created (or already reaped).
        let gone = dir.path().join("worktrees").join("vanished");
        let id = write_snapshot(dir.path(), CoderState::NeedsApproval, Some(gone));

        let outcomes = adopt_orphaned_sessions(dir.path());
        assert_eq!(outcomes, vec![(id.clone(), AdoptionOutcome::Failed)]);

        let s = reload(dir.path(), &id);
        assert_eq!(s.state, CoderState::Failed);
        assert_eq!(s.error.as_deref(), Some("daemon restarted mid-session"));
    }

    #[test]
    fn adoption_leaves_terminal_snapshots_alone() {
        let dir = tempfile::tempdir().unwrap();
        let merged = write_snapshot(dir.path(), CoderState::Merged, None);
        let failed = write_snapshot(dir.path(), CoderState::Failed, None);
        let abandoned = write_snapshot(dir.path(), CoderState::Abandoned, None);

        let outcomes = adopt_orphaned_sessions(dir.path());
        assert!(
            outcomes.is_empty(),
            "terminal snapshots must not be adopted"
        );

        // Merged stays merged, no spurious error stamped on it.
        assert_eq!(reload(dir.path(), &merged).state, CoderState::Merged);
        assert_eq!(reload(dir.path(), &failed).state, CoderState::Failed);
        assert_eq!(reload(dir.path(), &abandoned).state, CoderState::Abandoned);
    }

    /// A daemon restart is the most common way a session dies, and it is the
    /// machinery dying — no check ever judged the work. Without an explicit
    /// kind the summary defaults to `"error"`, so a board (and any scorer
    /// reading the snapshot) renders a restart as a red verdict on the task.
    #[test]
    fn adopted_after_restart_carries_the_infrastructure_failure_kind() {
        let dir = tempfile::tempdir().unwrap();
        let running = write_snapshot(dir.path(), CoderState::Running, None);
        // …including the needs_approval orphan whose worktree is gone.
        let gone = dir.path().join("worktrees").join("vanished");
        let approval = write_snapshot(dir.path(), CoderState::NeedsApproval, Some(gone));

        adopt_orphaned_sessions(dir.path());

        for id in [&running, &approval] {
            let s = reload(dir.path(), id);
            assert_eq!(s.state, CoderState::Failed, "{id}");
            assert_eq!(
                s.failure_kind.as_deref(),
                Some("infrastructure"),
                "{id}: a restart is not a judged loss"
            );
        }
    }

    #[test]
    fn adoption_is_a_noop_on_missing_dir() {
        let dir = tempfile::tempdir().unwrap();
        let missing = dir.path().join("never-created");
        assert!(adopt_orphaned_sessions(&missing).is_empty());
    }

    // --- needs_you derivation (wire contract §1) --------------------------

    /// Every row of the §1 table, including the `null` default. The four kinds
    /// are what a board renders as "this one wants you"; getting one wrong
    /// either hides a blocked session or nags about a busy one.
    #[test]
    fn needs_you_covers_all_four_kinds_and_null() {
        use CoderState::*;
        // contract: the gate is decided by state alone.
        assert_eq!(
            needs_you_from(ContractProposed, false, false, None),
            Some(NeedsYou::Contract)
        );
        // approval: likewise.
        assert_eq!(
            needs_you_from(NeedsApproval, false, false, None),
            Some(NeedsYou::Approval)
        );
        // question: running + a parked question.
        assert_eq!(
            needs_you_from(Running, true, false, None),
            Some(NeedsYou::Question)
        );
        // auth: running + an unresolved auth_required.
        assert_eq!(
            needs_you_from(Running, false, true, None),
            Some(NeedsYou::Auth)
        );
        // null: running with neither, and every other state.
        assert_eq!(needs_you_from(Running, false, false, None), None);
        for state in [Created, ContractConfirmed, Merged, Failed, Abandoned] {
            assert_eq!(needs_you_from(state, false, false, None), None, "{state:?}");
            // A stale gate flag must not resurrect a terminal session as
            // "waiting on you" — the state is what decides.
            assert_eq!(needs_you_from(state, true, true, None), None, "{state:?}");
        }
    }

    /// A parked question outranks an outstanding sign-in: the question is a
    /// literal prompt with a waiter behind it, while an auth event only means
    /// the loop is polling for a credential.
    #[test]
    fn a_parked_question_outranks_an_outstanding_sign_in() {
        assert_eq!(
            needs_you_from(CoderState::Running, true, true, None),
            Some(NeedsYou::Question)
        );
    }

    /// The daemon owns the wording so two clients cannot describe one state
    /// differently — the `overlap_disclosure` precedent.
    #[test]
    fn needs_you_labels_and_wire_values_round_trip() {
        for (kind, wire, label) in [
            (
                NeedsYou::Contract,
                "contract",
                "contract awaiting confirmation",
            ),
            (NeedsYou::Question, "question", "question waiting"),
            (NeedsYou::Approval, "approval", "diff ready for approval"),
            (NeedsYou::Auth, "auth", "sign-in needed"),
        ] {
            assert_eq!(kind.as_str(), wire);
            assert_eq!(kind.label(), label);
            assert_eq!(NeedsYou::parse(wire), Some(kind));
        }
        assert_eq!(NeedsYou::parse("nonsense"), None);
    }

    /// The gate carries the prompt so a summary can render *what* is being
    /// asked without replaying the event stream — and drops it the moment the
    /// question is answered or cleared.
    #[test]
    fn the_input_gate_carries_and_releases_its_prompt() {
        let gate = UserInputGate::new();
        assert!(!gate.is_pending());
        assert_eq!(gate.pending_prompt(), None);

        let _rx = gate.park("Which database should this target?");
        assert!(gate.is_pending());
        assert_eq!(
            gate.pending_prompt().as_deref(),
            Some("Which database should this target?")
        );

        gate.fulfill("postgres".into()).unwrap();
        assert!(!gate.is_pending());
        assert_eq!(gate.pending_prompt(), None);

        let _rx = gate.park("again?");
        gate.clear();
        assert_eq!(gate.pending_prompt(), None);
    }

    /// An OLD on-disk snapshot — written before `failure_kind` / `needs_you` /
    /// `discussion_id` existed — must still deserialize. A daemon upgrade that
    /// bricked `coder.list` on every pre-upgrade session would be a far worse
    /// bug than the missing fields it was adding.
    #[test]
    fn an_old_format_snapshot_still_deserializes() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("coder-legacy.json");
        // Verbatim shape of a pre-board snapshot: no failure_kind, no
        // needs_you, no discussion_id.
        std::fs::write(
            &path,
            r#"{
              "id": "coder-legacy",
              "repo": "/tmp/repo",
              "intent": "make it work",
              "engine": "native",
              "state": "failed",
              "iterations": 3,
              "max_iterations": 8,
              "keep_workspace_on_failure": false,
              "last_check_results": [],
              "created_at": 100,
              "updated_at": 200,
              "error": "contract not satisfied after 3 iteration(s)"
            }"#,
        )
        .unwrap();

        let loaded = CoderSession::load(&path).expect("legacy snapshot must still load");
        assert_eq!(loaded.id, "coder-legacy");
        assert_eq!(loaded.state, CoderState::Failed);
        // The new fields default rather than failing the parse.
        assert_eq!(loaded.failure_kind, None);
        assert_eq!(loaded.discussion_id, None);
        // ...and `list` (what coder.list reads) picks it up unchanged.
        let listed = CoderSession::list(dir.path());
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].id, "coder-legacy");
    }

    /// The new fields survive a write→read round trip, which is what makes a
    /// post-restart summary able to say *why* a session failed.
    #[test]
    fn attention_fields_survive_a_snapshot_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        let mut s = CoderSession::new(
            "/tmp/repo",
            "intent",
            EngineChoice::Native,
            4,
            Some(dir.path().to_path_buf()),
        );
        s.state = CoderState::Failed;
        s.failure_kind = Some("budget_exhausted".into());
        s.discussion_id = Some("disc-abc".into());
        s.result_branch = Some("car/coder/ab12cd34".into());
        s.model = Some("parslee/reasoning".into());
        s.persist().unwrap();

        let loaded = CoderSession::load(&dir.path().join(format!("{}.json", s.id))).unwrap();
        assert_eq!(loaded.failure_kind.as_deref(), Some("budget_exhausted"));
        assert_eq!(loaded.discussion_id.as_deref(), Some("disc-abc"));
        assert_eq!(loaded.result_branch.as_deref(), Some("car/coder/ab12cd34"));
        assert_eq!(loaded.model.as_deref(), Some("parslee/reasoning"));
    }

    #[test]
    fn contract_revision_rejected_is_named_and_ws_shaped() {
        let kind = CoderEventKind::ContractRevisionRejected {
            request: "also verify the Windows path".into(),
            reason: "the redrafted contract is invalid: contract has no checks".into(),
        };
        assert_eq!(coder_event_name(&kind), "coder.contract_revision_rejected");
        let v = serde_json::to_value(CoderEvent {
            session_id: "coder-x".into(),
            seq: 4,
            ts: 1,
            kind,
        })
        .unwrap();
        assert_eq!(v["type"], "contract_revision_rejected");
        assert_eq!(v["request"], "also verify the Windows path");
        assert!(v["reason"].as_str().unwrap().contains("no checks"));
    }
}