car_server_core/coder/

discuss.rs

//! `coder.discuss.*` — a repo-grounded, strictly **read-only** conversation
//! that can be distilled into a run intent.
//!
//! The gap this closes: `coder.start` demands a well-formed intent before
//! anything exists to react to. An operator who is still working out *what*
//! they want has no surface between "I have a vague idea" and "here is a
//! contract-worthy sentence" — so they either guess (and burn a session on a
//! badly-aimed contract) or go think somewhere else with none of the repo in
//! front of them.
//!
//! A discussion is grounded in the repo through the same
//! [`AssistantService`] that backs `car do`,
//! bound with `bind_default_substrate(prefer_local = true, full_access = false,
//! …)` — i.e. [`PermissionTier::ReadOnly`], where every write and every shell
//! escalates to an approval gate.
//!
//! [`PermissionTier::ReadOnly`]: car_policy::permission::PermissionTier
//!
//! ## What is actually enforced
//!
//! Two independent mechanisms, both required — an earlier version of this doc
//! claimed the discussion "never touches the repo", which overstated the first
//! and ignored that reads were unbounded:
//!
//! 1. **No mutation.** `write_file`, `edit_file` and `shell` are in the
//!    ReadOnly tier's gated set, so each escalates to the approval gate — and
//!    this surface **auto-DENIES** every escalation rather than prompting a
//!    human. A discussion cannot write a file, run a command, create a branch,
//!    or provision a worktree. The refusal is visible as a
//!    `tool_result { ok: false }`, never silent.
//! 2. **No read *path* outside the repo.** The read tools (`read_file`,
//!    `list_dir`, `find_files`, `grep_files`) are NOT gated — they are the
//!    point of a grounded discussion — so mutation-gating alone left them
//!    pointed at the whole filesystem. The discussion's bound environment
//!    therefore sets [`BoundEnvironment::clamp_reads`], pinning those four
//!    inside the repo root. Without it, a prompt-injected repo file could ask
//!    for `grep_files {"path":"/Users/<user>","pattern":"sk-ant-"}` and the
//!    hits would stream to every `coder.discuss.event` subscriber. Scoped to
//!    this surface only; the general assistant's read reach is unchanged.
//!
//!    Read the claim precisely: the clamp is **lexical**, not a resolved-path
//!    check. `coder::policy::stays_under` normalizes `.` / `..` textually and
//!    compares the result against the root (see its own
//!    `stays_under_is_lexical_and_strict` test), and the file walk stats
//!    entries with `metadata()`, which follows symlinks. So a symlink
//!    *committed inside the repo* and pointing outward reads through the clamp
//!    — its path stays under the root, its target does not. What the clamp
//!    stops is the model **naming** a path outside the repo, which is the
//!    prompt-injection vector above; it is not a containment boundary against
//!    the repo's own contents. Treat the repo as trusted-to-the-extent-you-
//!    trust-what-is-committed-in-it.
//!
//! [`BoundEnvironment::clamp_reads`]: crate::assistant::BoundEnvironment::clamp_reads
//!
//! `coder.start` is the only thing that starts work, and
//! `coder.discuss.promote` deliberately starts nothing — it hands back a
//! distilled intent the operator may edit first.
//!
//! ## Lifetime
//!
//! Model transcripts and action receipts are checkpointed through the daemon's
//! existing assistant oplog. The live runtime, substrate, replay stream and
//! connection ownership are process-local: `start { resume_id }` reconstructs
//! them from a checkpoint and a local repository/owner binding. Discussions
//! are **owned by the connection that opened
//! them**: only that connection may send to, subscribe to, promote or close
//! them, and closing it closes the discussion and cancels any in-flight turn,
//! because a detached turn would keep billing model tokens to nobody. Bounded
//! three ways — [`MAX_OPEN_DISCUSSIONS`], [`DISCUSSION_IDLE_TTL_SECS`], and the
//! per-discussion buffer/transcript caps.
//!
//! ## Event fanout
//!
//! One **drain task per discussion** owns the subscriber set. Emits, attaches
//! and detaches are commands on its channel, so a single task serializes them:
//! `seq` is assigned under the buffer lock by the only writer (no out-of-order
//! buffer), an attach replays everything buffered *before* the next emit is
//! processed (no gap, no duplicate), and — unlike the `coder.event` path — no
//! lock is ever held across a send.
//!
//! The drain does not send, though: **each subscriber owns a bounded queue and
//! its own sender task**. That is the part that makes a wedged subscriber
//! merely its own problem. When the drain itself performed the sends, an
//! untimed write to a half-open socket blocked the drain, so the *next* `Emit`
//! command sat unprocessed — and since every `entry.emit(…).await` inside
//! [`run_turn`] waits for its `seq`, one wedged board stalled the whole
//! **turn**, not just its stream. Now the drain only `try_send`s into each
//! subscriber's queue: a subscriber that cannot keep up (queue full, or a send
//! past [`DISCUSS_SEND_TIMEOUT`]) is **shed**, and the turn never waits on a
//! socket.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex as StdMutex, MutexGuard};

use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tokio::sync::{mpsc, oneshot};

use super::discussion_record::DiscussionRecord;
use crate::assistant::governance::AssistantDurability;
use crate::assistant::{
    bind_default_substrate, build_assistant_runtime_with_tools, prompt, AssistantConfig,
    AssistantService,
};
use crate::coder::native_loop::TurnGenerator;
use crate::handler::JsonRpcMessage;
use crate::session::{ClientSession, ServerState, WsChannel};

const DISCUSSION_MUTATION_REFUSAL: &str = "This conversation stage is read-only; this is not a file-permission problem or a user refusing the requested change. Use prepare_coding_task with the requested change and constraints so the user can review it and start coding. Do not change file permissions or ask the user to do so.";

/// Turn cap for one discussion reply. A discussion reads and reasons; it never
/// edits, so it has no repair loop to spend turns on.
const DISCUSS_MAX_TURNS: u32 = 12;

/// Attempts allowed when distilling a transcript into an intent. Same bounded
/// shape as `derive_contract`: the output is structured JSON, so a malformed
/// reply is worth one retry, not an unbounded loop.
const PROMOTE_MAX_ATTEMPTS: u32 = 3;

/// Concurrent open discussions per daemon. Each pins an `AssistantService`, a
/// `Runtime`, and an open runtime session, so they are not free; a board opens
/// one at a time and an operator juggling more than a handful has lost track.
///
/// Enforced by [`ServerState::coder_discussion_slots`], a semaphore whose
/// permit is taken before any of `start_discussion`'s async work and lives
/// inside the admitted [`DiscussionEntry`] — NOT by counting the registry, which
/// was a TOCTOU check that bounded nothing under pipelined starts.
///
/// [`ServerState::coder_discussion_slots`]: crate::session::ServerState
pub(crate) const MAX_OPEN_DISCUSSIONS: usize = 8;

/// A discussion with no activity for this long is reaped on the next
/// `coder.discuss.start`. Long enough to step away from a train of thought,
/// short enough that a forgotten one does not pin a runtime overnight.
const DISCUSSION_IDLE_TTL_SECS: u64 = 60 * 60;

/// Retained events per discussion. The oldest are dropped past this; a replay
/// from a trimmed cursor returns what survives (`events_replayed` says how
/// much) rather than growing without bound on a long conversation.
const DISCUSS_EVENT_BUFFER_MAX: usize = 2000;

/// Transcript turns retained for distillation. `promote` is a summarization
/// call, so the recent exchange is what carries the intent; keeping everything
/// eventually builds a prompt no model window holds.
const TRANSCRIPT_MAX_TURNS: usize = 40;

/// Turns handed to `distill`. The most recent slice of the retained transcript
/// — the tail is where the operator converged.
const DISTILL_WINDOW_TURNS: usize = 12;

/// Byte cap on one operator message.
///
/// tungstenite accepts up to 64 MiB per frame, and an accepted message is
/// cloned into the transcript, cloned again into the event buffer, and rendered
/// into the distill prompt — so without a cap, 40 sequential 50 MB sends retain
/// gigabytes per discussion and make `promote` build a prompt no window holds.
/// `summarize_repo` is head-capped for exactly this reason; operator text needs
/// the same. Generous for prose — this is a conversation, not a file upload.
const DISCUSS_MESSAGE_MAX_BYTES: usize = 64 * 1024;

/// Depth of one subscriber's outbound frame queue.
///
/// Must exceed [`DISCUSS_EVENT_BUFFER_MAX`] so a legitimate
/// `subscribe { from_seq: 0 }` replay — up to a full buffer, queued in one go —
/// is never mistaken for a slow consumer. Past that, a subscriber this far
/// behind is not reading.
const DISCUSS_SUBSCRIBER_QUEUE: usize = DISCUSS_EVENT_BUFFER_MAX + 128;

/// How long one frame may take to reach a subscriber's socket before that
/// subscriber is shed. A half-open peer never fails a write — it parks forever,
/// holding the socket's write half. Matches the coder fanout's deadline.
const DISCUSS_SEND_TIMEOUT: std::time::Duration = crate::coder::rpc::FANOUT_WRITE_TIMEOUT;

/// Live discussions keyed by `discussion_id`.
pub type DiscussionMap = HashMap<String, Arc<DiscussionEntry>>;

/// Take a `std` lock without letting a poisoned mutex become permanent.
///
/// A panic anywhere under one of these locks would otherwise brick the
/// discussion for its whole lifetime — and the first panic is swallowed by the
/// detached turn task, so the operator would see an inexplicably dead
/// conversation with no error. The data behind each of these is a plain
/// `Vec`/`Option`; a torn write is not a safety problem here.
fn lock<T>(m: &StdMutex<T>) -> MutexGuard<'_, T> {
    m.lock().unwrap_or_else(|e| e.into_inner())
}

/// One event in a discussion's stream. `seq` is monotonic per discussion so a
/// client can resume from a cursor, exactly like `CoderEvent`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DiscussEvent {
    pub discussion_id: String,
    pub seq: u64,
    pub ts: u64,
    #[serde(flatten)]
    pub kind: DiscussEventKind,
}

/// What happened in a discussion. Serialized with `"type":"snake_case_name"`,
/// tagged the same way [`CoderEventKind`](super::session::CoderEventKind) is.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum DiscussEventKind {
    UserMessage {
        text: String,
    },
    /// A streaming chunk of the model's reply.
    AssistantDelta {
        text: String,
    },
    /// The complete assistant turn.
    AssistantMessage {
        text: String,
    },
    ToolCall {
        tool: String,
        params_preview: String,
    },
    ToolResult {
        tool: String,
        ok: bool,
        preview: String,
    },
    TaskPrepared {
        proposed_intent: String,
        constraints: Vec<String>,
    },
    TurnComplete {},
    Error {
        message: String,
    },
}

/// A command for a discussion's drain task — the single owner of its
/// subscriber set and the only thing that writes its buffer or sends a frame.
enum StreamCmd {
    Emit(DiscussEventKind, oneshot::Sender<u64>),
    Attach {
        client_id: String,
        channel: Arc<WsChannel>,
        from_seq: u64,
        replayed: oneshot::Sender<u64>,
    },
    Detach(String),
}

/// The in-flight turn's handle and the discussion's terminal `closed` latch,
/// deliberately behind one lock.
///
/// They were separate, and the gap between them orphaned model loops: `close`
/// read `turn_task` (still `None`, because `send_message` stores the handle
/// only *after* its first `emit().await`), found nothing to abort, and removed
/// the entry from the registry — then `send_message` resumed and spawned a turn
/// against a discussion nothing could reach any more. It billed up to
/// [`DISCUSS_MAX_TURNS`] turns against a live provider with no way to stop it.
/// Publishing "this discussion is closed" and "here is the turn to abort"
/// through the same lock closes that window in both directions: a close either
/// aborts the running turn or latches `closed` so the turn is never spawned.
#[derive(Default)]
struct TurnSlot {
    /// Set once, terminally, by [`DiscussionEntry::cancel_turn`]. Every caller
    /// of `cancel_turn` also removes the entry from the registry, so there is
    /// no legitimate reopen.
    closed: bool,
    handle: Option<tokio::task::JoinHandle<()>>,
}

/// Clears `in_flight` on EVERY exit path, including a cancelled or panicking
/// handler future.
///
/// `in_flight` was set by CAS in `send_message` and cleared only at the tail of
/// the spawned turn task. Anything that dropped the handler future between
/// those two points — the daemon's handler deadline is the reachable one, since
/// `coder.discuss.send` is not deadline-exempt — left it `true` with no turn
/// running. The discussion then answered "still answering the previous message"
/// to every `send` and "still answering" to every `promote`, forever, and
/// `reap_idle` runs only on the next `discuss.start`, so on a quiet daemon it
/// was never reclaimed either. A latch that only one code path can release is
/// a latch that leaks; this releases in `Drop`.
struct InFlightGuard(Arc<DiscussionEntry>);

impl Drop for InFlightGuard {
    fn drop(&mut self) {
        self.0.in_flight.store(false, Ordering::SeqCst);
        self.0.touch();
    }
}

/// Keeps the operator's turn in the transcript only if a reply turn was
/// actually dispatched for it.
///
/// `send_message` records the turn before the `emit().await` it may be
/// cancelled at, and before the dispatch that may be refused. `InFlightGuard`
/// frees the discussion on those paths, but the transcript was left ending in
/// an operator question with no reply — and that is exactly the input the
/// `is_answering()` guards on `promote` and `coder.start { discussion_id }`
/// exist to keep out of distillation. Those guards read "not answering", so a
/// stranded question sails through them and the model invents a confident
/// intent from a question nobody answered. Recording after the dispatch would
/// let the spawned turn's `Assistant` row land first, so the row goes in early
/// and comes back out on every path that did not dispatch.
struct TurnRecordGuard {
    entry: Arc<DiscussionEntry>,
    text: String,
    dispatched: bool,
}

impl Drop for TurnRecordGuard {
    fn drop(&mut self) {
        if !self.dispatched {
            self.entry.rollback_turn("Operator", &self.text);
        }
    }
}

/// One live discussion.
pub struct DiscussionEntry {
    pub id: String,
    /// The git repo the conversation is grounded in.
    pub repo: PathBuf,
    /// Cheap repo orientation, returned by `coder.discuss.start` so a caller
    /// can show what the discussion can see.
    pub repo_summary: String,
    project_context: String,
    pub created_at: u64,
    pub model: Option<String>,
    /// The connection that opened this discussion. Closing it closes the
    /// discussion — see the module docs on lifetime.
    owner_client_id: String,
    principal: String,
    record_root: PathBuf,
    /// Replay buffer. Written **only** by the drain task, so it is always in
    /// `seq` order; readable elsewhere for inspection.
    pub events: Arc<tokio::sync::Mutex<Vec<DiscussEvent>>>,
    /// Commands to the drain task.
    cmds: mpsc::UnboundedSender<StreamCmd>,
    /// Completed operator turns (what `turns` reports in `coder.discuss.list`).
    turns: AtomicU64,
    /// Whether a reply turn is running right now. A discussion is a
    /// conversation: two overlapping turns interleave into one model thread and
    /// silently lose one of them, so a second `send` is refused rather than
    /// queued.
    in_flight: AtomicBool,
    start_lock: Arc<tokio::sync::Mutex<()>>,
    /// Last activity, for the idle TTL.
    last_active: AtomicU64,
    /// The in-flight turn's task plus the terminal `closed` latch, under ONE
    /// lock. See [`TurnSlot`] for why they cannot be separate.
    turn_task: StdMutex<TurnSlot>,
    /// This discussion's open-slot reservation, taken before any of
    /// `start_discussion`'s async work and released when the entry drops.
    _slot: tokio::sync::OwnedSemaphorePermit,
    /// The grounded, read-only conversational service.
    service: Arc<AssistantService>,
    durability: Arc<crate::assistant::durability::LocalAssistantDurability>,
    /// The model seam used for distillation (`promote`). Same injection style
    /// `derive_contract` uses, so promote is testable with a scripted model.
    generator: Arc<dyn TurnGenerator>,
    /// Role-tagged plain-text transcript, kept for distillation. Deliberately
    /// separate from the service's own message thread: promote must see the
    /// conversation, not the tool plumbing. Capped at [`TRANSCRIPT_MAX_TURNS`].
    transcript: StdMutex<Vec<(&'static str, String)>>,
    /// The most recent `promote` result, cached so `coder.start
    /// { discussion_id }` can fold the agreed constraints into contract
    /// derivation without a second distillation call.
    last_promote: StdMutex<Option<(String, Vec<String>)>>,
    task_proposal: super::task_proposal::PreparedTask,
}

impl DiscussionEntry {
    fn save_pending_task(&self, proposal: Option<(String, Vec<String>)>) -> Result<(), String> {
        let mut record = DiscussionRecord::load(&self.record_root, &self.id, &self.principal)?;
        record.pending_task = proposal;
        record.save_model(&self.record_root)
    }

