car-server-core 0.47.0

//! `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`](crate::assistant::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
//!
//! Discussions are in-memory only and do **not** survive a daemon restart. The
//! model thread, the bound runtime and the substrate are all process-local;
//! persisting the transcript alone would resume a conversation whose grounding
//! no longer exists. They are also **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 crate::assistant::{
    bind_default_substrate, build_assistant_runtime, prompt, AssistantConfig, AssistantService,
};
use crate::coder::native_loop::TurnGenerator;
use crate::handler::JsonRpcMessage;
use crate::session::{ClientSession, ServerState, WsChannel};

/// 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,
    },
    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,
    pub created_at: u64,
    /// The connection that opened this discussion. Closing it closes the
    /// discussion — see the module docs on lifetime.
    owner_client_id: String,
    /// 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,
    /// 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>,
    /// 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>)>>,
}

impl DiscussionEntry {
    /// 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.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> {
    // `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 || {
        let repo = probe
            .canonicalize()
            .map_err(|e| format!("repo path {}: {e}", probe.display()))?;
        if !super::rpc::is_git_repo(&repo) {
            return Err(format!(
                "{} is not a git repository — discuss needs a repo to ground itself in",
                repo.display()
            ));
        }
        Ok(repo)
    })
    .await
    .map_err(|e| format!("repo probe failed: {e}"))??;

    // 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 = tokio::task::spawn_blocking(move || super::rpc::summarize_repo(&summarize))
        .await
        .map_err(|e| format!("repo summary 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 asm = build_assistant_runtime(engine, env, None, None, None, None).await;
    let system = format!(
        "{}\n\nYou are in a DISCUSSION about this repository, not a work session. \
         You have read-only access, scoped to this repository: you can read and reason \
         about the code here, but any attempt to write a file, run a shell command, or \
         read outside {} WILL be refused. Do not propose to make the change yourself — \
         help the operator decide what the change should be, what it must not break, and \
         how they would know it worked. Be concrete and cite real paths from the repo.",
        prompt::chat_prompt(&asm.description, &asm.tools),
        repo.display()
    );
    let cfg = AssistantConfig {
        model: None,
        strict_model: false,
        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_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,
        // Off, like every other call site — #813 gates the shape change on a
        // `car-bench` A/B, and a discussion is not where that gets decided.
        value_store_previews: false,
    };
    let service = Arc::new(AssistantService::new(
        generator.clone(),
        Arc::new(asm.runtime),
        cfg,
        system,
    ));

    let id = format!("disc-{}", uuid::Uuid::new_v4().simple());
    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(),
        created_at: now_secs(),
        owner_client_id: owner_client_id.to_string(),
        events,
        cmds,
        turns: AtomicU64::new(0),
        in_flight: AtomicBool::new(false),
        last_active: AtomicU64::new(now_secs()),
        turn_task: StdMutex::new(TurnSlot::default()),
        _slot: slot,
        service,
        generator,
        transcript: StdMutex::new(Vec::new()),
        last_promote: StdMutex::new(None),
    });
    state
        .coder_discussions
        .lock()
        .await
        .insert(id.clone(), entry);

    Ok(json!({
        "discussion_id": id,
        "repo": repo,
        "repo_summary": repo_summary,
    }))
}

/// 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}' — discussions are in-memory and do not \
                 survive a daemon restart; start a new one with coder.discuss.start"
            )
        })
}

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

/// Drive one assistant turn, translating its wire events into discussion
/// events and auto-denying every approval escalation.
async fn run_turn(entry: Arc<DiscussionEntry>, text: String) {
    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(&id, &text, None, 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.resolve_approval(approval_id, false);
                        }
                        entry
                            .emit(DiscussEventKind::ToolResult {
                                tool,
                                ok: false,
                                preview: "refused: a discussion is read-only — it cannot write \
                                          files or run commands. Describe the change instead; \
                                          `coder.start` is what performs it."
                                    .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;
                        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;
                    }
                    _ => {}
                }
            }
        })
        .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 (intent, constraints) = distill(
        &entry.generator,
        &entry.distill_transcript(),
        &entry.repo_summary,
    )
    .await?;
    *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}"
    ))
}

/// 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* is not: the
/// operator already supplied the intent, so the run proceeds with no extra
/// constraints rather than being blocked by a model hiccup.
///
/// 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?;
    let cached = entry.constraints();
    if !cached.is_empty() {
        return Ok(cached);
    }
    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"
        ));
    }
    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(),
        &entry.repo_summary,
    )
    .await
    {
        Ok((intent, constraints)) => {
            *lock(&entry.last_promote) = Some((intent, constraints.clone()));
            Ok(constraints)
        }
        Err(e) => {
            tracing::warn!(discussion_id, "discussion constraints unavailable: {e}");
            Ok(Vec::new())
        }
    }
}

/// 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,
}

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();
    start_discussion(state, &params.repo, &session.client_id, engine, generator).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}"))?;
    promote(state, &params.discussion_id, &session.client_id).await
}

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
}

/// `coder.discuss.list` — this connection's open discussions.
///
/// 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)));
    Ok(json!({ "discussions": rows }))
}

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

    /// 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()),
            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");
    }

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

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

    /// 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 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!([]),
                ),
            ],
            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;

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

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

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

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