car-server-core 0.47.0

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

/// `~/.car/coder` — session snapshots, event journals, and worktrees.
pub fn default_state_dir() -> Result<PathBuf, String> {
    let home = std::env::var_os("HOME")
        .or_else(|| std::env::var_os("USERPROFILE"))
        .ok_or("cannot resolve home directory (HOME/USERPROFILE unset)")?;
    Ok(PathBuf::from(home).join(".car").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,
    Failed,
    Abandoned,
}

impl CoderState {
    pub fn is_terminal(&self) -> bool {
        matches!(self, Self::Merged | 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::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,
    /// 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::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::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),
            "auth" => Some(Self::Auth),
            _ => None,
        }
    }
}

/// 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,
) -> Option<NeedsYou> {
    match state {
        CoderState::ContractProposed => Some(NeedsYou::Contract),
        CoderState::NeedsApproval => 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)
        )
}

/// 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,
    },
    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,
    },
    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,
    ) {
        let Some(journal) = &self.journal else { return };
        let data = car_engine::goal::turn_completed_data(
            decision,
            stop_reason,
            was_truncated,
            turns,
            model,
        );
        if let Ok(mut log) = journal.lock() {
            log.append(EventKind::TurnCompleted, Some(&self.session_id), None, data);
        }
    }

    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::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::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",
    }
}

/// 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>,
    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"` | `"error"`. Persisted so a
    /// summary read from disk after a daemon restart still distinguishes "ran
    /// out of clock" from "nobody signed in" from "the work was judged and
    /// rejected" — a live-only derivation would go blank exactly when the
    /// operator comes back to look.
    #[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>,
    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,
            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,
            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
    }
}

/// 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;
        }
        session.state = CoderState::Failed;
        session.error = Some("daemon restarted mid-session".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::*;

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

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

    #[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),
            Some(NeedsYou::Contract)
        );
        // approval: likewise.
        assert_eq!(
            needs_you_from(NeedsApproval, false, false),
            Some(NeedsYou::Approval)
        );
        // question: running + a parked question.
        assert_eq!(
            needs_you_from(Running, true, false),
            Some(NeedsYou::Question)
        );
        // auth: running + an unresolved auth_required.
        assert_eq!(needs_you_from(Running, false, true), Some(NeedsYou::Auth));
        // null: running with neither, and every other state.
        assert_eq!(needs_you_from(Running, false, false), None);
        for state in [Created, ContractConfirmed, Merged, Failed, Abandoned] {
            assert_eq!(needs_you_from(state, false, false), 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, "{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),
            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"));
    }
}