    /// Constraints agreed in this discussion, from the last `promote`.
    pub fn constraints(&self) -> Vec<String> {
        lock(&self.last_promote)
            .as_ref()
            .map(|(_, c)| c.clone())
            .unwrap_or_default()
    }

    /// Whether a reply turn is running right now.
    pub fn is_answering(&self) -> bool {
        self.in_flight.load(Ordering::SeqCst)
    }

    fn touch(&self) {
        self.last_active.store(now_secs(), Ordering::SeqCst);
    }

    fn idle_secs(&self) -> u64 {
        now_secs().saturating_sub(self.last_active.load(Ordering::SeqCst))
    }

    fn record_turn(&self, role: &'static str, text: &str) {
        if text.trim().is_empty() {
            return;
        }
        let mut t = lock(&self.transcript);
        t.push((role, text.to_string()));
        // Bounded: drop from the front, keeping the recent exchange.
        let len = t.len();
        if len > TRANSCRIPT_MAX_TURNS {
            t.drain(..len - TRANSCRIPT_MAX_TURNS);
        }
    }

    /// Undo the most recent [`record_turn`](Self::record_turn) when it is still
    /// the tail and still ours. Matching on both role and text is what keeps a
    /// rollback from eating someone else's row if the transcript moved on.
    fn rollback_turn(&self, role: &'static str, text: &str) {
        let mut t = lock(&self.transcript);
        if t.last().is_some_and(|(r, s)| *r == role && s == text) {
            t.pop();
        }
    }

    /// The most recent turns, rendered for distillation.
    fn distill_transcript(&self) -> String {
        let t = lock(&self.transcript);
        let start = t.len().saturating_sub(DISTILL_WINDOW_TURNS);
        t[start..]
            .iter()
            .map(|(role, text)| format!("{role}: {text}"))
            .collect::<Vec<_>>()
            .join("\n\n")
    }

    fn transcript_is_empty(&self) -> bool {
        lock(&self.transcript).is_empty()
    }

    /// Append an event to the stream, returning its assigned `seq`.
    ///
    /// The drain assigns the seq under the buffer lock, so the buffer is always
    /// ordered; this only waits for that assignment, never for a WS send.
    async fn emit(&self, kind: DiscussEventKind) -> u64 {
        let (tx, rx) = oneshot::channel();
        if self.cmds.send(StreamCmd::Emit(kind, tx)).is_err() {
            return 0; // drain gone (discussion closed) — nothing to stream to
        }
        rx.await.unwrap_or(0)
    }

    /// Stop an in-flight turn and latch the discussion closed: signal the loop,
    /// drop the task, and make sure no turn that is still being dispatched can
    /// start behind us.
    ///
    /// Terminal by construction — every caller (`close`, disconnect teardown,
    /// `reap_idle`) also removes the entry from the registry.
    fn cancel_turn(&self) {
        self.durability.revoke();
        self.service.cancel(&self.id);
        {
            let mut slot = lock(&self.turn_task);
            slot.closed = true;
            if let Some(handle) = slot.handle.take() {
                handle.abort();
            }
        }
        self.in_flight.store(false, Ordering::SeqCst);
    }

    /// Spawn the reply turn under the same lock `cancel_turn` latches, so a
    /// close that raced the dispatch either aborts this turn or prevents it.
    ///
    /// Returns `false` when the discussion was closed before the dispatch
    /// reached this point — the turn is then never spawned at all.
    fn spawn_turn<F>(&self, make: F) -> bool
    where
        F: FnOnce() -> tokio::task::JoinHandle<()>,
    {
        let mut slot = lock(&self.turn_task);
        if slot.closed {
            return false;
        }
        // No await under this guard: `tokio::spawn` only queues the task.
        slot.handle = Some(make());
        true
    }

