//! `coder.discuss.*` — a repo-grounded, strictly **read-only** conversation
//! that can be distilled into a run intent.
//!
//! The gap this closes: `coder.start` demands a well-formed intent before
//! anything exists to react to. An operator who is still working out *what*
//! they want has no surface between "I have a vague idea" and "here is a
//! contract-worthy sentence" — so they either guess (and burn a session on a
//! badly-aimed contract) or go think somewhere else with none of the repo in
//! front of them.
//!
//! A discussion is grounded in the repo through the same
//! [`AssistantService`] that backs `car do`,
//! bound with `bind_default_substrate(prefer_local = true, full_access = false,
//! …)` — i.e. [`PermissionTier::ReadOnly`], where every write and every shell
//! escalates to an approval gate.
//!
//! [`PermissionTier::ReadOnly`]: car_policy::permission::PermissionTier
//!
//! ## What is actually enforced
//!
//! Two independent mechanisms, both required — an earlier version of this doc
//! claimed the discussion "never touches the repo", which overstated the first
//! and ignored that reads were unbounded:
//!
//! 1. **No mutation.** `write_file`, `edit_file` and `shell` are in the
//! ReadOnly tier's gated set, so each escalates to the approval gate — and
//! this surface **auto-DENIES** every escalation rather than prompting a
//! human. A discussion cannot write a file, run a command, create a branch,
//! or provision a worktree. The refusal is visible as a
//! `tool_result { ok: false }`, never silent.
//! 2. **No read *path* outside the repo.** The read tools (`read_file`,
//! `list_dir`, `find_files`, `grep_files`) are NOT gated — they are the
//! point of a grounded discussion — so mutation-gating alone left them
//! pointed at the whole filesystem. The discussion's bound environment
//! therefore sets [`BoundEnvironment::clamp_reads`], pinning those four
//! inside the repo root. Without it, a prompt-injected repo file could ask
//! for `grep_files {"path":"/Users/<user>","pattern":"sk-ant-"}` and the
//! hits would stream to every `coder.discuss.event` subscriber. Scoped to
//! this surface only; the general assistant's read reach is unchanged.
//!
//! Read the claim precisely: the clamp is **lexical**, not a resolved-path
//! check. `coder::policy::stays_under` normalizes `.` / `..` textually and
//! compares the result against the root (see its own
//! `stays_under_is_lexical_and_strict` test), and the file walk stats
//! entries with `metadata()`, which follows symlinks. So a symlink
//! *committed inside the repo* and pointing outward reads through the clamp
//! — its path stays under the root, its target does not. What the clamp
//! stops is the model **naming** a path outside the repo, which is the
//! prompt-injection vector above; it is not a containment boundary against
//! the repo's own contents. Treat the repo as trusted-to-the-extent-you-
//! trust-what-is-committed-in-it.
//!
//! [`BoundEnvironment::clamp_reads`]: crate::assistant::BoundEnvironment::clamp_reads
//!
//! `coder.start` is the only thing that starts work, and
//! `coder.discuss.promote` deliberately starts nothing — it hands back a
//! distilled intent the operator may edit first.
//!
//! ## Lifetime
//!
//! Model transcripts and action receipts are checkpointed through the daemon's
//! existing assistant oplog. The live runtime, substrate, replay stream and
//! connection ownership are process-local: `start { resume_id }` reconstructs
//! them from a checkpoint and a local repository/owner binding. Discussions
//! are **owned by the connection that opened
//! them**: only that connection may send to, subscribe to, promote or close
//! them, and closing it closes the discussion and cancels any in-flight turn,
//! because a detached turn would keep billing model tokens to nobody. Bounded
//! three ways — [`MAX_OPEN_DISCUSSIONS`], [`DISCUSSION_IDLE_TTL_SECS`], and the
//! per-discussion buffer/transcript caps.
//!
//! ## Event fanout
//!
//! One **drain task per discussion** owns the subscriber set. Emits, attaches
//! and detaches are commands on its channel, so a single task serializes them:
//! `seq` is assigned under the buffer lock by the only writer (no out-of-order
//! buffer), an attach replays everything buffered *before* the next emit is
//! processed (no gap, no duplicate), and — unlike the `coder.event` path — no
//! lock is ever held across a send.
//!
//! The drain does not send, though: **each subscriber owns a bounded queue and
//! its own sender task**. That is the part that makes a wedged subscriber
//! merely its own problem. When the drain itself performed the sends, an
//! untimed write to a half-open socket blocked the drain, so the *next* `Emit`
//! command sat unprocessed — and since every `entry.emit(…).await` inside
//! [`run_turn`] waits for its `seq`, one wedged board stalled the whole
//! **turn**, not just its stream. Now the drain only `try_send`s into each
//! subscriber's queue: a subscriber that cannot keep up (queue full, or a send
//! past [`DISCUSS_SEND_TIMEOUT`]) is **shed**, and the turn never waits on a
//! socket.
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex as StdMutex, MutexGuard};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tokio::sync::{mpsc, oneshot};
use super::discussion_record::DiscussionRecord;
use crate::assistant::governance::AssistantDurability;
use crate::assistant::{
bind_default_substrate, build_assistant_runtime_with_tools, prompt, AssistantConfig,
AssistantService,
};
use crate::coder::native_loop::TurnGenerator;
use crate::handler::JsonRpcMessage;
use crate::session::{ClientSession, ServerState, WsChannel};
const DISCUSSION_MUTATION_REFUSAL: &str = "This conversation stage is read-only; this is not a file-permission problem or a user refusing the requested change. Use prepare_coding_task with the requested change and constraints so the user can review it and start coding. Do not change file permissions or ask the user to do so.";
/// Turn cap for one discussion reply. A discussion reads and reasons; it never
/// edits, so it has no repair loop to spend turns on.
const DISCUSS_MAX_TURNS: u32 = 12;
/// Attempts allowed when distilling a transcript into an intent. Same bounded
/// shape as `derive_contract`: the output is structured JSON, so a malformed
/// reply is worth one retry, not an unbounded loop.
const PROMOTE_MAX_ATTEMPTS: u32 = 3;
/// Concurrent open discussions per daemon. Each pins an `AssistantService`, a
/// `Runtime`, and an open runtime session, so they are not free; a board opens
/// one at a time and an operator juggling more than a handful has lost track.
///
/// Enforced by [`ServerState::coder_discussion_slots`], a semaphore whose
/// permit is taken before any of `start_discussion`'s async work and lives
/// inside the admitted [`DiscussionEntry`] — NOT by counting the registry, which
/// was a TOCTOU check that bounded nothing under pipelined starts.
///
/// [`ServerState::coder_discussion_slots`]: crate::session::ServerState
pub(crate) const MAX_OPEN_DISCUSSIONS: usize = 8;
/// A discussion with no activity for this long is reaped on the next
/// `coder.discuss.start`. Long enough to step away from a train of thought,
/// short enough that a forgotten one does not pin a runtime overnight.
const DISCUSSION_IDLE_TTL_SECS: u64 = 60 * 60;
/// Retained events per discussion. The oldest are dropped past this; a replay
/// from a trimmed cursor returns what survives (`events_replayed` says how
/// much) rather than growing without bound on a long conversation.
const DISCUSS_EVENT_BUFFER_MAX: usize = 2000;
/// Transcript turns retained for distillation. `promote` is a summarization
/// call, so the recent exchange is what carries the intent; keeping everything
/// eventually builds a prompt no model window holds.
const TRANSCRIPT_MAX_TURNS: usize = 40;
/// Turns handed to `distill`. The most recent slice of the retained transcript
/// — the tail is where the operator converged.
const DISTILL_WINDOW_TURNS: usize = 12;
/// Byte cap on one operator message.
///
/// tungstenite accepts up to 64 MiB per frame, and an accepted message is
/// cloned into the transcript, cloned again into the event buffer, and rendered
/// into the distill prompt — so without a cap, 40 sequential 50 MB sends retain
/// gigabytes per discussion and make `promote` build a prompt no window holds.
/// `summarize_repo` is head-capped for exactly this reason; operator text needs
/// the same. Generous for prose — this is a conversation, not a file upload.
const DISCUSS_MESSAGE_MAX_BYTES: usize = 64 * 1024;
/// Depth of one subscriber's outbound frame queue.
///
/// Must exceed [`DISCUSS_EVENT_BUFFER_MAX`] so a legitimate
/// `subscribe { from_seq: 0 }` replay — up to a full buffer, queued in one go —
/// is never mistaken for a slow consumer. Past that, a subscriber this far
/// behind is not reading.
const DISCUSS_SUBSCRIBER_QUEUE: usize = DISCUSS_EVENT_BUFFER_MAX + 128;
/// How long one frame may take to reach a subscriber's socket before that
/// subscriber is shed. A half-open peer never fails a write — it parks forever,
/// holding the socket's write half. Matches the coder fanout's deadline.
const DISCUSS_SEND_TIMEOUT: std::time::Duration = crate::coder::rpc::FANOUT_WRITE_TIMEOUT;
/// Live discussions keyed by `discussion_id`.
pub type DiscussionMap = HashMap<String, Arc<DiscussionEntry>>;
/// Take a `std` lock without letting a poisoned mutex become permanent.
///
/// A panic anywhere under one of these locks would otherwise brick the
/// discussion for its whole lifetime — and the first panic is swallowed by the
/// detached turn task, so the operator would see an inexplicably dead
/// conversation with no error. The data behind each of these is a plain
/// `Vec`/`Option`; a torn write is not a safety problem here.
fn lock<T>(m: &StdMutex<T>) -> MutexGuard<'_, T> {
m.lock().unwrap_or_else(|e| e.into_inner())
}
/// One event in a discussion's stream. `seq` is monotonic per discussion so a
/// client can resume from a cursor, exactly like `CoderEvent`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DiscussEvent {
pub discussion_id: String,
pub seq: u64,
pub ts: u64,
#[serde(flatten)]
pub kind: DiscussEventKind,
}
/// What happened in a discussion. Serialized with `"type":"snake_case_name"`,
/// tagged the same way [`CoderEventKind`](super::session::CoderEventKind) is.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum DiscussEventKind {
UserMessage {
text: String,
},
/// A streaming chunk of the model's reply.
AssistantDelta {
text: String,
},
/// The complete assistant turn.
AssistantMessage {
text: String,
},
ToolCall {
tool: String,
params_preview: String,
},
ToolResult {
tool: String,
ok: bool,
preview: String,
},
TaskPrepared {
proposed_intent: String,
constraints: Vec<String>,
},
TurnComplete {},
Error {
message: String,
},
}
/// A command for a discussion's drain task — the single owner of its
/// subscriber set and the only thing that writes its buffer or sends a frame.
enum StreamCmd {
Emit(DiscussEventKind, oneshot::Sender<u64>),
Attach {
client_id: String,
channel: Arc<WsChannel>,
from_seq: u64,
replayed: oneshot::Sender<u64>,
},
Detach(String),
}
/// The in-flight turn's handle and the discussion's terminal `closed` latch,
/// deliberately behind one lock.
///
/// They were separate, and the gap between them orphaned model loops: `close`
/// read `turn_task` (still `None`, because `send_message` stores the handle
/// only *after* its first `emit().await`), found nothing to abort, and removed
/// the entry from the registry — then `send_message` resumed and spawned a turn
/// against a discussion nothing could reach any more. It billed up to
/// [`DISCUSS_MAX_TURNS`] turns against a live provider with no way to stop it.
/// Publishing "this discussion is closed" and "here is the turn to abort"
/// through the same lock closes that window in both directions: a close either
/// aborts the running turn or latches `closed` so the turn is never spawned.
#[derive(Default)]
struct TurnSlot {
/// Set once, terminally, by [`DiscussionEntry::cancel_turn`]. Every caller
/// of `cancel_turn` also removes the entry from the registry, so there is
/// no legitimate reopen.
closed: bool,
handle: Option<tokio::task::JoinHandle<()>>,
}
/// Clears `in_flight` on EVERY exit path, including a cancelled or panicking
/// handler future.
///
/// `in_flight` was set by CAS in `send_message` and cleared only at the tail of
/// the spawned turn task. Anything that dropped the handler future between
/// those two points — the daemon's handler deadline is the reachable one, since
/// `coder.discuss.send` is not deadline-exempt — left it `true` with no turn
/// running. The discussion then answered "still answering the previous message"
/// to every `send` and "still answering" to every `promote`, forever, and
/// `reap_idle` runs only on the next `discuss.start`, so on a quiet daemon it
/// was never reclaimed either. A latch that only one code path can release is
/// a latch that leaks; this releases in `Drop`.
struct InFlightGuard(Arc<DiscussionEntry>);
impl Drop for InFlightGuard {
fn drop(&mut self) {
self.0.in_flight.store(false, Ordering::SeqCst);
self.0.touch();
}
}
/// Keeps the operator's turn in the transcript only if a reply turn was
/// actually dispatched for it.
///
/// `send_message` records the turn before the `emit().await` it may be
/// cancelled at, and before the dispatch that may be refused. `InFlightGuard`
/// frees the discussion on those paths, but the transcript was left ending in
/// an operator question with no reply — and that is exactly the input the
/// `is_answering()` guards on `promote` and `coder.start { discussion_id }`
/// exist to keep out of distillation. Those guards read "not answering", so a
/// stranded question sails through them and the model invents a confident
/// intent from a question nobody answered. Recording after the dispatch would
/// let the spawned turn's `Assistant` row land first, so the row goes in early
/// and comes back out on every path that did not dispatch.
struct TurnRecordGuard {
entry: Arc<DiscussionEntry>,
text: String,
dispatched: bool,
}
impl Drop for TurnRecordGuard {
fn drop(&mut self) {
if !self.dispatched {
self.entry.rollback_turn("Operator", &self.text);
}
}
}
/// One live discussion.
pub struct DiscussionEntry {
pub id: String,
/// The git repo the conversation is grounded in.
pub repo: PathBuf,
/// Cheap repo orientation, returned by `coder.discuss.start` so a caller
/// can show what the discussion can see.
pub repo_summary: String,
project_context: String,
pub created_at: u64,
pub model: Option<String>,
/// The connection that opened this discussion. Closing it closes the
/// discussion — see the module docs on lifetime.
owner_client_id: String,
principal: String,
record_root: PathBuf,
/// Replay buffer. Written **only** by the drain task, so it is always in
/// `seq` order; readable elsewhere for inspection.
pub events: Arc<tokio::sync::Mutex<Vec<DiscussEvent>>>,
/// Commands to the drain task.
cmds: mpsc::UnboundedSender<StreamCmd>,
/// Completed operator turns (what `turns` reports in `coder.discuss.list`).
turns: AtomicU64,
/// Whether a reply turn is running right now. A discussion is a
/// conversation: two overlapping turns interleave into one model thread and
/// silently lose one of them, so a second `send` is refused rather than
/// queued.
in_flight: AtomicBool,
start_lock: Arc<tokio::sync::Mutex<()>>,
/// Last activity, for the idle TTL.
last_active: AtomicU64,
/// The in-flight turn's task plus the terminal `closed` latch, under ONE
/// lock. See [`TurnSlot`] for why they cannot be separate.
turn_task: StdMutex<TurnSlot>,
/// This discussion's open-slot reservation, taken before any of
/// `start_discussion`'s async work and released when the entry drops.
_slot: tokio::sync::OwnedSemaphorePermit,
/// The grounded, read-only conversational service.
service: Arc<AssistantService>,
durability: Arc<crate::assistant::durability::LocalAssistantDurability>,
/// The model seam used for distillation (`promote`). Same injection style
/// `derive_contract` uses, so promote is testable with a scripted model.
generator: Arc<dyn TurnGenerator>,
/// Role-tagged plain-text transcript, kept for distillation. Deliberately
/// separate from the service's own message thread: promote must see the
/// conversation, not the tool plumbing. Capped at [`TRANSCRIPT_MAX_TURNS`].
transcript: StdMutex<Vec<(&'static str, String)>>,
/// The most recent `promote` result, cached so `coder.start
/// { discussion_id }` can fold the agreed constraints into contract
/// derivation without a second distillation call.
last_promote: StdMutex<Option<(String, Vec<String>)>>,
task_proposal: super::task_proposal::PreparedTask,
}
impl DiscussionEntry {
fn save_pending_task(&self, proposal: Option<(String, Vec<String>)>) -> Result<(), String> {
let mut record = DiscussionRecord::load(&self.record_root, &self.id, &self.principal)?;
record.pending_task = proposal;
record.save_model(&self.record_root)
}
/// Constraints agreed in this discussion, from the last `promote`.
pub fn constraints(&self) -> Vec<String> {
lock(&self.last_promote)
.as_ref()
.map(|(_, c)| c.clone())
.unwrap_or_default()
}
/// Whether a reply turn is running right now.
pub fn is_answering(&self) -> bool {
self.in_flight.load(Ordering::SeqCst)
}
fn touch(&self) {
self.last_active.store(now_secs(), Ordering::SeqCst);
}
fn idle_secs(&self) -> u64 {
now_secs().saturating_sub(self.last_active.load(Ordering::SeqCst))
}
fn record_turn(&self, role: &'static str, text: &str) {
if text.trim().is_empty() {
return;
}
let mut t = lock(&self.transcript);
t.push((role, text.to_string()));
// Bounded: drop from the front, keeping the recent exchange.
let len = t.len();
if len > TRANSCRIPT_MAX_TURNS {
t.drain(..len - TRANSCRIPT_MAX_TURNS);
}
}
/// Undo the most recent [`record_turn`](Self::record_turn) when it is still
/// the tail and still ours. Matching on both role and text is what keeps a
/// rollback from eating someone else's row if the transcript moved on.
fn rollback_turn(&self, role: &'static str, text: &str) {
let mut t = lock(&self.transcript);
if t.last().is_some_and(|(r, s)| *r == role && s == text) {
t.pop();
}
}
/// The most recent turns, rendered for distillation.
fn distill_transcript(&self) -> String {
let t = lock(&self.transcript);
let start = t.len().saturating_sub(DISTILL_WINDOW_TURNS);
t[start..]
.iter()
.map(|(role, text)| format!("{role}: {text}"))
.collect::<Vec<_>>()
.join("\n\n")
}
fn transcript_is_empty(&self) -> bool {
lock(&self.transcript).is_empty()
}
/// Append an event to the stream, returning its assigned `seq`.
///
/// The drain assigns the seq under the buffer lock, so the buffer is always
/// ordered; this only waits for that assignment, never for a WS send.
async fn emit(&self, kind: DiscussEventKind) -> u64 {
let (tx, rx) = oneshot::channel();
if self.cmds.send(StreamCmd::Emit(kind, tx)).is_err() {
return 0; // drain gone (discussion closed) — nothing to stream to
}
rx.await.unwrap_or(0)
}
/// Stop an in-flight turn and latch the discussion closed: signal the loop,
/// drop the task, and make sure no turn that is still being dispatched can
/// start behind us.
///
/// Terminal by construction — every caller (`close`, disconnect teardown,
/// `reap_idle`) also removes the entry from the registry.
fn cancel_turn(&self) {
self.durability.revoke();
self.service.cancel(&self.id);
{
let mut slot = lock(&self.turn_task);
slot.closed = true;
if let Some(handle) = slot.handle.take() {
handle.abort();
}
}
self.in_flight.store(false, Ordering::SeqCst);
}
/// Spawn the reply turn under the same lock `cancel_turn` latches, so a
/// close that raced the dispatch either aborts this turn or prevents it.
///
/// Returns `false` when the discussion was closed before the dispatch
/// reached this point — the turn is then never spawned at all.
fn spawn_turn<F>(&self, make: F) -> bool
where
F: FnOnce() -> tokio::task::JoinHandle<()>,
{
let mut slot = lock(&self.turn_task);
if slot.closed {
return false;
}
// No await under this guard: `tokio::spawn` only queues the task.
slot.handle = Some(make());
true
}
fn summary_row(&self) -> Value {
json!({
"discussion_id": self.id,
"repo": self.repo,
"created_at": self.created_at,
"turns": self.turns.load(Ordering::SeqCst),
})
}
}
fn now_secs() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
fn event_frame(event: &DiscussEvent) -> Option<String> {
serde_json::to_string(&json!({
"jsonrpc": "2.0",
"method": "coder.discuss.event",
"params": event,
}))
.ok()
}
/// One subscriber's outbound lane: a bounded frame queue plus the task that
/// drains it onto that subscriber's socket.
///
/// One lane per subscriber is what decouples the stream from the turn. The
/// drain hands frames over with `try_send` and never awaits a socket, so no
/// subscriber can delay the `seq` reply the turn is blocked on.
struct Subscriber {
frames: mpsc::Sender<String>,
task: tokio::task::JoinHandle<()>,
}
impl Drop for Subscriber {
/// Abort rather than let the queue drain: the task may be parked on a
/// half-open socket's write mutex, and that parked future is precisely what
/// keeps the write half alive past teardown.
fn drop(&mut self) {
self.task.abort();
}
}
fn spawn_subscriber(channel: Arc<WsChannel>) -> Subscriber {
let (frames, mut rx) = mpsc::channel::<String>(DISCUSS_SUBSCRIBER_QUEUE);
let task = tokio::spawn(async move {
while let Some(frame) = rx.recv().await {
if tokio::time::timeout(
DISCUSS_SEND_TIMEOUT,
crate::coder::rpc::send_frame(&channel, &frame),
)
.await
.is_err()
{
// Wedged socket. Ending the task drops the channel handle and
// closes the queue, so the drain sheds this subscriber on its
// next `try_send` instead of queueing for a peer that is gone.
break;
}
}
});
Subscriber { frames, task }
}
/// The per-discussion drain: the single writer of the buffer and the single
/// owner of the subscriber set.
///
/// Because one task handles emits and attaches in order, an attach replays
/// everything buffered so far and is registered before the next emit is
/// processed — no gap and no duplicate — without holding any lock across a
/// handoff. The drain itself never touches a socket: it `try_send`s into each
/// subscriber's own queue, so a subscriber that has stopped reading is shed
/// rather than allowed to stall the buffer, the next `Emit`, or the turn
/// waiting on that `Emit`'s `seq`.
fn spawn_discuss_drain(
discussion_id: String,
events: Arc<tokio::sync::Mutex<Vec<DiscussEvent>>>,
) -> mpsc::UnboundedSender<StreamCmd> {
let (tx, mut rx) = mpsc::unbounded_channel::<StreamCmd>();
tokio::spawn(async move {
let mut subscribers: HashMap<String, Subscriber> = HashMap::new();
let mut next_seq: u64 = 0;
while let Some(cmd) = rx.recv().await {
match cmd {
StreamCmd::Emit(kind, reply) => {
let seq = next_seq;
next_seq += 1;
let event = DiscussEvent {
discussion_id: discussion_id.clone(),
seq,
ts: now_secs(),
kind,
};
let frame = event_frame(&event);
{
let mut buffer = events.lock().await;
buffer.push(event);
let len = buffer.len();
if len > DISCUSS_EVENT_BUFFER_MAX {
buffer.drain(..len - DISCUSS_EVENT_BUFFER_MAX);
}
} // lock released BEFORE the handoff
let _ = reply.send(seq);
if let Some(frame) = &frame {
// `try_send`, never `send`: a full queue means this
// subscriber is not draining, and waiting for it is how
// one wedged board used to stall the whole turn.
subscribers.retain(|client_id, s| {
let ok = s.frames.try_send(frame.clone()).is_ok();
if !ok {
tracing::warn!(
discussion_id = %discussion_id,
client_id = %client_id,
"discussion subscriber is not draining; dropping it"
);
}
ok
});
}
}
StreamCmd::Attach {
client_id,
channel,
from_seq,
replayed,
} => {
// Clone the frames under the lock, release, then queue.
let frames: Vec<String> = {
let buffer = events.lock().await;
buffer
.iter()
.filter(|e| e.seq >= from_seq)
.filter_map(event_frame)
.collect()
};
let subscriber = spawn_subscriber(channel);
// The queue is sized to hold a whole buffer replay, so this
// only short-circuits if the peer's lane already died.
let mut n = 0u64;
for frame in frames {
if subscriber.frames.try_send(frame).is_err() {
break;
}
n += 1;
}
subscribers.insert(client_id, subscriber);
let _ = replayed.send(n);
}
StreamCmd::Detach(client_id) => {
subscribers.remove(&client_id);
}
}
}
// Discussion closed: every lane's task is aborted by `Subscriber::drop`.
});
tx
}
// ---------------------------------------------------------------------------
// Orchestration (generation-injectable, transport-free)
// ---------------------------------------------------------------------------
/// Provision a discussion grounded in `repo`, owned by `owner_client_id`.
///
/// `engine` builds the read-only assistant runtime (tools, substrate, gates);
/// `generator` is the model seam both the conversation and `promote` run on.
/// Split so tests can drive a scripted model against a real temp repo.
pub async fn start_discussion(
state: &Arc<ServerState>,
repo: &Path,
owner_client_id: &str,
engine: Arc<car_inference::InferenceEngine>,
generator: Arc<dyn TurnGenerator>,
) -> Result<Value, String> {
open_discussion(
state,
repo,
owner_client_id,
engine,
generator,
owner_client_id,
None,
)
.await
}
pub(super) async fn open_discussion(
state: &Arc<ServerState>,
repo: &Path,
owner_client_id: &str,
engine: Arc<car_inference::InferenceEngine>,
generator: Arc<dyn TurnGenerator>,
principal: &str,
resume_id: Option<&str>,
) -> Result<Value, String> {
open_discussion_with_model(
state,
repo,
owner_client_id,
engine,
generator,
principal,
resume_id,
None,
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn open_discussion_with_model(
state: &Arc<ServerState>,
repo: &Path,
owner_client_id: &str,
engine: Arc<car_inference::InferenceEngine>,
generator: Arc<dyn TurnGenerator>,
principal: &str,
resume_id: Option<&str>,
requested_model: Option<&str>,
) -> Result<Value, String> {
let _recovery = match resume_id {
Some(_) => Some(state.coder_discussion_recovery.lock().await),
None => None,
};
let saved = resume_id
.map(|id| DiscussionRecord::load(&state.journal_dir, id, principal))
.transpose()?;
if let Some(record) = &saved {
if state
.coder_discussions
.lock()
.await
.contains_key(&record.id)
{
return Err(
"conversation is already open; close its other window before resuming".into(),
);
}
}
// `canonicalize` and the `git rev-parse` probe are blocking syscalls (the
// probe forks), so they go to a blocking worker rather than parking a tokio
// runtime thread on fork/exec.
let probe = repo.to_path_buf();
let repo = tokio::task::spawn_blocking(move || {
// The same root `coder.start` keys tasks by; a subdirectory here would
// fail the conversation's own task admission (repo mismatch).
super::rpc::repo_toplevel(&probe)
.map_err(|e| format!("{e} — discuss needs a repo to ground itself in"))
})
.await
.map_err(|e| format!("repo probe failed: {e}"))??;
if saved.as_ref().is_some_and(|record| record.repo != repo) {
return Err("saved conversation belongs to a different repository".into());
}
if let (Some(record), Some(requested)) = (&saved, requested_model) {
let requested = requested.trim();
let requested = if requested.is_empty() || requested == "auto" {
None
} else {
Some(requested)
};
if requested != record.model.as_deref() {
let runs =
coding_runs(state, &record.id, &repo, super::rpc::coder_state_dir()?).await?;
if runs.iter().any(|run| {
!matches!(
run["state"].as_str(),
Some("merged" | "reported" | "failed" | "abandoned")
)
}) {
return Err(
"Finish or stop this conversation's active task before changing its model."
.into(),
);
}
}
}
// Reap idle discussions before enforcing the cap, so a forgotten one from
// this morning never blocks a new one this afternoon.
reap_idle(state).await;
// RESERVE the slot before any of the work below. Counting the registry here
// and inserting after `bind_default_substrate` + `build_assistant_runtime`
// was a TOCTOU check: the daemon runs a connection's requests concurrently,
// so N pipelined starts all read the same count, all passed, and all built
// a runtime — the cap bounded nothing. The permit lives in the entry and
// comes back if any step below fails.
let slot = state
.coder_discussion_slots
.clone()
.try_acquire_owned()
.map_err(|_| {
format!(
"{MAX_OPEN_DISCUSSIONS} discussions are already open — close one with \
coder.discuss.close before starting another"
)
})?;
let summarize = repo.clone();
let (repo_summary, project_context) = tokio::task::spawn_blocking(move || {
(
super::rpc::summarize_repo(&summarize),
super::project_context::project_context(&summarize).unwrap_or_default(),
)
})
.await
.map_err(|e| format!("repo context failed: {e}"))?;
// prefer_local = true, full_access = false ⇒ PermissionTier::ReadOnly:
// every write/shell escalates to the approval gate, which this surface
// auto-denies (see the `approval_pending` arm in `run_turn`). No Docker
// preflight either — a discussion must open promptly.
let mut env = bind_default_substrate(true, false, &repo, None).await;
// ...and the read tools are pinned to the repo too. Mutation-gating alone
// left `read_file`/`list_dir`/`find_files`/`grep_files` pointed at the
// whole filesystem, whose output streams to every subscriber.
env.clamp_reads = true;
let task_proposal = Arc::new(StdMutex::new(None));
let proposal_tool = Arc::new(super::task_proposal::TaskProposalTool(
task_proposal.clone(),
));
let mut asm = build_assistant_runtime_with_tools(
engine.clone(),
env,
None,
None,
None,
None,
false,
vec![super::task_proposal::TaskProposalTool::definition()],
vec![proposal_tool],
)
.await?;
// Advertise the tools useful before a coding task starts. The general
// assistant also knows about mail, calendars, media and file mutations;
// offering those here invites calls this conversation will refuse and
// needlessly enlarges every inference request. Keep the shared executor
// and approval gates intact for stale/hallucinated calls.
asm.tools.retain(|tool| {
matches!(
tool["name"].as_str(),
Some(
"read_file"
| "list_dir"
| "find_files"
| "grep_files"
| "calculate"
| "web_search"
| "http_request"
| "prepare_coding_task"
)
)
});
let system = format!(
"{}\n\n{project_context}\n\nRepository guidance applies to your analysis and proposed work; it does not expand the read-only permissions below.\n\nYou are the coding assistant for repository {}. \
When the user asks you to implement or fix something, call prepare_coding_task with \
their requested change and constraints. This is how you begin implementation from \
this conversation. The interface will show an editable task and verification review \
before execution. For questions and planning-only requests, inspect the repository \
and answer without preparing a task. Use real paths. \
The current conversation tools can inspect the repository and prepare coding tasks; \
file edits and shell commands run in the subsequent coding task. A refused write \
in this stage is NOT evidence that the file is read-only or that the user declined \
implementation. Do not repeat earlier file-permission claims without current evidence \
or ask the user to change permissions to start coding. Use prepare_coding_task instead. \
Reads outside the repository remain refused. Do not require the user to know a \
command or runtime API. Preparing is not execution: never say files have changed \
or work has started.",
prompt::chat_prompt(&asm.identity, &asm.description, &asm.tools),
repo.display()
);
let model = match requested_model {
Some(value) if value.trim().is_empty() || value.trim() == "auto" => None,
Some(value) => Some(value.trim().to_string()),
None => saved.as_ref().and_then(|record| record.model.clone()),
};
if let Some(model) = &model {
let schema = engine.model_schema(model).ok_or_else(|| {
format!("Unknown model '{model}'. Use `car models list --capability tool_use` to choose a model, or --model auto to clear the saved choice.")
})?;
if !schema
.capabilities
.contains(&car_inference::schema::ModelCapability::ToolUse)
{
return Err(format!(
"Model '{model}' cannot call repository tools. Choose a tool-capable model with `car models list --capability tool_use`, or use --model auto."
));
}
}
let generator: Arc<dyn TurnGenerator> = match &model {
Some(model) => Arc::new(super::discussion_model::DiscussionModel {
inner: generator,
model: model.clone(),
}),
None => generator,
};
let cfg = AssistantConfig {
model: model.clone(),
strict_model: model.is_some(),
max_turns: DISCUSS_MAX_TURNS,
tools: asm.tools.clone(),
gated_tools: asm.gated_tools.clone(),
approval_policy: None,
// A discussion writes nothing — including durable memory. Leaving the
// proactive-memory bank unbound keeps `remember` out of the loop's
// automatic pass; the tool itself is gated and auto-denied anyway.
proactive_memory: None,
tool_memory: None,
tool_labels: None,
// A discussion has no task list: it executes nothing, so there is no
// run for #814's per-turn state block to describe.
todos: None,
// The shipped default, like every other production call site — #813's
// A/B has been run and chose it; a discussion is not where that gets
// re-decided.
value_store_previews: crate::assistant::agent_loop::VALUE_STORE_PREVIEWS_DEFAULT,
response_format: None,
context_window_override: None,
refuse_unadvertised_tools: false,
response_format_validator: None,
delegate_budget: None,
};
let mut record = saved.unwrap_or_else(|| DiscussionRecord {
id: format!("disc-{}", uuid::Uuid::new_v4().simple()),
repo: repo.clone(),
principal: principal.to_string(),
created_at: now_secs(),
model: model.clone(),
pending_task: None,
});
record.model = model.clone();
let id = record.id.clone();
let durability = Arc::new(crate::assistant::durability::LocalAssistantDurability::new(
state.sync_subsystem()?,
id.clone(),
repo.clone(),
));
let mut messages = if resume_id.is_some() {
durability
.load_checkpoint(&id)
.await?
.ok_or("saved conversation has no durable transcript")?
.messages
} else {
Vec::new()
};
// Rebind current runtime instructions while retaining exact provider/tool
// history. Interrupted tool exchanges are reconciled by AssistantService.
if let Some(car_inference::Message::System { content }) = messages.first_mut() {
*content = system.clone();
} else {
messages.insert(
0,
car_inference::Message::System {
content: system.clone(),
},
);
}
durability
.checkpoint(
&id,
&messages,
if resume_id.is_some() {
"conversation_resume"
} else {
"conversation_open"
},
None,
)
.await?;
if resume_id.is_none() {
record.save(&state.journal_dir)?;
} else if requested_model.is_some() {
record.save_model(&state.journal_dir)?;
}
let service = Arc::new(AssistantService::new_durable(
generator.clone(),
Arc::new(asm.runtime),
cfg,
system,
durability.clone(),
repo.clone(),
));
let events = Arc::new(tokio::sync::Mutex::new(Vec::new()));
let cmds = spawn_discuss_drain(id.clone(), events.clone());
let entry = Arc::new(DiscussionEntry {
id: id.clone(),
repo: repo.clone(),
repo_summary: repo_summary.clone(),
project_context,
created_at: record.created_at,
model: model.clone(),
owner_client_id: owner_client_id.to_string(),
principal: principal.to_string(),
record_root: state.journal_dir.clone(),
events,
cmds,
turns: AtomicU64::new(0),
in_flight: AtomicBool::new(false),
start_lock: Arc::new(tokio::sync::Mutex::new(())),
last_active: AtomicU64::new(now_secs()),
turn_task: StdMutex::new(TurnSlot::default()),
_slot: slot,
service,
durability,
generator,
transcript: StdMutex::new(Vec::new()),
last_promote: StdMutex::new(record.pending_task.clone()),
task_proposal,
});
// Project recorded conversation and tool activity without executing it.
// Checkpoints do not retain a typed success flag for every tool result, so
// do not invent result status from arbitrary tool-output text. Full history
// beyond model-context compaction still requires the presentation log.
for message in messages {
let (role, kind, text) = match message {
car_inference::Message::User { content } => ("Operator", true, content),
car_inference::Message::Assistant {
content,
tool_calls,
..
} if tool_calls.is_empty() => ("Assistant", false, content),
car_inference::Message::Assistant { tool_calls, .. } => {
for call in tool_calls {
entry
.emit(DiscussEventKind::ToolCall {
tool: call.name,
params_preview: preview(
&serde_json::to_string(&call.arguments).unwrap_or_default(),
),
})
.await;
}
continue;
}
_ => continue,
};
if text.trim().is_empty() {
continue;
}
entry.record_turn(role, &text);
if kind {
entry.emit(DiscussEventKind::UserMessage { text }).await;
} else {
entry.turns.fetch_add(1, Ordering::SeqCst);
entry
.emit(DiscussEventKind::AssistantMessage { text })
.await;
}
}
if let Some((proposed_intent, constraints)) = record.pending_task {
entry
.emit(DiscussEventKind::TaskPrepared {
proposed_intent,
constraints,
})
.await;
}
state
.coder_discussions
.lock()
.await
.insert(id.clone(), entry);
Ok(json!({
"discussion_id": id,
"repo": repo,
"repo_summary": repo_summary,
"persistent": true,
"resumed": resume_id.is_some(),
"model": model,
}))
}
/// Close discussions idle past [`DISCUSSION_IDLE_TTL_SECS`].
async fn reap_idle(state: &Arc<ServerState>) {
let stale: Vec<Arc<DiscussionEntry>> = {
let open = state.coder_discussions.lock().await;
open.values()
.filter(|e| e.idle_secs() > DISCUSSION_IDLE_TTL_SECS)
.cloned()
.collect()
};
for entry in stale {
entry.cancel_turn();
state.coder_discussions.lock().await.remove(&entry.id);
}
}
async fn get_discussion(
state: &Arc<ServerState>,
discussion_id: &str,
) -> Result<Arc<DiscussionEntry>, String> {
state
.coder_discussions
.lock()
.await
.get(discussion_id)
.cloned()
.ok_or_else(|| {
format!(
"no open discussion '{discussion_id}' — reopen a saved conversation with \
coder.discuss.start {{repo, resume_id}}, or start a new one"
)
})
}
pub(super) async fn selected_model(
state: &Arc<ServerState>,
discussion_id: &str,
) -> Result<Option<String>, String> {
Ok(get_discussion(state, discussion_id).await?.model.clone())
}
/// Resolve a discussion **and prove the caller owns it**.
///
/// Ownership was recorded but only ever consulted by disconnect teardown, so
/// every `coder.discuss.*` method resolved by id alone: any connected client
/// could send into, subscribe to, promote, or close another connection's
/// discussion — closing one mid-turn was the sharp end, since it cancels a turn
/// the owner is watching. Discussions are already per-connection and die with
/// their connection, so refusing here is the same model, enforced.
pub(crate) async fn get_owned_discussion(
state: &Arc<ServerState>,
discussion_id: &str,
client_id: &str,
) -> Result<Arc<DiscussionEntry>, String> {
let entry = get_discussion(state, discussion_id).await?;
if entry.owner_client_id != client_id {
return Err(format!(
"discussion '{discussion_id}' belongs to another connection — a discussion is \
owned by the connection that opened it and closes with it; start your own with \
coder.discuss.start"
));
}
Ok(entry)
}
/// Send one operator message and run the reply turn.
///
/// Returns once the turn is dispatched and has emitted its first event,
/// carrying that event's `seq` (the `user_message`), so a caller that has not
/// yet subscribed can resume from exactly there without missing or replaying a
/// frame. A refused dispatch emits nothing at all.
///
/// **One turn at a time.** A `send` arriving while a turn is in flight is
/// REFUSED, not queued: both turns clone the same model thread and the last one
/// to finish overwrites the other, so the earlier exchange vanishes from the
/// conversation — and from what `promote` later distills. Refusing is the
/// honest answer; the caller retries when `turn_complete` lands.
pub async fn send_message(
state: &Arc<ServerState>,
discussion_id: &str,
client_id: &str,
text: &str,
) -> Result<Value, String> {
let entry = get_owned_discussion(state, discussion_id, client_id).await?;
if text.trim().is_empty() {
return Err("discuss message is empty".to_string());
}
if text.len() > DISCUSS_MESSAGE_MAX_BYTES {
return Err(format!(
"that message is {} bytes; the limit is {DISCUSS_MESSAGE_MAX_BYTES}. A discussion \
keeps every message in its transcript, its replay buffer, and its distillation \
prompt — point at a file in the repo instead of pasting it",
text.len()
));
}
if entry
.in_flight
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
.is_err()
{
return Err(format!(
"{discussion_id} is still answering the previous message — wait for \
`turn_complete` before sending another"
));
}
// Armed IMMEDIATELY after the CAS: if this future is dropped before the
// turn owns it, the guard's Drop is the only thing that stops the
// discussion latching "still answering" forever with nothing running.
//
// Held in an `Option` so a REFUSED dispatch leaves it here rather than
// dropping it inside the `spawn_turn(…)` expression: it then drops at this
// function's scope exit, AFTER `recorded` (declared below, so it drops
// first) has rolled the transcript row back. Otherwise `in_flight` reads
// false while the stranded operator row is still visible — the reverse of
// the cancellation path's order.
let mut guard = Some(InFlightGuard(entry.clone()));
entry.touch();
entry.record_turn("Operator", text);
// ...and armed with it, for the same reason: a dispatch refused by a racing
// `close` must not leave the transcript ending in an operator question no
// turn will ever answer.
let mut recorded = TurnRecordGuard {
entry: entry.clone(),
text: text.to_string(),
dispatched: false,
};
let task_entry = entry.clone();
let task_state = state.clone();
let prompt_text = text.to_string();
// The `user_message` is emitted INSIDE the turn, as its first act — not
// here, before the dispatch is known to have happened. Emitting it first
// put it in the replay buffer and on every subscriber even when the
// dispatch was refused and `TurnRecordGuard` rolled the transcript row
// back: the board then rendered the operator's question followed by
// permanent silence. Emitting from the turn makes the event and the
// transcript row commit or roll back together, and makes the ordering
// (`user_message` before any assistant delta for this turn) structural
// rather than a scheduling accident.
let (seq_tx, seq_rx) = oneshot::channel::<u64>();
// Spawned under the turn-slot lock, so a `close` that raced this dispatch
// either aborts the turn or stops it being spawned at all.
let dispatched = entry.spawn_turn(|| {
// Invalidate before spawning: a fast reply can prepare the next task
// before send_message returns, and must not have its result erased.
*lock(&entry.last_promote) = None;
let guard = guard.take();
tokio::spawn(async move {
// The guard moves into the turn; it releases `in_flight` when the
// turn ends, is aborted, or panics.
let _guard = guard;
let seq = task_entry
.emit(DiscussEventKind::UserMessage {
text: prompt_text.clone(),
})
.await;
let _ = seq_tx.send(seq);
run_turn(task_state, task_entry, prompt_text).await;
})
});
if !dispatched {
return Err(format!(
"{discussion_id} was closed while your message was being dispatched — nothing is \
running; start a new discussion"
));
}
// A turn is running for this message now, so the transcript row stays.
recorded.dispatched = true;
// The turn's first event, reported so a caller that has not yet subscribed
// can resume from exactly there. 0 if the turn was aborted before it got
// that far — same answer `emit` gives when the drain is already gone.
let first_seq = seq_rx.await.unwrap_or(0);
Ok(json!({ "ok": true, "seq": first_seq }))
}
/// Read the same live/persisted session records used by the coding UI. This is
/// a projection, not a second task store. Binding both repository and discussion
/// prevents unrelated task history from entering a model request.
async fn coding_context(
state: &Arc<ServerState>,
entry: &DiscussionEntry,
) -> Result<String, String> {
let dir = super::rpc::coder_state_dir()?;
let mut rows = coding_runs(state, &entry.id, &entry.repo, dir).await?;
let total = rows.len();
rows.truncate(8);
if total > rows.len() {
rows.push(json!({"older_runs_omitted": total - rows.len()}));
}
if rows.is_empty() {
return Ok(String::new());
}
Ok(format!(
"Linked coding runs (current observations; older observations may be stale):\n{}\nThese runs use isolated worktrees. Repository reads still target the conversation repository; do not assume its checkout contains a run's changes. A passing check or published branch does not establish deployment. Ask for review of retained work before proposing to start over. After checkout delivery, a new task captures current checkout files, including later manual edits and deletions. After branch delivery, it starts from the recorded result commit. A retained native worktree with execution_stopped=true can be reopened by the next task. Other unfinished work still requires recovery checks.",
serde_json::to_string(&rows).map_err(|e| e.to_string())?
))
}
async fn coding_runs(
state: &Arc<ServerState>,
discussion_id: &str,
repo: &Path,
dir: PathBuf,
) -> Result<Vec<Value>, String> {
let entries: Vec<_> = state
.coder_sessions
.lock()
.await
.values()
.cloned()
.collect();
let mut rows = Vec::new();
let mut live_ids = std::collections::HashSet::new();
for entry in entries {
let session = entry.session.lock().await;
live_ids.insert(session.id.clone());
if session.discussion_id.as_deref() == Some(discussion_id) && session.repo == repo {
rows.push(coding_run_row(&session, true));
}
}
let discussion_id = discussion_id.to_string();
let repo = repo.to_path_buf();
let saved = tokio::task::spawn_blocking(move || {
super::session::CoderSession::list(&dir)
.into_iter()
.filter(|session| {
!live_ids.contains(&session.id)
&& session.discussion_id.as_deref() == Some(discussion_id.as_str())
&& session.repo == repo
})
.map(|session| coding_run_row(&session, false))
.collect::<Vec<_>>()
})
.await
.map_err(|e| format!("read linked coding runs: {e}"))?;
rows.extend(saved);
let superseded: std::collections::HashSet<String> = rows
.iter()
.filter_map(|row| row["resumed_from"].as_str().map(str::to_string))
.collect();
rows.retain(|row| {
!row["session_id"]
.as_str()
.is_some_and(|id| superseded.contains(id))
});
rows.sort_by_key(|row| std::cmp::Reverse(row["updated_at"].as_u64().unwrap_or(0)));
Ok(rows)
}
fn coding_run_row(session: &super::session::CoderSession, live: bool) -> Value {
let omitted_guidance = session.steering_messages.len().saturating_sub(8);
let guidance: Vec<Value> = session.steering_messages[omitted_guidance..]
.iter()
.map(|text| {
json!({
"text": text.chars().take(1000).collect::<String>(),
"truncated": text.chars().count() > 1000,
})
})
.collect();
let checks: Vec<Value> = session.last_check_results.iter().take(20).map(|check| json!({
"name": check.name.chars().take(160).collect::<String>(),
"passed": check.passed,
"exit_code": check.exit_code,
"timed_out": check.timed_out,
"deadline_clamped": check.deadline_clamped,
"output_tail": check.output_tail.chars().rev().take(800).collect::<String>().chars().rev().collect::<String>(),
})).collect();
json!({
"session_id": session.id, "state": session.state.as_str(), "live": live,
"intent": session.intent.chars().take(1000).collect::<String>(),
"updated_at": session.updated_at, "result_branch": session.result_branch,
"result_commit": session.result_commit,
"result_delivery": session.result_delivery,
"resumed_from": session.resumed_from,
"execution_stopped": session.execution_stopped,
"operator_guidance": guidance,
"operator_guidance_omitted": omitted_guidance,
"engine": session.engine.label(),
"worktree": session.workspace_path.as_ref().filter(|path| path.is_dir()),
"error": session.error.as_ref().map(|error| error.chars().take(1000).collect::<String>()),
"failure_kind": session.failure_kind, "checks": checks,
"checks_omitted": session.last_check_results.len().saturating_sub(checks.len()),
})
}
/// Recover only a native task whose execution returned or was joined. The
/// start admission guard must remain held through adoption and registration.
pub(super) async fn retained_workspace(
state: &Arc<ServerState>,
discussion_id: &str,
repo: &Path,
dir: PathBuf,
) -> Result<Option<(String, PathBuf)>, String> {
let rows = coding_runs(state, discussion_id, repo, dir.clone()).await?;
let retained: Vec<_> = rows
.iter()
.filter(|row| {
matches!(row["state"].as_str(), Some("failed" | "abandoned"))
&& row["worktree"].is_string()
})
.collect();
if retained.len() > 1 {
return Err("Multiple unfinished worktrees belong to this conversation. Review them before choosing work to continue.".into());
}
let Some(row) = retained.first() else {
return Ok(None);
};
if row["engine"] != "native" || row["execution_stopped"] != true {
return Err("The previous task's execution has not been confirmed stopped. Its work is retained; automatic recovery cannot safely reopen it yet.".into());
}
let path = PathBuf::from(
row["worktree"]
.as_str()
.ok_or("missing retained worktree")?,
)
.canonicalize()
.map_err(|e| format!("retained worktree: {e}"))?;
let root = dir
.join("worktrees")
.canonicalize()
.map_err(|e| e.to_string())?;
if path.parent() != Some(root.as_path()) {
return Err("retained worktree is outside this daemon's workspace directory".into());
}
Ok(Some((
row["session_id"]
.as_str()
.ok_or("missing retained task id")?
.to_string(),
path,
)))
}
/// Branch deliveries continue from the saved revision. Checkout deliveries
/// continue from current files (including later user edits), captured by task
/// admission. Explicit caller bases win; legacy unknown results still refuse.
pub(super) async fn followup_base(
state: &Arc<ServerState>,
discussion_id: &str,
repo: &Path,
dir: PathBuf,
) -> Result<Option<String>, String> {
let rows = coding_runs(state, discussion_id, repo, dir).await?;
followup_base_from(&rows)
}
fn followup_base_from(rows: &[Value]) -> Result<Option<String>, String> {
for run in rows {
match run["state"].as_str() {
Some("merged") if rows.iter().any(|other| other["state"] == "merged"
&& other["updated_at"] == run["updated_at"]
&& other["result_commit"] != run["result_commit"]) => {
return Err("Multiple deliveries have the same recorded timestamp. Choose an explicit base revision; CAR cannot safely infer their order.".into());
}
Some("merged") => {
let commit = run["result_commit"].as_str().ok_or_else(|| "The previous run predates saved result revisions. Choose an explicit base revision before continuing; CAR will not silently start again from repository HEAD.".to_string())?;
// Applied results already live in the checkout. Reusing their
// old tree would erase later user edits/deletions from the next
// task's view. None asks admission to capture current inputs.
return Ok((run["result_delivery"] != "checkout").then(|| commit.to_string()));
},
Some("failed" | "abandoned") if !run["worktree"].is_null() => return Err(format!(
"The previous run has unfinished changes at {}. Review that work before starting over; resuming that worktree is not yet supported.", run["worktree"].as_str().unwrap_or("the retained worktree")
)),
_ => {},
}
}
Ok(None)
}
/// Serialize task admission for one conversation through registration. The
/// caller holds this guard until start_session has published its live entry.
/// Unlike prompt advice, this prevents two concurrent builds of the same turn.
pub(super) async fn claim_coding_start(
state: &Arc<ServerState>,
discussion_id: &str,
repo: &Path,
dir: PathBuf,
) -> Result<tokio::sync::OwnedMutexGuard<()>, String> {
if get_discussion(state, discussion_id).await?.repo != repo {
return Err("The task repository differs from this conversation's repository.".into());
}
claim_coding_start_at(state, discussion_id, dir).await
}
async fn claim_coding_start_at(
state: &Arc<ServerState>,
discussion_id: &str,
dir: PathBuf,
) -> Result<tokio::sync::OwnedMutexGuard<()>, String> {
let entry = get_discussion(state, discussion_id).await?;
let guard = entry.start_lock.clone().lock_owned().await;
// Recheck after waiting: closing a discussion must not leave a queued
// launch with a stale entry that is no longer authorized to start.
let current = get_discussion(state, discussion_id).await?;
if !Arc::ptr_eq(&entry, ¤t) {
return Err("conversation was reopened; retry the task".into());
}
let runs = coding_runs(state, discussion_id, &entry.repo, dir).await?;
if let Some(run) = runs.iter().find(|row| {
row.get("state").is_some_and(|state| {
!matches!(
state.as_str(),
Some("merged" | "reported" | "failed" | "abandoned")
)
})
}) {
return Err(format!("Conversation already has unfinished work in {} ({}). Open it from /sessions to continue or cancel it before starting another task.", run["session_id"].as_str().unwrap_or("unknown"), run["state"].as_str().unwrap_or("unknown")));
}
Ok(guard)
}
/// Drive one assistant turn, translating its wire events into discussion
/// events and auto-denying every approval escalation.
async fn run_turn(state: Arc<ServerState>, entry: Arc<DiscussionEntry>, text: String) {
*lock(&entry.task_proposal) = None;
if let Err(message) = entry.save_pending_task(None) {
entry.emit(DiscussEventKind::Error { message }).await;
entry.emit(DiscussEventKind::TurnComplete {}).await;
return;
}
let context = match coding_context(&state, &entry).await {
Ok(context) => context,
Err(error) => {
entry.emit(DiscussEventKind::Error { message: error }).await;
entry.emit(DiscussEventKind::TurnComplete {}).await;
return;
}
};
let sink_entry = entry.clone();
let assembled: Arc<StdMutex<String>> = Arc::new(StdMutex::new(String::new()));
let sink_assembled = assembled.clone();
let service = entry.service.clone();
// The sink resolves approvals on the same service it streams from, so it
// needs its own handle rather than borrowing the one being called.
let sink_service = service.clone();
let id = entry.id.clone();
service
.handle_turn_with_context(
&id,
&text,
None,
None,
Some(&context),
move |payload: Value| {
let entry = sink_entry.clone();
let assembled = sink_assembled.clone();
let service = sink_service.clone();
async move {
let kind = payload.get("kind").and_then(Value::as_str).unwrap_or("");
match kind {
"token" => {
let delta = payload
.get("delta")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
if delta.is_empty() {
return;
}
lock(&assembled).push_str(&delta);
entry
.emit(DiscussEventKind::AssistantDelta { text: delta })
.await;
}
"tool_call" => {
let tool = payload
.get("tool")
.and_then(Value::as_str)
.unwrap_or("tool")
.to_string();
let params_preview = payload
.get("params")
.map(|p| preview(&p.to_string()))
.unwrap_or_default();
entry
.emit(DiscussEventKind::ToolCall {
tool,
params_preview,
})
.await;
}
// The no-mutation boundary, enforced here rather than left
// to a human: a discussion never writes, so an escalation is
// answered immediately with "no" instead of parking a
// prompt nobody asked for (and timing out five minutes
// later, which is what the unresolved gate would do).
"approval_pending" => {
let tool = payload
.get("tool")
.and_then(Value::as_str)
.unwrap_or("tool")
.to_string();
if let Some(approval_id) =
payload.get("approval_id").and_then(Value::as_str)
{
service.deny_approval(
approval_id,
DISCUSSION_MUTATION_REFUSAL.to_string(),
);
}
entry
.emit(DiscussEventKind::ToolResult {
tool,
ok: false,
preview: DISCUSSION_MUTATION_REFUSAL.to_string(),
})
.await;
}
"done" => {
let text = payload
.get("text")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let text = if text.trim().is_empty() {
lock(&assembled).clone()
} else {
text
};
entry.record_turn("Assistant", &text);
entry.turns.fetch_add(1, Ordering::SeqCst);
entry
.emit(DiscussEventKind::AssistantMessage { text })
.await;
let proposal = lock(&entry.task_proposal).take();
if let Some((proposed_intent, constraints)) = proposal {
if let Err(message) = entry.save_pending_task(Some((
proposed_intent.clone(),
constraints.clone(),
))) {
entry.emit(DiscussEventKind::Error { message }).await;
entry.emit(DiscussEventKind::TurnComplete {}).await;
return;
}
*lock(&entry.last_promote) =
Some((proposed_intent.clone(), constraints.clone()));
entry
.emit(DiscussEventKind::TaskPrepared {
proposed_intent,
constraints,
})
.await;
}
entry.emit(DiscussEventKind::TurnComplete {}).await;
}
"error" => {
let message = payload
.get("error")
.and_then(Value::as_str)
.unwrap_or("discussion turn failed")
.to_string();
entry.emit(DiscussEventKind::Error { message }).await;
entry.emit(DiscussEventKind::TurnComplete {}).await;
}
// The third terminal kind. A discussion runs the same
// `AssistantService` as chat, so a turn can be refused on
// the Parslee account here too — and without this arm the
// frame fell through to `_ => {}`: no error, no remedy, and
// crucially no `TurnComplete`, which is the event
// `docs/websocket-protocol.md` tells a client to wait for
// before sending again. The client was left holding a turn
// that had already ended.
//
// Rendered through `Error` rather than a new variant: the
// message IS the remedy, in the copy a person reads, and a
// client that already renders discussion errors shows it
// without changing.
"auth_required" => {
let message = payload
.get("message")
.and_then(Value::as_str)
.unwrap_or("this discussion needs a Parslee sign-in")
.to_string();
entry.emit(DiscussEventKind::Error { message }).await;
entry.emit(DiscussEventKind::TurnComplete {}).await;
}
_ => {}
}
}
},
)
.await;
}
fn preview(s: &str) -> String {
const CAP: usize = 200;
if s.chars().count() <= CAP {
return s.to_string();
}
let mut out: String = s.chars().take(CAP).collect();
out.push('…');
out
}
/// Distill the discussion into a run intent + the constraints agreed in it.
///
/// **Starts nothing.** No worktree, no branch, no session — the caller shows
/// `proposed_intent` to the operator, who may edit it before calling
/// `coder.start`. Callable repeatedly on an open discussion.
///
/// Refuses while a turn is streaming: distilling then would run on the
/// operator's question with no answer beside it, and the model would happily
/// invent a confident intent from an unanswered question — which then feeds
/// `coder.start { discussion_id }` and contract derivation.
pub async fn promote(
state: &Arc<ServerState>,
discussion_id: &str,
client_id: &str,
) -> Result<Value, String> {
let entry = get_owned_discussion(state, discussion_id, client_id).await?;
if entry.is_answering() {
return Err(format!(
"{discussion_id} is still answering — try again in a moment"
));
}
if entry.transcript_is_empty() {
return Err(
"this discussion has no turns yet — say what you are trying to do first".to_string(),
);
}
let _start_guard = claim_coding_start(
state,
discussion_id,
&entry.repo,
super::rpc::coder_state_dir()?,
)
.await?;
let context = coding_context(state, &entry).await?;
let repo_context = format!(
"{}\n{}\n{}",
entry.repo_summary, entry.project_context, context
);
let (intent, constraints) =
distill(&entry.generator, &entry.distill_transcript(), &repo_context).await?;
entry.save_pending_task(Some((intent.clone(), constraints.clone())))?;
*lock(&entry.last_promote) = Some((intent.clone(), constraints.clone()));
entry.touch();
Ok(json!({
"discussion_id": entry.id,
"proposed_intent": intent,
"constraints": constraints,
}))
}
/// The distillation call. Generation is injected exactly the way
/// `derive_app_contract` injects it into `derive_contract`, so the prompt +
/// parse + bounded-retry shape is testable with a scripted model.
async fn distill(
generator: &Arc<dyn TurnGenerator>,
transcript: &str,
repo_summary: &str,
) -> Result<(String, Vec<String>), String> {
let mut last_err = String::from("no attempt was made");
for _ in 0..PROMOTE_MAX_ATTEMPTS {
let prompt = format!(
"A developer has been discussing a change to a codebase. Distill the discussion \
into ONE actionable coding intent plus the constraints they agreed on.\n\n\
REPOSITORY\n{repo_summary}\n\n\
DISCUSSION (most recent turns)\n{transcript}\n\n\
Return ONLY a JSON object, no prose and no code fences:\n\
{{\n \"proposed_intent\": \"one paragraph, imperative, what to change and why\",\n \
\"constraints\": [\"a thing the change must not break or must respect\"]\n}}\n\n\
Rules:\n\
- `proposed_intent` is an INSTRUCTION, not a summary of the conversation. Never \
quote the transcript back.\n\
- Include only constraints actually agreed in the discussion. If none were, \
return an empty array — do not invent any.\n"
);
let text = match generator
.generate(car_inference::GenerateRequest {
prompt,
params: car_inference::GenerateParams {
temperature: 0.0,
max_tokens: 1024,
thinking: car_inference::tasks::generate::ThinkingMode::Off,
..Default::default()
},
..Default::default()
})
.await
{
Ok(r) => r.text,
Err(e) => {
last_err = format!("generation failed: {e}");
continue;
}
};
let value = match super::contract::extract_json_object(&text) {
Ok(v) => v,
Err(e) => {
last_err = format!("output did not parse: {e}");
continue;
}
};
let intent = value
.get("proposed_intent")
.and_then(Value::as_str)
.unwrap_or_default()
.trim()
.to_string();
if intent.is_empty() {
last_err = "the model returned no proposed_intent".to_string();
continue;
}
let constraints: Vec<String> = value
.get("constraints")
.and_then(Value::as_array)
.map(|a| {
a.iter()
.filter_map(Value::as_str)
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect()
})
.unwrap_or_default();
return Ok((intent, constraints));
}
Err(format!(
"could not distill this discussion into an intent after {PROMOTE_MAX_ATTEMPTS} \
attempts: {last_err}"
))
}
/// Accepting a prepared task consumes its saved editor before task creation.
/// A failed start leaves the current client editor available for retry.
pub(super) async fn consume_prepared_task(
state: &Arc<ServerState>,
id: &str,
) -> Result<(), String> {
get_discussion(state, id).await?.save_pending_task(None)
}
/// Constraints to fold into `derive_contract` for a `coder.start
/// { discussion_id }`.
///
/// An unknown id is a hard error — a run that silently drops its grounding is
/// worse than one that refuses to start. A distillation failure also refuses
/// the start: the supplied intent need not repeat every preference agreed in
/// the conversation. The operator can retry without losing that grounding.
///
/// Refused while a turn is streaming, for the same reason `promote` is: the
/// distillation would run on the operator's question with no answer beside it,
/// and these constraints go straight into contract derivation.
pub async fn constraints_for_start(
state: &Arc<ServerState>,
discussion_id: &str,
) -> Result<Vec<String>, String> {
let entry = get_discussion(state, discussion_id).await?;
if entry.is_answering() {
return Err(format!(
"{discussion_id} is still answering — wait for `turn_complete` before starting a \
run from it, or the constraints would be distilled from a question with no \
answer beside it"
));
}
let cached = entry.constraints();
if !cached.is_empty() {
return Ok(cached);
}
if lock(&entry.last_promote).is_some() {
// Promoted already, and it genuinely agreed no constraints.
return Ok(Vec::new());
}
if entry.transcript_is_empty() {
return Ok(Vec::new());
}
match distill(
&entry.generator,
&entry.distill_transcript(),
&format!("{}\n{}", entry.repo_summary, entry.project_context),
)
.await
{
Ok((intent, constraints)) => {
*lock(&entry.last_promote) = Some((intent, constraints.clone()));
Ok(constraints)
}
Err(e) => {
tracing::warn!(discussion_id, "discussion constraints unavailable: {e}");
Err(format!(
"Could not carry the conversation's requirements into this task: {e}. \
No task was started. Retry Build when model access is available."
))
}
}
}
/// Close a discussion: cancel any in-flight turn, free its runtime, end its
/// drain.
pub async fn close(
state: &Arc<ServerState>,
discussion_id: &str,
client_id: &str,
) -> Result<Value, String> {
// Ownership first, and against the live registry: a foreign `close` must
// not be able to cancel a turn its owner is watching.
get_owned_discussion(state, discussion_id, client_id).await?;
let entry = state.coder_discussions.lock().await.remove(discussion_id);
let Some(entry) = entry else {
return Err(format!("no open discussion '{discussion_id}'"));
};
// Actually stop the model: without this the turn keeps running against a
// live provider, billing tokens to a conversation nobody can read. This
// also latches the discussion closed, so a `send` parked mid-dispatch never
// spawns its turn behind us.
entry.cancel_turn();
Ok(json!({ "ok": true }))
}
/// Drop a disconnecting client's discussion state (called from
/// `remove_session`).
///
/// A discussion is owned by the connection that opened it (module docs), so
/// this closes it outright rather than only unsubscribing — otherwise every
/// closed board leaks an `AssistantService`, a `Runtime`, an open runtime
/// session, and an unbounded transcript for the daemon's lifetime. Other
/// clients' subscriptions to a surviving discussion are just detached.
pub async fn drop_subscriptions_for_client(state: &ServerState, client_id: &str) {
let (owned, others): (Vec<_>, Vec<_>) = {
let open = state.coder_discussions.lock().await;
open.values()
.cloned()
.partition(|e| e.owner_client_id == client_id)
};
for entry in &others {
let _ = entry.cmds.send(StreamCmd::Detach(client_id.to_string()));
}
if owned.is_empty() {
return;
}
let mut open = state.coder_discussions.lock().await;
for entry in owned {
entry.cancel_turn();
open.remove(&entry.id);
}
}
// ---------------------------------------------------------------------------
// JSON-RPC handlers (thin parsing wrappers)
// ---------------------------------------------------------------------------
#[derive(Deserialize)]
struct StartParams {
repo: PathBuf,
#[serde(default)]
resume_id: Option<String>,
#[serde(default)]
model: Option<String>,
}
pub async fn handle_discuss_start(
req: &JsonRpcMessage,
state: &Arc<ServerState>,
session: &Arc<ClientSession>,
) -> Result<Value, String> {
let params: StartParams =
serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
let engine = crate::handler::get_inference_engine(state).clone();
let generator: Arc<dyn TurnGenerator> = engine.clone();
if let Some(model) = params
.model
.as_deref()
.map(str::trim)
.filter(|m| !m.is_empty() && *m != "auto")
{
if !engine.knows_model(model) {
return Err(format!(
"Unknown model '{model}'. Use `car models list` to choose an available model id."
));
}
}
// The daemon is single-principal for operator connections; supervised
// agents have a separately authenticated stable identity. Never derive
// archive ownership from a caller-supplied parameter or a connection UUID.
let principal = discussion_principal(session).await;
open_discussion_with_model(
state,
¶ms.repo,
&session.client_id,
engine,
generator,
&principal,
params.resume_id.as_deref(),
params.model.as_deref(),
)
.await
}
#[derive(Deserialize)]
struct SendParams {
discussion_id: String,
text: String,
}
pub async fn handle_discuss_send(
req: &JsonRpcMessage,
state: &Arc<ServerState>,
session: &Arc<ClientSession>,
) -> Result<Value, String> {
let params: SendParams =
serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
send_message(
state,
¶ms.discussion_id,
&session.client_id,
¶ms.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, ¶ms.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, ¶ms.discussion_id).await {
let _ = entry
.cmds
.send(StreamCmd::Detach(session.client_id.clone()));
}
Ok(json!({ "ok": true }))
}
pub async fn handle_discuss_promote(
req: &JsonRpcMessage,
state: &Arc<ServerState>,
session: &Arc<ClientSession>,
) -> Result<Value, String> {
let params: DiscussionIdParams =
serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
// Model generation has a deep polling stack in debug builds. Poll it in
// its own task, rather than underneath the large RPC dispatch future.
// JoinSet owns cancellation: a deadline or disconnected caller dropping
// this handler also aborts generation instead of leaving a detached bill.
let state = state.clone();
let client_id = session.client_id.clone();
let mut generation = tokio::task::JoinSet::new();
generation.spawn(async move { promote(&state, ¶ms.discussion_id, &client_id).await });
generation
.join_next()
.await
.ok_or("conversation planning task did not start")?
.map_err(|error| format!("conversation planning task failed: {error}"))?
}
pub async fn handle_discuss_close(
req: &JsonRpcMessage,
state: &Arc<ServerState>,
session: &Arc<ClientSession>,
) -> Result<Value, String> {
let params: DiscussionIdParams =
serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
close(state, ¶ms.discussion_id, &session.client_id).await
}
async fn discussion_principal(session: &Arc<ClientSession>) -> String {
match session.agent_id.lock().await.as_deref() {
Some(id) => format!("agent:{id}"),
None => "operator".to_string(),
}
}
/// `coder.discuss.list` — this connection's open discussions, plus the
/// authenticated principal's saved conversations that are not currently open.
///
/// Scoped to the caller, like every other `coder.discuss.*` method: a
/// discussion is owned by the connection that opened it, and listing another
/// connection's discussions would hand out ids the caller cannot use anyway.
pub async fn handle_discuss_list(
state: &Arc<ServerState>,
session: &Arc<ClientSession>,
) -> Result<Value, String> {
let mut rows: Vec<Value> = state
.coder_discussions
.lock()
.await
.values()
.filter(|e| e.owner_client_id == session.client_id)
.map(|e| e.summary_row())
.collect();
rows.sort_by_key(|v| std::cmp::Reverse(v["created_at"].as_u64().unwrap_or(0)));
let principal = discussion_principal(session).await;
let records = DiscussionRecord::list(&state.journal_dir, &principal)?;
let live = state.coder_discussions.lock().await;
let saved: Vec<Value> = records
.into_iter()
.filter(|record| !live.contains_key(&record.id))
.map(|record| {
json!({
"discussion_id": record.id,
"repo": record.repo,
"created_at": record.created_at,
})
})
.collect();
Ok(json!({ "discussions": rows, "saved": saved }))
}
#[cfg(test)]
mod tests {
use super::*;
use async_trait::async_trait;
use car_inference::{GenerateRequest, InferenceResult};
use std::sync::atomic::AtomicUsize;
fn turn(text: &str, tool_calls: Value) -> InferenceResult {
serde_json::from_value(json!({
"text": text, "tool_calls": tool_calls,
"trace_id": "t", "model_used": "scripted", "latency_ms": 0,
}))
.expect("scripted InferenceResult shape")
}
struct Script {
turns: Vec<InferenceResult>,
cursor: AtomicUsize,
}
#[async_trait]
impl TurnGenerator for Script {
async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
let i = self.cursor.fetch_add(1, Ordering::SeqCst);
self.turns
.get(i)
.cloned()
.ok_or_else(|| "script exhausted".to_string())
}
}
/// A generator that blocks until released — lets a test observe a turn
/// while it is genuinely in flight.
///
/// Released with `notify_one`, never `notify_waiters`: the turn is spawned,
/// so the test can reach the release before the task has registered as a
/// waiter, and `notify_waiters` wakes only waiters that already exist.
/// `notify_one` stores a permit, so the ordering does not matter.
struct Blocking {
gate: Arc<tokio::sync::Notify>,
}
#[async_trait]
impl TurnGenerator for Blocking {
async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
self.gate.notified().await;
Ok(turn("done at last", json!([])))
}
}
/// Counts invocations — for asserting a turn NEVER reached the model.
struct Counting {
calls: Arc<AtomicUsize>,
}
#[async_trait]
impl TurnGenerator for Counting {
async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
self.calls.fetch_add(1, Ordering::SeqCst);
Ok(turn("counted", json!([])))
}
}
fn init_repo(dir: &Path) {
for args in [
vec!["init", "-q", "-b", "main"],
vec![
"-c",
"user.name=t",
"-c",
"user.email=t@t",
"commit",
"-q",
"--allow-empty",
"-m",
"init",
],
] {
let out = std::process::Command::new("git")
.arg("-C")
.arg(dir)
.args(&args)
.output()
.unwrap();
assert!(
out.status.success(),
"{}",
String::from_utf8_lossy(&out.stderr)
);
}
}
fn engine(root: &Path) -> Arc<car_inference::InferenceEngine> {
let mut cfg = car_inference::InferenceConfig::default();
cfg.models_dir = root.join("models");
Arc::new(car_inference::InferenceEngine::new(cfg))
}
/// A standalone daemon state plus the journal dir it writes to — the
/// caller keeps the `TempDir` alive for the length of the test.
fn state() -> (Arc<ServerState>, tempfile::TempDir) {
let journal = tempfile::tempdir().unwrap();
let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));
(state, journal)
}
async fn start(
state: &Arc<ServerState>,
repo: &Path,
generator: Arc<dyn TurnGenerator>,
) -> String {
let started = start_discussion(state, repo, "owner-1", engine(repo), generator)
.await
.unwrap();
started["discussion_id"].as_str().unwrap().to_string()
}
/// A `ClientSession` over a drain sink — enough for the handlers that need
/// a connection identity, without a tungstenite handshake.
async fn client(state: &Arc<ServerState>, id: &str) -> Arc<ClientSession> {
state
.create_session(id, Arc::new(crate::session::WsChannel::test_stub()))
.await
.unwrap()
}
/// A WS sink that keeps every frame instead of writing it, so a test can
/// read exactly what a subscriber's lane delivered. `test_stub` drains to
/// nowhere, which is enough for membership checks but says nothing about
/// what arrived.
struct CaptureSink(Arc<StdMutex<Vec<String>>>);
impl futures::Sink<tokio_tungstenite::tungstenite::Message> for CaptureSink {
type Error = tokio_tungstenite::tungstenite::Error;
fn poll_ready(
self: std::pin::Pin<&mut Self>,
_: &mut std::task::Context<'_>,
) -> std::task::Poll<Result<(), Self::Error>> {
std::task::Poll::Ready(Ok(()))
}
fn start_send(
self: std::pin::Pin<&mut Self>,
item: tokio_tungstenite::tungstenite::Message,
) -> Result<(), Self::Error> {
if let tokio_tungstenite::tungstenite::Message::Text(text) = item {
lock(&self.0).push(text.to_string());
}
Ok(())
}
fn poll_flush(
self: std::pin::Pin<&mut Self>,
_: &mut std::task::Context<'_>,
) -> std::task::Poll<Result<(), Self::Error>> {
std::task::Poll::Ready(Ok(()))
}
fn poll_close(
self: std::pin::Pin<&mut Self>,
_: &mut std::task::Context<'_>,
) -> std::task::Poll<Result<(), Self::Error>> {
std::task::Poll::Ready(Ok(()))
}
}
/// A real `WsChannel` over [`CaptureSink`], plus the frames it collected.
/// Locking its `write` half is a half-open peer: writes stop completing and
/// never fail, exactly what wedges a subscriber's lane.
fn capturing_channel() -> (Arc<WsChannel>, Arc<StdMutex<Vec<String>>>) {
let frames = Arc::new(StdMutex::new(Vec::new()));
let sink: crate::session::WsSink = Box::pin(CaptureSink(frames.clone()));
let channel = Arc::new(WsChannel {
write: tokio::sync::Mutex::new(sink),
pending: tokio::sync::Mutex::new(HashMap::new()),
active_actions: tokio::sync::Mutex::new(HashMap::new()),
next_id: AtomicU64::new(0),
});
(channel, frames)
}
fn rpc_req(params: Value) -> JsonRpcMessage {
serde_json::from_value(json!({ "jsonrpc": "2.0", "id": 1, "params": params }))
.expect("JsonRpcMessage shape")
}
/// The `seq` of every `coder.discuss.event` frame a lane delivered.
fn delivered_seqs(frames: &Arc<StdMutex<Vec<String>>>) -> Vec<u64> {
lock(frames)
.iter()
.map(|f| serde_json::from_str::<Value>(f).expect("a lane frame must be JSON"))
.inspect(|v| assert_eq!(v["method"], "coder.discuss.event", "unexpected frame: {v}"))
.map(|v| {
v["params"]["seq"]
.as_u64()
.expect("every event carries a seq")
})
.collect()
}
async fn wait_for_turn_complete(entry: &Arc<DiscussionEntry>) {
for _ in 0..400 {
{
let events = entry.events.lock().await;
if events
.iter()
.any(|e| matches!(e.kind, DiscussEventKind::TurnComplete {}))
{
return;
}
}
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
}
panic!("discussion turn never completed");
}
async fn wait_for_idle(entry: &Arc<DiscussionEntry>) {
tokio::time::timeout(std::time::Duration::from_secs(10), async {
while entry.is_answering() {
tokio::task::yield_now().await;
}
})
.await
.expect("discussion turn did not release its guard");
}
#[tokio::test]
async fn conversation_recovers_after_restart_with_same_identity_and_full_model_history() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
std::fs::write(repo.path().join("AGENTS.md"), "Initial repository rule.").unwrap();
std::fs::write(repo.path().join("CLAUDE.md"), "Preserve exported names.").unwrap();
let (state, journal) = state();
let script = || {
Arc::new(Script {
turns: vec![turn("Remember the export regression.", json!([]))],
cursor: AtomicUsize::new(0),
}) as Arc<dyn TurnGenerator>
};
let started = open_discussion(
&state,
repo.path(),
"connection-1",
engine(repo.path()),
script(),
"operator",
None,
)
.await
.unwrap();
let id = started["discussion_id"].as_str().unwrap().to_string();
let entry = get_discussion(&state, &id).await.unwrap();
let checkpoint = entry
.durability
.load_checkpoint(&id)
.await
.unwrap()
.unwrap();
let system = serde_json::to_string(&checkpoint.messages[0]).unwrap();
assert!(system.contains("Initial repository rule."));
assert!(system.contains("Preserve exported names."));
assert!(system.contains("does not expand the read-only permissions"));
send_message(
&state,
&id,
"connection-1",
"Investigate the export regression.",
)
.await
.unwrap();
wait_for_turn_complete(&entry).await;
wait_for_idle(&entry).await;
let err = open_discussion(
&state,
repo.path(),
"connection-2",
engine(repo.path()),
script(),
"operator",
Some(&id),
)
.await
.unwrap_err();
assert!(err.contains("already open"));
close(&state, &id, "connection-1").await.unwrap();
assert!(entry
.durability
.checkpoint(&id, &[], "late writer", None)
.await
.is_err());
drop(entry);
drop(state);
std::fs::write(repo.path().join("AGENTS.md"), "Updated repository rule.").unwrap();
let restarted = Arc::new(ServerState::standalone(journal.path().to_path_buf()));
assert!(open_discussion(
&restarted,
repo.path(),
"foreign",
engine(repo.path()),
script(),
"agent:foreign",
Some(&id)
)
.await
.is_err());
let resumed = open_discussion(
&restarted,
repo.path(),
"connection-2",
engine(repo.path()),
script(),
"operator",
Some(&id),
)
.await
.unwrap();
assert_eq!(resumed["discussion_id"], id);
assert_eq!(resumed["resumed"], true);
let entry = get_discussion(&restarted, &id).await.unwrap();
assert_eq!(entry.turns.load(Ordering::SeqCst), 1);
assert!(lock(&entry.transcript)
.iter()
.any(|(_, text)| text == "Remember the export regression."));
send_message(
&restarted,
&id,
"connection-2",
"Now explain the next step.",
)
.await
.unwrap();
wait_for_idle(&entry).await;
let checkpoint = entry
.durability
.load_checkpoint(&id)
.await
.unwrap()
.unwrap();
let history = serde_json::to_string(&checkpoint.messages).unwrap();
assert!(history.contains("Updated repository rule."));
assert!(!history.contains("Initial repository rule."));
assert!(history.contains("Preserve exported names."));
assert!(history.contains("Investigate the export regression."));
assert!(history.contains("Remember the export regression."));
assert!(history.contains("Now explain the next step."));
assert!(get_owned_discussion(&restarted, &id, "connection-1")
.await
.is_err());
close(&restarted, &id, "connection-2").await.unwrap();
}
#[tokio::test]
async fn discuss_start_rejects_a_non_git_directory() {
let dir = tempfile::tempdir().unwrap();
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
let err = start_discussion(&state, dir.path(), "owner-1", engine(dir.path()), script)
.await
.unwrap_err();
assert!(
err.contains("is not a git repository")
&& err.contains("discuss needs a repo to ground itself in"),
"operator-readable non-repo error, got: {err}"
);
}
#[tokio::test]
async fn conversation_advertises_relevant_tools_without_mutations() {
struct Capture(Arc<StdMutex<Vec<Value>>>);
#[async_trait]
impl TurnGenerator for Capture {
async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
*lock(&self.0) = req.tools.unwrap_or_default();
Ok(turn("Here is how the repository is organized.", json!([])))
}
}
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let tools = Arc::new(StdMutex::new(Vec::new()));
let id = start(&state, repo.path(), Arc::new(Capture(tools.clone()))).await;
send_message(&state, &id, "owner-1", "Explain the repository")
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
wait_for_idle(&entry).await;
let captured = lock(&tools);
let names: Vec<_> = captured
.iter()
.filter_map(|def| def["name"].as_str())
.collect();
for required in [
"read_file",
"list_dir",
"find_files",
"grep_files",
"prepare_coding_task",
"web_search",
"http_request",
] {
assert!(names.contains(&required), "missing {required}: {names:?}");
}
assert!(
names.len() <= 8,
"unrelated tools leaked into coding conversation: {names:?}"
);
assert!(!names.contains(&"write_file"));
assert!(!names.contains(&"shell"));
assert!(
lock(&entry.last_promote).is_none(),
"a question must not automatically prepare a task"
);
eprintln!(
"conversation tool payload: {} tools, {} JSON bytes",
names.len(),
serde_json::to_vec(&*captured).unwrap().len()
);
}
#[tokio::test]
async fn project_policy_can_refuse_task_preparation() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let policies = repo.path().join(".car/policies");
std::fs::create_dir_all(&policies).unwrap();
std::fs::write(
policies.join("rules.toml"),
"deny_tool = [\"prepare_coding_task\"]\n",
)
.unwrap();
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![
turn(
"",
json!([{"id":"prepare-denied", "name":"prepare_coding_task",
"arguments":{"intent":"Fix parser", "constraints":[]}}]),
),
turn(
"Task preparation is unavailable under repository policy.",
json!([]),
),
],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
send_message(&state, &id, "owner-1", "Fix parser")
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
wait_for_turn_complete(&entry).await;
assert!(lock(&entry.last_promote).is_none());
assert!(!entry
.events
.lock()
.await
.iter()
.any(|event| matches!(event.kind, DiscussEventKind::TaskPrepared { .. })));
}
#[tokio::test]
async fn implementation_request_prepares_task_without_starting_execution() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![
turn(
"",
json!([{"id":"prepare-1", "name":"prepare_coding_task",
"arguments":{"intent":"Fix the parser", "constraints":["Preserve public APIs"]}}]),
),
turn("The parser task is ready for review.", json!([])),
],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
send_message(
&state,
&id,
"owner-1",
"Please fix the parser and preserve public APIs.",
)
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
wait_for_turn_complete(&entry).await;
assert_eq!(entry.constraints(), vec!["Preserve public APIs"]);
let events = entry.events.lock().await;
assert!(events.iter().any(|event| matches!(&event.kind,
DiscussEventKind::TaskPrepared { proposed_intent, constraints }
if proposed_intent == "Fix the parser" && constraints == &["Preserve public APIs"])));
assert!(
state.coder_sessions.lock().await.is_empty(),
"preparing must not create a coding session"
);
assert!(!repo.path().join("parser.rs").exists());
drop(events);
wait_for_idle(&entry).await;
let journal = state.journal_dir.clone();
close(&state, &id, "owner-1").await.unwrap();
drop(entry);
drop(state);
let restarted = Arc::new(ServerState::standalone(journal));
let no_turns = || -> Arc<dyn TurnGenerator> {
Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
})
};
open_discussion(
&restarted,
repo.path(),
"owner-2",
engine(repo.path()),
no_turns(),
"owner-1",
Some(&id),
)
.await
.unwrap();
let resumed = get_discussion(&restarted, &id).await.unwrap();
assert_eq!(resumed.constraints(), vec!["Preserve public APIs"]);
assert!(
resumed
.events
.lock()
.await
.iter()
.any(|event| matches!(&event.kind,
DiscussEventKind::ToolCall { tool, params_preview }
if tool == "prepare_coding_task" && params_preview.contains("Fix the parser"))),
"resume must show the recorded tool call without executing it again"
);
assert!(restarted.coder_sessions.lock().await.is_empty());
assert!(resumed.events.lock().await.iter().any(|event| matches!(&event.kind,
DiscussEventKind::TaskPrepared { proposed_intent, .. } if proposed_intent == "Fix the parser")));
consume_prepared_task(&restarted, &id).await.unwrap();
close(&restarted, &id, "owner-2").await.unwrap();
open_discussion(
&restarted,
repo.path(),
"owner-3",
engine(repo.path()),
no_turns(),
"owner-1",
Some(&id),
)
.await
.unwrap();
let consumed = get_discussion(&restarted, &id).await.unwrap();
assert!(!consumed
.events
.lock()
.await
.iter()
.any(|event| matches!(event.kind, DiscussEventKind::TaskPrepared { .. })));
}
/// The load-bearing property: a discussion NEVER writes in the target repo.
#[tokio::test]
async fn a_discussion_writes_nothing_in_the_repo() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
std::fs::write(repo.path().join("keep.txt"), "original").unwrap();
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![
turn(
"",
json!([{
"id": "c1", "name": "write_file",
"arguments": {"path": "sneaky.txt", "content": "written by a discussion"}
}]),
),
turn(
"",
json!([{
"id": "c2", "name": "shell",
"arguments": {"command": "printf x > shelled.txt"}
}]),
),
turn(
"I cannot edit from a discussion; here is what I would change.",
json!([]),
),
],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
assert!(id.starts_with("disc-"));
send_message(
&state,
&id,
"owner-1",
"can you just make the change for me?",
)
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
wait_for_turn_complete(&entry).await;
assert!(
!repo.path().join("sneaky.txt").exists(),
"a discussion must not create files in the repo"
);
assert!(
!repo.path().join("shelled.txt").exists(),
"a discussion must not run shell commands that write"
);
assert_eq!(
std::fs::read_to_string(repo.path().join("keep.txt")).unwrap(),
"original"
);
let events = entry.events.lock().await;
assert!(
events.iter().any(|e| matches!(
&e.kind,
DiscussEventKind::ToolResult { ok, preview, .. }
if !ok && preview.contains("read-only")
)),
"the denial must surface as a tool_result"
);
drop(events);
wait_for_idle(&entry).await;
let checkpoint = entry
.durability
.load_checkpoint(&id)
.await
.unwrap()
.unwrap();
let refusals: Vec<_> = checkpoint
.messages
.iter()
.filter_map(|message| {
if let car_inference::Message::ToolResult { content, .. } = message {
Some(content.as_str())
} else {
None
}
})
.collect();
assert!(refusals
.iter()
.any(|text| text.contains("prepare_coding_task")
&& text.contains("not a file-permission problem")));
assert!(!refusals
.iter()
.any(|text| text.contains("declined by user")));
}
/// The other half of the boundary: a discussion cannot READ outside its
/// repo. Mutation-gating alone left the read tools pointed at the whole
/// filesystem, and their output streams to every subscriber.
#[tokio::test]
async fn a_discussion_cannot_read_outside_the_repo() {
let outside = tempfile::tempdir().unwrap();
let secret_path = outside.path().join("credentials.txt");
std::fs::write(&secret_path, "sk-ant-SUPERSECRETVALUE").unwrap();
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![
// Absolute path outside the repo — the exfiltration attempt.
turn(
"",
json!([{
"id": "c1", "name": "read_file",
"arguments": {"path": secret_path.to_string_lossy()}
}]),
),
// ...and the directory-scanning variant.
turn(
"",
json!([{
"id": "c2", "name": "grep_files",
"arguments": {"path": outside.path().to_string_lossy(), "pattern": "sk-ant-"}
}]),
),
turn("I can only read inside this repository.", json!([])),
],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
send_message(
&state,
&id,
"owner-1",
"what credentials does this project use?",
)
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
wait_for_turn_complete(&entry).await;
let events = entry.events.lock().await;
let stream = serde_json::to_string(&*events).unwrap();
assert!(
!stream.contains("SUPERSECRETVALUE"),
"a discussion must never stream content from outside its repo: {stream}"
);
}
#[tokio::test]
async fn selected_model_survives_reopen_and_auto_clears_it() {
struct Recording {
requests: Arc<StdMutex<Vec<(Option<String>, bool)>>>,
}
#[async_trait]
impl TurnGenerator for Recording {
async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
lock(&self.requests).push((req.model, req.params.strict_model));
Ok(turn("A grounded reply.", json!([])))
}
}
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, journal) = state();
let mut cfg = car_inference::InferenceConfig::default();
cfg.models_dir = journal.path().join("models");
let engine = Arc::new(car_inference::InferenceEngine::new(cfg));
let requests = Arc::new(StdMutex::new(Vec::new()));
let generator: Arc<dyn TurnGenerator> = Arc::new(Recording {
requests: requests.clone(),
});
let rejected = open_discussion_with_model(
&state,
repo.path(),
"owner",
engine.clone(),
generator.clone(),
"operator",
None,
Some("qwen/qwen3-embedding-0.6b:q8_0"),
)
.await
.unwrap_err();
assert!(
rejected.contains("cannot call repository tools"),
"{rejected}"
);
assert!(requests.lock().unwrap().is_empty());
let first = open_discussion_with_model(
&state,
repo.path(),
"owner",
engine.clone(),
generator.clone(),
"operator",
None,
Some("anthropic/claude-opus-4-6:latest"),
)
.await
.unwrap();
let id = first["discussion_id"].as_str().unwrap();
assert_eq!(first["model"], "anthropic/claude-opus-4-6:latest");
for selection in [None, Some("auto")] {
send_message(&state, id, "owner", "What is here?")
.await
.unwrap();
let entry = get_discussion(&state, id).await.unwrap();
wait_for_idle(&entry).await;
close(&state, id, "owner").await.unwrap();
let reopened = open_discussion_with_model(
&state,
repo.path(),
"owner",
engine.clone(),
generator.clone(),
"operator",
Some(id),
selection,
)
.await
.unwrap();
if selection.is_none() {
assert_eq!(reopened["model"], "anthropic/claude-opus-4-6:latest");
} else {
assert!(reopened["model"].is_null());
}
}
send_message(&state, id, "owner", "Continue.")
.await
.unwrap();
wait_for_idle(&get_discussion(&state, id).await.unwrap()).await;
close(&state, id, "owner").await.unwrap();
let captured = lock(&requests);
assert_eq!(
*captured,
vec![
(Some("anthropic/claude-opus-4-6:latest".into()), true),
(Some("anthropic/claude-opus-4-6:latest".into()), true),
(None, false)
]
);
assert!(DiscussionRecord::load(&state.journal_dir, id, "operator")
.unwrap()
.model
.is_none());
}
#[tokio::test]
async fn starting_refuses_to_drop_requirements_when_distillation_fails() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
// The reply succeeds; all later generation attempts fail.
turns: vec![turn("I will preserve the public API.", json!([]))],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
send_message(&state, &id, "owner-1", "Preserve the public API.")
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
wait_for_idle(&entry).await;
let error = constraints_for_start(&state, &id).await.unwrap_err();
assert!(error.contains("No task was started"), "{error}");
assert!(error.contains("Retry Build"), "{error}");
assert!(lock(&entry.last_promote).is_none());
assert!(!entry.transcript_is_empty());
assert!(state.coder_sessions.lock().await.is_empty());
close(&state, &id, "owner-1").await.unwrap();
}
#[tokio::test]
async fn promote_distills_an_intent_and_starts_nothing() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![
turn("The Windows path is the risky one.", json!([])),
turn(
r#"{"proposed_intent":"Make the config loader resolve paths on Windows.",
"constraints":["do not change the POSIX behavior"]}"#,
json!([]),
),
turn("Understood, preserve both platforms.", json!([])),
],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
send_message(
&state,
&id,
"owner-1",
"what is fragile about the config loader?",
)
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
wait_for_turn_complete(&entry).await;
wait_for_idle(&entry).await;
let promoted = promote(&state, &id, "owner-1").await.unwrap();
assert_eq!(
promoted["proposed_intent"],
"Make the config loader resolve paths on Windows."
);
assert_eq!(
promoted["constraints"],
json!(["do not change the POSIX behavior"])
);
assert!(state.coder_sessions.lock().await.is_empty());
assert_eq!(
constraints_for_start(&state, &id).await.unwrap(),
vec!["do not change the POSIX behavior".to_string()]
);
send_message(
&state,
&id,
"owner-1",
"Also preserve Windows compatibility.",
)
.await
.unwrap();
assert!(
lock(&entry.last_promote).is_none(),
"a new turn must invalidate the old plan constraints"
);
wait_for_idle(&entry).await;
close(&state, &id, "owner-1").await.unwrap();
}
/// A discussion turn refused on the Parslee account must still END.
///
/// `coder.discuss` runs the same `AssistantService` as chat, so it sees the
/// same terminal `auth_required` frame — and before this arm existed the
/// frame fell through to `_ => {}`: no error, no remedy, and no
/// `TurnComplete`. `docs/websocket-protocol.md` tells a client to wait for
/// `TurnComplete` before sending again, so the client was left holding a
/// turn that had already finished. The second `send` at the end is the
/// point: it proves the discussion is usable afterwards, not just that an
/// event was emitted.
#[tokio::test]
async fn an_account_refusal_ends_the_discussion_turn_with_its_remedy() {
struct SignedOut;
#[async_trait]
impl TurnGenerator for SignedOut {
async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
panic!("the assistant loop must generate through the typed seam")
}
async fn generate_assistant(
&self,
_req: GenerateRequest,
) -> Result<InferenceResult, crate::coder::native_loop::AssistantGenerateError>
{
Err(crate::coder::native_loop::AssistantGenerateError::from(
car_inference::InferenceError::CredentialUnavailable {
provider: "parslee".into(),
model: "parslee/advisor".into(),
reason: car_inference::CredentialFailure::SignedOut,
detail: "no account is signed in. Run `car auth login`".into(),
},
))
}
}
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let generator: Arc<dyn TurnGenerator> = Arc::new(SignedOut);
let id = start(&state, repo.path(), generator).await;
send_message(&state, &id, "owner-1", "what is fragile here?")
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
wait_for_turn_complete(&entry).await;
let kinds: Vec<DiscussEventKind> = {
let events = entry.events.lock().await;
events.iter().map(|e| e.kind.clone()).collect()
};
let message = kinds
.iter()
.find_map(|k| match k {
DiscussEventKind::Error { message } => Some(message.clone()),
_ => None,
})
.expect("the refusal must reach the client as a terminal error");
assert_eq!(
message,
crate::assistant::AUTH_REQUIRED_SIGNED_OUT_MESSAGE,
"the remedy is the daemon's approved copy, verbatim"
);
assert!(
matches!(kinds.last(), Some(DiscussEventKind::TurnComplete {})),
"the turn must end with TurnComplete: {kinds:?}"
);
// …and the discussion is usable again, which is what `TurnComplete`
// promises a client.
send_message(&state, &id, "owner-1", "and the second question?")
.await
.expect("a completed turn must accept the next message");
}
/// A second `send` while a turn is streaming is refused, not silently
/// interleaved — and `promote` refuses too rather than distilling a
/// question with no answer beside it.
#[tokio::test]
async fn a_turn_in_flight_blocks_a_second_send_and_promote() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let gate = Arc::new(tokio::sync::Notify::new());
let generator: Arc<dyn TurnGenerator> = Arc::new(Blocking { gate: gate.clone() });
let id = start(&state, repo.path(), generator).await;
send_message(&state, &id, "owner-1", "first question")
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
for _ in 0..200 {
if entry.is_answering() {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
assert!(entry.is_answering(), "the turn should be in flight");
let err = send_message(&state, &id, "owner-1", "second question")
.await
.unwrap_err();
assert!(
err.contains("still answering"),
"a concurrent send must be refused, not silently lose a turn: {err}"
);
let err = promote(&state, &id, "owner-1").await.unwrap_err();
assert!(
err.contains("still answering"),
"promote must not distill a half-finished turn: {err}"
);
gate.notify_one();
wait_for_turn_complete(&entry).await;
}
/// Closing cancels the in-flight turn rather than leaving it billing tokens
/// to a conversation nobody can read.
#[tokio::test]
async fn close_cancels_an_in_flight_turn() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let gate = Arc::new(tokio::sync::Notify::new());
let generator: Arc<dyn TurnGenerator> = Arc::new(Blocking { gate });
let id = start(&state, repo.path(), generator).await;
send_message(&state, &id, "owner-1", "a broad question")
.await
.unwrap();
let entry = get_discussion(&state, &id).await.unwrap();
for _ in 0..200 {
if entry.is_answering() {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
close(&state, &id, "owner-1").await.unwrap();
assert!(!entry.is_answering(), "close must stop the turn");
assert!(state.coder_discussions.lock().await.is_empty());
}
/// A `send` whose handler future is dropped after the turn was dispatched
/// must NOT leave the discussion latched as answering.
///
/// `coder.discuss.send` is not deadline-exempt, so the daemon's handler
/// deadline cancels this future at its one remaining await — the turn's
/// first-event cursor. `in_flight` is set by CAS before that and cleared
/// only at the tail of the spawned turn task, so the question is whether
/// that task exists. It does: the dispatch is complete before this await is
/// ever reached, so the drop costs the caller its `seq` reply and nothing
/// else. The turn answers the message, releases `in_flight`, and the
/// discussion is usable again — rather than answering "still answering the
/// previous message" forever with nothing running (`reap_idle` runs only on
/// the next `discuss.start`, so a quiet daemon never reclaimed that).
#[tokio::test]
async fn a_send_cancelled_after_dispatch_leaves_the_discussion_usable() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![
turn("answered anyway", json!([])),
turn("answered on the retry", json!([])),
],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
let entry = get_discussion(&state, &id).await.unwrap();
// Cancellation IS "the future is dropped at an .await point" — that is
// all `tokio::time::timeout` does to a handler. Poll once to get past
// the CAS and the dispatch, park on the turn's first-event cursor, then
// drop it there.
let mut send = Box::pin(send_message(
&state,
&id,
"owner-1",
"the message whose reply frame gets cancelled",
));
assert!(
matches!(futures::poll!(send.as_mut()), std::task::Poll::Pending),
"the fixture needs the send parked on its cursor"
);
assert!(
entry.is_answering(),
"the fixture needs the CAS to have run"
);
drop(send);
// The turn was already dispatched, so it runs and releases the latch.
wait_for_turn_complete(&entry).await;
for _ in 0..200 {
if !entry.is_answering() {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
assert!(
!entry.is_answering(),
"a cancelled handler must not strand `in_flight`"
);
// ...and the discussion still works.
send_message(&state, &id, "owner-1", "second try")
.await
.expect("the discussion must still accept a message");
}
/// A `send` whose dispatch is refused by a `close` must never reach the
/// model.
///
/// `cancel_turn` used to read `turn_task` before `send_message` stored it —
/// the store happened only after the first `emit().await` — so a close in
/// that window found nothing to abort, removed the entry from the registry,
/// and then `send_message` resumed and spawned a 12-turn model loop against
/// a discussion nothing could reach. The turn slot latch is what closed
/// that: `close` latches it, the dispatch checks it under the same lock,
/// and a send that arrives after the latch is REFUSED. Here the latch is
/// set without removing the registry entry, so the send reaches the
/// dispatch and is refused exactly there.
#[tokio::test]
async fn a_close_racing_a_dispatching_send_never_starts_the_turn() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let calls = Arc::new(AtomicUsize::new(0));
let script: Arc<dyn TurnGenerator> = Arc::new(Counting {
calls: calls.clone(),
});
let id = start(&state, repo.path(), script).await;
let entry = get_discussion(&state, &id).await.unwrap();
entry.cancel_turn();
let err = send_message(&state, &id, "owner-1", "a broad question")
.await
.unwrap_err();
assert!(
err.contains("closed while your message was being dispatched"),
"the caller must be told the send did not run: {err}"
);
assert_eq!(
calls.load(Ordering::SeqCst),
0,
"a closed discussion must never reach the model"
);
assert!(!entry.is_answering());
close(&state, &id, "owner-1").await.unwrap();
assert!(state.coder_discussions.lock().await.is_empty());
}
/// A discussion is owned by the connection that opened it — and that is now
/// enforced, not merely recorded. Every method resolved by id alone, so any
/// connected client could send into, promote, or close another's
/// discussion; closing one mid-turn cancels a turn its owner is watching.
#[tokio::test]
async fn another_connection_cannot_drive_a_discussion() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
for err in [
send_message(&state, &id, "intruder", "run this for me")
.await
.unwrap_err(),
promote(&state, &id, "intruder").await.unwrap_err(),
close(&state, &id, "intruder").await.unwrap_err(),
// The path `coder.discuss.subscribe` and `coder.start
// { discussion_id }` both resolve through.
match get_owned_discussion(&state, &id, "intruder").await {
Ok(_) => panic!("a foreign client must not resolve another's discussion"),
Err(e) => e,
},
] {
assert!(
err.contains("belongs to another connection"),
"a foreign client must be refused: {err}"
);
}
// Untouched, and still the owner's to close.
assert_eq!(state.coder_discussions.lock().await.len(), 1);
close(&state, &id, "owner-1").await.unwrap();
}
/// Operator text is retained in the transcript, the replay buffer and the
/// distill prompt, so it needs the byte cap `summarize_repo` already has.
#[tokio::test]
async fn an_oversized_message_is_refused() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
let entry = get_discussion(&state, &id).await.unwrap();
let err = send_message(
&state,
&id,
"owner-1",
&"x".repeat(DISCUSS_MESSAGE_MAX_BYTES + 1),
)
.await
.unwrap_err();
assert!(err.contains("the limit is"), "{err}");
// Refused BEFORE the latch, so the discussion is still usable.
assert!(!entry.is_answering());
assert!(entry.transcript_is_empty());
}
/// A disconnecting client's discussions are freed, not leaked for the
/// daemon's lifetime.
#[tokio::test]
async fn disconnect_closes_the_owning_clients_discussions() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
assert_eq!(state.coder_discussions.lock().await.len(), 1);
// A different client disconnecting leaves it alone...
drop_subscriptions_for_client(&state, "someone-else").await;
assert_eq!(state.coder_discussions.lock().await.len(), 1);
// ...its owner disconnecting closes it.
drop_subscriptions_for_client(&state, "owner-1").await;
assert!(state.coder_discussions.lock().await.is_empty());
assert!(get_discussion(&state, &id).await.is_err());
}
/// The cap is a slot RESERVATION, so the test holds the slots directly
/// rather than building eight full assistant runtimes — each
/// `start_discussion` binds a substrate and registers ~40 tools, and doing
/// that eight times to assert a length check cost minutes of CI for
/// nothing.
#[tokio::test]
async fn open_discussions_are_capped() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let held: Vec<_> = (0..MAX_OPEN_DISCUSSIONS)
.map(|_| {
state
.coder_discussion_slots
.clone()
.try_acquire_owned()
.expect("a fresh daemon has every slot free")
})
.collect();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
let err = start_discussion(&state, repo.path(), "owner-1", engine(repo.path()), script)
.await
.unwrap_err();
assert!(err.contains("already open"), "{err}");
// ...and a freed slot admits the next one.
drop(held);
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
start_discussion(&state, repo.path(), "owner-1", engine(repo.path()), script)
.await
.expect("a released slot must be reusable");
}
/// The cap must hold under CONCURRENT starts, which is what it did not do:
/// the count was read, the registry lock released, and two awaits (bind the
/// substrate, build the runtime) ran before the insert — and the daemon
/// runs a connection's requests concurrently, so N pipelined starts all
/// read `len() == 0`, all passed a cap of 8, and all built a runtime.
///
/// One slot is left free and four starts race for it: exactly one may win,
/// and the three losers must fail BEFORE building anything.
#[tokio::test]
async fn concurrent_starts_cannot_exceed_the_open_discussion_cap() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let _held: Vec<_> = (0..MAX_OPEN_DISCUSSIONS - 1)
.map(|_| {
state
.coder_discussion_slots
.clone()
.try_acquire_owned()
.unwrap()
})
.collect();
let mut racers = Vec::new();
for _ in 0..4 {
let state = state.clone();
let repo = repo.path().to_path_buf();
racers.push(tokio::spawn(async move {
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
start_discussion(&state, &repo, "owner-1", engine(&repo), script).await
}));
}
let mut admitted = 0;
let mut refused = 0;
for racer in racers {
match racer.await.unwrap() {
Ok(_) => admitted += 1,
Err(e) => {
assert!(e.contains("already open"), "unexpected refusal: {e}");
refused += 1;
}
}
}
assert_eq!(admitted, 1, "exactly one racer may take the last slot");
assert_eq!(refused, 3);
assert_eq!(
state.coder_discussions.lock().await.len(),
1,
"the registry must never exceed the cap"
);
}
#[tokio::test]
async fn unknown_discussion_ids_are_clear_errors() {
let (state, _journal) = state();
for err in [
send_message(&state, "disc-nope", "owner-1", "hi")
.await
.unwrap_err(),
promote(&state, "disc-nope", "owner-1").await.unwrap_err(),
constraints_for_start(&state, "disc-nope")
.await
.unwrap_err(),
] {
assert!(err.contains("disc-nope"), "must name the id, got: {err}");
}
assert!(close(&state, "disc-nope", "owner-1").await.is_err());
}
#[test]
fn checkout_followup_uses_current_files_but_requires_a_known_delivery() {
assert_eq!(
followup_base_from(&[
json!({"state":"merged", "result_delivery":"checkout", "result_commit":"saved"})
])
.unwrap(),
None
);
assert!(
followup_base_from(&[json!({"state":"merged", "result_delivery":"checkout"})]).is_err()
);
assert_eq!(
followup_base_from(&[
json!({"state":"merged", "result_delivery":"branch", "result_commit":"saved"})
])
.unwrap(),
Some("saved".into())
);
}
#[test]
fn followup_never_substitutes_head_for_known_work() {
assert!(followup_base_from(&[
json!({"state": "merged", "updated_at": 1, "result_commit": "one"}),
json!({"state": "merged", "updated_at": 1, "result_commit": "two"}),
])
.is_err());
assert!(
followup_base_from(&[json!({"state": "merged", "result_branch": "moved"})]).is_err()
);
assert!(
followup_base_from(&[json!({"state": "failed", "worktree": "/retained"})]).is_err()
);
let rows = vec![
json!({"state": "failed", "worktree": null}),
json!({"state": "merged", "result_commit": "fixed-revision"}),
];
assert_eq!(
followup_base_from(&rows).unwrap().as_deref(),
Some("fixed-revision")
);
}
#[tokio::test]
async fn linked_work_is_grounded_scoped_and_blocks_overlapping_starts() {
use super::super::router::EngineChoice;
use super::super::session::{CoderSession, CoderState};
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let dir = tempfile::tempdir().unwrap();
let id = start(
&state,
repo.path(),
Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
}),
)
.await;
let entry = get_discussion(&state, &id).await.unwrap();
assert!(
claim_coding_start(&state, &id, Path::new("/different"), dir.path().into())
.await
.unwrap_err()
.contains("repository")
);
let guard = claim_coding_start_at(&state, &id, dir.path().into())
.await
.unwrap();
assert!(tokio::time::timeout(
std::time::Duration::from_millis(20),
claim_coding_start_at(&state, &id, dir.path().into())
)
.await
.is_err());
let mut run = CoderSession::new(
&entry.repo,
"repair export",
EngineChoice::Native,
3,
Some(dir.path().into()),
);
run.discussion_id = Some(id.clone());
run.state = CoderState::NeedsApproval;
run.workspace_path = Some(repo.path().into());
run.persist().unwrap();
drop(guard);
let error = claim_coding_start_at(&state, &id, dir.path().into())
.await
.unwrap_err();
assert!(error.contains(&run.id));
assert!(error.contains("unfinished work"));
run.state = CoderState::Failed;
run.error = Some("export check failed".into());
run.persist().unwrap();
let mut unrelated = CoderSession::new(
&entry.repo,
"private other conversation",
EngineChoice::Native,
3,
Some(dir.path().into()),
);
unrelated.discussion_id = Some("other".into());
unrelated.persist().unwrap();
let rows = coding_runs(&state, &id, &entry.repo, dir.path().into())
.await
.unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0]["error"], "export check failed");
assert_eq!(rows[0]["live"], false);
assert!(
retained_workspace(&state, &id, &entry.repo, dir.path().into())
.await
.unwrap_err()
.contains("not been confirmed stopped")
);
run.execution_stopped = true;
run.persist().unwrap();
assert!(
retained_workspace(&state, &id, &entry.repo, dir.path().into())
.await
.is_err(),
"a stopped task cannot adopt a path outside the daemon's worktrees"
);
assert!(
coding_runs(&state, &id, Path::new("/different"), dir.path().into())
.await
.unwrap()
.is_empty()
);
assert!(claim_coding_start_at(&state, &id, dir.path().into())
.await
.is_ok());
close(&state, &id, "owner-1").await.unwrap();
}
#[tokio::test]
async fn saved_list_survives_restart_and_filters_principals_and_live_owners() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, journal) = state();
let owner = client(&state, "operator-1").await;
let other = client(&state, "agent-1").await;
*other.agent_id.lock().await = Some("foreign".into());
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
let opened = open_discussion(
&state,
repo.path(),
&owner.client_id,
engine(repo.path()),
script,
"operator",
None,
)
.await
.unwrap();
let id = opened["discussion_id"].as_str().unwrap();
assert!(handle_discuss_list(&state, &owner).await.unwrap()["saved"]
.as_array()
.unwrap()
.is_empty());
close(&state, id, &owner.client_id).await.unwrap();
let restarted = Arc::new(ServerState::standalone(journal.path().to_path_buf()));
let saved = handle_discuss_list(&restarted, &owner).await.unwrap();
assert_eq!(saved["saved"][0]["discussion_id"], id);
assert!(saved["discussions"].as_array().unwrap().is_empty());
assert!(
handle_discuss_list(&restarted, &other).await.unwrap()["saved"]
.as_array()
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn list_and_close_track_open_discussions() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
let owner = client(&state, "owner-1").await;
let listed = handle_discuss_list(&state, &owner).await.unwrap();
assert_eq!(listed["discussions"].as_array().unwrap().len(), 1);
assert_eq!(listed["discussions"][0]["discussion_id"], id.as_str());
assert_eq!(listed["discussions"][0]["turns"], 0);
// ...and it is scoped to the owning connection.
let stranger = client(&state, "someone-else").await;
let listed = handle_discuss_list(&state, &stranger).await.unwrap();
assert!(
listed["discussions"].as_array().unwrap().is_empty(),
"another connection must not see this discussion: {listed}"
);
assert_eq!(
close(&state, &id, "owner-1").await.unwrap(),
json!({ "ok": true })
);
let listed = handle_discuss_list(&state, &owner).await.unwrap();
assert!(listed["discussions"].as_array().unwrap().is_empty());
}
/// The stated guarantee, measured where a client actually lives: what a
/// SUBSCRIBER receives across an attach is contiguous from its cursor —
/// no gap, no duplicate — even when emits are racing the attach.
///
/// Asserting on the buffer proves only that the drain is the single writer.
/// The property clients depend on spans three more hops the buffer never
/// touches: the replay clone at attach, the per-subscriber queue, and that
/// lane's sender task. An attach that registered before replaying would
/// duplicate here and an attach that replayed before registering would drop
/// whatever emitted in between, and the buffer would look perfect either
/// way.
///
/// **The replay hop has to actually run.** `handle_discuss_subscribe` has
/// exactly one await before it enqueues `Attach`, and it resolves on the
/// first poll; on the current-thread test runtime the "racing" emitter had
/// therefore never been polled when the attach landed, so
/// `events_replayed` was 0 on every run and `replayed <= 30` was satisfied
/// by nothing having been replayed at all. Mutating the replay filter to
/// `e.seq > from_seq` — the off-by-one that drops the first event of every
/// real client resume — left the test green. So: yield until the emitter
/// has genuinely produced events, assert the replay is non-empty, and
/// attach a second time from a NON-ZERO cursor, where an off-by-one is a
/// wrong first seq rather than a merely smaller count.
#[tokio::test]
async fn a_subscriber_receives_every_seq_exactly_once_across_its_attach() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
let entry = get_discussion(&state, &id).await.unwrap();
let (channel, frames) = capturing_channel();
let owner = state
.create_session("owner-1", channel.clone())
.await
.unwrap();
// Emitted WHILE the attach is in flight: each of these lands on one
// side or the other of the `Attach` command, and the subscriber must
// see it exactly once either way.
let racing = {
let entry = entry.clone();
tokio::spawn(async move {
for i in 0..30u64 {
entry
.emit(DiscussEventKind::AssistantDelta {
text: format!("during-{i}"),
})
.await;
}
})
};
// Let the emitter actually get ahead of the attach. Without this the
// attach wins every poll and there is no race to observe.
while entry.events.lock().await.is_empty() {
tokio::task::yield_now().await;
}
let subscribed = handle_discuss_subscribe(
&rpc_req(json!({ "discussion_id": id, "from_seq": 0 })),
&state,
&owner,
)
.await
.unwrap();
racing.await.unwrap();
// ...and after it, live through the same lane.
for i in 0..20u64 {
entry
.emit(DiscussEventKind::AssistantDelta {
text: format!("after-{i}"),
})
.await;
}
const TOTAL: usize = 50;
let replayed = subscribed["events_replayed"].as_u64().unwrap();
assert!(
replayed > 0,
"the attach replayed nothing, so this test never exercised the \
replay hop it exists to cover"
);
assert!(
replayed <= 30,
"replay cannot exceed what was emitted before the attach: {replayed}"
);
let mut seqs = Vec::new();
for _ in 0..400 {
seqs = delivered_seqs(&frames);
if seqs.len() >= TOTAL {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
}
assert_eq!(
seqs,
(0..TOTAL as u64).collect::<Vec<_>>(),
"a subscriber must receive seq 0..{TOTAL} once each, in order"
);
// ...and a resume from a non-zero cursor is inclusive of that cursor.
// Every seq is in the buffer now, so this is exact: an off-by-one in
// the replay filter shows up as a missing FIRST event, not as a count
// that merely looks plausible.
const RESUME_FROM: u64 = 17;
let (resumed_channel, resumed_frames) = capturing_channel();
// Same client id: a discussion is owned by the connection that opened
// it, and re-attaching replaces that connection's lane.
let resumed = state
.create_session("owner-1", resumed_channel)
.await
.unwrap();
let reattached = handle_discuss_subscribe(
&rpc_req(json!({ "discussion_id": id, "from_seq": RESUME_FROM })),
&state,
&resumed,
)
.await
.unwrap();
assert_eq!(
reattached["events_replayed"].as_u64().unwrap(),
TOTAL as u64 - RESUME_FROM,
"a resume from {RESUME_FROM} must replay seq {RESUME_FROM}..{TOTAL}"
);
let mut resumed_seqs = Vec::new();
for _ in 0..400 {
resumed_seqs = delivered_seqs(&resumed_frames);
if resumed_seqs.len() >= TOTAL - RESUME_FROM as usize {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
}
assert_eq!(
resumed_seqs,
(RESUME_FROM..TOTAL as u64).collect::<Vec<_>>(),
"a resume must start AT its cursor, not one past it"
);
}
/// A send that IS dispatched still puts the `user_message` on the
/// stream first, ahead of every assistant delta for that turn. Moving the
/// emit into the turn must not reorder it behind the turn's own output.
#[tokio::test]
async fn a_dispatched_send_emits_the_user_message_before_any_delta() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![turn("here is what I would change", json!([]))],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
let entry = get_discussion(&state, &id).await.unwrap();
let sent = send_message(&state, &id, "owner-1", "what should this change do?")
.await
.unwrap();
assert_eq!(
sent["seq"], 0,
"the reported cursor is the user_message's own seq"
);
wait_for_turn_complete(&entry).await;
let events = entry.events.lock().await;
assert!(
matches!(events[0].kind, DiscussEventKind::UserMessage { .. }),
"the operator's message must be the turn's first event, got: {:?}",
events[0].kind
);
assert!(
events.len() > 1,
"the turn produced nothing to order against"
);
assert!(
!events[1..]
.iter()
.any(|e| matches!(e.kind, DiscussEventKind::UserMessage { .. })),
"exactly one user_message per send"
);
}
/// A subscriber that has stopped reading is its own problem: it is SHED,
/// and the turn it was watching completes anyway.
///
/// Half of this was never pinned. When the drain performed the sends
/// itself, a half-open board (no FIN, no RST — writes park forever) held
/// the drain for `DISCUSS_SEND_TIMEOUT` per event, so the next `Emit` sat
/// unprocessed and every `entry.emit(…).await` inside `run_turn` waited on
/// it: one dead board stalled the whole TURN. The turn here must complete
/// while the wedge is still in place, on a clock well inside that deadline.
#[tokio::test]
async fn a_wedged_subscriber_is_shed_and_the_turn_still_completes() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![turn("here is what I would change", json!([]))],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
let entry = get_discussion(&state, &id).await.unwrap();
let (channel, _frames) = capturing_channel();
let owner = state
.create_session("owner-1", channel.clone())
.await
.unwrap();
let unsubscribed = Arc::strong_count(&channel);
handle_discuss_subscribe(
&rpc_req(json!({ "discussion_id": id, "from_seq": 0 })),
&state,
&owner,
)
.await
.unwrap();
assert_eq!(
Arc::strong_count(&channel),
unsubscribed + 1,
"the lane must hold this subscriber's channel"
);
// Half-open from here on: writes never fail, they just never finish.
let stuck = channel.write.lock().await;
let started = std::time::Instant::now();
send_message(&state, &id, "owner-1", "what should this change do?")
.await
.unwrap();
let mut completed = false;
for _ in 0..120 {
if entry
.events
.lock()
.await
.iter()
.any(|e| matches!(e.kind, DiscussEventKind::TurnComplete {}))
{
completed = true;
break;
}
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
}
assert!(
completed && started.elapsed() < DISCUSS_SEND_TIMEOUT,
"the turn must not wait on a wedged subscriber's socket ({:?} elapsed)",
started.elapsed()
);
// ...and the lane is shed rather than carried: its queue fills, the
// drain's `try_send` fails, and dropping the `Subscriber` aborts the
// task parked on that socket — releasing the channel handle it pinned.
for i in 0..(DISCUSS_SUBSCRIBER_QUEUE + 64) {
entry
.emit(DiscussEventKind::AssistantDelta {
text: format!("overflow-{i}"),
})
.await;
}
let mut shed = false;
for _ in 0..200 {
if Arc::strong_count(&channel) == unsubscribed {
shed = true;
break;
}
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
}
assert!(
shed,
"a subscriber that is not draining must be shed, not retained"
);
drop(stuck);
}
/// A `send` whose dispatch is refused must leave no unanswered operator
/// question behind — not in the transcript, and not on the wire.
///
/// `InFlightGuard` frees the discussion on that path, so `is_answering()`
/// reads false — and `promote` and `coder.start { discussion_id }` gate on
/// exactly that. The transcript still ended in a question no turn answered,
/// which sailed through both guards and became the distillation input those
/// guards exist to prevent: a confident intent invented from a question
/// nobody replied to.
///
/// The `user_message` event had the same hole for the same reason: it was
/// emitted BEFORE the dispatch, so a refused send still put the operator's
/// question in the replay buffer and on every subscriber while the
/// transcript row rolled back — and the board's discussion pane rendered
/// that question followed by permanent silence. The emit now happens inside
/// the turn, so it and the transcript row commit or roll back together.
#[tokio::test]
async fn a_refused_send_leaves_no_unanswered_turn_in_the_transcript() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let calls = Arc::new(AtomicUsize::new(0));
let script: Arc<dyn TurnGenerator> = Arc::new(Counting {
calls: calls.clone(),
});
let id = start(&state, repo.path(), script).await;
let entry = get_discussion(&state, &id).await.unwrap();
// Latch the turn slot closed WITHOUT removing the registry entry, so
// the send reaches the dispatch and is refused THERE — the window a
// racing `close` actually wins.
entry.cancel_turn();
let err = send_message(&state, &id, "owner-1", "should we rewrite the scheduler?")
.await
.unwrap_err();
assert!(
err.contains("closed while your message was being dispatched"),
"expected a refused dispatch, got: {err}"
);
assert!(
!entry.is_answering(),
"a refused dispatch must not strand `in_flight`"
);
assert!(
entry.transcript_is_empty(),
"a question no turn will answer must not survive in the transcript: {:?}",
lock(&entry.transcript)
);
assert!(
!entry
.events
.lock()
.await
.iter()
.any(|e| matches!(e.kind, DiscussEventKind::UserMessage { .. })),
"...nor reach the replay buffer and every subscriber"
);
// ...and the guards that read the transcript agree.
let err = promote(&state, &id, "owner-1").await.unwrap_err();
assert!(
err.contains("no turns yet"),
"promote must refuse an empty discussion rather than distill a stranded \
question: {err}"
);
assert!(
constraints_for_start(&state, &id).await.unwrap().is_empty(),
"coder.start must not distill constraints from a stranded question"
);
assert_eq!(
calls.load(Ordering::SeqCst),
0,
"no turn ran, so nothing reached the model"
);
}
/// The drain assigns `seq` under the buffer lock as the only writer, so the
/// buffer is strictly ordered even when emits are produced concurrently.
#[tokio::test]
async fn concurrent_emits_stay_seq_ordered_in_the_buffer() {
let repo = tempfile::tempdir().unwrap();
init_repo(repo.path());
let (state, _journal) = state();
let script: Arc<dyn TurnGenerator> = Arc::new(Script {
turns: vec![],
cursor: AtomicUsize::new(0),
});
let id = start(&state, repo.path(), script).await;
let entry = get_discussion(&state, &id).await.unwrap();
let mut tasks = Vec::new();
for i in 0..50 {
let e = entry.clone();
tasks.push(tokio::spawn(async move {
e.emit(DiscussEventKind::AssistantDelta {
text: format!("chunk-{i}"),
})
.await
}));
}
for t in tasks {
t.await.unwrap();
}
let events = entry.events.lock().await;
assert_eq!(events.len(), 50);
for (i, e) in events.iter().enumerate() {
assert_eq!(e.seq, i as u64, "buffer must be in seq order");
}
}
#[test]
fn discuss_event_json_shape_is_ws_friendly() {
let e = DiscussEvent {
discussion_id: "disc-x".into(),
seq: 7,
ts: 1,
kind: DiscussEventKind::AssistantDelta {
text: "hello".into(),
},
};
let v = serde_json::to_value(&e).unwrap();
assert_eq!(v["type"], "assistant_delta");
assert_eq!(v["text"], "hello");
assert_eq!(v["seq"], 7);
assert_eq!(v["discussion_id"], "disc-x");
let v = serde_json::to_value(DiscussEvent {
discussion_id: "disc-x".into(),
seq: 8,
ts: 1,
kind: DiscussEventKind::TurnComplete {},
})
.unwrap();
assert_eq!(v["type"], "turn_complete");
}
}