    fn summary_row(&self) -> Value {
        json!({
            "discussion_id": self.id,
            "repo": self.repo,
            "created_at": self.created_at,
            "turns": self.turns.load(Ordering::SeqCst),
        })
    }
}

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

fn event_frame(event: &DiscussEvent) -> Option<String> {
    serde_json::to_string(&json!({
        "jsonrpc": "2.0",
        "method": "coder.discuss.event",
        "params": event,
    }))
    .ok()
}

/// One subscriber's outbound lane: a bounded frame queue plus the task that
/// drains it onto that subscriber's socket.
///
/// One lane per subscriber is what decouples the stream from the turn. The
/// drain hands frames over with `try_send` and never awaits a socket, so no
/// subscriber can delay the `seq` reply the turn is blocked on.
struct Subscriber {
    frames: mpsc::Sender<String>,
    task: tokio::task::JoinHandle<()>,
}

impl Drop for Subscriber {
    /// Abort rather than let the queue drain: the task may be parked on a
    /// half-open socket's write mutex, and that parked future is precisely what
    /// keeps the write half alive past teardown.
    fn drop(&mut self) {
        self.task.abort();
    }
}

fn spawn_subscriber(channel: Arc<WsChannel>) -> Subscriber {
    let (frames, mut rx) = mpsc::channel::<String>(DISCUSS_SUBSCRIBER_QUEUE);
    let task = tokio::spawn(async move {
        while let Some(frame) = rx.recv().await {
            if tokio::time::timeout(
                DISCUSS_SEND_TIMEOUT,
                crate::coder::rpc::send_frame(&channel, &frame),
            )
            .await
            .is_err()
            {
                // Wedged socket. Ending the task drops the channel handle and
                // closes the queue, so the drain sheds this subscriber on its
                // next `try_send` instead of queueing for a peer that is gone.
                break;
            }
        }
    });
    Subscriber { frames, task }
}

/// The per-discussion drain: the single writer of the buffer and the single
/// owner of the subscriber set.
///
/// Because one task handles emits and attaches in order, an attach replays
/// everything buffered so far and is registered before the next emit is
/// processed — no gap and no duplicate — without holding any lock across a
/// handoff. The drain itself never touches a socket: it `try_send`s into each
/// subscriber's own queue, so a subscriber that has stopped reading is shed
/// rather than allowed to stall the buffer, the next `Emit`, or the turn
/// waiting on that `Emit`'s `seq`.
fn spawn_discuss_drain(
    discussion_id: String,
    events: Arc<tokio::sync::Mutex<Vec<DiscussEvent>>>,
) -> mpsc::UnboundedSender<StreamCmd> {
    let (tx, mut rx) = mpsc::unbounded_channel::<StreamCmd>();
    tokio::spawn(async move {
        let mut subscribers: HashMap<String, Subscriber> = HashMap::new();
        let mut next_seq: u64 = 0;
        while let Some(cmd) = rx.recv().await {
            match cmd {
                StreamCmd::Emit(kind, reply) => {
                    let seq = next_seq;
                    next_seq += 1;
                    let event = DiscussEvent {
                        discussion_id: discussion_id.clone(),
                        seq,
                        ts: now_secs(),
                        kind,
                    };
                    let frame = event_frame(&event);
                    {
                        let mut buffer = events.lock().await;
                        buffer.push(event);
                        let len = buffer.len();
                        if len > DISCUSS_EVENT_BUFFER_MAX {
                            buffer.drain(..len - DISCUSS_EVENT_BUFFER_MAX);
                        }
                    } // lock released BEFORE the handoff
                    let _ = reply.send(seq);
                    if let Some(frame) = &frame {
                        // `try_send`, never `send`: a full queue means this
                        // subscriber is not draining, and waiting for it is how
                        // one wedged board used to stall the whole turn.
                        subscribers.retain(|client_id, s| {
                            let ok = s.frames.try_send(frame.clone()).is_ok();
                            if !ok {
                                tracing::warn!(
                                    discussion_id = %discussion_id,
                                    client_id = %client_id,
                                    "discussion subscriber is not draining; dropping it"
                                );
                            }
                            ok
                        });
                    }
                }
                StreamCmd::Attach {
                    client_id,
                    channel,
                    from_seq,
                    replayed,
                } => {
                    // Clone the frames under the lock, release, then queue.
                    let frames: Vec<String> = {
                        let buffer = events.lock().await;
                        buffer
                            .iter()
                            .filter(|e| e.seq >= from_seq)
                            .filter_map(event_frame)
                            .collect()
                    };
                    let subscriber = spawn_subscriber(channel);
                    // The queue is sized to hold a whole buffer replay, so this
                    // only short-circuits if the peer's lane already died.
                    let mut n = 0u64;
                    for frame in frames {
                        if subscriber.frames.try_send(frame).is_err() {
                            break;
                        }
                        n += 1;
                    }
                    subscribers.insert(client_id, subscriber);
                    let _ = replayed.send(n);
                }
                StreamCmd::Detach(client_id) => {
                    subscribers.remove(&client_id);
                }
            }
        }
        // Discussion closed: every lane's task is aborted by `Subscriber::drop`.
    });
    tx
}

// ---------------------------------------------------------------------------
// Orchestration (generation-injectable, transport-free)
// ---------------------------------------------------------------------------

/// Provision a discussion grounded in `repo`, owned by `owner_client_id`.
///
/// `engine` builds the read-only assistant runtime (tools, substrate, gates);
/// `generator` is the model seam both the conversation and `promote` run on.
/// Split so tests can drive a scripted model against a real temp repo.
pub async fn start_discussion(
    state: &Arc<ServerState>,
    repo: &Path,
    owner_client_id: &str,
    engine: Arc<car_inference::InferenceEngine>,
    generator: Arc<dyn TurnGenerator>,
) -> Result<Value, String> {
    open_discussion(
        state,
        repo,
        owner_client_id,
        engine,
        generator,
        owner_client_id,
        None,
    )
    .await
}

pub(super) async fn open_discussion(
    state: &Arc<ServerState>,
    repo: &Path,
    owner_client_id: &str,
    engine: Arc<car_inference::InferenceEngine>,
    generator: Arc<dyn TurnGenerator>,
    principal: &str,
    resume_id: Option<&str>,
) -> Result<Value, String> {
    open_discussion_with_model(
        state,
        repo,
        owner_client_id,
        engine,
        generator,
        principal,
        resume_id,
        None,
    )
    .await
}

#[allow(clippy::too_many_arguments)]
async fn open_discussion_with_model(
    state: &Arc<ServerState>,
    repo: &Path,
    owner_client_id: &str,
    engine: Arc<car_inference::InferenceEngine>,
    generator: Arc<dyn TurnGenerator>,
    principal: &str,
    resume_id: Option<&str>,
    requested_model: Option<&str>,
) -> Result<Value, String> {
    let _recovery = match resume_id {
        Some(_) => Some(state.coder_discussion_recovery.lock().await),
        None => None,
    };
    let saved = resume_id
        .map(|id| DiscussionRecord::load(&state.journal_dir, id, principal))
        .transpose()?;
    if let Some(record) = &saved {
        if state
            .coder_discussions
            .lock()
            .await
            .contains_key(&record.id)
        {
            return Err(
                "conversation is already open; close its other window before resuming".into(),
            );
        }
    }
    // `canonicalize` and the `git rev-parse` probe are blocking syscalls (the
    // probe forks), so they go to a blocking worker rather than parking a tokio
    // runtime thread on fork/exec.
    let probe = repo.to_path_buf();
    let repo = tokio::task::spawn_blocking(move || {
        // The same root `coder.start` keys tasks by; a subdirectory here would
        // fail the conversation's own task admission (repo mismatch).
        super::rpc::repo_toplevel(&probe)
            .map_err(|e| format!("{e} — discuss needs a repo to ground itself in"))
    })
    .await
    .map_err(|e| format!("repo probe failed: {e}"))??;

    if saved.as_ref().is_some_and(|record| record.repo != repo) {
        return Err("saved conversation belongs to a different repository".into());
    }
    if let (Some(record), Some(requested)) = (&saved, requested_model) {
        let requested = requested.trim();
        let requested = if requested.is_empty() || requested == "auto" {
            None
        } else {
            Some(requested)
        };
        if requested != record.model.as_deref() {
            let runs =
                coding_runs(state, &record.id, &repo, super::rpc::coder_state_dir()?).await?;
            if runs.iter().any(|run| {
                !matches!(
                    run["state"].as_str(),
                    Some("merged" | "reported" | "failed" | "abandoned")
                )
            }) {
                return Err(
                    "Finish or stop this conversation's active task before changing its model."
                        .into(),
                );
            }
        }
    }

    // Reap idle discussions before enforcing the cap, so a forgotten one from
    // this morning never blocks a new one this afternoon.
    reap_idle(state).await;
    // RESERVE the slot before any of the work below. Counting the registry here
    // and inserting after `bind_default_substrate` + `build_assistant_runtime`
    // was a TOCTOU check: the daemon runs a connection's requests concurrently,
    // so N pipelined starts all read the same count, all passed, and all built
    // a runtime — the cap bounded nothing. The permit lives in the entry and
    // comes back if any step below fails.
    let slot = state
        .coder_discussion_slots
        .clone()
        .try_acquire_owned()
        .map_err(|_| {
            format!(
                "{MAX_OPEN_DISCUSSIONS} discussions are already open — close one with \
                 coder.discuss.close before starting another"
            )
        })?;

    let summarize = repo.clone();
    let (repo_summary, project_context) = tokio::task::spawn_blocking(move || {
        (
            super::rpc::summarize_repo(&summarize),
            super::project_context::project_context(&summarize).unwrap_or_default(),
        )
    })
    .await
    .map_err(|e| format!("repo context failed: {e}"))?;

    // prefer_local = true, full_access = false ⇒ PermissionTier::ReadOnly:
    // every write/shell escalates to the approval gate, which this surface
    // auto-denies (see the `approval_pending` arm in `run_turn`). No Docker
    // preflight either — a discussion must open promptly.
    let mut env = bind_default_substrate(true, false, &repo, None).await;
    // ...and the read tools are pinned to the repo too. Mutation-gating alone
    // left `read_file`/`list_dir`/`find_files`/`grep_files` pointed at the
    // whole filesystem, whose output streams to every subscriber.
    env.clamp_reads = true;
    let task_proposal = Arc::new(StdMutex::new(None));
    let proposal_tool = Arc::new(super::task_proposal::TaskProposalTool(
        task_proposal.clone(),
    ));
    let mut asm = build_assistant_runtime_with_tools(
        engine.clone(),
        env,
        None,
        None,
        None,
        None,
        false,
        vec![super::task_proposal::TaskProposalTool::definition()],
        vec![proposal_tool],
    )
    .await?;
    // Advertise the tools useful before a coding task starts. The general
    // assistant also knows about mail, calendars, media and file mutations;
    // offering those here invites calls this conversation will refuse and
    // needlessly enlarges every inference request. Keep the shared executor
    // and approval gates intact for stale/hallucinated calls.
    asm.tools.retain(|tool| {
        matches!(
            tool["name"].as_str(),
            Some(
                "read_file"
                    | "list_dir"
                    | "find_files"
                    | "grep_files"
                    | "calculate"
                    | "web_search"
                    | "http_request"
                    | "prepare_coding_task"
            )
        )
    });
    let system = format!(
        "{}\n\n{project_context}\n\nRepository guidance applies to your analysis and proposed work; it does not expand the read-only permissions below.\n\nYou are the coding assistant for repository {}. \
         When the user asks you to implement or fix something, call prepare_coding_task with \
         their requested change and constraints. This is how you begin implementation from \
         this conversation. The interface will show an editable task and verification review \
         before execution. For questions and planning-only requests, inspect the repository \
         and answer without preparing a task. Use real paths. \
         The current conversation tools can inspect the repository and prepare coding tasks; \
         file edits and shell commands run in the subsequent coding task. A refused write \
         in this stage is NOT evidence that the file is read-only or that the user declined \
         implementation. Do not repeat earlier file-permission claims without current evidence \
         or ask the user to change permissions to start coding. Use prepare_coding_task instead. \
         Reads outside the repository remain refused. Do not require the user to know a \
         command or runtime API. Preparing is not execution: never say files have changed \
         or work has started.",
        prompt::chat_prompt(&asm.identity, &asm.description, &asm.tools),
        repo.display()
    );
    let model = match requested_model {
        Some(value) if value.trim().is_empty() || value.trim() == "auto" => None,
        Some(value) => Some(value.trim().to_string()),
        None => saved.as_ref().and_then(|record| record.model.clone()),
    };
    if let Some(model) = &model {
        let schema = engine.model_schema(model).ok_or_else(|| {
            format!("Unknown model '{model}'. Use `car models list --capability tool_use` to choose a model, or --model auto to clear the saved choice.")
        })?;
        if !schema
            .capabilities
            .contains(&car_inference::schema::ModelCapability::ToolUse)
        {
            return Err(format!(
                "Model '{model}' cannot call repository tools. Choose a tool-capable model with `car models list --capability tool_use`, or use --model auto."
            ));
        }
    }
    let generator: Arc<dyn TurnGenerator> = match &model {
        Some(model) => Arc::new(super::discussion_model::DiscussionModel {
            inner: generator,
            model: model.clone(),
        }),
        None => generator,
    };
    let cfg = AssistantConfig {
        model: model.clone(),
        strict_model: model.is_some(),
        max_turns: DISCUSS_MAX_TURNS,
        tools: asm.tools.clone(),
        gated_tools: asm.gated_tools.clone(),
        approval_policy: None,
        // A discussion writes nothing — including durable memory. Leaving the
        // proactive-memory bank unbound keeps `remember` out of the loop's
        // automatic pass; the tool itself is gated and auto-denied anyway.
        proactive_memory: None,
        tool_memory: None,
        tool_labels: None,
        // A discussion has no task list: it executes nothing, so there is no
        // run for #814's per-turn state block to describe.
        todos: None,
        // The shipped default, like every other production call site — #813's
        // A/B has been run and chose it; a discussion is not where that gets
        // re-decided.
        value_store_previews: crate::assistant::agent_loop::VALUE_STORE_PREVIEWS_DEFAULT,
        response_format: None,
        context_window_override: None,
        refuse_unadvertised_tools: false,
        response_format_validator: None,
        delegate_budget: None,
    };
    let mut record = saved.unwrap_or_else(|| DiscussionRecord {
        id: format!("disc-{}", uuid::Uuid::new_v4().simple()),
        repo: repo.clone(),
        principal: principal.to_string(),
        created_at: now_secs(),
        model: model.clone(),
        pending_task: None,
    });
    record.model = model.clone();
    let id = record.id.clone();
    let durability = Arc::new(crate::assistant::durability::LocalAssistantDurability::new(
        state.sync_subsystem()?,
        id.clone(),
        repo.clone(),
    ));
    let mut messages = if resume_id.is_some() {
        durability
            .load_checkpoint(&id)
            .await?
            .ok_or("saved conversation has no durable transcript")?
            .messages
    } else {
        Vec::new()
    };
    // Rebind current runtime instructions while retaining exact provider/tool
    // history. Interrupted tool exchanges are reconciled by AssistantService.
    if let Some(car_inference::Message::System { content }) = messages.first_mut() {
        *content = system.clone();
    } else {
        messages.insert(
            0,
            car_inference::Message::System {
                content: system.clone(),
            },
        );
    }
    durability
        .checkpoint(
            &id,
            &messages,
            if resume_id.is_some() {
                "conversation_resume"
            } else {
                "conversation_open"
            },
            None,
        )
        .await?;
    if resume_id.is_none() {
        record.save(&state.journal_dir)?;
    } else if requested_model.is_some() {
        record.save_model(&state.journal_dir)?;
    }
    let service = Arc::new(AssistantService::new_durable(
        generator.clone(),
        Arc::new(asm.runtime),
        cfg,
        system,
        durability.clone(),
        repo.clone(),
    ));

    let events = Arc::new(tokio::sync::Mutex::new(Vec::new()));
    let cmds = spawn_discuss_drain(id.clone(), events.clone());
    let entry = Arc::new(DiscussionEntry {
        id: id.clone(),
        repo: repo.clone(),
        repo_summary: repo_summary.clone(),
        project_context,
        created_at: record.created_at,
        model: model.clone(),
        owner_client_id: owner_client_id.to_string(),
        principal: principal.to_string(),
        record_root: state.journal_dir.clone(),
        events,
        cmds,
        turns: AtomicU64::new(0),
        in_flight: AtomicBool::new(false),
        start_lock: Arc::new(tokio::sync::Mutex::new(())),
        last_active: AtomicU64::new(now_secs()),
        turn_task: StdMutex::new(TurnSlot::default()),
        _slot: slot,
        service,
        durability,
        generator,
        transcript: StdMutex::new(Vec::new()),
        last_promote: StdMutex::new(record.pending_task.clone()),
        task_proposal,
    });
    // Project recorded conversation and tool activity without executing it.
    // Checkpoints do not retain a typed success flag for every tool result, so
    // do not invent result status from arbitrary tool-output text. Full history
    // beyond model-context compaction still requires the presentation log.
    for message in messages {
        let (role, kind, text) = match message {
            car_inference::Message::User { content } => ("Operator", true, content),
            car_inference::Message::Assistant {
                content,
                tool_calls,
                ..
            } if tool_calls.is_empty() => ("Assistant", false, content),
            car_inference::Message::Assistant { tool_calls, .. } => {
                for call in tool_calls {
                    entry
                        .emit(DiscussEventKind::ToolCall {
                            tool: call.name,
                            params_preview: preview(
                                &serde_json::to_string(&call.arguments).unwrap_or_default(),
                            ),
                        })
                        .await;
                }
                continue;
            }
            _ => continue,
        };
        if text.trim().is_empty() {
            continue;
        }
        entry.record_turn(role, &text);
        if kind {
            entry.emit(DiscussEventKind::UserMessage { text }).await;
        } else {
            entry.turns.fetch_add(1, Ordering::SeqCst);
            entry
                .emit(DiscussEventKind::AssistantMessage { text })
                .await;
        }
    }
    if let Some((proposed_intent, constraints)) = record.pending_task {
        entry
            .emit(DiscussEventKind::TaskPrepared {
                proposed_intent,
                constraints,
            })
            .await;
    }
    state
        .coder_discussions
        .lock()
        .await
        .insert(id.clone(), entry);

    Ok(json!({
        "discussion_id": id,
        "repo": repo,
        "repo_summary": repo_summary,
        "persistent": true,
        "resumed": resume_id.is_some(),
        "model": model,
    }))
}

/// Close discussions idle past [`DISCUSSION_IDLE_TTL_SECS`].
async fn reap_idle(state: &Arc<ServerState>) {
    let stale: Vec<Arc<DiscussionEntry>> = {
        let open = state.coder_discussions.lock().await;
        open.values()
            .filter(|e| e.idle_secs() > DISCUSSION_IDLE_TTL_SECS)
            .cloned()
            .collect()
    };
    for entry in stale {
        entry.cancel_turn();
        state.coder_discussions.lock().await.remove(&entry.id);
    }
}

async fn get_discussion(
    state: &Arc<ServerState>,
    discussion_id: &str,
) -> Result<Arc<DiscussionEntry>, String> {
    state
        .coder_discussions
        .lock()
        .await
        .get(discussion_id)
        .cloned()
        .ok_or_else(|| {
            format!(
                "no open discussion '{discussion_id}' — reopen a saved conversation with \
                 coder.discuss.start {{repo, resume_id}}, or start a new one"
            )
        })
}

pub(super) async fn selected_model(
    state: &Arc<ServerState>,
    discussion_id: &str,
) -> Result<Option<String>, String> {
    Ok(get_discussion(state, discussion_id).await?.model.clone())
}

/// Resolve a discussion **and prove the caller owns it**.
///
/// Ownership was recorded but only ever consulted by disconnect teardown, so
/// every `coder.discuss.*` method resolved by id alone: any connected client
/// could send into, subscribe to, promote, or close another connection's
/// discussion — closing one mid-turn was the sharp end, since it cancels a turn
/// the owner is watching. Discussions are already per-connection and die with
/// their connection, so refusing here is the same model, enforced.
pub(crate) async fn get_owned_discussion(
    state: &Arc<ServerState>,
    discussion_id: &str,
    client_id: &str,
) -> Result<Arc<DiscussionEntry>, String> {
    let entry = get_discussion(state, discussion_id).await?;
    if entry.owner_client_id != client_id {
        return Err(format!(
            "discussion '{discussion_id}' belongs to another connection — a discussion is \
             owned by the connection that opened it and closes with it; start your own with \
             coder.discuss.start"
        ));
    }
    Ok(entry)
}

/// Send one operator message and run the reply turn.
///
/// Returns once the turn is dispatched and has emitted its first event,
/// carrying that event's `seq` (the `user_message`), so a caller that has not
/// yet subscribed can resume from exactly there without missing or replaying a
/// frame. A refused dispatch emits nothing at all.
///
/// **One turn at a time.** A `send` arriving while a turn is in flight is
/// REFUSED, not queued: both turns clone the same model thread and the last one
/// to finish overwrites the other, so the earlier exchange vanishes from the
/// conversation — and from what `promote` later distills. Refusing is the
/// honest answer; the caller retries when `turn_complete` lands.
pub async fn send_message(
    state: &Arc<ServerState>,
    discussion_id: &str,
    client_id: &str,
    text: &str,
) -> Result<Value, String> {
    let entry = get_owned_discussion(state, discussion_id, client_id).await?;
    if text.trim().is_empty() {
        return Err("discuss message is empty".to_string());
    }
    if text.len() > DISCUSS_MESSAGE_MAX_BYTES {
        return Err(format!(
            "that message is {} bytes; the limit is {DISCUSS_MESSAGE_MAX_BYTES}. A discussion \
             keeps every message in its transcript, its replay buffer, and its distillation \
             prompt — point at a file in the repo instead of pasting it",
            text.len()
        ));
    }
    if entry
        .in_flight
        .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
        .is_err()
    {
        return Err(format!(
            "{discussion_id} is still answering the previous message — wait for \
             `turn_complete` before sending another"
        ));
    }
    // Armed IMMEDIATELY after the CAS: if this future is dropped before the
    // turn owns it, the guard's Drop is the only thing that stops the
    // discussion latching "still answering" forever with nothing running.
    //
    // Held in an `Option` so a REFUSED dispatch leaves it here rather than
    // dropping it inside the `spawn_turn(…)` expression: it then drops at this
    // function's scope exit, AFTER `recorded` (declared below, so it drops
    // first) has rolled the transcript row back. Otherwise `in_flight` reads
    // false while the stranded operator row is still visible — the reverse of
    // the cancellation path's order.
    let mut guard = Some(InFlightGuard(entry.clone()));
    entry.touch();
    entry.record_turn("Operator", text);
    // ...and armed with it, for the same reason: a dispatch refused by a racing
    // `close` must not leave the transcript ending in an operator question no
    // turn will ever answer.
    let mut recorded = TurnRecordGuard {
        entry: entry.clone(),
        text: text.to_string(),
        dispatched: false,
    };

    let task_entry = entry.clone();
    let task_state = state.clone();
    let prompt_text = text.to_string();
    // The `user_message` is emitted INSIDE the turn, as its first act — not
    // here, before the dispatch is known to have happened. Emitting it first
    // put it in the replay buffer and on every subscriber even when the
    // dispatch was refused and `TurnRecordGuard` rolled the transcript row
    // back: the board then rendered the operator's question followed by
    // permanent silence. Emitting from the turn makes the event and the
    // transcript row commit or roll back together, and makes the ordering
    // (`user_message` before any assistant delta for this turn) structural
    // rather than a scheduling accident.
    let (seq_tx, seq_rx) = oneshot::channel::<u64>();
    // Spawned under the turn-slot lock, so a `close` that raced this dispatch
    // either aborts the turn or stops it being spawned at all.
    let dispatched = entry.spawn_turn(|| {
        // Invalidate before spawning: a fast reply can prepare the next task
        // before send_message returns, and must not have its result erased.
        *lock(&entry.last_promote) = None;
        let guard = guard.take();
        tokio::spawn(async move {
            // The guard moves into the turn; it releases `in_flight` when the
            // turn ends, is aborted, or panics.
            let _guard = guard;
            let seq = task_entry
                .emit(DiscussEventKind::UserMessage {
                    text: prompt_text.clone(),
                })
                .await;
            let _ = seq_tx.send(seq);
            run_turn(task_state, task_entry, prompt_text).await;
        })
    });
    if !dispatched {
        return Err(format!(
            "{discussion_id} was closed while your message was being dispatched — nothing is \
             running; start a new discussion"
        ));
    }
    // A turn is running for this message now, so the transcript row stays.
    recorded.dispatched = true;

    // The turn's first event, reported so a caller that has not yet subscribed
    // can resume from exactly there. 0 if the turn was aborted before it got
    // that far — same answer `emit` gives when the drain is already gone.
    let first_seq = seq_rx.await.unwrap_or(0);
    Ok(json!({ "ok": true, "seq": first_seq }))
}

/// Read the same live/persisted session records used by the coding UI. This is
/// a projection, not a second task store. Binding both repository and discussion
/// prevents unrelated task history from entering a model request.
async fn coding_context(
    state: &Arc<ServerState>,
    entry: &DiscussionEntry,
) -> Result<String, String> {
    let dir = super::rpc::coder_state_dir()?;
    let mut rows = coding_runs(state, &entry.id, &entry.repo, dir).await?;
    let total = rows.len();
    rows.truncate(8);
    if total > rows.len() {
        rows.push(json!({"older_runs_omitted": total - rows.len()}));
    }
    if rows.is_empty() {
        return Ok(String::new());
    }
    Ok(format!(
        "Linked coding runs (current observations; older observations may be stale):\n{}\nThese runs use isolated worktrees. Repository reads still target the conversation repository; do not assume its checkout contains a run's changes. A passing check or published branch does not establish deployment. Ask for review of retained work before proposing to start over. After checkout delivery, a new task captures current checkout files, including later manual edits and deletions. After branch delivery, it starts from the recorded result commit. A retained native worktree with execution_stopped=true can be reopened by the next task. Other unfinished work still requires recovery checks.",
        serde_json::to_string(&rows).map_err(|e| e.to_string())?
    ))
}

async fn coding_runs(
    state: &Arc<ServerState>,
    discussion_id: &str,
    repo: &Path,
    dir: PathBuf,
) -> Result<Vec<Value>, String> {
    let entries: Vec<_> = state
        .coder_sessions
        .lock()
        .await
        .values()
        .cloned()
        .collect();
    let mut rows = Vec::new();
    let mut live_ids = std::collections::HashSet::new();
    for entry in entries {
        let session = entry.session.lock().await;
        live_ids.insert(session.id.clone());
        if session.discussion_id.as_deref() == Some(discussion_id) && session.repo == repo {
            rows.push(coding_run_row(&session, true));
        }
    }
    let discussion_id = discussion_id.to_string();
    let repo = repo.to_path_buf();
    let saved = tokio::task::spawn_blocking(move || {
        super::session::CoderSession::list(&dir)
            .into_iter()
            .filter(|session| {
                !live_ids.contains(&session.id)
                    && session.discussion_id.as_deref() == Some(discussion_id.as_str())
                    && session.repo == repo
            })
            .map(|session| coding_run_row(&session, false))
            .collect::<Vec<_>>()
    })
    .await
    .map_err(|e| format!("read linked coding runs: {e}"))?;
    rows.extend(saved);
    let superseded: std::collections::HashSet<String> = rows
        .iter()
        .filter_map(|row| row["resumed_from"].as_str().map(str::to_string))
        .collect();
    rows.retain(|row| {
        !row["session_id"]
            .as_str()
            .is_some_and(|id| superseded.contains(id))
    });
    rows.sort_by_key(|row| std::cmp::Reverse(row["updated_at"].as_u64().unwrap_or(0)));
    Ok(rows)
}

fn coding_run_row(session: &super::session::CoderSession, live: bool) -> Value {
    let omitted_guidance = session.steering_messages.len().saturating_sub(8);
    let guidance: Vec<Value> = session.steering_messages[omitted_guidance..]
        .iter()
        .map(|text| {
            json!({
                "text": text.chars().take(1000).collect::<String>(),
                "truncated": text.chars().count() > 1000,
            })
        })
        .collect();
    let checks: Vec<Value> = session.last_check_results.iter().take(20).map(|check| json!({
        "name": check.name.chars().take(160).collect::<String>(),
        "passed": check.passed,
        "exit_code": check.exit_code,
        "timed_out": check.timed_out,
        "deadline_clamped": check.deadline_clamped,
        "output_tail": check.output_tail.chars().rev().take(800).collect::<String>().chars().rev().collect::<String>(),
    })).collect();
    json!({
        "session_id": session.id, "state": session.state.as_str(), "live": live,
        "intent": session.intent.chars().take(1000).collect::<String>(),
        "updated_at": session.updated_at, "result_branch": session.result_branch,
        "result_commit": session.result_commit,
        "result_delivery": session.result_delivery,
        "resumed_from": session.resumed_from,
        "execution_stopped": session.execution_stopped,
        "operator_guidance": guidance,
        "operator_guidance_omitted": omitted_guidance,
        "engine": session.engine.label(),
        "worktree": session.workspace_path.as_ref().filter(|path| path.is_dir()),
        "error": session.error.as_ref().map(|error| error.chars().take(1000).collect::<String>()),
        "failure_kind": session.failure_kind, "checks": checks,
        "checks_omitted": session.last_check_results.len().saturating_sub(checks.len()),
    })
}

/// Recover only a native task whose execution returned or was joined. The
/// start admission guard must remain held through adoption and registration.
pub(super) async fn retained_workspace(
    state: &Arc<ServerState>,
    discussion_id: &str,
    repo: &Path,
    dir: PathBuf,
) -> Result<Option<(String, PathBuf)>, String> {
    let rows = coding_runs(state, discussion_id, repo, dir.clone()).await?;
    let retained: Vec<_> = rows
        .iter()
        .filter(|row| {
            matches!(row["state"].as_str(), Some("failed" | "abandoned"))
                && row["worktree"].is_string()
        })
        .collect();
    if retained.len() > 1 {
        return Err("Multiple unfinished worktrees belong to this conversation. Review them before choosing work to continue.".into());
    }
    let Some(row) = retained.first() else {
        return Ok(None);
    };
    if row["engine"] != "native" || row["execution_stopped"] != true {
        return Err("The previous task's execution has not been confirmed stopped. Its work is retained; automatic recovery cannot safely reopen it yet.".into());
    }
    let path = PathBuf::from(
        row["worktree"]
            .as_str()
            .ok_or("missing retained worktree")?,
    )
    .canonicalize()
    .map_err(|e| format!("retained worktree: {e}"))?;
    let root = dir
        .join("worktrees")
        .canonicalize()
        .map_err(|e| e.to_string())?;
    if path.parent() != Some(root.as_path()) {
        return Err("retained worktree is outside this daemon's workspace directory".into());
    }
    Ok(Some((
        row["session_id"]
            .as_str()
            .ok_or("missing retained task id")?
            .to_string(),
        path,
    )))
}

/// Branch deliveries continue from the saved revision. Checkout deliveries
/// continue from current files (including later user edits), captured by task
/// admission. Explicit caller bases win; legacy unknown results still refuse.
pub(super) async fn followup_base(
    state: &Arc<ServerState>,
    discussion_id: &str,
    repo: &Path,
    dir: PathBuf,
) -> Result<Option<String>, String> {
    let rows = coding_runs(state, discussion_id, repo, dir).await?;
    followup_base_from(&rows)
}

fn followup_base_from(rows: &[Value]) -> Result<Option<String>, String> {
    for run in rows {
        match run["state"].as_str() {
            Some("merged") if rows.iter().any(|other| other["state"] == "merged"
                && other["updated_at"] == run["updated_at"]
                && other["result_commit"] != run["result_commit"]) => {
                return Err("Multiple deliveries have the same recorded timestamp. Choose an explicit base revision; CAR cannot safely infer their order.".into());
            }
            Some("merged") => {
                let commit = run["result_commit"].as_str().ok_or_else(|| "The previous run predates saved result revisions. Choose an explicit base revision before continuing; CAR will not silently start again from repository HEAD.".to_string())?;
                // Applied results already live in the checkout. Reusing their
                // old tree would erase later user edits/deletions from the next
                // task's view. None asks admission to capture current inputs.
                return Ok((run["result_delivery"] != "checkout").then(|| commit.to_string()));
            },
            Some("failed" | "abandoned") if !run["worktree"].is_null() => return Err(format!(
                "The previous run has unfinished changes at {}. Review that work before starting over; resuming that worktree is not yet supported.", run["worktree"].as_str().unwrap_or("the retained worktree")
            )),
            _ => {},
        }
    }
    Ok(None)
}

/// Serialize task admission for one conversation through registration. The
/// caller holds this guard until start_session has published its live entry.
/// Unlike prompt advice, this prevents two concurrent builds of the same turn.
pub(super) async fn claim_coding_start(
    state: &Arc<ServerState>,
    discussion_id: &str,
    repo: &Path,
    dir: PathBuf,
) -> Result<tokio::sync::OwnedMutexGuard<()>, String> {
    if get_discussion(state, discussion_id).await?.repo != repo {
        return Err("The task repository differs from this conversation's repository.".into());
    }
    claim_coding_start_at(state, discussion_id, dir).await
}

async fn claim_coding_start_at(
    state: &Arc<ServerState>,
    discussion_id: &str,
    dir: PathBuf,
) -> Result<tokio::sync::OwnedMutexGuard<()>, String> {
    let entry = get_discussion(state, discussion_id).await?;
    let guard = entry.start_lock.clone().lock_owned().await;
    // Recheck after waiting: closing a discussion must not leave a queued
    // launch with a stale entry that is no longer authorized to start.
    let current = get_discussion(state, discussion_id).await?;
    if !Arc::ptr_eq(&entry, &current) {
        return Err("conversation was reopened; retry the task".into());
    }
    let runs = coding_runs(state, discussion_id, &entry.repo, dir).await?;
    if let Some(run) = runs.iter().find(|row| {
        row.get("state").is_some_and(|state| {
            !matches!(
                state.as_str(),
                Some("merged" | "reported" | "failed" | "abandoned")
            )
        })
    }) {
        return Err(format!("Conversation already has unfinished work in {} ({}). Open it from /sessions to continue or cancel it before starting another task.", run["session_id"].as_str().unwrap_or("unknown"), run["state"].as_str().unwrap_or("unknown")));
    }
    Ok(guard)
}

/// Drive one assistant turn, translating its wire events into discussion
/// events and auto-denying every approval escalation.
async fn run_turn(state: Arc<ServerState>, entry: Arc<DiscussionEntry>, text: String) {
    *lock(&entry.task_proposal) = None;
    if let Err(message) = entry.save_pending_task(None) {
        entry.emit(DiscussEventKind::Error { message }).await;
        entry.emit(DiscussEventKind::TurnComplete {}).await;
        return;
    }
    let context = match coding_context(&state, &entry).await {
        Ok(context) => context,
        Err(error) => {
            entry.emit(DiscussEventKind::Error { message: error }).await;
            entry.emit(DiscussEventKind::TurnComplete {}).await;
            return;
        }
    };
    let sink_entry = entry.clone();
    let assembled: Arc<StdMutex<String>> = Arc::new(StdMutex::new(String::new()));
    let sink_assembled = assembled.clone();

    let service = entry.service.clone();
    // The sink resolves approvals on the same service it streams from, so it
    // needs its own handle rather than borrowing the one being called.
    let sink_service = service.clone();
    let id = entry.id.clone();
    service
        .handle_turn_with_context(
            &id,
            &text,
            None,
            None,
            Some(&context),
            move |payload: Value| {
                let entry = sink_entry.clone();
                let assembled = sink_assembled.clone();
                let service = sink_service.clone();
                async move {
                    let kind = payload.get("kind").and_then(Value::as_str).unwrap_or("");
                    match kind {
                        "token" => {
                            let delta = payload
                                .get("delta")
                                .and_then(Value::as_str)
                                .unwrap_or_default()
                                .to_string();
                            if delta.is_empty() {
                                return;
                            }
                            lock(&assembled).push_str(&delta);
                            entry
                                .emit(DiscussEventKind::AssistantDelta { text: delta })
                                .await;
                        }
                        "tool_call" => {
                            let tool = payload
                                .get("tool")
                                .and_then(Value::as_str)
                                .unwrap_or("tool")
                                .to_string();
                            let params_preview = payload
                                .get("params")
                                .map(|p| preview(&p.to_string()))
                                .unwrap_or_default();
                            entry
                                .emit(DiscussEventKind::ToolCall {
                                    tool,
                                    params_preview,
                                })
                                .await;
                        }
                        // The no-mutation boundary, enforced here rather than left
                        // to a human: a discussion never writes, so an escalation is
                        // answered immediately with "no" instead of parking a
                        // prompt nobody asked for (and timing out five minutes
                        // later, which is what the unresolved gate would do).
                        "approval_pending" => {
                            let tool = payload
                                .get("tool")
                                .and_then(Value::as_str)
                                .unwrap_or("tool")
                                .to_string();
                            if let Some(approval_id) =
                                payload.get("approval_id").and_then(Value::as_str)
                            {
                                service.deny_approval(
                                    approval_id,
                                    DISCUSSION_MUTATION_REFUSAL.to_string(),
                                );
                            }
                            entry
                                .emit(DiscussEventKind::ToolResult {
                                    tool,
                                    ok: false,
                                    preview: DISCUSSION_MUTATION_REFUSAL.to_string(),
                                })
                                .await;
                        }
                        "done" => {
                            let text = payload
                                .get("text")
                                .and_then(Value::as_str)
                                .unwrap_or_default()
                                .to_string();
                            let text = if text.trim().is_empty() {
                                lock(&assembled).clone()
                            } else {
                                text
                            };
                            entry.record_turn("Assistant", &text);
                            entry.turns.fetch_add(1, Ordering::SeqCst);
                            entry
                                .emit(DiscussEventKind::AssistantMessage { text })
                                .await;
                            let proposal = lock(&entry.task_proposal).take();
                            if let Some((proposed_intent, constraints)) = proposal {
                                if let Err(message) = entry.save_pending_task(Some((
                                    proposed_intent.clone(),
                                    constraints.clone(),
                                ))) {
                                    entry.emit(DiscussEventKind::Error { message }).await;
                                    entry.emit(DiscussEventKind::TurnComplete {}).await;
                                    return;
                                }
                                *lock(&entry.last_promote) =
                                    Some((proposed_intent.clone(), constraints.clone()));
                                entry
                                    .emit(DiscussEventKind::TaskPrepared {
                                        proposed_intent,
                                        constraints,
                                    })
                                    .await;
                            }
                            entry.emit(DiscussEventKind::TurnComplete {}).await;
                        }
                        "error" => {
                            let message = payload
                                .get("error")
                                .and_then(Value::as_str)
                                .unwrap_or("discussion turn failed")
                                .to_string();
                            entry.emit(DiscussEventKind::Error { message }).await;
                            entry.emit(DiscussEventKind::TurnComplete {}).await;
                        }
                        // The third terminal kind. A discussion runs the same
                        // `AssistantService` as chat, so a turn can be refused on
                        // the Parslee account here too — and without this arm the
                        // frame fell through to `_ => {}`: no error, no remedy, and
                        // crucially no `TurnComplete`, which is the event
                        // `docs/websocket-protocol.md` tells a client to wait for
                        // before sending again. The client was left holding a turn
                        // that had already ended.
                        //
                        // Rendered through `Error` rather than a new variant: the
                        // message IS the remedy, in the copy a person reads, and a
                        // client that already renders discussion errors shows it
                        // without changing.
                        "auth_required" => {
                            let message = payload
                                .get("message")
                                .and_then(Value::as_str)
                                .unwrap_or("this discussion needs a Parslee sign-in")
                                .to_string();
                            entry.emit(DiscussEventKind::Error { message }).await;
                            entry.emit(DiscussEventKind::TurnComplete {}).await;
                        }
                        _ => {}
                    }
                }
            },
        )
        .await;
}

fn preview(s: &str) -> String {
    const CAP: usize = 200;
    if s.chars().count() <= CAP {
        return s.to_string();
    }
    let mut out: String = s.chars().take(CAP).collect();
    out.push('…');
    out
}

/// Distill the discussion into a run intent + the constraints agreed in it.
///
/// **Starts nothing.** No worktree, no branch, no session — the caller shows
/// `proposed_intent` to the operator, who may edit it before calling
/// `coder.start`. Callable repeatedly on an open discussion.
///
/// Refuses while a turn is streaming: distilling then would run on the
/// operator's question with no answer beside it, and the model would happily
/// invent a confident intent from an unanswered question — which then feeds
/// `coder.start { discussion_id }` and contract derivation.
pub async fn promote(
    state: &Arc<ServerState>,
    discussion_id: &str,
    client_id: &str,
) -> Result<Value, String> {
    let entry = get_owned_discussion(state, discussion_id, client_id).await?;
    if entry.is_answering() {
        return Err(format!(
            "{discussion_id} is still answering — try again in a moment"
        ));
    }
    if entry.transcript_is_empty() {
        return Err(
            "this discussion has no turns yet — say what you are trying to do first".to_string(),
        );
    }
    let _start_guard = claim_coding_start(
        state,
        discussion_id,
        &entry.repo,
        super::rpc::coder_state_dir()?,
    )
    .await?;
    let context = coding_context(state, &entry).await?;
    let repo_context = format!(
        "{}\n{}\n{}",
        entry.repo_summary, entry.project_context, context
    );
    let (intent, constraints) =
        distill(&entry.generator, &entry.distill_transcript(), &repo_context).await?;
    entry.save_pending_task(Some((intent.clone(), constraints.clone())))?;
    *lock(&entry.last_promote) = Some((intent.clone(), constraints.clone()));
    entry.touch();
    Ok(json!({
        "discussion_id": entry.id,
        "proposed_intent": intent,
        "constraints": constraints,
    }))
}

/// The distillation call. Generation is injected exactly the way
/// `derive_app_contract` injects it into `derive_contract`, so the prompt +
/// parse + bounded-retry shape is testable with a scripted model.
async fn distill(
    generator: &Arc<dyn TurnGenerator>,
    transcript: &str,
    repo_summary: &str,
) -> Result<(String, Vec<String>), String> {
    let mut last_err = String::from("no attempt was made");
    for _ in 0..PROMOTE_MAX_ATTEMPTS {
        let prompt = format!(
            "A developer has been discussing a change to a codebase. Distill the discussion \
             into ONE actionable coding intent plus the constraints they agreed on.\n\n\
             REPOSITORY\n{repo_summary}\n\n\
             DISCUSSION (most recent turns)\n{transcript}\n\n\
             Return ONLY a JSON object, no prose and no code fences:\n\
             {{\n  \"proposed_intent\": \"one paragraph, imperative, what to change and why\",\n  \
             \"constraints\": [\"a thing the change must not break or must respect\"]\n}}\n\n\
             Rules:\n\
             - `proposed_intent` is an INSTRUCTION, not a summary of the conversation. Never \
             quote the transcript back.\n\
             - Include only constraints actually agreed in the discussion. If none were, \
             return an empty array — do not invent any.\n"
        );
        let text = match generator
            .generate(car_inference::GenerateRequest {
                prompt,
                params: car_inference::GenerateParams {
                    temperature: 0.0,
                    max_tokens: 1024,
                    thinking: car_inference::tasks::generate::ThinkingMode::Off,
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
        {
            Ok(r) => r.text,
            Err(e) => {
                last_err = format!("generation failed: {e}");
                continue;
            }
        };
        let value = match super::contract::extract_json_object(&text) {
            Ok(v) => v,
            Err(e) => {
                last_err = format!("output did not parse: {e}");
                continue;
            }
        };
        let intent = value
            .get("proposed_intent")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .trim()
            .to_string();
        if intent.is_empty() {
            last_err = "the model returned no proposed_intent".to_string();
            continue;
        }
        let constraints: Vec<String> = value
            .get("constraints")
            .and_then(Value::as_array)
            .map(|a| {
                a.iter()
                    .filter_map(Value::as_str)
                    .map(str::trim)
                    .filter(|s| !s.is_empty())
                    .map(str::to_string)
                    .collect()
            })
            .unwrap_or_default();
        return Ok((intent, constraints));
    }
    Err(format!(
        "could not distill this discussion into an intent after {PROMOTE_MAX_ATTEMPTS} \
         attempts: {last_err}"
    ))
}

/// Accepting a prepared task consumes its saved editor before task creation.
/// A failed start leaves the current client editor available for retry.
pub(super) async fn consume_prepared_task(
    state: &Arc<ServerState>,
    id: &str,
) -> Result<(), String> {
    get_discussion(state, id).await?.save_pending_task(None)
}

/// Constraints to fold into `derive_contract` for a `coder.start
/// { discussion_id }`.
///
/// An unknown id is a hard error — a run that silently drops its grounding is
/// worse than one that refuses to start. A distillation failure also refuses
/// the start: the supplied intent need not repeat every preference agreed in
/// the conversation. The operator can retry without losing that grounding.
///
/// Refused while a turn is streaming, for the same reason `promote` is: the
/// distillation would run on the operator's question with no answer beside it,
/// and these constraints go straight into contract derivation.
pub async fn constraints_for_start(
    state: &Arc<ServerState>,
    discussion_id: &str,
) -> Result<Vec<String>, String> {
    let entry = get_discussion(state, discussion_id).await?;
    if entry.is_answering() {
        return Err(format!(
            "{discussion_id} is still answering — wait for `turn_complete` before starting a \
             run from it, or the constraints would be distilled from a question with no \
             answer beside it"
        ));
    }
    let cached = entry.constraints();
    if !cached.is_empty() {
        return Ok(cached);
    }
    if lock(&entry.last_promote).is_some() {
        // Promoted already, and it genuinely agreed no constraints.
        return Ok(Vec::new());
    }
    if entry.transcript_is_empty() {
        return Ok(Vec::new());
    }
    match distill(
        &entry.generator,
        &entry.distill_transcript(),
        &format!("{}\n{}", entry.repo_summary, entry.project_context),
    )
    .await
    {
        Ok((intent, constraints)) => {
            *lock(&entry.last_promote) = Some((intent, constraints.clone()));
            Ok(constraints)
        }
        Err(e) => {
            tracing::warn!(discussion_id, "discussion constraints unavailable: {e}");
            Err(format!(
                "Could not carry the conversation's requirements into this task: {e}. \
                 No task was started. Retry Build when model access is available."
            ))
        }
    }
}

/// Close a discussion: cancel any in-flight turn, free its runtime, end its
/// drain.
pub async fn close(
    state: &Arc<ServerState>,
    discussion_id: &str,
    client_id: &str,
) -> Result<Value, String> {
    // Ownership first, and against the live registry: a foreign `close` must
    // not be able to cancel a turn its owner is watching.
    get_owned_discussion(state, discussion_id, client_id).await?;
    let entry = state.coder_discussions.lock().await.remove(discussion_id);
    let Some(entry) = entry else {
        return Err(format!("no open discussion '{discussion_id}'"));
    };
    // Actually stop the model: without this the turn keeps running against a
    // live provider, billing tokens to a conversation nobody can read. This
    // also latches the discussion closed, so a `send` parked mid-dispatch never
    // spawns its turn behind us.
    entry.cancel_turn();
    Ok(json!({ "ok": true }))
}

/// Drop a disconnecting client's discussion state (called from
/// `remove_session`).
///
/// A discussion is owned by the connection that opened it (module docs), so
/// this closes it outright rather than only unsubscribing — otherwise every
/// closed board leaks an `AssistantService`, a `Runtime`, an open runtime
/// session, and an unbounded transcript for the daemon's lifetime. Other
/// clients' subscriptions to a surviving discussion are just detached.
pub async fn drop_subscriptions_for_client(state: &ServerState, client_id: &str) {
    let (owned, others): (Vec<_>, Vec<_>) = {
        let open = state.coder_discussions.lock().await;
        open.values()
            .cloned()
            .partition(|e| e.owner_client_id == client_id)
    };
    for entry in &others {
        let _ = entry.cmds.send(StreamCmd::Detach(client_id.to_string()));
    }
    if owned.is_empty() {
        return;
    }
    let mut open = state.coder_discussions.lock().await;
    for entry in owned {
        entry.cancel_turn();
        open.remove(&entry.id);
    }
}

// ---------------------------------------------------------------------------
// JSON-RPC handlers (thin parsing wrappers)
// ---------------------------------------------------------------------------

#[derive(Deserialize)]
struct StartParams {
    repo: PathBuf,
    #[serde(default)]
    resume_id: Option<String>,
    #[serde(default)]
    model: Option<String>,
}

pub async fn handle_discuss_start(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    let params: StartParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let engine = crate::handler::get_inference_engine(state).clone();
    let generator: Arc<dyn TurnGenerator> = engine.clone();
    if let Some(model) = params
        .model
        .as_deref()
        .map(str::trim)
        .filter(|m| !m.is_empty() && *m != "auto")
    {
        if !engine.knows_model(model) {
            return Err(format!(
                "Unknown model '{model}'. Use `car models list` to choose an available model id."
            ));
        }
    }
    // The daemon is single-principal for operator connections; supervised
    // agents have a separately authenticated stable identity. Never derive
    // archive ownership from a caller-supplied parameter or a connection UUID.
    let principal = discussion_principal(session).await;
    open_discussion_with_model(
        state,
        &params.repo,
        &session.client_id,
        engine,
        generator,
        &principal,
        params.resume_id.as_deref(),
        params.model.as_deref(),
    )
    .await
}

#[derive(Deserialize)]
struct SendParams {
    discussion_id: String,
    text: String,
}

pub async fn handle_discuss_send(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    let params: SendParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    send_message(
        state,
        &params.discussion_id,
        &session.client_id,
        &params.text,
    )
    .await
}

#[derive(Deserialize)]
struct DiscussionIdParams {
    discussion_id: String,
}

#[derive(Deserialize)]
struct SubscribeParams {
    discussion_id: String,
    #[serde(default)]
    from_seq: u64,
}

pub async fn handle_discuss_subscribe(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    let params: SubscribeParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let entry = get_owned_discussion(state, &params.discussion_id, &session.client_id).await?;
    // A read counts as activity: a discussion an operator is actively watching
    // must not be eligible for the idle reaper.
    entry.touch();
    // Replay + register happen inside the drain task, which is the only owner
    // — so they are ordered against live emits without holding a lock across
    // any send.
    let (tx, rx) = oneshot::channel();
    entry
        .cmds
        .send(StreamCmd::Attach {
            client_id: session.client_id.clone(),
            channel: session.channel.clone(),
            from_seq: params.from_seq,
            replayed: tx,
        })
        .map_err(|_| format!("discussion '{}' is closing", params.discussion_id))?;
    let replayed = rx.await.unwrap_or(0);
    Ok(json!({ "events_replayed": replayed }))
}

pub async fn handle_discuss_unsubscribe(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    let params: DiscussionIdParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    if let Ok(entry) = get_discussion(state, &params.discussion_id).await {
        let _ = entry
            .cmds
            .send(StreamCmd::Detach(session.client_id.clone()));
    }
    Ok(json!({ "ok": true }))
}

pub async fn handle_discuss_promote(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    let params: DiscussionIdParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    // Model generation has a deep polling stack in debug builds. Poll it in
    // its own task, rather than underneath the large RPC dispatch future.
    // JoinSet owns cancellation: a deadline or disconnected caller dropping
    // this handler also aborts generation instead of leaving a detached bill.
    let state = state.clone();
    let client_id = session.client_id.clone();
    let mut generation = tokio::task::JoinSet::new();
    generation.spawn(async move { promote(&state, &params.discussion_id, &client_id).await });
    generation
        .join_next()
        .await
        .ok_or("conversation planning task did not start")?
        .map_err(|error| format!("conversation planning task failed: {error}"))?
}

pub async fn handle_discuss_close(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    let params: DiscussionIdParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    close(state, &params.discussion_id, &session.client_id).await
}

async fn discussion_principal(session: &Arc<ClientSession>) -> String {
    match session.agent_id.lock().await.as_deref() {
        Some(id) => format!("agent:{id}"),
        None => "operator".to_string(),
    }
}

/// `coder.discuss.list` — this connection's open discussions, plus the
/// authenticated principal's saved conversations that are not currently open.
///
/// Scoped to the caller, like every other `coder.discuss.*` method: a
/// discussion is owned by the connection that opened it, and listing another
/// connection's discussions would hand out ids the caller cannot use anyway.
pub async fn handle_discuss_list(
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    let mut rows: Vec<Value> = state
        .coder_discussions
        .lock()
        .await
        .values()
        .filter(|e| e.owner_client_id == session.client_id)
        .map(|e| e.summary_row())
        .collect();
    rows.sort_by_key(|v| std::cmp::Reverse(v["created_at"].as_u64().unwrap_or(0)));
    let principal = discussion_principal(session).await;
    let records = DiscussionRecord::list(&state.journal_dir, &principal)?;
    let live = state.coder_discussions.lock().await;
    let saved: Vec<Value> = records
        .into_iter()
        .filter(|record| !live.contains_key(&record.id))
        .map(|record| {
            json!({
                "discussion_id": record.id,
                "repo": record.repo,
                "created_at": record.created_at,
            })
        })
        .collect();
    Ok(json!({ "discussions": rows, "saved": saved }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use car_inference::{GenerateRequest, InferenceResult};
    use std::sync::atomic::AtomicUsize;

    fn turn(text: &str, tool_calls: Value) -> InferenceResult {
        serde_json::from_value(json!({
            "text": text, "tool_calls": tool_calls,
            "trace_id": "t", "model_used": "scripted", "latency_ms": 0,
        }))
        .expect("scripted InferenceResult shape")
    }

    struct Script {
        turns: Vec<InferenceResult>,
        cursor: AtomicUsize,
    }

    #[async_trait]
    impl TurnGenerator for Script {
        async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
            let i = self.cursor.fetch_add(1, Ordering::SeqCst);
            self.turns
                .get(i)
                .cloned()
                .ok_or_else(|| "script exhausted".to_string())
        }
    }

    /// A generator that blocks until released — lets a test observe a turn
    /// while it is genuinely in flight.
    ///
    /// Released with `notify_one`, never `notify_waiters`: the turn is spawned,
    /// so the test can reach the release before the task has registered as a
    /// waiter, and `notify_waiters` wakes only waiters that already exist.
    /// `notify_one` stores a permit, so the ordering does not matter.
    struct Blocking {
        gate: Arc<tokio::sync::Notify>,
    }

    #[async_trait]
    impl TurnGenerator for Blocking {
        async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
            self.gate.notified().await;
            Ok(turn("done at last", json!([])))
        }
    }

    /// Counts invocations — for asserting a turn NEVER reached the model.
    struct Counting {
        calls: Arc<AtomicUsize>,
    }

    #[async_trait]
    impl TurnGenerator for Counting {
        async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Ok(turn("counted", json!([])))
        }
    }

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

    fn engine(root: &Path) -> Arc<car_inference::InferenceEngine> {
        let mut cfg = car_inference::InferenceConfig::default();
        cfg.models_dir = root.join("models");
        Arc::new(car_inference::InferenceEngine::new(cfg))
    }

    /// A standalone daemon state plus the journal dir it writes to — the
    /// caller keeps the `TempDir` alive for the length of the test.
    fn state() -> (Arc<ServerState>, tempfile::TempDir) {
        let journal = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));
        (state, journal)
    }

    async fn start(
        state: &Arc<ServerState>,
        repo: &Path,
        generator: Arc<dyn TurnGenerator>,
    ) -> String {
        let started = start_discussion(state, repo, "owner-1", engine(repo), generator)
            .await
            .unwrap();
        started["discussion_id"].as_str().unwrap().to_string()
    }

    /// A `ClientSession` over a drain sink — enough for the handlers that need
    /// a connection identity, without a tungstenite handshake.
    async fn client(state: &Arc<ServerState>, id: &str) -> Arc<ClientSession> {
        state
            .create_session(id, Arc::new(crate::session::WsChannel::test_stub()))
            .await
            .unwrap()
    }

    /// A WS sink that keeps every frame instead of writing it, so a test can
    /// read exactly what a subscriber's lane delivered. `test_stub` drains to
    /// nowhere, which is enough for membership checks but says nothing about
    /// what arrived.
    struct CaptureSink(Arc<StdMutex<Vec<String>>>);

    impl futures::Sink<tokio_tungstenite::tungstenite::Message> for CaptureSink {
        type Error = tokio_tungstenite::tungstenite::Error;

        fn poll_ready(
            self: std::pin::Pin<&mut Self>,
            _: &mut std::task::Context<'_>,
        ) -> std::task::Poll<Result<(), Self::Error>> {
            std::task::Poll::Ready(Ok(()))
        }

        fn start_send(
            self: std::pin::Pin<&mut Self>,
            item: tokio_tungstenite::tungstenite::Message,
        ) -> Result<(), Self::Error> {
            if let tokio_tungstenite::tungstenite::Message::Text(text) = item {
                lock(&self.0).push(text.to_string());
            }
            Ok(())
        }

        fn poll_flush(
            self: std::pin::Pin<&mut Self>,
            _: &mut std::task::Context<'_>,
        ) -> std::task::Poll<Result<(), Self::Error>> {
            std::task::Poll::Ready(Ok(()))
        }

        fn poll_close(
            self: std::pin::Pin<&mut Self>,
            _: &mut std::task::Context<'_>,
        ) -> std::task::Poll<Result<(), Self::Error>> {
            std::task::Poll::Ready(Ok(()))
        }
    }

    /// A real `WsChannel` over [`CaptureSink`], plus the frames it collected.
    /// Locking its `write` half is a half-open peer: writes stop completing and
    /// never fail, exactly what wedges a subscriber's lane.
    fn capturing_channel() -> (Arc<WsChannel>, Arc<StdMutex<Vec<String>>>) {
        let frames = Arc::new(StdMutex::new(Vec::new()));
        let sink: crate::session::WsSink = Box::pin(CaptureSink(frames.clone()));
        let channel = Arc::new(WsChannel {
            write: tokio::sync::Mutex::new(sink),
            pending: tokio::sync::Mutex::new(HashMap::new()),
            active_actions: tokio::sync::Mutex::new(HashMap::new()),
            next_id: AtomicU64::new(0),
        });
        (channel, frames)
    }

    fn rpc_req(params: Value) -> JsonRpcMessage {
        serde_json::from_value(json!({ "jsonrpc": "2.0", "id": 1, "params": params }))
            .expect("JsonRpcMessage shape")
    }

    /// The `seq` of every `coder.discuss.event` frame a lane delivered.
    fn delivered_seqs(frames: &Arc<StdMutex<Vec<String>>>) -> Vec<u64> {
        lock(frames)
            .iter()
            .map(|f| serde_json::from_str::<Value>(f).expect("a lane frame must be JSON"))
            .inspect(|v| assert_eq!(v["method"], "coder.discuss.event", "unexpected frame: {v}"))
            .map(|v| {
                v["params"]["seq"]
                    .as_u64()
                    .expect("every event carries a seq")
            })
            .collect()
    }

    async fn wait_for_turn_complete(entry: &Arc<DiscussionEntry>) {
        for _ in 0..400 {
            {
                let events = entry.events.lock().await;
                if events
                    .iter()
                    .any(|e| matches!(e.kind, DiscussEventKind::TurnComplete {}))
                {
                    return;
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        panic!("discussion turn never completed");
    }

    async fn wait_for_idle(entry: &Arc<DiscussionEntry>) {
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            while entry.is_answering() {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("discussion turn did not release its guard");
    }

    #[tokio::test]
    async fn conversation_recovers_after_restart_with_same_identity_and_full_model_history() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        std::fs::write(repo.path().join("AGENTS.md"), "Initial repository rule.").unwrap();
        std::fs::write(repo.path().join("CLAUDE.md"), "Preserve exported names.").unwrap();
        let (state, journal) = state();
        let script = || {
            Arc::new(Script {
                turns: vec![turn("Remember the export regression.", json!([]))],
                cursor: AtomicUsize::new(0),
            }) as Arc<dyn TurnGenerator>
        };
        let started = open_discussion(
            &state,
            repo.path(),
            "connection-1",
            engine(repo.path()),
            script(),
            "operator",
            None,
        )
        .await
        .unwrap();
        let id = started["discussion_id"].as_str().unwrap().to_string();
        let entry = get_discussion(&state, &id).await.unwrap();
        let checkpoint = entry
            .durability
            .load_checkpoint(&id)
            .await
            .unwrap()
            .unwrap();
        let system = serde_json::to_string(&checkpoint.messages[0]).unwrap();
        assert!(system.contains("Initial repository rule."));
        assert!(system.contains("Preserve exported names."));
        assert!(system.contains("does not expand the read-only permissions"));
        send_message(
            &state,
            &id,
            "connection-1",
            "Investigate the export regression.",
        )
        .await
        .unwrap();
        wait_for_turn_complete(&entry).await;
        wait_for_idle(&entry).await;
        let err = open_discussion(
            &state,
            repo.path(),
            "connection-2",
            engine(repo.path()),
            script(),
            "operator",
            Some(&id),
        )
        .await
        .unwrap_err();
        assert!(err.contains("already open"));
        close(&state, &id, "connection-1").await.unwrap();
        assert!(entry
            .durability
            .checkpoint(&id, &[], "late writer", None)
            .await
            .is_err());
        drop(entry);
        drop(state);

        std::fs::write(repo.path().join("AGENTS.md"), "Updated repository rule.").unwrap();
        let restarted = Arc::new(ServerState::standalone(journal.path().to_path_buf()));
        assert!(open_discussion(
            &restarted,
            repo.path(),
            "foreign",
            engine(repo.path()),
            script(),
            "agent:foreign",
            Some(&id)
        )
        .await
        .is_err());
        let resumed = open_discussion(
            &restarted,
            repo.path(),
            "connection-2",
            engine(repo.path()),
            script(),
            "operator",
            Some(&id),
        )
        .await
        .unwrap();
        assert_eq!(resumed["discussion_id"], id);
        assert_eq!(resumed["resumed"], true);
        let entry = get_discussion(&restarted, &id).await.unwrap();
        assert_eq!(entry.turns.load(Ordering::SeqCst), 1);
        assert!(lock(&entry.transcript)
            .iter()
            .any(|(_, text)| text == "Remember the export regression."));
        send_message(
            &restarted,
            &id,
            "connection-2",
            "Now explain the next step.",
        )
        .await
        .unwrap();
        wait_for_idle(&entry).await;
        let checkpoint = entry
            .durability
            .load_checkpoint(&id)
            .await
            .unwrap()
            .unwrap();
        let history = serde_json::to_string(&checkpoint.messages).unwrap();
        assert!(history.contains("Updated repository rule."));
        assert!(!history.contains("Initial repository rule."));
        assert!(history.contains("Preserve exported names."));
        assert!(history.contains("Investigate the export regression."));
        assert!(history.contains("Remember the export regression."));
        assert!(history.contains("Now explain the next step."));
        assert!(get_owned_discussion(&restarted, &id, "connection-1")
            .await
            .is_err());
        close(&restarted, &id, "connection-2").await.unwrap();
    }

    #[tokio::test]
    async fn discuss_start_rejects_a_non_git_directory() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        let err = start_discussion(&state, dir.path(), "owner-1", engine(dir.path()), script)
            .await
            .unwrap_err();
        assert!(
            err.contains("is not a git repository")
                && err.contains("discuss needs a repo to ground itself in"),
            "operator-readable non-repo error, got: {err}"
        );
    }

    #[tokio::test]
    async fn conversation_advertises_relevant_tools_without_mutations() {
        struct Capture(Arc<StdMutex<Vec<Value>>>);
        #[async_trait]
        impl TurnGenerator for Capture {
            async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
                *lock(&self.0) = req.tools.unwrap_or_default();
                Ok(turn("Here is how the repository is organized.", json!([])))
            }
        }
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let tools = Arc::new(StdMutex::new(Vec::new()));
        let id = start(&state, repo.path(), Arc::new(Capture(tools.clone()))).await;
        send_message(&state, &id, "owner-1", "Explain the repository")
            .await
            .unwrap();
        let entry = get_discussion(&state, &id).await.unwrap();
        wait_for_idle(&entry).await;
        let captured = lock(&tools);
        let names: Vec<_> = captured
            .iter()
            .filter_map(|def| def["name"].as_str())
            .collect();
        for required in [
            "read_file",
            "list_dir",
            "find_files",
            "grep_files",
            "prepare_coding_task",
            "web_search",
            "http_request",
        ] {
            assert!(names.contains(&required), "missing {required}: {names:?}");
        }
        assert!(
            names.len() <= 8,
            "unrelated tools leaked into coding conversation: {names:?}"
        );
        assert!(!names.contains(&"write_file"));
        assert!(!names.contains(&"shell"));
        assert!(
            lock(&entry.last_promote).is_none(),
            "a question must not automatically prepare a task"
        );
        eprintln!(
            "conversation tool payload: {} tools, {} JSON bytes",
            names.len(),
            serde_json::to_vec(&*captured).unwrap().len()
        );
    }

    #[tokio::test]
    async fn project_policy_can_refuse_task_preparation() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let policies = repo.path().join(".car/policies");
        std::fs::create_dir_all(&policies).unwrap();
        std::fs::write(
            policies.join("rules.toml"),
            "deny_tool = [\"prepare_coding_task\"]\n",
        )
        .unwrap();
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    "",
                    json!([{"id":"prepare-denied", "name":"prepare_coding_task",
                    "arguments":{"intent":"Fix parser", "constraints":[]}}]),
                ),
                turn(
                    "Task preparation is unavailable under repository policy.",
                    json!([]),
                ),
            ],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        send_message(&state, &id, "owner-1", "Fix parser")
            .await
            .unwrap();
        let entry = get_discussion(&state, &id).await.unwrap();
        wait_for_turn_complete(&entry).await;
        assert!(lock(&entry.last_promote).is_none());
        assert!(!entry
            .events
            .lock()
            .await
            .iter()
            .any(|event| matches!(event.kind, DiscussEventKind::TaskPrepared { .. })));
    }

    #[tokio::test]
    async fn implementation_request_prepares_task_without_starting_execution() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    "",
                    json!([{"id":"prepare-1", "name":"prepare_coding_task",
                    "arguments":{"intent":"Fix the parser", "constraints":["Preserve public APIs"]}}]),
                ),
                turn("The parser task is ready for review.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        send_message(
            &state,
            &id,
            "owner-1",
            "Please fix the parser and preserve public APIs.",
        )
        .await
        .unwrap();
        let entry = get_discussion(&state, &id).await.unwrap();
        wait_for_turn_complete(&entry).await;
        assert_eq!(entry.constraints(), vec!["Preserve public APIs"]);
        let events = entry.events.lock().await;
        assert!(events.iter().any(|event| matches!(&event.kind,
            DiscussEventKind::TaskPrepared { proposed_intent, constraints }
            if proposed_intent == "Fix the parser" && constraints == &["Preserve public APIs"])));
        assert!(
            state.coder_sessions.lock().await.is_empty(),
            "preparing must not create a coding session"
        );
        assert!(!repo.path().join("parser.rs").exists());
        drop(events);
        wait_for_idle(&entry).await;
        let journal = state.journal_dir.clone();
        close(&state, &id, "owner-1").await.unwrap();
        drop(entry);
        drop(state);
        let restarted = Arc::new(ServerState::standalone(journal));
        let no_turns = || -> Arc<dyn TurnGenerator> {
            Arc::new(Script {
                turns: vec![],
                cursor: AtomicUsize::new(0),
            })
        };
        open_discussion(
            &restarted,
            repo.path(),
            "owner-2",
            engine(repo.path()),
            no_turns(),
            "owner-1",
            Some(&id),
        )
        .await
        .unwrap();
        let resumed = get_discussion(&restarted, &id).await.unwrap();
        assert_eq!(resumed.constraints(), vec!["Preserve public APIs"]);
        assert!(
            resumed
                .events
                .lock()
                .await
                .iter()
                .any(|event| matches!(&event.kind,
            DiscussEventKind::ToolCall { tool, params_preview }
            if tool == "prepare_coding_task" && params_preview.contains("Fix the parser"))),
            "resume must show the recorded tool call without executing it again"
        );
        assert!(restarted.coder_sessions.lock().await.is_empty());
        assert!(resumed.events.lock().await.iter().any(|event| matches!(&event.kind,
            DiscussEventKind::TaskPrepared { proposed_intent, .. } if proposed_intent == "Fix the parser")));
        consume_prepared_task(&restarted, &id).await.unwrap();
        close(&restarted, &id, "owner-2").await.unwrap();
        open_discussion(
            &restarted,
            repo.path(),
            "owner-3",
            engine(repo.path()),
            no_turns(),
            "owner-1",
            Some(&id),
        )
        .await
        .unwrap();
        let consumed = get_discussion(&restarted, &id).await.unwrap();
        assert!(!consumed
            .events
            .lock()
            .await
            .iter()
            .any(|event| matches!(event.kind, DiscussEventKind::TaskPrepared { .. })));
    }

    /// The load-bearing property: a discussion NEVER writes in the target repo.
    #[tokio::test]
    async fn a_discussion_writes_nothing_in_the_repo() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        std::fs::write(repo.path().join("keep.txt"), "original").unwrap();
        let (state, _journal) = state();

        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    "",
                    json!([{
                        "id": "c1", "name": "write_file",
                        "arguments": {"path": "sneaky.txt", "content": "written by a discussion"}
                    }]),
                ),
                turn(
                    "",
                    json!([{
                        "id": "c2", "name": "shell",
                        "arguments": {"command": "printf x > shelled.txt"}
                    }]),
                ),
                turn(
                    "I cannot edit from a discussion; here is what I would change.",
                    json!([]),
                ),
            ],
            cursor: AtomicUsize::new(0),
        });

        let id = start(&state, repo.path(), script).await;
        assert!(id.starts_with("disc-"));
        send_message(
            &state,
            &id,
            "owner-1",
            "can you just make the change for me?",
        )
        .await
        .unwrap();
        let entry = get_discussion(&state, &id).await.unwrap();
        wait_for_turn_complete(&entry).await;

        assert!(
            !repo.path().join("sneaky.txt").exists(),
            "a discussion must not create files in the repo"
        );
        assert!(
            !repo.path().join("shelled.txt").exists(),
            "a discussion must not run shell commands that write"
        );
        assert_eq!(
            std::fs::read_to_string(repo.path().join("keep.txt")).unwrap(),
            "original"
        );

        let events = entry.events.lock().await;
        assert!(
            events.iter().any(|e| matches!(
                &e.kind,
                DiscussEventKind::ToolResult { ok, preview, .. }
                    if !ok && preview.contains("read-only")
            )),
            "the denial must surface as a tool_result"
        );
        drop(events);
        wait_for_idle(&entry).await;
        let checkpoint = entry
            .durability
            .load_checkpoint(&id)
            .await
            .unwrap()
            .unwrap();
        let refusals: Vec<_> = checkpoint
            .messages
            .iter()
            .filter_map(|message| {
                if let car_inference::Message::ToolResult { content, .. } = message {
                    Some(content.as_str())
                } else {
                    None
                }
            })
            .collect();
        assert!(refusals
            .iter()
            .any(|text| text.contains("prepare_coding_task")
                && text.contains("not a file-permission problem")));
        assert!(!refusals
            .iter()
            .any(|text| text.contains("declined by user")));
    }

    /// The other half of the boundary: a discussion cannot READ outside its
    /// repo. Mutation-gating alone left the read tools pointed at the whole
    /// filesystem, and their output streams to every subscriber.
    #[tokio::test]
    async fn a_discussion_cannot_read_outside_the_repo() {
        let outside = tempfile::tempdir().unwrap();
        let secret_path = outside.path().join("credentials.txt");
        std::fs::write(&secret_path, "sk-ant-SUPERSECRETVALUE").unwrap();

        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();

        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                // Absolute path outside the repo — the exfiltration attempt.
                turn(
                    "",
                    json!([{
                        "id": "c1", "name": "read_file",
                        "arguments": {"path": secret_path.to_string_lossy()}
                    }]),
                ),
                // ...and the directory-scanning variant.
                turn(
                    "",
                    json!([{
                        "id": "c2", "name": "grep_files",
                        "arguments": {"path": outside.path().to_string_lossy(), "pattern": "sk-ant-"}
                    }]),
                ),
                turn("I can only read inside this repository.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });

        let id = start(&state, repo.path(), script).await;
        send_message(
            &state,
            &id,
            "owner-1",
            "what credentials does this project use?",
        )
        .await
        .unwrap();
        let entry = get_discussion(&state, &id).await.unwrap();
        wait_for_turn_complete(&entry).await;

        let events = entry.events.lock().await;
        let stream = serde_json::to_string(&*events).unwrap();
        assert!(
            !stream.contains("SUPERSECRETVALUE"),
            "a discussion must never stream content from outside its repo: {stream}"
        );
    }

    #[tokio::test]
    async fn selected_model_survives_reopen_and_auto_clears_it() {
        struct Recording {
            requests: Arc<StdMutex<Vec<(Option<String>, bool)>>>,
        }
        #[async_trait]
        impl TurnGenerator for Recording {
            async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
                lock(&self.requests).push((req.model, req.params.strict_model));
                Ok(turn("A grounded reply.", json!([])))
            }
        }
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, journal) = state();
        let mut cfg = car_inference::InferenceConfig::default();
        cfg.models_dir = journal.path().join("models");
        let engine = Arc::new(car_inference::InferenceEngine::new(cfg));
        let requests = Arc::new(StdMutex::new(Vec::new()));
        let generator: Arc<dyn TurnGenerator> = Arc::new(Recording {
            requests: requests.clone(),
        });
        let rejected = open_discussion_with_model(
            &state,
            repo.path(),
            "owner",
            engine.clone(),
            generator.clone(),
            "operator",
            None,
            Some("qwen/qwen3-embedding-0.6b:q8_0"),
        )
        .await
        .unwrap_err();
        assert!(
            rejected.contains("cannot call repository tools"),
            "{rejected}"
        );
        assert!(requests.lock().unwrap().is_empty());
        let first = open_discussion_with_model(
            &state,
            repo.path(),
            "owner",
            engine.clone(),
            generator.clone(),
            "operator",
            None,
            Some("anthropic/claude-opus-4-6:latest"),
        )
        .await
        .unwrap();
        let id = first["discussion_id"].as_str().unwrap();
        assert_eq!(first["model"], "anthropic/claude-opus-4-6:latest");
        for selection in [None, Some("auto")] {
            send_message(&state, id, "owner", "What is here?")
                .await
                .unwrap();
            let entry = get_discussion(&state, id).await.unwrap();
            wait_for_idle(&entry).await;
            close(&state, id, "owner").await.unwrap();
            let reopened = open_discussion_with_model(
                &state,
                repo.path(),
                "owner",
                engine.clone(),
                generator.clone(),
                "operator",
                Some(id),
                selection,
            )
            .await
            .unwrap();
            if selection.is_none() {
                assert_eq!(reopened["model"], "anthropic/claude-opus-4-6:latest");
            } else {
                assert!(reopened["model"].is_null());
            }
        }
        send_message(&state, id, "owner", "Continue.")
            .await
            .unwrap();
        wait_for_idle(&get_discussion(&state, id).await.unwrap()).await;
        close(&state, id, "owner").await.unwrap();
        let captured = lock(&requests);
        assert_eq!(
            *captured,
            vec![
                (Some("anthropic/claude-opus-4-6:latest".into()), true),
                (Some("anthropic/claude-opus-4-6:latest".into()), true),
                (None, false)
            ]
        );
        assert!(DiscussionRecord::load(&state.journal_dir, id, "operator")
            .unwrap()
            .model
            .is_none());
    }

    #[tokio::test]
    async fn starting_refuses_to_drop_requirements_when_distillation_fails() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            // The reply succeeds; all later generation attempts fail.
            turns: vec![turn("I will preserve the public API.", json!([]))],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        send_message(&state, &id, "owner-1", "Preserve the public API.")
            .await
            .unwrap();
        let entry = get_discussion(&state, &id).await.unwrap();
        wait_for_idle(&entry).await;

        let error = constraints_for_start(&state, &id).await.unwrap_err();
        assert!(error.contains("No task was started"), "{error}");
        assert!(error.contains("Retry Build"), "{error}");
        assert!(lock(&entry.last_promote).is_none());
        assert!(!entry.transcript_is_empty());
        assert!(state.coder_sessions.lock().await.is_empty());
        close(&state, &id, "owner-1").await.unwrap();
    }

    #[tokio::test]
    async fn promote_distills_an_intent_and_starts_nothing() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();

        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn("The Windows path is the risky one.", json!([])),
                turn(
                    r#"{"proposed_intent":"Make the config loader resolve paths on Windows.",
                        "constraints":["do not change the POSIX behavior"]}"#,
                    json!([]),
                ),
                turn("Understood, preserve both platforms.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });

        let id = start(&state, repo.path(), script).await;
        send_message(
            &state,
            &id,
            "owner-1",
            "what is fragile about the config loader?",
        )
        .await
        .unwrap();
        let entry = get_discussion(&state, &id).await.unwrap();
        wait_for_turn_complete(&entry).await;

        wait_for_idle(&entry).await;
        let promoted = promote(&state, &id, "owner-1").await.unwrap();
        assert_eq!(
            promoted["proposed_intent"],
            "Make the config loader resolve paths on Windows."
        );
        assert_eq!(
            promoted["constraints"],
            json!(["do not change the POSIX behavior"])
        );
        assert!(state.coder_sessions.lock().await.is_empty());
        assert_eq!(
            constraints_for_start(&state, &id).await.unwrap(),
            vec!["do not change the POSIX behavior".to_string()]
        );
        send_message(
            &state,
            &id,
            "owner-1",
            "Also preserve Windows compatibility.",
        )
        .await
        .unwrap();
        assert!(
            lock(&entry.last_promote).is_none(),
            "a new turn must invalidate the old plan constraints"
        );
        wait_for_idle(&entry).await;
        close(&state, &id, "owner-1").await.unwrap();
    }

    /// A discussion turn refused on the Parslee account must still END.
    ///
    /// `coder.discuss` runs the same `AssistantService` as chat, so it sees the
    /// same terminal `auth_required` frame — and before this arm existed the
    /// frame fell through to `_ => {}`: no error, no remedy, and no
    /// `TurnComplete`. `docs/websocket-protocol.md` tells a client to wait for
    /// `TurnComplete` before sending again, so the client was left holding a
    /// turn that had already finished. The second `send` at the end is the
    /// point: it proves the discussion is usable afterwards, not just that an
    /// event was emitted.
    #[tokio::test]
    async fn an_account_refusal_ends_the_discussion_turn_with_its_remedy() {
        struct SignedOut;
        #[async_trait]
        impl TurnGenerator for SignedOut {
            async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
                panic!("the assistant loop must generate through the typed seam")
            }

            async fn generate_assistant(
                &self,
                _req: GenerateRequest,
            ) -> Result<InferenceResult, crate::coder::native_loop::AssistantGenerateError>
            {
                Err(crate::coder::native_loop::AssistantGenerateError::from(
                    car_inference::InferenceError::CredentialUnavailable {
                        provider: "parslee".into(),
                        model: "parslee/advisor".into(),
                        reason: car_inference::CredentialFailure::SignedOut,
                        detail: "no account is signed in. Run `car auth login`".into(),
                    },
                ))
            }
        }

        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let generator: Arc<dyn TurnGenerator> = Arc::new(SignedOut);
        let id = start(&state, repo.path(), generator).await;

        send_message(&state, &id, "owner-1", "what is fragile here?")
            .await
            .unwrap();
        let entry = get_discussion(&state, &id).await.unwrap();
        wait_for_turn_complete(&entry).await;

        let kinds: Vec<DiscussEventKind> = {
            let events = entry.events.lock().await;
            events.iter().map(|e| e.kind.clone()).collect()
        };
        let message = kinds
            .iter()
            .find_map(|k| match k {
                DiscussEventKind::Error { message } => Some(message.clone()),
                _ => None,
            })
            .expect("the refusal must reach the client as a terminal error");
        assert_eq!(
            message,
            crate::assistant::AUTH_REQUIRED_SIGNED_OUT_MESSAGE,
            "the remedy is the daemon's approved copy, verbatim"
        );
        assert!(
            matches!(kinds.last(), Some(DiscussEventKind::TurnComplete {})),
            "the turn must end with TurnComplete: {kinds:?}"
        );

        // …and the discussion is usable again, which is what `TurnComplete`
        // promises a client.
        send_message(&state, &id, "owner-1", "and the second question?")
            .await
            .expect("a completed turn must accept the next message");
    }

    /// A second `send` while a turn is streaming is refused, not silently
    /// interleaved — and `promote` refuses too rather than distilling a
    /// question with no answer beside it.
    #[tokio::test]
    async fn a_turn_in_flight_blocks_a_second_send_and_promote() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let gate = Arc::new(tokio::sync::Notify::new());
        let generator: Arc<dyn TurnGenerator> = Arc::new(Blocking { gate: gate.clone() });

        let id = start(&state, repo.path(), generator).await;
        send_message(&state, &id, "owner-1", "first question")
            .await
            .unwrap();

        let entry = get_discussion(&state, &id).await.unwrap();
        for _ in 0..200 {
            if entry.is_answering() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        assert!(entry.is_answering(), "the turn should be in flight");

        let err = send_message(&state, &id, "owner-1", "second question")
            .await
            .unwrap_err();
        assert!(
            err.contains("still answering"),
            "a concurrent send must be refused, not silently lose a turn: {err}"
        );
        let err = promote(&state, &id, "owner-1").await.unwrap_err();
        assert!(
            err.contains("still answering"),
            "promote must not distill a half-finished turn: {err}"
        );

        gate.notify_one();
        wait_for_turn_complete(&entry).await;
    }

    /// Closing cancels the in-flight turn rather than leaving it billing tokens
    /// to a conversation nobody can read.
    #[tokio::test]
    async fn close_cancels_an_in_flight_turn() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let gate = Arc::new(tokio::sync::Notify::new());
        let generator: Arc<dyn TurnGenerator> = Arc::new(Blocking { gate });

        let id = start(&state, repo.path(), generator).await;
        send_message(&state, &id, "owner-1", "a broad question")
            .await
            .unwrap();
        let entry = get_discussion(&state, &id).await.unwrap();
        for _ in 0..200 {
            if entry.is_answering() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }

        close(&state, &id, "owner-1").await.unwrap();
        assert!(!entry.is_answering(), "close must stop the turn");
        assert!(state.coder_discussions.lock().await.is_empty());
    }

    /// A `send` whose handler future is dropped after the turn was dispatched
    /// must NOT leave the discussion latched as answering.
    ///
    /// `coder.discuss.send` is not deadline-exempt, so the daemon's handler
    /// deadline cancels this future at its one remaining await — the turn's
    /// first-event cursor. `in_flight` is set by CAS before that and cleared
    /// only at the tail of the spawned turn task, so the question is whether
    /// that task exists. It does: the dispatch is complete before this await is
    /// ever reached, so the drop costs the caller its `seq` reply and nothing
    /// else. The turn answers the message, releases `in_flight`, and the
    /// discussion is usable again — rather than answering "still answering the
    /// previous message" forever with nothing running (`reap_idle` runs only on
    /// the next `discuss.start`, so a quiet daemon never reclaimed that).
    #[tokio::test]
    async fn a_send_cancelled_after_dispatch_leaves_the_discussion_usable() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn("answered anyway", json!([])),
                turn("answered on the retry", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        let entry = get_discussion(&state, &id).await.unwrap();

        // Cancellation IS "the future is dropped at an .await point" — that is
        // all `tokio::time::timeout` does to a handler. Poll once to get past
        // the CAS and the dispatch, park on the turn's first-event cursor, then
        // drop it there.
        let mut send = Box::pin(send_message(
            &state,
            &id,
            "owner-1",
            "the message whose reply frame gets cancelled",
        ));
        assert!(
            matches!(futures::poll!(send.as_mut()), std::task::Poll::Pending),
            "the fixture needs the send parked on its cursor"
        );
        assert!(
            entry.is_answering(),
            "the fixture needs the CAS to have run"
        );
        drop(send);

        // The turn was already dispatched, so it runs and releases the latch.
        wait_for_turn_complete(&entry).await;
        for _ in 0..200 {
            if !entry.is_answering() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        assert!(
            !entry.is_answering(),
            "a cancelled handler must not strand `in_flight`"
        );

        // ...and the discussion still works.
        send_message(&state, &id, "owner-1", "second try")
            .await
            .expect("the discussion must still accept a message");
    }

    /// A `send` whose dispatch is refused by a `close` must never reach the
    /// model.
    ///
    /// `cancel_turn` used to read `turn_task` before `send_message` stored it —
    /// the store happened only after the first `emit().await` — so a close in
    /// that window found nothing to abort, removed the entry from the registry,
    /// and then `send_message` resumed and spawned a 12-turn model loop against
    /// a discussion nothing could reach. The turn slot latch is what closed
    /// that: `close` latches it, the dispatch checks it under the same lock,
    /// and a send that arrives after the latch is REFUSED. Here the latch is
    /// set without removing the registry entry, so the send reaches the
    /// dispatch and is refused exactly there.
    #[tokio::test]
    async fn a_close_racing_a_dispatching_send_never_starts_the_turn() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let calls = Arc::new(AtomicUsize::new(0));
        let script: Arc<dyn TurnGenerator> = Arc::new(Counting {
            calls: calls.clone(),
        });
        let id = start(&state, repo.path(), script).await;
        let entry = get_discussion(&state, &id).await.unwrap();

        entry.cancel_turn();

        let err = send_message(&state, &id, "owner-1", "a broad question")
            .await
            .unwrap_err();
        assert!(
            err.contains("closed while your message was being dispatched"),
            "the caller must be told the send did not run: {err}"
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            0,
            "a closed discussion must never reach the model"
        );
        assert!(!entry.is_answering());

        close(&state, &id, "owner-1").await.unwrap();
        assert!(state.coder_discussions.lock().await.is_empty());
    }

    /// A discussion is owned by the connection that opened it — and that is now
    /// enforced, not merely recorded. Every method resolved by id alone, so any
    /// connected client could send into, promote, or close another's
    /// discussion; closing one mid-turn cancels a turn its owner is watching.
    #[tokio::test]
    async fn another_connection_cannot_drive_a_discussion() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;

        for err in [
            send_message(&state, &id, "intruder", "run this for me")
                .await
                .unwrap_err(),
            promote(&state, &id, "intruder").await.unwrap_err(),
            close(&state, &id, "intruder").await.unwrap_err(),
            // The path `coder.discuss.subscribe` and `coder.start
            // { discussion_id }` both resolve through.
            match get_owned_discussion(&state, &id, "intruder").await {
                Ok(_) => panic!("a foreign client must not resolve another's discussion"),
                Err(e) => e,
            },
        ] {
            assert!(
                err.contains("belongs to another connection"),
                "a foreign client must be refused: {err}"
            );
        }

        // Untouched, and still the owner's to close.
        assert_eq!(state.coder_discussions.lock().await.len(), 1);
        close(&state, &id, "owner-1").await.unwrap();
    }

    /// Operator text is retained in the transcript, the replay buffer and the
    /// distill prompt, so it needs the byte cap `summarize_repo` already has.
    #[tokio::test]
    async fn an_oversized_message_is_refused() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        let entry = get_discussion(&state, &id).await.unwrap();

        let err = send_message(
            &state,
            &id,
            "owner-1",
            &"x".repeat(DISCUSS_MESSAGE_MAX_BYTES + 1),
        )
        .await
        .unwrap_err();
        assert!(err.contains("the limit is"), "{err}");
        // Refused BEFORE the latch, so the discussion is still usable.
        assert!(!entry.is_answering());
        assert!(entry.transcript_is_empty());
    }

    /// A disconnecting client's discussions are freed, not leaked for the
    /// daemon's lifetime.
    #[tokio::test]
    async fn disconnect_closes_the_owning_clients_discussions() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        assert_eq!(state.coder_discussions.lock().await.len(), 1);

        // A different client disconnecting leaves it alone...
        drop_subscriptions_for_client(&state, "someone-else").await;
        assert_eq!(state.coder_discussions.lock().await.len(), 1);

        // ...its owner disconnecting closes it.
        drop_subscriptions_for_client(&state, "owner-1").await;
        assert!(state.coder_discussions.lock().await.is_empty());
        assert!(get_discussion(&state, &id).await.is_err());
    }

    /// The cap is a slot RESERVATION, so the test holds the slots directly
    /// rather than building eight full assistant runtimes — each
    /// `start_discussion` binds a substrate and registers ~40 tools, and doing
    /// that eight times to assert a length check cost minutes of CI for
    /// nothing.
    #[tokio::test]
    async fn open_discussions_are_capped() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let held: Vec<_> = (0..MAX_OPEN_DISCUSSIONS)
            .map(|_| {
                state
                    .coder_discussion_slots
                    .clone()
                    .try_acquire_owned()
                    .expect("a fresh daemon has every slot free")
            })
            .collect();

        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        let err = start_discussion(&state, repo.path(), "owner-1", engine(repo.path()), script)
            .await
            .unwrap_err();
        assert!(err.contains("already open"), "{err}");

        // ...and a freed slot admits the next one.
        drop(held);
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        start_discussion(&state, repo.path(), "owner-1", engine(repo.path()), script)
            .await
            .expect("a released slot must be reusable");
    }

    /// The cap must hold under CONCURRENT starts, which is what it did not do:
    /// the count was read, the registry lock released, and two awaits (bind the
    /// substrate, build the runtime) ran before the insert — and the daemon
    /// runs a connection's requests concurrently, so N pipelined starts all
    /// read `len() == 0`, all passed a cap of 8, and all built a runtime.
    ///
    /// One slot is left free and four starts race for it: exactly one may win,
    /// and the three losers must fail BEFORE building anything.
    #[tokio::test]
    async fn concurrent_starts_cannot_exceed_the_open_discussion_cap() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let _held: Vec<_> = (0..MAX_OPEN_DISCUSSIONS - 1)
            .map(|_| {
                state
                    .coder_discussion_slots
                    .clone()
                    .try_acquire_owned()
                    .unwrap()
            })
            .collect();

        let mut racers = Vec::new();
        for _ in 0..4 {
            let state = state.clone();
            let repo = repo.path().to_path_buf();
            racers.push(tokio::spawn(async move {
                let script: Arc<dyn TurnGenerator> = Arc::new(Script {
                    turns: vec![],
                    cursor: AtomicUsize::new(0),
                });
                start_discussion(&state, &repo, "owner-1", engine(&repo), script).await
            }));
        }

        let mut admitted = 0;
        let mut refused = 0;
        for racer in racers {
            match racer.await.unwrap() {
                Ok(_) => admitted += 1,
                Err(e) => {
                    assert!(e.contains("already open"), "unexpected refusal: {e}");
                    refused += 1;
                }
            }
        }
        assert_eq!(admitted, 1, "exactly one racer may take the last slot");
        assert_eq!(refused, 3);
        assert_eq!(
            state.coder_discussions.lock().await.len(),
            1,
            "the registry must never exceed the cap"
        );
    }

    #[tokio::test]
    async fn unknown_discussion_ids_are_clear_errors() {
        let (state, _journal) = state();
        for err in [
            send_message(&state, "disc-nope", "owner-1", "hi")
                .await
                .unwrap_err(),
            promote(&state, "disc-nope", "owner-1").await.unwrap_err(),
            constraints_for_start(&state, "disc-nope")
                .await
                .unwrap_err(),
        ] {
            assert!(err.contains("disc-nope"), "must name the id, got: {err}");
        }
        assert!(close(&state, "disc-nope", "owner-1").await.is_err());
    }

    #[test]
    fn checkout_followup_uses_current_files_but_requires_a_known_delivery() {
        assert_eq!(
            followup_base_from(&[
                json!({"state":"merged", "result_delivery":"checkout", "result_commit":"saved"})
            ])
            .unwrap(),
            None
        );
        assert!(
            followup_base_from(&[json!({"state":"merged", "result_delivery":"checkout"})]).is_err()
        );
        assert_eq!(
            followup_base_from(&[
                json!({"state":"merged", "result_delivery":"branch", "result_commit":"saved"})
            ])
            .unwrap(),
            Some("saved".into())
        );
    }

    #[test]
    fn followup_never_substitutes_head_for_known_work() {
        assert!(followup_base_from(&[
            json!({"state": "merged", "updated_at": 1, "result_commit": "one"}),
            json!({"state": "merged", "updated_at": 1, "result_commit": "two"}),
        ])
        .is_err());
        assert!(
            followup_base_from(&[json!({"state": "merged", "result_branch": "moved"})]).is_err()
        );
        assert!(
            followup_base_from(&[json!({"state": "failed", "worktree": "/retained"})]).is_err()
        );
        let rows = vec![
            json!({"state": "failed", "worktree": null}),
            json!({"state": "merged", "result_commit": "fixed-revision"}),
        ];
        assert_eq!(
            followup_base_from(&rows).unwrap().as_deref(),
            Some("fixed-revision")
        );
    }

    #[tokio::test]
    async fn linked_work_is_grounded_scoped_and_blocks_overlapping_starts() {
        use super::super::router::EngineChoice;
        use super::super::session::{CoderSession, CoderState};
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let dir = tempfile::tempdir().unwrap();
        let id = start(
            &state,
            repo.path(),
            Arc::new(Script {
                turns: vec![],
                cursor: AtomicUsize::new(0),
            }),
        )
        .await;
        let entry = get_discussion(&state, &id).await.unwrap();
        assert!(
            claim_coding_start(&state, &id, Path::new("/different"), dir.path().into())
                .await
                .unwrap_err()
                .contains("repository")
        );
        let guard = claim_coding_start_at(&state, &id, dir.path().into())
            .await
            .unwrap();
        assert!(tokio::time::timeout(
            std::time::Duration::from_millis(20),
            claim_coding_start_at(&state, &id, dir.path().into())
        )
        .await
        .is_err());
        let mut run = CoderSession::new(
            &entry.repo,
            "repair export",
            EngineChoice::Native,
            3,
            Some(dir.path().into()),
        );
        run.discussion_id = Some(id.clone());
        run.state = CoderState::NeedsApproval;
        run.workspace_path = Some(repo.path().into());
        run.persist().unwrap();
        drop(guard);
        let error = claim_coding_start_at(&state, &id, dir.path().into())
            .await
            .unwrap_err();
        assert!(error.contains(&run.id));
        assert!(error.contains("unfinished work"));
        run.state = CoderState::Failed;
        run.error = Some("export check failed".into());
        run.persist().unwrap();
        let mut unrelated = CoderSession::new(
            &entry.repo,
            "private other conversation",
            EngineChoice::Native,
            3,
            Some(dir.path().into()),
        );
        unrelated.discussion_id = Some("other".into());
        unrelated.persist().unwrap();
        let rows = coding_runs(&state, &id, &entry.repo, dir.path().into())
            .await
            .unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0]["error"], "export check failed");
        assert_eq!(rows[0]["live"], false);
        assert!(
            retained_workspace(&state, &id, &entry.repo, dir.path().into())
                .await
                .unwrap_err()
                .contains("not been confirmed stopped")
        );
        run.execution_stopped = true;
        run.persist().unwrap();
        assert!(
            retained_workspace(&state, &id, &entry.repo, dir.path().into())
                .await
                .is_err(),
            "a stopped task cannot adopt a path outside the daemon's worktrees"
        );
        assert!(
            coding_runs(&state, &id, Path::new("/different"), dir.path().into())
                .await
                .unwrap()
                .is_empty()
        );
        assert!(claim_coding_start_at(&state, &id, dir.path().into())
            .await
            .is_ok());
        close(&state, &id, "owner-1").await.unwrap();
    }

    #[tokio::test]
    async fn saved_list_survives_restart_and_filters_principals_and_live_owners() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, journal) = state();
        let owner = client(&state, "operator-1").await;
        let other = client(&state, "agent-1").await;
        *other.agent_id.lock().await = Some("foreign".into());
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        let opened = open_discussion(
            &state,
            repo.path(),
            &owner.client_id,
            engine(repo.path()),
            script,
            "operator",
            None,
        )
        .await
        .unwrap();
        let id = opened["discussion_id"].as_str().unwrap();
        assert!(handle_discuss_list(&state, &owner).await.unwrap()["saved"]
            .as_array()
            .unwrap()
            .is_empty());
        close(&state, id, &owner.client_id).await.unwrap();
        let restarted = Arc::new(ServerState::standalone(journal.path().to_path_buf()));
        let saved = handle_discuss_list(&restarted, &owner).await.unwrap();
        assert_eq!(saved["saved"][0]["discussion_id"], id);
        assert!(saved["discussions"].as_array().unwrap().is_empty());
        assert!(
            handle_discuss_list(&restarted, &other).await.unwrap()["saved"]
                .as_array()
                .unwrap()
                .is_empty()
        );
    }

    #[tokio::test]
    async fn list_and_close_track_open_discussions() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        let owner = client(&state, "owner-1").await;

        let listed = handle_discuss_list(&state, &owner).await.unwrap();
        assert_eq!(listed["discussions"].as_array().unwrap().len(), 1);
        assert_eq!(listed["discussions"][0]["discussion_id"], id.as_str());
        assert_eq!(listed["discussions"][0]["turns"], 0);

        // ...and it is scoped to the owning connection.
        let stranger = client(&state, "someone-else").await;
        let listed = handle_discuss_list(&state, &stranger).await.unwrap();
        assert!(
            listed["discussions"].as_array().unwrap().is_empty(),
            "another connection must not see this discussion: {listed}"
        );

        assert_eq!(
            close(&state, &id, "owner-1").await.unwrap(),
            json!({ "ok": true })
        );
        let listed = handle_discuss_list(&state, &owner).await.unwrap();
        assert!(listed["discussions"].as_array().unwrap().is_empty());
    }

    /// The stated guarantee, measured where a client actually lives: what a
    /// SUBSCRIBER receives across an attach is contiguous from its cursor —
    /// no gap, no duplicate — even when emits are racing the attach.
    ///
    /// Asserting on the buffer proves only that the drain is the single writer.
    /// The property clients depend on spans three more hops the buffer never
    /// touches: the replay clone at attach, the per-subscriber queue, and that
    /// lane's sender task. An attach that registered before replaying would
    /// duplicate here and an attach that replayed before registering would drop
    /// whatever emitted in between, and the buffer would look perfect either
    /// way.
    ///
    /// **The replay hop has to actually run.** `handle_discuss_subscribe` has
    /// exactly one await before it enqueues `Attach`, and it resolves on the
    /// first poll; on the current-thread test runtime the "racing" emitter had
    /// therefore never been polled when the attach landed, so
    /// `events_replayed` was 0 on every run and `replayed <= 30` was satisfied
    /// by nothing having been replayed at all. Mutating the replay filter to
    /// `e.seq > from_seq` — the off-by-one that drops the first event of every
    /// real client resume — left the test green. So: yield until the emitter
    /// has genuinely produced events, assert the replay is non-empty, and
    /// attach a second time from a NON-ZERO cursor, where an off-by-one is a
    /// wrong first seq rather than a merely smaller count.
    #[tokio::test]
    async fn a_subscriber_receives_every_seq_exactly_once_across_its_attach() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        let entry = get_discussion(&state, &id).await.unwrap();

        let (channel, frames) = capturing_channel();
        let owner = state
            .create_session("owner-1", channel.clone())
            .await
            .unwrap();

        // Emitted WHILE the attach is in flight: each of these lands on one
        // side or the other of the `Attach` command, and the subscriber must
        // see it exactly once either way.
        let racing = {
            let entry = entry.clone();
            tokio::spawn(async move {
                for i in 0..30u64 {
                    entry
                        .emit(DiscussEventKind::AssistantDelta {
                            text: format!("during-{i}"),
                        })
                        .await;
                }
            })
        };
        // Let the emitter actually get ahead of the attach. Without this the
        // attach wins every poll and there is no race to observe.
        while entry.events.lock().await.is_empty() {
            tokio::task::yield_now().await;
        }
        let subscribed = handle_discuss_subscribe(
            &rpc_req(json!({ "discussion_id": id, "from_seq": 0 })),
            &state,
            &owner,
        )
        .await
        .unwrap();
        racing.await.unwrap();

        // ...and after it, live through the same lane.
        for i in 0..20u64 {
            entry
                .emit(DiscussEventKind::AssistantDelta {
                    text: format!("after-{i}"),
                })
                .await;
        }

        const TOTAL: usize = 50;
        let replayed = subscribed["events_replayed"].as_u64().unwrap();
        assert!(
            replayed > 0,
            "the attach replayed nothing, so this test never exercised the \
             replay hop it exists to cover"
        );
        assert!(
            replayed <= 30,
            "replay cannot exceed what was emitted before the attach: {replayed}"
        );

        let mut seqs = Vec::new();
        for _ in 0..400 {
            seqs = delivered_seqs(&frames);
            if seqs.len() >= TOTAL {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert_eq!(
            seqs,
            (0..TOTAL as u64).collect::<Vec<_>>(),
            "a subscriber must receive seq 0..{TOTAL} once each, in order"
        );

        // ...and a resume from a non-zero cursor is inclusive of that cursor.
        // Every seq is in the buffer now, so this is exact: an off-by-one in
        // the replay filter shows up as a missing FIRST event, not as a count
        // that merely looks plausible.
        const RESUME_FROM: u64 = 17;
        let (resumed_channel, resumed_frames) = capturing_channel();
        // Same client id: a discussion is owned by the connection that opened
        // it, and re-attaching replaces that connection's lane.
        let resumed = state
            .create_session("owner-1", resumed_channel)
            .await
            .unwrap();
        let reattached = handle_discuss_subscribe(
            &rpc_req(json!({ "discussion_id": id, "from_seq": RESUME_FROM })),
            &state,
            &resumed,
        )
        .await
        .unwrap();
        assert_eq!(
            reattached["events_replayed"].as_u64().unwrap(),
            TOTAL as u64 - RESUME_FROM,
            "a resume from {RESUME_FROM} must replay seq {RESUME_FROM}..{TOTAL}"
        );

        let mut resumed_seqs = Vec::new();
        for _ in 0..400 {
            resumed_seqs = delivered_seqs(&resumed_frames);
            if resumed_seqs.len() >= TOTAL - RESUME_FROM as usize {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert_eq!(
            resumed_seqs,
            (RESUME_FROM..TOTAL as u64).collect::<Vec<_>>(),
            "a resume must start AT its cursor, not one past it"
        );
    }

    /// A send that IS dispatched still puts the `user_message` on the
    /// stream first, ahead of every assistant delta for that turn. Moving the
    /// emit into the turn must not reorder it behind the turn's own output.
    #[tokio::test]
    async fn a_dispatched_send_emits_the_user_message_before_any_delta() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![turn("here is what I would change", json!([]))],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        let entry = get_discussion(&state, &id).await.unwrap();

        let sent = send_message(&state, &id, "owner-1", "what should this change do?")
            .await
            .unwrap();
        assert_eq!(
            sent["seq"], 0,
            "the reported cursor is the user_message's own seq"
        );
        wait_for_turn_complete(&entry).await;

        let events = entry.events.lock().await;
        assert!(
            matches!(events[0].kind, DiscussEventKind::UserMessage { .. }),
            "the operator's message must be the turn's first event, got: {:?}",
            events[0].kind
        );
        assert!(
            events.len() > 1,
            "the turn produced nothing to order against"
        );
        assert!(
            !events[1..]
                .iter()
                .any(|e| matches!(e.kind, DiscussEventKind::UserMessage { .. })),
            "exactly one user_message per send"
        );
    }

    /// A subscriber that has stopped reading is its own problem: it is SHED,
    /// and the turn it was watching completes anyway.
    ///
    /// Half of this was never pinned. When the drain performed the sends
    /// itself, a half-open board (no FIN, no RST — writes park forever) held
    /// the drain for `DISCUSS_SEND_TIMEOUT` per event, so the next `Emit` sat
    /// unprocessed and every `entry.emit(…).await` inside `run_turn` waited on
    /// it: one dead board stalled the whole TURN. The turn here must complete
    /// while the wedge is still in place, on a clock well inside that deadline.
    #[tokio::test]
    async fn a_wedged_subscriber_is_shed_and_the_turn_still_completes() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![turn("here is what I would change", json!([]))],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        let entry = get_discussion(&state, &id).await.unwrap();

        let (channel, _frames) = capturing_channel();
        let owner = state
            .create_session("owner-1", channel.clone())
            .await
            .unwrap();
        let unsubscribed = Arc::strong_count(&channel);
        handle_discuss_subscribe(
            &rpc_req(json!({ "discussion_id": id, "from_seq": 0 })),
            &state,
            &owner,
        )
        .await
        .unwrap();
        assert_eq!(
            Arc::strong_count(&channel),
            unsubscribed + 1,
            "the lane must hold this subscriber's channel"
        );

        // Half-open from here on: writes never fail, they just never finish.
        let stuck = channel.write.lock().await;

        let started = std::time::Instant::now();
        send_message(&state, &id, "owner-1", "what should this change do?")
            .await
            .unwrap();
        let mut completed = false;
        for _ in 0..120 {
            if entry
                .events
                .lock()
                .await
                .iter()
                .any(|e| matches!(e.kind, DiscussEventKind::TurnComplete {}))
            {
                completed = true;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert!(
            completed && started.elapsed() < DISCUSS_SEND_TIMEOUT,
            "the turn must not wait on a wedged subscriber's socket ({:?} elapsed)",
            started.elapsed()
        );

        // ...and the lane is shed rather than carried: its queue fills, the
        // drain's `try_send` fails, and dropping the `Subscriber` aborts the
        // task parked on that socket — releasing the channel handle it pinned.
        for i in 0..(DISCUSS_SUBSCRIBER_QUEUE + 64) {
            entry
                .emit(DiscussEventKind::AssistantDelta {
                    text: format!("overflow-{i}"),
                })
                .await;
        }
        let mut shed = false;
        for _ in 0..200 {
            if Arc::strong_count(&channel) == unsubscribed {
                shed = true;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert!(
            shed,
            "a subscriber that is not draining must be shed, not retained"
        );
        drop(stuck);
    }

    /// A `send` whose dispatch is refused must leave no unanswered operator
    /// question behind — not in the transcript, and not on the wire.
    ///
    /// `InFlightGuard` frees the discussion on that path, so `is_answering()`
    /// reads false — and `promote` and `coder.start { discussion_id }` gate on
    /// exactly that. The transcript still ended in a question no turn answered,
    /// which sailed through both guards and became the distillation input those
    /// guards exist to prevent: a confident intent invented from a question
    /// nobody replied to.
    ///
    /// The `user_message` event had the same hole for the same reason: it was
    /// emitted BEFORE the dispatch, so a refused send still put the operator's
    /// question in the replay buffer and on every subscriber while the
    /// transcript row rolled back — and the board's discussion pane rendered
    /// that question followed by permanent silence. The emit now happens inside
    /// the turn, so it and the transcript row commit or roll back together.
    #[tokio::test]
    async fn a_refused_send_leaves_no_unanswered_turn_in_the_transcript() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let calls = Arc::new(AtomicUsize::new(0));
        let script: Arc<dyn TurnGenerator> = Arc::new(Counting {
            calls: calls.clone(),
        });
        let id = start(&state, repo.path(), script).await;
        let entry = get_discussion(&state, &id).await.unwrap();

        // Latch the turn slot closed WITHOUT removing the registry entry, so
        // the send reaches the dispatch and is refused THERE — the window a
        // racing `close` actually wins.
        entry.cancel_turn();
        let err = send_message(&state, &id, "owner-1", "should we rewrite the scheduler?")
            .await
            .unwrap_err();
        assert!(
            err.contains("closed while your message was being dispatched"),
            "expected a refused dispatch, got: {err}"
        );

        assert!(
            !entry.is_answering(),
            "a refused dispatch must not strand `in_flight`"
        );
        assert!(
            entry.transcript_is_empty(),
            "a question no turn will answer must not survive in the transcript: {:?}",
            lock(&entry.transcript)
        );
        assert!(
            !entry
                .events
                .lock()
                .await
                .iter()
                .any(|e| matches!(e.kind, DiscussEventKind::UserMessage { .. })),
            "...nor reach the replay buffer and every subscriber"
        );
        // ...and the guards that read the transcript agree.
        let err = promote(&state, &id, "owner-1").await.unwrap_err();
        assert!(
            err.contains("no turns yet"),
            "promote must refuse an empty discussion rather than distill a stranded \
             question: {err}"
        );
        assert!(
            constraints_for_start(&state, &id).await.unwrap().is_empty(),
            "coder.start must not distill constraints from a stranded question"
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            0,
            "no turn ran, so nothing reached the model"
        );
    }

    /// The drain assigns `seq` under the buffer lock as the only writer, so the
    /// buffer is strictly ordered even when emits are produced concurrently.
    #[tokio::test]
    async fn concurrent_emits_stay_seq_ordered_in_the_buffer() {
        let repo = tempfile::tempdir().unwrap();
        init_repo(repo.path());
        let (state, _journal) = state();
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
        });
        let id = start(&state, repo.path(), script).await;
        let entry = get_discussion(&state, &id).await.unwrap();

        let mut tasks = Vec::new();
        for i in 0..50 {
            let e = entry.clone();
            tasks.push(tokio::spawn(async move {
                e.emit(DiscussEventKind::AssistantDelta {
                    text: format!("chunk-{i}"),
                })
                .await
            }));
        }
        for t in tasks {
            t.await.unwrap();
        }

        let events = entry.events.lock().await;
        assert_eq!(events.len(), 50);
        for (i, e) in events.iter().enumerate() {
            assert_eq!(e.seq, i as u64, "buffer must be in seq order");
        }
    }

    #[test]
    fn discuss_event_json_shape_is_ws_friendly() {
        let e = DiscussEvent {
            discussion_id: "disc-x".into(),
            seq: 7,
            ts: 1,
            kind: DiscussEventKind::AssistantDelta {
                text: "hello".into(),
            },
        };
        let v = serde_json::to_value(&e).unwrap();
        assert_eq!(v["type"], "assistant_delta");
        assert_eq!(v["text"], "hello");
        assert_eq!(v["seq"], 7);
        assert_eq!(v["discussion_id"], "disc-x");

        let v = serde_json::to_value(DiscussEvent {
            discussion_id: "disc-x".into(),
            seq: 8,
            ts: 1,
            kind: DiscussEventKind::TurnComplete {},
        })
        .unwrap();
        assert_eq!(v["type"], "turn_complete");
    }
}