car-server-core 0.49.0

//! The flagship assistant as three MCP tools: `assistant_start`,
//! `assistant_poll`, `assistant_cancel` (car#972 §6).
//!
//! ## Why a run handle and not a blocking call
//!
//! An assistant run takes minutes. A blocking `tools/call` times out in most
//! hosts, and the host has no way to say "stop" while it waits. So `start`
//! returns a handle immediately, `poll` reads incremental progress from a
//! buffer, and `cancel` targets the handle.
//!
//! A handle costs nothing on the endpoint's wire posture. `handle_mcp_post`
//! reads no session header and this changes none of that: the `run_id` is
//! application state the **client** carries between calls, exactly like the
//! opaque `resources/list` cursor. Progress *notifications* would have cost the
//! stateless POST — that is why this is a poll.
//!
//! ## Not a second envelope
//!
//! A poll returns the `car.do/1` events and, at the end, the `car.do/1`
//! document — built by [`car_server_core::assistant::do_json`], the same code
//! `car do --json` writes, including its `SUMMARY_CAP` / `RECEIPT_SAMPLE`
//! truncation. That truncation exists because a delegating caller pays for
//! every byte in the user's context, which is true of an MCP tool result
//! verbatim. The only field this layer adds is a monotonic `seq` per event.
//!
//! [`car_server_core::assistant::do_json`]: crate::assistant::do_json
//!
//! ## Daemon only
//!
//! These are registered on the daemon's HTTP endpoint through
//! [`car_mcp::Server::register_tool`] and nowhere else. `car-mcp-server` (the
//! stdio binary an editor plugin launches) is `car-mcp` + `car-telemetry`: it
//! has no `Runtime`, no inference engine, and no daemon state, and giving it
//! one would mean shipping the whole daemon inside the plugin's MCP binary. A
//! stdio client asking for `assistant_start` gets "unknown tool", which is
//! accurate, and `car do --json` is the delegation path there.
//!
//! ## Lifetime, and what it is not
//!
//! The registry is in memory. A `run_id` is a handle on a **live run**, not a
//! durable record: a daemon restart makes every handle unknown, and `poll`
//! says so rather than returning an empty `running` forever. Three bounds keep
//! it honest — [`MAX_OPEN_RUNS`] concurrent runs, [`RUN_IDLE_TTL_SECS`] since
//! the last poll, and [`RUN_EVENT_BUFFER_MAX`] retained events with the head
//! trimmed and the loss *stated* (`events_skipped`), never a silent gap.
//!
//! ## Recursion (car#972 §7)
//!
//! [`car_external_agents::recursion`]'s guard reads `$CAR_INVOKED_BY` from the
//! process environment, which works when the host *spawns* CAR. The daemon is
//! spawned by the supervisor and serves many callers, so a caller that is
//! itself an adapter names itself per call with `invoked_by`. That id is merged
//! with the daemon's own chain — read once at registry construction, never
//! per request — and the result is **recorded on the run and echoed by
//! `poll`**.
//!
//! Recorded, not enforced: nothing on this surface spawns an external agent, so
//! there is no adapter for the guard to refuse here. The ancestry's job is to
//! be the chain an agent *further down* is judged against, and to make a loop
//! visible in a poll result. What closes the cycle on this surface is the
//! posture below.
//!
//! The cycle §7 names — CAR → `shell` → `claude -p` → CAR — is closed here by
//! the execution posture rather than by the ancestry: the default run is a
//! sandbox with no network and no host binaries, and a `local: true` run binds
//! at [`PermissionTier::ReadOnly`] with no approval gate, so `shell`,
//! `write_file`, and `edit_file` are all refused. Like the guard itself, this
//! is a cost-and-hang control that depends on the posture it describes; it is
//! not a security boundary.
//!
//! [`PermissionTier::ReadOnly`]: car_policy::permission::PermissionTier

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

use car_external_agents::recursion::seed_ancestry_in;
use car_inference::tasks::generate::Message;
use car_inference::InferenceEngine;
use car_mcp::{RegisterError, ToolError, ToolHandler};
use car_policy::permission::PermissionTier;
use serde_json::{json, Value};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};

use crate::assistant::do_json::{
    startup_error_doc, EventSink, GoalReport, JsonEmitter, SandboxPosture,
};
use crate::assistant::{
    bind_default_substrate, build_assistant_runtime, prompt, run_assistant_goal_loop,
    run_assistant_loop_cancellable, AssistantConfig, AssistantEvent, DEFAULT_ASSISTANT_IMAGE,
};
use crate::coder::native_loop::TurnGenerator;
use crate::session::ServerState;

/// Concurrently **executing** assistant runs per daemon. Each pins a
/// `Runtime`, a substrate (usually a container), and a model session, so they
/// are not free.
///
/// The same number `coder.discuss` uses, for the same reason and enforced the
/// same way: a semaphore permit taken **before** any of `start`'s async work —
/// not by counting the registry, which `discuss` records as a TOCTOU check
/// that bounded nothing under pipelined starts.
///
/// The permit lives in the run's *task*, not in its registry entry, so it comes
/// back the moment the run stops rather than when the record is finally reaped.
/// A finished run still occupies the registry until it is polled and aged out —
/// but it costs nothing, and a client that polled eight runs to completion must
/// not have to wait out the idle TTL to start a ninth.
pub const MAX_OPEN_RUNS: usize = 8;

/// A run nobody has polled for this long is cancelled and dropped. Long enough
/// that a host doing something else between polls is fine; short enough that an
/// abandoned run does not bill model tokens to nobody all afternoon.
pub const RUN_IDLE_TTL_SECS: u64 = 60 * 60;

/// Retained events per run. Past this the oldest are dropped and `poll` reports
/// `events_skipped` — a trimmed head is stated, never a silent gap. Same cap
/// and same discipline as the discuss replay buffer.
pub const RUN_EVENT_BUFFER_MAX: usize = 2000;

/// Turn cap when the caller does not set one — the `car do --max-turns`
/// default, because this is the same agent.
const DEFAULT_MAX_TURNS: u32 = 50;

/// Hard ceiling on a caller-supplied `max_turns`. The cap is a backstop against
/// a runaway loop; letting a caller raise it without limit removes the backstop.
const MAX_MAX_TURNS: u32 = 200;

/// Goal-mode re-drive budget — the `car do --goal-max-iterations` default.
const GOAL_MAX_ITERATIONS: u32 = 10;

/// What `start` tells the caller to wait before its first `poll`.
const POLL_AFTER_MS: u64 = 2000;

/// How often the background reaper looks for idle runs.
const REAP_INTERVAL_SECS: u64 = 60;

/// The registry's bounds, so a test can drive the TTL and the buffer trim
/// without mocking the clock or queueing 2000 real events.
#[derive(Clone, Copy)]
pub struct RunBounds {
    pub max_open_runs: usize,
    pub idle_ttl_secs: u64,
    pub event_buffer_max: usize,
}

impl Default for RunBounds {
    fn default() -> Self {
        Self {
            max_open_runs: MAX_OPEN_RUNS,
            idle_ttl_secs: RUN_IDLE_TTL_SECS,
            event_buffer_max: RUN_EVENT_BUFFER_MAX,
        }
    }
}

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

/// Take a `std` lock without letting a poisoned mutex become permanent — the
/// same reasoning as `coder::discuss::lock`. A panic under one of these would
/// otherwise wedge the run for its whole lifetime — and the detached run task
/// swallows the panic itself, so nothing else would report it. What the caller
/// sees if a run dies anyway is [`RunEntry::settle_if_task_died`]'s error
/// document; this keeps one panic from poisoning the *next* run's bookkeeping.
fn lock<T>(m: &StdMutex<T>) -> std::sync::MutexGuard<'_, T> {
    m.lock().unwrap_or_else(|e| e.into_inner())
}

/// The state of a run **handle**, which is not the state of the run's work.
///
/// `Ok` means the loop reached a terminal outcome; whether that outcome was a
/// success, a `max_turns` stop, or a goal that was never met is the `status`
/// field of the `car.do/1` document, which `poll` returns alongside this.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum RunStatus {
    Running,
    Ok,
    Error,
    Cancelled,
}

impl RunStatus {
    fn as_str(self) -> &'static str {
        match self {
            RunStatus::Running => "running",
            RunStatus::Ok => "ok",
            RunStatus::Error => "error",
            RunStatus::Cancelled => "cancelled",
        }
    }

    fn is_terminal(self) -> bool {
        !matches!(self, RunStatus::Running)
    }
}

/// The handle state and the terminal document behind ONE lock, so the
/// terminal-state guard is a single critical section.
///
/// Adopted from `car_a2a::store`'s `update_state` guard, and for the reason
/// A2A documents hitting: a run that has been cancelled may still be unwinding,
/// and the task's own settle must not clobber `cancelled` back to `ok`.
#[derive(Default)]
struct RunOutcome {
    status: Option<RunStatus>,
    doc: Option<Value>,
}

/// The `car.do/1` event stream for one run, with a monotonic `seq` and a
/// trimmed head.
struct EventBuffer {
    events: VecDeque<Value>,
    /// Seq the next event will take.
    next_seq: u64,
    /// Seq of the oldest event still retained. Rises as the head is trimmed,
    /// which is what makes `events_skipped` computable rather than guessed.
    first_seq: u64,
    max: usize,
}

impl EventBuffer {
    fn new(max: usize) -> Self {
        Self {
            events: VecDeque::new(),
            next_seq: 0,
            first_seq: 0,
            max,
        }
    }

    fn push(&mut self, mut event: Value) {
        let seq = self.next_seq;
        self.next_seq += 1;
        if let Some(obj) = event.as_object_mut() {
            obj.insert("seq".to_string(), json!(seq));
        }
        self.events.push_back(event);
        while self.events.len() > self.max {
            self.events.pop_front();
            self.first_seq += 1;
        }
    }

    /// Events at or after `since_seq`, plus how many were dropped from the head
    /// before the caller could read them.
    fn since(&self, since_seq: u64) -> (Vec<Value>, u64) {
        let events = self
            .events
            .iter()
            .filter(|e| e["seq"].as_u64().unwrap_or(0) >= since_seq)
            .cloned()
            .collect();
        (events, self.first_seq.saturating_sub(since_seq))
    }
}

/// One live run.
struct RunEntry {
    id: String,
    /// The invocation chain this run was started from — the process ancestry
    /// plus whatever the caller named itself as. Echoed by `poll` so the
    /// guard's answer travels with the run.
    ancestry: Vec<String>,
    created_at: u64,
    /// Last `poll` (or the start), for the idle TTL.
    last_poll: AtomicU64,
    outcome: StdMutex<RunOutcome>,
    events: StdMutex<EventBuffer>,
    /// Checked by the loop at each turn boundary — so a cancel lands between
    /// model calls, never inside one.
    cancel: Arc<AtomicBool>,
    /// The run's task, so the reaper can stop one parked in a model call. Not
    /// an ownership cycle: a `JoinHandle` is a handle, and the runtime drops
    /// the future (releasing its `Arc<RunEntry>` and its slot permit) when the
    /// task ends.
    task: StdMutex<Option<tokio::task::JoinHandle<()>>>,
}

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

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

    fn status(&self) -> RunStatus {
        lock(&self.outcome).status.unwrap_or(RunStatus::Running)
    }

    /// Record the terminal state, unless one is already recorded.
    ///
    /// The guard is the point: a cancelled run whose task is still unwinding
    /// will settle its own outcome a moment later, and that write must not
    /// turn `cancelled` into `ok`.
    fn settle(&self, status: RunStatus, doc: Option<Value>) {
        let mut out = lock(&self.outcome);
        if out.status.is_some() {
            return;
        }
        out.status = Some(status);
        out.doc = doc;
    }

    /// Settle a run whose task has ended without recording any outcome.
    ///
    /// The only way that happens is a panic. `tokio::spawn` swallows a panic
    /// into the `JoinHandle`, nothing joins one, and `run_task` records its
    /// outcome as its last statement — so a panicked run would otherwise report
    /// `running` forever, *and* report it forever: every `poll` calls
    /// [`Self::touch`], which keeps the idle reaper away from the entry for as
    /// long as an attentive client keeps asking. The module promises `poll`
    /// says what happened rather than returning an empty `running`; this is
    /// what makes that true when the run dies instead of finishing.
    ///
    /// No race with the normal path: `is_finished()` only flips once the future
    /// has completed, and on the normal path `settle` ran inside it.
    fn settle_if_task_died(&self) {
        if self.status().is_terminal() {
            return;
        }
        if !lock(&self.task).as_ref().is_some_and(|h| h.is_finished()) {
            return;
        }
        tracing::error!(
            run_id = %self.id,
            "assistant run task ended without an outcome; reporting it as an error"
        );
        self.settle(
            RunStatus::Error,
            Some(startup_error_doc(
                "run_task_died",
                "the run's task ended without producing a result, which means it panicked. \
                 Nothing was left running; the events already returned are all there are.",
                &[
                    "Start again with assistant_start.",
                    "Check the daemon log for the panic.",
                ],
            )),
        );
    }

    /// Ask the run to stop. It stops at the next turn boundary, not now — the
    /// loop checks the flag between model calls.
    fn request_cancel(&self) {
        self.cancel.store(true, Ordering::SeqCst);
    }

    /// Stop a run nobody is waiting for: request the cancel AND drop the task.
    ///
    /// Only the reaper does this. `assistant_cancel` deliberately does not: a
    /// caller who cancels still wants the document describing what the run had
    /// done, and the loop produces one (`status: "cancelled"`, with receipts)
    /// if it is allowed to unwind. Nobody is going to read a reaped run's
    /// document, so there the priority is to stop spending immediately.
    fn abandon(&self) {
        self.request_cancel();
        self.settle(RunStatus::Cancelled, None);
        if let Some(handle) = lock(&self.task).take() {
            handle.abort();
        }
    }
}

/// Appends the emitter's `car.do/1` events into a run's buffer.
///
/// Holds a `Weak` so an already-reaped run's still-unwinding task drops its
/// events instead of refilling a buffer nobody can reach.
struct RunSink(std::sync::Weak<RunEntry>);

impl EventSink for RunSink {
    fn emit(&self, event: Value) {
        if let Some(entry) = self.0.upgrade() {
            lock(&entry.events).push(event);
        }
    }
}

/// The model seam: the engine that builds the runtime, and the generator the
/// loop drives.
///
/// Two handles rather than one because they are the same object in production
/// (`InferenceEngine` implements [`TurnGenerator`]) and different in tests,
/// where a scripted generator answers turns while a real-but-unused engine
/// supplies the runtime — the pattern `coder::discuss` tests use.
#[derive(Clone)]
struct ModelSeam {
    engine: Arc<InferenceEngine>,
    generator: Arc<dyn TurnGenerator>,
}

/// Live assistant runs, keyed by `run_id`.
pub struct AssistantRunRegistry {
    state: Arc<ServerState>,
    runs: tokio::sync::Mutex<HashMap<String, Arc<RunEntry>>>,
    slots: Arc<Semaphore>,
    bounds: RunBounds,
    /// Test override. `None` means the daemon's shared engine, resolved on the
    /// first `start` rather than at registration — initializing inference at
    /// boot would spawn the offload worker for a daemon that may never be asked
    /// to run anything.
    model: Option<ModelSeam>,
    /// Where execution traces go, or `None` for a test that must not write into
    /// the user's real trajectory history. Same explicit `Option` and the same
    /// reasoning as [`build_assistant_runtime`]'s own parameter.
    trajectories: Option<PathBuf>,
    /// The daemon's **own** invocation chain, read from `$CAR_INVOKED_BY` once
    /// here rather than on every `start`.
    ///
    /// Once, because the daemon's environment is fixed when the supervisor
    /// spawns it — re-reading it per request would make each run's ancestry a
    /// function of process-global state that no caller controls, and would make
    /// every test of this surface depend on the shell it was launched from.
    /// (This repo ships `CAR_INVOKED_BY=claude-code` in `plugins/car/.mcp.json`,
    /// so "the shell it was launched from" is not hypothetical.)
    base_ancestry: Vec<String>,
}

impl AssistantRunRegistry {
    /// The daemon's registry, with a background reaper.
    pub fn new(state: Arc<ServerState>) -> Arc<Self> {
        let registry = Arc::new(Self {
            state,
            runs: tokio::sync::Mutex::new(HashMap::new()),
            slots: Arc::new(Semaphore::new(MAX_OPEN_RUNS)),
            bounds: RunBounds::default(),
            model: None,
            trajectories: Some(car_memgine::TrajectoryStore::default_path()),
            base_ancestry: car_external_agents::recursion::ancestry(),
        });
        // Weak, so the reaper is not what keeps the registry alive: when the
        // daemon drops it, the next tick ends the task.
        let weak = Arc::downgrade(&registry);
        tokio::spawn(async move {
            let mut ticker =
                tokio::time::interval(std::time::Duration::from_secs(REAP_INTERVAL_SECS));
            loop {
                ticker.tick().await;
                match weak.upgrade() {
                    Some(registry) => registry.reap_idle().await,
                    None => break,
                }
            }
        });
        registry
    }

    fn seam(&self) -> ModelSeam {
        self.model.clone().unwrap_or_else(|| {
            let engine = crate::handler::get_inference_engine(&self.state).clone();
            ModelSeam {
                generator: engine.clone(),
                engine,
            }
        })
    }

    /// Cancel and drop runs nobody has polled inside the TTL — **including
    /// runs still executing**, which is the point: an abandoned run is billing
    /// model tokens to nobody.
    pub(crate) async fn reap_idle(&self) {
        let stale: Vec<Arc<RunEntry>> = {
            let runs = self.runs.lock().await;
            runs.values()
                .filter(|e| e.idle_secs() > self.bounds.idle_ttl_secs)
                .cloned()
                .collect()
        };
        for entry in stale {
            tracing::info!(
                run_id = %entry.id,
                idle_secs = entry.idle_secs(),
                "reaping an assistant run nobody has polled"
            );
            entry.abandon();
            self.runs.lock().await.remove(&entry.id);
        }
    }

    /// `assistant_start` — bind an environment, spawn the run, return a handle.
    pub async fn start(&self, args: &Value) -> Result<Value, ToolError> {
        let mut req = StartArgs::parse(args)?;
        // A caller-supplied root that does not exist is a run that would bind,
        // start, and then fail every file and shell call for a reason the model
        // cannot see. Checked before a slot is reserved, and reported as an
        // execution error rather than a protocol one so the model reads it and
        // can correct the path itself.
        if !req.cwd.is_dir() {
            return Err(refused(&format!(
                "cwd is not a directory: {}. Pass the absolute path of the project the run \
                 should work in.",
                req.cwd.display()
            )));
        }

        // Reap before enforcing the cap, so this morning's forgotten run never
        // blocks this afternoon's.
        self.reap_idle().await;
        // RESERVE the slot before any async work below. Counting the registry
        // and inserting afterwards is the TOCTOU `coder::discuss` records:
        // pipelined starts all read the same count, all pass, and the cap
        // bounds nothing.
        let slot = self.slots.clone().try_acquire_owned().map_err(|_| {
            refused(&format!(
                "{} assistant runs are already executing, which is the limit. Wait for one \
                 to reach a terminal status (assistant_poll) or stop one with \
                 assistant_cancel, then start again — starts are refused, never queued.",
                self.bounds.max_open_runs
            ))
        })?;

        let env = bind_default_substrate(req.local, false, &req.cwd, None).await;
        // Captured before `env` is consumed by `build_assistant_runtime`: the
        // envelope has to report what the run was BOUND to, not what was asked
        // for. A run that silently fell back to the local host is materially
        // different from one that chose it.
        let posture = SandboxPosture {
            sandboxed: env.sandboxed,
            image: env.sandboxed.then(|| DEFAULT_ASSISTANT_IMAGE.to_string()),
            tier: format!("{:?}", env.tier),
            root: env.root.display().to_string(),
            fallback_notice: env.fallback_notice.clone(),
        };
        // A goal check is a `shell` call on the substrate. At ReadOnly — a
        // `local: true` run, or a sandbox run that fell back because Docker is
        // not there — shell is gated and this surface has no approval gate, so
        // the check would be refused every iteration and the run would burn its
        // whole budget failing a test it was never allowed to run. Refuse up
        // front and say which it is.
        if req.until.is_some() && matches!(env.tier, PermissionTier::ReadOnly) {
            return Err(refused(&format!(
                "`until` needs to run a shell command to decide completion, and this run \
                 bound at ReadOnly ({}), where shell is refused. Drop `until`, or start \
                 without `local` so the run gets a sandbox with a real shell.",
                env.fallback_notice
                    .as_deref()
                    .unwrap_or("local: true was requested")
            )));
        }

        let seam = self.seam();
        let asm = build_assistant_runtime(
            seam.engine.clone(),
            env,
            None,
            None,
            None,
            self.trajectories.clone(),
        )
        .await
        .map_err(|e| refused(&format!("could not assemble the assistant runtime: {e}")))?;
        req.system = prompt::batch_prompt(&asm.description, &asm.tools);

        let cfg = AssistantConfig {
            model: req.model.clone(),
            strict_model: false,
            max_turns: req.max_turns,
            tools: asm.tools.clone(),
            gated_tools: asm.gated_tools.clone(),
            // No approval transport exists on an MCP tool call, so a gated tool
            // is denied rather than queued for a human who is not there. That
            // is what makes `local: true` a read-only run.
            approval_policy: None,
            proactive_memory: Some(asm.proactive_memory.clone()),
            tool_labels: None,
            todos: Some(Arc::clone(&asm.todos)),
            value_store_previews: false,
        };

        let id = format!("mcp-run-{}", uuid::Uuid::new_v4().simple());
        let entry = Arc::new(RunEntry {
            id: id.clone(),
            ancestry: seed_ancestry_in(&self.base_ancestry, req.invoked_by.as_deref()),
            created_at: now_secs(),
            last_poll: AtomicU64::new(now_secs()),
            outcome: StdMutex::new(RunOutcome::default()),
            events: StdMutex::new(EventBuffer::new(self.bounds.event_buffer_max)),
            cancel: Arc::new(AtomicBool::new(false)),
            task: StdMutex::new(None),
        });
        self.runs.lock().await.insert(id.clone(), entry.clone());

        let sandbox = posture.to_json();
        let handle = tokio::spawn(run_task(
            entry.clone(),
            seam.generator,
            asm.runtime,
            cfg,
            posture,
            req,
            slot,
        ));
        *lock(&entry.task) = Some(handle);

        Ok(json!({
            "run_id": id,
            "status": RunStatus::Running.as_str(),
            "poll_after_ms": POLL_AFTER_MS,
            // Reported at start, not only at the end: a caller that asked for a
            // sandbox and silently got the local host should learn that before
            // it hands the run anything sensitive.
            "sandbox": sandbox,
            "ancestry": entry.ancestry,
        }))
    }

    /// `assistant_poll` — everything since `since_seq`, plus the run's state.
    pub async fn poll(&self, args: &Value) -> Result<Value, ToolError> {
        let run_id = str_arg(args, "run_id")?.ok_or_else(|| missing("run_id"))?;
        let since_seq = u64_arg(args, "since_seq")?.unwrap_or(0);
        let entry = match self.runs.lock().await.get(&run_id).cloned() {
            Some(e) => e,
            // A tool *execution* error, so the model reads it and can react. A
            // restarted daemon is the common cause and it is not the caller
            // having got the protocol wrong.
            None => {
                return Err(refused(&format!(
                    "run not found: {run_id} — the daemon may have restarted. A run handle \
                     lives in memory and does not survive one. Start again with \
                     assistant_start."
                )))
            }
        };
        entry.touch();
        // Before reading the outcome, not after: a run whose task panicked has
        // no outcome to read, and this is the only place anything notices.
        entry.settle_if_task_died();

        // `next_seq` is read under the SAME lock as the events it follows. Two
        // locks would let an event land in between: the caller would be told to
        // resume past a seq it was never given, which is a silent gap of
        // exactly the kind `events_skipped` exists to make impossible.
        let (events, events_skipped, next_seq) = {
            let buffer = lock(&entry.events);
            let (events, skipped) = buffer.since(since_seq);
            (events, skipped, buffer.next_seq)
        };
        let (status, doc) = {
            let out = lock(&entry.outcome);
            (out.status.unwrap_or(RunStatus::Running), out.doc.clone())
        };

        let mut result = json!({
            "run_id": entry.id,
            "status": status.as_str(),
            "events": events,
            "next_seq": next_seq,
            // Stated rather than implied: a poll that silently skipped 300
            // events reads as a complete stream to anyone who does not check.
            "events_skipped": events_skipped,
            "ancestry": entry.ancestry,
            "created_at": entry.created_at,
        });
        match status {
            RunStatus::Running => {
                result["poll_after_ms"] = json!(POLL_AFTER_MS);
            }
            // The `car.do/1` document, verbatim — `summary`, `turns`,
            // `receipts`, `ungrounded_claims`, `sandbox`, and `goal` when the
            // run had an `until`. Absent only on a reaped run, which was
            // stopped before it could produce one.
            _ => {
                if let Some(doc) = doc {
                    result["result"] = doc;
                }
            }
        }
        Ok(result)
    }

    /// `assistant_cancel` — ask a run to stop at its next turn boundary.
    pub async fn cancel(&self, args: &Value) -> Result<Value, ToolError> {
        let run_id = str_arg(args, "run_id")?.ok_or_else(|| missing("run_id"))?;
        let entry = self.runs.lock().await.get(&run_id).cloned();
        let status = match entry {
            None => "unknown",
            // Idempotent, and a no-op success: a run that already finished was
            // not going to do anything else anyway, and reporting that as a
            // failure would make every "cancel then poll" sequence look broken.
            Some(entry) if entry.status().is_terminal() => "already_terminal",
            Some(entry) => {
                entry.touch();
                entry.request_cancel();
                "cancelled"
            }
        };
        Ok(json!({ "run_id": run_id, "status": status }))
    }
}

/// Drive one run to a terminal state and settle its document.
///
/// `_slot` rides along so the concurrency permit is released exactly when this
/// task ends — including when the reaper aborts it.
#[allow(clippy::too_many_arguments)]
async fn run_task(
    entry: Arc<RunEntry>,
    generator: Arc<dyn TurnGenerator>,
    runtime: car_engine::Runtime,
    cfg: AssistantConfig,
    posture: SandboxPosture,
    req: StartArgs,
    _slot: OwnedSemaphorePermit,
) {
    let emitter = JsonEmitter::new(posture, Arc::new(RunSink(Arc::downgrade(&entry))));
    emitter.started(&req.task, cfg.model.as_deref().unwrap_or("(router)"));

    let description = req.system.clone();
    let (outcome, goal) = match req.until.clone() {
        None => {
            let mut messages = vec![
                Message::System {
                    content: description,
                },
                Message::User {
                    content: req.task.clone(),
                },
            ];
            let outcome = run_assistant_loop_cancellable(
                generator.as_ref(),
                &runtime,
                &cfg,
                &mut messages,
                &entry.cancel,
                None,
                None,
                |ev: AssistantEvent| emitter.on_assistant_event(&ev),
            )
            .await;
            (outcome, None)
        }
        Some(check) => {
            let (outcome, report) = goal_run(
                &entry,
                &emitter,
                generator.as_ref(),
                &runtime,
                &cfg,
                &req,
                &check,
                description,
            )
            .await;
            (outcome, Some(report))
        }
    };

    let cancelled = outcome.status == "cancelled";
    let doc = emitter.finish(&outcome, goal.as_ref());
    let status = if doc["status"] == "error" {
        RunStatus::Error
    } else if cancelled {
        RunStatus::Cancelled
    } else {
        RunStatus::Ok
    };
    entry.settle(status, Some(doc));
}

/// Goal mode: re-drive the agent until `check` exits 0 on the substrate. The
/// completion decision is a real command the runtime runs and audits, not a
/// model reading its own transcript — the same loop `car do --until` drives.
#[allow(clippy::too_many_arguments)]
async fn goal_run(
    entry: &Arc<RunEntry>,
    emitter: &JsonEmitter,
    generator: &dyn TurnGenerator,
    runtime: &car_engine::Runtime,
    cfg: &AssistantConfig,
    req: &StartArgs,
    check: &str,
    system: String,
) -> (crate::assistant::AssistantOutcome, GoalReport) {
    use car_verify::goal::{GoalCondition, GoalGovernor, GoalSpec, GoalStatus};

    let spec = GoalSpec {
        goal: req.task.clone(),
        condition: GoalCondition::Command {
            id: "goal_check".into(),
            expect_exit: 0,
        },
        governor: GoalGovernor {
            max_turns: Some(GOAL_MAX_ITERATIONS),
            ..Default::default()
        },
    };
    let mut messages = vec![Message::System {
        content: format!(
            "{system}\n\nYou are working toward a goal. Completion is verified \
             deterministically by running this shell command:\n  {check}\nIt is done \
             only when that command exits 0. Keep working until it does."
        ),
    }];
    let result = run_assistant_goal_loop(
        generator,
        runtime,
        cfg,
        &mut messages,
        &entry.cancel,
        None,
        &spec,
        |_outcome| {
            let cmd = check.to_string();
            async move {
                let exit = goal_check_exit(runtime, cfg, &cmd).await;
                let mut g = car_engine::GoalGather::default();
                g.command_exits.insert("goal_check".into(), exit);
                g
            }
        },
        |ev: AssistantEvent| emitter.on_assistant_event(&ev),
    )
    .await;

    let report = GoalReport {
        check: check.to_string(),
        passed: matches!(result.run.status, GoalStatus::Achieved),
        // Reported alongside `passed`, never folded into it: a goal a model
        // judge signed off is not the same result as one the check confirmed.
        grounded: result.run.grounded,
        iterations: result.run.iterations,
        halt: match &result.run.status {
            GoalStatus::Achieved => None,
            GoalStatus::Halted { halt } => Some(halt.as_str().to_string()),
        },
    };
    (result.outcome, report)
}

/// Run the completion check through the runtime, so it is validated, policed,
/// and audited like any other tool call.
async fn goal_check_exit(
    runtime: &car_engine::Runtime,
    cfg: &AssistantConfig,
    command: &str,
) -> i32 {
    // Should not be reachable — `start` refuses `until` when shell is gated —
    // but a gated check that silently "passed" would be the worst possible
    // failure of a completion check.
    if cfg.gated_tools.iter().any(|tool| tool == "shell") {
        return 1;
    }
    let proposal: car_ir::ActionProposal = serde_json::from_value(json!({
        "source": "goal-check",
        "actions": [{
            "id": "goal_check",
            "type": "tool_call",
            "tool": "shell",
            "parameters": { "command": command },
        }],
    }))
    .expect("static shell-check proposal shape");
    let exec = runtime.execute(&proposal).await;
    exec.results
        .first()
        .and_then(|r| r.output.as_ref())
        .and_then(|o| o.get("exit_code"))
        .and_then(|v| v.as_i64())
        .unwrap_or(1) as i32
}

// ---------------------------------------------------------------------------
// Arguments
// ---------------------------------------------------------------------------

/// Parsed `assistant_start` arguments.
#[derive(Clone)]
struct StartArgs {
    task: String,
    cwd: PathBuf,
    until: Option<String>,
    max_turns: u32,
    local: bool,
    model: Option<String>,
    invoked_by: Option<String>,
    /// Filled in after the runtime is assembled — the system prompt describing
    /// the bound environment and the tools.
    system: String,
}

impl StartArgs {
    fn parse(args: &Value) -> Result<Self, ToolError> {
        let task = str_arg(args, "task")?
            .filter(|t| !t.trim().is_empty())
            .ok_or_else(|| missing("task"))?;
        // The daemon's own cwd is the fallback, not the answer: a daemon
        // started by a login item is rooted wherever the login item was, which
        // is almost never the project. A caller-supplied root is the
        // `coder.discuss.start { repo }` precedent.
        let cwd = match str_arg(args, "cwd")?.filter(|c| !c.trim().is_empty()) {
            Some(c) => PathBuf::from(c),
            None => std::env::current_dir().map_err(|e| {
                refused(&format!(
                    "no cwd was given and the daemon's is unresolvable: {e}"
                ))
            })?,
        };
        let max_turns = u64_arg(args, "max_turns")?
            .map(|n| (n as u32).clamp(1, MAX_MAX_TURNS))
            .unwrap_or(DEFAULT_MAX_TURNS);
        Ok(Self {
            task,
            cwd,
            until: str_arg(args, "until")?.filter(|c| !c.trim().is_empty()),
            max_turns,
            local: args.get("local").and_then(Value::as_bool).unwrap_or(false),
            model: str_arg(args, "model")?.filter(|m| !m.trim().is_empty()),
            invoked_by: str_arg(args, "invoked_by")?,
            system: String::new(),
        })
    }
}

fn missing(field: &str) -> ToolError {
    ToolError::InvalidParams(format!("missing {field}"))
}

/// A refusal the model should see: [`ToolError::Internal`] comes back as a
/// normal result carrying `isError: true`, not as a JSON-RPC error the client
/// swallows.
fn refused(message: &str) -> ToolError {
    ToolError::Internal(message.to_string())
}

fn str_arg(args: &Value, key: &str) -> Result<Option<String>, ToolError> {
    match args.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::String(s)) => Ok(Some(s.clone())),
        Some(_) => Err(ToolError::InvalidParams(format!("{key} must be a string"))),
    }
}

fn u64_arg(args: &Value, key: &str) -> Result<Option<u64>, ToolError> {
    match args.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(v) => v.as_u64().map(Some).ok_or_else(|| {
            ToolError::InvalidParams(format!("{key} must be a non-negative integer"))
        }),
    }
}

// ---------------------------------------------------------------------------
// Registration
// ---------------------------------------------------------------------------

macro_rules! tool_handler {
    ($name:ident, $method:ident) => {
        struct $name(Arc<AssistantRunRegistry>);

        #[async_trait::async_trait]
        impl ToolHandler for $name {
            async fn call(&self, args: Value) -> Result<String, ToolError> {
                let v = self.0.$method(&args).await?;
                serde_json::to_string(&v).map_err(|e| ToolError::Internal(e.to_string()))
            }
        }
    };
}

tool_handler!(StartTool, start);
tool_handler!(PollTool, poll);
tool_handler!(CancelTool, cancel);

/// Register `assistant_start` / `assistant_poll` / `assistant_cancel` on
/// `server`, backed by a fresh run registry over `state`.
///
/// Called by the daemon only. Every schema carries all four annotation hints,
/// which [`car_mcp::Server::register_tool`] enforces — the seam does not route
/// around the gate the built-ins pass.
pub fn register_assistant_tools(
    server: &mut car_mcp::Server,
    state: Arc<ServerState>,
) -> Result<(), RegisterError> {
    let registry = AssistantRunRegistry::new(state);
    server.register_tool(start_schema(), Arc::new(StartTool(registry.clone())))?;
    server.register_tool(poll_schema(), Arc::new(PollTool(registry.clone())))?;
    server.register_tool(cancel_schema(), Arc::new(CancelTool(registry)))?;
    Ok(())
}

fn start_schema() -> Value {
    json!({
        "name": "assistant_start",
        "description": "Start a CAR assistant run (the agent behind `car do`) and return a \
                        run handle immediately. Poll it with assistant_poll; stop it with \
                        assistant_cancel. A run takes minutes, so this never blocks. The \
                        handle lives in the daemon's memory and does NOT survive a daemon \
                        restart. By default the run executes in a Docker sandbox with no \
                        network; `local: true` runs on the host read-only — writes and \
                        shell are refused, because a tool call has no way to ask a human \
                        for approval. If your host is itself an agent CLI, set `invoked_by` \
                        to its adapter id (claude-code, codex, gemini) so the run records \
                        the invocation chain it is part of.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "task": { "type": "string", "description": "What the assistant should do." },
                "cwd": {
                    "type": "string",
                    "description": "Working directory for the run. Defaults to the daemon's, which is usually not your project.",
                },
                "until": {
                    "type": "string",
                    "description": "Goal mode: keep working until this shell command exits 0. Requires the sandbox (a local run cannot use shell).",
                },
                "max_turns": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": MAX_MAX_TURNS,
                    "description": "Safety cap on agent turns. Default 50.",
                },
                "local": {
                    "type": "boolean",
                    "description": "Run on the host instead of the sandbox. Read-only: writes and shell are refused.",
                },
                "model": { "type": "string", "description": "Pin a model. Default: CAR's router picks." },
                "invoked_by": {
                    "type": "string",
                    "description": "Your own adapter id if you are an agent CLI: claude-code, codex, or gemini. Recorded on the run and echoed by assistant_poll as `ancestry`.",
                },
            },
            "required": ["task"],
        },
        "annotations": {
            "readOnlyHint": false,
            // It runs an autonomous agent with a real shell. Whether a given
            // run overwrites anything is not knowable up front, and a host
            // deciding whether to prompt should assume the worse case.
            "destructiveHint": true,
            "idempotentHint": false,
            // Not because of the web tools — `web_search` and `http_request`
            // declare `full_access`, which exceeds every tier this surface
            // binds, so they are gated and, with no approval transport, always
            // denied. It is open-world because the caller hands an autonomous
            // agent a free-text task: what it reads and touches inside its root
            // is decided by the model at runtime, not by these arguments.
            "openWorldHint": true,
        },
    })
}

fn poll_schema() -> Value {
    json!({
        "name": "assistant_poll",
        "description": "Read progress from an assistant run. Returns events at or after \
                        `since_seq` plus `next_seq` to pass to the following poll. `status` \
                        is running | ok | error | cancelled and describes the HANDLE; once \
                        terminal, `result` carries the car.do/1 document (summary, turns, \
                        receipts, ungrounded_claims, sandbox) whose own `status` describes \
                        the WORK — success | max_turns | stalled | goal_pending | cancelled \
                        | error. `events_skipped` is non-zero when the buffer trimmed its \
                        head before you read it. Poll incrementally: a poll with \
                        `since_seq: 0` on a long run can return up to 2000 buffered events \
                        in one result, all of which land in your context. An unknown run_id \
                        means the run finished long ago or the daemon restarted.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "run_id": { "type": "string" },
                "since_seq": {
                    "type": "integer",
                    "minimum": 0,
                    "description": "First event seq to return. Use next_seq from the previous poll; 0 for the whole buffer.",
                },
            },
            "required": ["run_id"],
        },
        "annotations": {
            "readOnlyHint": true,
            "destructiveHint": false,
            "idempotentHint": true,
            "openWorldHint": false,
        },
    })
}

fn cancel_schema() -> Value {
    json!({
        "name": "assistant_cancel",
        "description": "Stop an assistant run. The run stops at its next TURN BOUNDARY, not \
                        mid-model-call, so expect one more turn's worth of activity — then \
                        poll for the car.do/1 document describing what it had done. Returns \
                        cancelled | already_terminal | unknown; cancelling a finished or \
                        unknown run is a successful no-op.",
        "inputSchema": {
            "type": "object",
            "properties": { "run_id": { "type": "string" } },
            "required": ["run_id"],
        },
        "annotations": {
            "readOnlyHint": false,
            "destructiveHint": false,
            "idempotentHint": true,
            "openWorldHint": false,
        },
    })
}

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

    impl AssistantRunRegistry {
        /// A registry with caller-chosen bounds and a scripted model, and NO
        /// background reaper — the tests call [`Self::reap_idle`] directly.
        ///
        /// Bounds are injected rather than the clock mocked: driving the TTL by
        /// setting it to zero and the buffer trim by setting it to four proves
        /// the same code paths a 3600-second TTL and a 2000-event buffer would,
        /// without a test that sleeps for an hour or scripts 2000 turns.
        fn for_test(state: Arc<ServerState>, bounds: RunBounds, model: ModelSeam) -> Arc<Self> {
            Arc::new(Self {
                state,
                runs: tokio::sync::Mutex::new(HashMap::new()),
                slots: Arc::new(Semaphore::new(bounds.max_open_runs)),
                bounds,
                model: Some(model),
                // Never the user's real trajectory store: a test that runs a
                // scripted agent must not skew the per-tool success rates
                // `verify.monte_carlo` reads back.
                trajectories: None,
                // Empty, not `recursion::ancestry()`: a test asserting on a
                // run's ancestry must not read `$CAR_INVOKED_BY`, which this
                // repo's own plugin manifest sets. The merge of a non-empty
                // base with a per-call `invoked_by` is covered where it lives,
                // by `recursion`'s `seed_ancestry_in` tests.
                base_ancestry: Vec::new(),
            })
        }
    }

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

    /// A turn that calls `calculate` — a tool that needs no substrate and is
    /// not approval-gated, so it exercises the receipt path without depending
    /// on Docker or a human.
    fn calculate_turn(text: &str, expression: &str) -> InferenceResult {
        turn(
            text,
            json!([{
                "id": "c1",
                "name": "calculate",
                "arguments": { "expression": expression },
            }]),
        )
    }

    /// A turn that calls `shell` — the tool `local: true` must refuse. The
    /// command is harmless on purpose: if the gate ever regresses, this test
    /// should fail on the assertion, not by doing something to the host.
    fn shell_turn(text: &str, command: &str) -> InferenceResult {
        turn(
            text,
            json!([{
                "id": "s1",
                "name": "shell",
                "arguments": { "command": command },
            }]),
        )
    }

    /// A model that panics instead of answering.
    ///
    /// The only way into the "task ended without settling" branch from a test.
    /// It prints a panic backtrace into the test output; that is the scripted
    /// panic, not a failure.
    struct Panics;

    #[async_trait]
    impl TurnGenerator for Panics {
        async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
            panic!("scripted panic inside a run task");
        }
    }

    /// A scripted model that optionally hands control to the test at each turn.
    ///
    /// `entered`/`release` are what make the cancellation test deterministic:
    /// the test waits until the model is genuinely mid-turn, cancels, and only
    /// then lets the turn finish — so the run reaches the next turn boundary
    /// with the flag already set, which is exactly the boundary the tool
    /// description promises cancellation lands on.
    struct Script {
        turns: Vec<InferenceResult>,
        cursor: AtomicUsize,
        entered: Option<Arc<tokio::sync::Notify>>,
        release: Option<Arc<tokio::sync::Notify>>,
    }

    impl Script {
        fn new(turns: Vec<InferenceResult>) -> Arc<Self> {
            Arc::new(Self {
                turns,
                cursor: AtomicUsize::new(0),
                entered: None,
                release: None,
            })
        }

        fn gated(
            turns: Vec<InferenceResult>,
            entered: Arc<tokio::sync::Notify>,
            release: Arc<tokio::sync::Notify>,
        ) -> Arc<Self> {
            Arc::new(Self {
                turns,
                cursor: AtomicUsize::new(0),
                entered: Some(entered),
                release: Some(release),
            })
        }
    }

    #[async_trait]
    impl TurnGenerator for Script {
        async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
            if let Some(entered) = &self.entered {
                // `notify_one` stores a permit, so it does not matter whether
                // the test is already waiting when this runs.
                entered.notify_one();
            }
            if let Some(release) = &self.release {
                release.notified().await;
            }
            let i = self.cursor.fetch_add(1, Ordering::SeqCst);
            self.turns
                .get(i)
                .cloned()
                .ok_or_else(|| "script exhausted".to_string())
        }
    }

    /// A real engine that is never asked to generate — it supplies the runtime
    /// while the `Script` answers turns, the pattern `coder::discuss`'s tests
    /// use. Pointed at a temp models dir so nothing reaches the user's cache.
    fn seam(root: &std::path::Path, generator: Arc<dyn TurnGenerator>) -> ModelSeam {
        let mut cfg = car_inference::InferenceConfig::default();
        cfg.models_dir = root.join("models");
        ModelSeam {
            engine: Arc::new(car_inference::InferenceEngine::new(cfg)),
            generator,
        }
    }

    fn state() -> (Arc<ServerState>, tempfile::TempDir) {
        let journal = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));
        (state, journal)
    }

    /// `local: true` throughout: a test must not depend on Docker being
    /// installed, and the scripted turns never call a gated tool anyway.
    fn start_args(cwd: &std::path::Path, task: &str) -> Value {
        json!({ "task": task, "cwd": cwd.display().to_string(), "local": true })
    }

    async fn poll(registry: &AssistantRunRegistry, run_id: &str, since: u64) -> Value {
        registry
            .poll(&json!({ "run_id": run_id, "since_seq": since }))
            .await
            .expect("poll")
    }

    /// Poll until the handle leaves `running`, or fail rather than hang.
    async fn await_terminal(registry: &AssistantRunRegistry, run_id: &str) -> Value {
        for _ in 0..500 {
            let v = poll(registry, run_id, 0).await;
            if v["status"] != "running" {
                return v;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        panic!("run {run_id} never reached a terminal status");
    }

    #[tokio::test]
    async fn start_poll_and_cancel_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let registry = AssistantRunRegistry::for_test(
            state,
            RunBounds::default(),
            seam(dir.path(), Script::new(vec![turn("all done", json!([]))])),
        );

        let started = registry
            .start(&start_args(dir.path(), "say you are done"))
            .await
            .expect("start");
        let run_id = started["run_id"].as_str().expect("run_id").to_string();
        assert!(run_id.starts_with("mcp-run-"), "{run_id}");
        assert_eq!(started["status"], "running");
        assert_eq!(started["poll_after_ms"], POLL_AFTER_MS);

        let done = await_terminal(&registry, &run_id).await;
        assert_eq!(done["status"], "ok", "{done}");
        assert_eq!(done["events_skipped"], 0);
        // The terminal payload IS the car.do/1 document, not a second envelope
        // wrapping it.
        assert_eq!(done["result"]["schema"], "car.do/1");
        assert_eq!(done["result"]["summary"], "all done");
        assert!(done["result"]["receipts"]["total"].is_number(), "{done}");
        // ...and the events are the car.do/1 JSONL events, plus a seq.
        let events = done["events"].as_array().expect("events");
        assert_eq!(events[0]["type"], "started");
        assert_eq!(events[0]["seq"], 0);
        assert!(
            events.iter().any(|e| e["type"] == "completed"),
            "{events:?}"
        );
        assert_eq!(done["next_seq"], events.len() as u64);

        // An incremental poll returns only what is new.
        let tail = poll(&registry, &run_id, done["next_seq"].as_u64().unwrap()).await;
        assert!(tail["events"].as_array().unwrap().is_empty(), "{tail}");
    }

    #[tokio::test]
    async fn cancel_lands_at_the_next_turn_boundary() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let entered = Arc::new(tokio::sync::Notify::new());
        let release = Arc::new(tokio::sync::Notify::new());
        let registry = AssistantRunRegistry::for_test(
            state,
            RunBounds::default(),
            seam(
                dir.path(),
                Script::gated(
                    vec![
                        calculate_turn("working", "1 + 1"),
                        turn("never reached", json!([])),
                    ],
                    entered.clone(),
                    release.clone(),
                ),
            ),
        );

        let started = registry
            .start(&start_args(dir.path(), "keep going"))
            .await
            .expect("start");
        let run_id = started["run_id"].as_str().unwrap().to_string();

        // Wait until the model is genuinely mid-turn, THEN cancel: the flag is
        // checked at the top of the next turn, never inside this one.
        entered.notified().await;
        let cancelled = registry
            .cancel(&json!({ "run_id": run_id }))
            .await
            .expect("cancel");
        assert_eq!(cancelled["status"], "cancelled");
        release.notify_one();

        let done = await_terminal(&registry, &run_id).await;
        assert_eq!(done["status"], "cancelled", "{done}");
        // The document still describes what the run had done before stopping —
        // the tool call from turn one is in the receipts.
        assert_eq!(done["result"]["status"], "cancelled");
        assert_eq!(done["result"]["receipts"]["total"], 1);

        // Cancelling a finished run is a successful no-op, not an error: a host
        // that always cancels after reading the result must not see a failure.
        let again = registry
            .cancel(&json!({ "run_id": run_id }))
            .await
            .expect("cancel again");
        assert_eq!(again["status"], "already_terminal");
    }

    #[tokio::test]
    async fn a_start_past_the_cap_is_refused_and_creates_no_run() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let entered = Arc::new(tokio::sync::Notify::new());
        let release = Arc::new(tokio::sync::Notify::new());
        // Two slots rather than the real eight: it is the same code path, and
        // eight live runtimes per test run is a lot of setup for a bound the
        // constant already parameterizes.
        let bounds = RunBounds {
            max_open_runs: 2,
            ..RunBounds::default()
        };
        let registry = AssistantRunRegistry::for_test(
            state,
            bounds,
            seam(
                dir.path(),
                Script::gated(
                    vec![turn("done", json!([]))],
                    entered.clone(),
                    release.clone(),
                ),
            ),
        );

        for _ in 0..2 {
            registry
                .start(&start_args(dir.path(), "hold a slot"))
                .await
                .expect("start within the cap");
        }
        let refused = registry
            .start(&start_args(dir.path(), "one too many"))
            .await
            .expect_err("past the cap");
        // An execution error, so the model reads the refusal and can act on it.
        assert!(refused.is_execution_error());
        let message = refused.message().to_string();
        assert!(message.contains('2'), "the cap must be named: {message}");
        assert!(
            message.contains("assistant_cancel"),
            "the way out must be named: {message}"
        );
        assert_eq!(registry.runs.lock().await.len(), 2, "no run was created");
    }

    /// The cap bounds *execution*, not retention.
    ///
    /// The permit lives in the run's task rather than its registry entry, so a
    /// client that polls a run to completion can start the next one straight
    /// away instead of waiting out the hour-long idle TTL for a record that is
    /// finished and costs nothing.
    #[tokio::test]
    async fn a_finished_run_does_not_keep_holding_its_slot() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let bounds = RunBounds {
            max_open_runs: 1,
            ..RunBounds::default()
        };
        let registry = AssistantRunRegistry::for_test(
            state,
            bounds,
            seam(
                dir.path(),
                Script::new(vec![turn("first", json!([])), turn("second", json!([]))]),
            ),
        );

        let first = registry
            .start(&start_args(dir.path(), "the first run"))
            .await
            .expect("start");
        let run_id = first["run_id"].as_str().unwrap().to_string();
        await_terminal(&registry, &run_id).await;
        // The finished record is still pollable...
        assert_eq!(registry.runs.lock().await.len(), 1);

        // ...and does not block the next run. Retried because the permit drops
        // when the task future ends, a scheduler tick after the status settles.
        for attempt in 0..100 {
            if registry
                .start(&start_args(dir.path(), "the second run"))
                .await
                .is_ok()
            {
                return;
            }
            assert!(attempt < 99, "a finished run never released its slot");
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
    }

    #[tokio::test]
    async fn a_run_nobody_polls_is_reaped_and_its_slot_returned() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let entered = Arc::new(tokio::sync::Notify::new());
        let release = Arc::new(tokio::sync::Notify::new());
        // One slot, and the REAL idle TTL — the test backdates the run's last
        // poll rather than shortening the constant, so what is exercised is the
        // shipped 3600-second bound and not a test-only zero.
        let bounds = RunBounds {
            max_open_runs: 1,
            ..RunBounds::default()
        };
        let registry = AssistantRunRegistry::for_test(
            state,
            bounds,
            seam(
                dir.path(),
                Script::gated(
                    vec![turn("done", json!([]))],
                    entered.clone(),
                    release.clone(),
                ),
            ),
        );

        let started = registry
            .start(&start_args(dir.path(), "abandon me"))
            .await
            .expect("start");
        let run_id = started["run_id"].as_str().unwrap().to_string();
        entered.notified().await; // genuinely executing, and never released

        registry.runs.lock().await[&run_id]
            .last_poll
            .store(now_secs() - RUN_IDLE_TTL_SECS - 1, Ordering::SeqCst);
        registry.reap_idle().await;
        assert!(registry.runs.lock().await.is_empty());
        // And the handle answers honestly rather than reporting an empty
        // "running" forever.
        let err = registry
            .poll(&json!({ "run_id": run_id }))
            .await
            .expect_err("reaped");
        assert!(err.message().contains("run not found"), "{}", err.message());

        // The slot comes back once the reaped task is actually gone — which is
        // what "the cap bounds live runs" has to mean.
        for attempt in 0..100 {
            if registry
                .start(&start_args(dir.path(), "the next run"))
                .await
                .is_ok()
            {
                return;
            }
            assert!(attempt < 99, "the reaped run never released its slot");
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
    }

    #[tokio::test]
    async fn a_trimmed_event_buffer_states_the_gap() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let bounds = RunBounds {
            event_buffer_max: 4,
            ..RunBounds::default()
        };
        let registry = AssistantRunRegistry::for_test(
            state,
            bounds,
            seam(
                dir.path(),
                Script::new(vec![
                    calculate_turn("step one", "1 + 1"),
                    calculate_turn("step two", "2 + 2"),
                    calculate_turn("step three", "3 + 3"),
                    turn("finished", json!([])),
                ]),
            ),
        );

        let started = registry
            .start(&start_args(dir.path(), "do several things"))
            .await
            .expect("start");
        let run_id = started["run_id"].as_str().unwrap().to_string();
        let done = await_terminal(&registry, &run_id).await;

        let events = done["events"].as_array().expect("events");
        assert!(
            events.len() <= 4,
            "buffer was not trimmed: {}",
            events.len()
        );
        assert!(
            done["events_skipped"].as_u64().unwrap() > 0,
            "a trimmed head must be stated, not silent: {done}"
        );
        // The surviving events keep their original seqs, so a caller can tell
        // exactly where the gap is rather than inferring it.
        assert!(events[0]["seq"].as_u64().unwrap() > 0, "{events:?}");
        assert_eq!(
            done["events_skipped"].as_u64().unwrap(),
            events[0]["seq"].as_u64().unwrap()
        );
    }

    #[tokio::test]
    async fn a_caller_that_names_itself_gets_an_ancestry_and_one_that_does_not_gets_none() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let registry = AssistantRunRegistry::for_test(
            state,
            RunBounds::default(),
            seam(dir.path(), Script::new(vec![turn("done", json!([]))])),
        );

        let mut args = start_args(dir.path(), "run under a host");
        args["invoked_by"] = json!("Claude-Code");
        let named = registry.start(&args).await.expect("start");
        assert_eq!(named["ancestry"], json!(["claude-code"]));
        // ...and it travels with the run rather than being re-read from the
        // daemon's process environment on each poll.
        let run_id = named["run_id"].as_str().unwrap().to_string();
        assert_eq!(
            await_terminal(&registry, &run_id).await["ancestry"],
            json!(["claude-code"])
        );

        // A caller that names nothing gets the registry's stored base chain,
        // which `for_test` sets empty. This assertion is on the registry's
        // field, NOT on `$CAR_INVOKED_BY` — which this repo's own
        // `plugins/car/.mcp.json` sets to `claude-code`, so an env-reading
        // version of this test failed for every agent-driven run.
        let anonymous = registry
            .start(&start_args(dir.path(), "run from nowhere"))
            .await
            .expect("start");
        assert_eq!(anonymous["ancestry"], json!([]));
    }

    /// The §7 claim the whole recursion story rests on: what closes
    /// CAR → `shell` → `claude -p` → CAR is the *posture*, not the ancestry.
    /// Every other test here scripts `calculate`, which is never gated, so
    /// without this one a regression in `bind_default_substrate`'s tier or in
    /// `build_assistant_runtime`'s `gated_tools` wiring would open the hole
    /// silently.
    #[tokio::test]
    async fn a_local_run_refuses_shell_and_still_settles() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let registry = AssistantRunRegistry::for_test(
            state,
            RunBounds::default(),
            seam(
                dir.path(),
                Script::new(vec![
                    shell_turn("let me look around", "true"),
                    turn("could not run that", json!([])),
                ]),
            ),
        );

        let started = registry
            .start(&start_args(dir.path(), "run a shell command"))
            .await
            .expect("start");
        // The tier is reported at start, and it is the tier the refusal below
        // depends on.
        assert_eq!(started["sandbox"]["tier"], "ReadOnly", "{started}");
        let run_id = started["run_id"].as_str().unwrap().to_string();

        let done = await_terminal(&registry, &run_id).await;
        // A refused tool does not sink the run: the model is told and the run
        // still produces a document.
        assert_eq!(done["status"], "ok", "{done}");
        assert_eq!(done["result"]["schema"], "car.do/1");
        let events = done["events"].as_array().expect("events");
        assert!(
            events
                .iter()
                .any(|e| e["type"] == "tool_failed" && e["data"]["tool"] == "shell"),
            "shell was not refused: {events:?}"
        );
        assert!(
            !events
                .iter()
                .any(|e| e["type"] == "tool_result" && e["data"]["tool"] == "shell"),
            "shell RAN on a local: true run: {events:?}"
        );
        // Refused at the gate, not attempted and failed: a refusal never
        // reaches the runtime, so it leaves no receipt. This is the assertion
        // that distinguishes "the posture closed the cycle" from "the command
        // happened to error".
        assert_eq!(done["result"]["receipts"]["total"], 0, "{done}");
    }

    #[tokio::test]
    async fn a_run_whose_task_panics_settles_instead_of_polling_forever() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let registry = AssistantRunRegistry::for_test(
            state,
            RunBounds::default(),
            seam(dir.path(), Arc::new(Panics)),
        );

        let started = registry
            .start(&start_args(dir.path(), "die mid-run"))
            .await
            .expect("start");
        let run_id = started["run_id"].as_str().unwrap().to_string();

        // Before the fix this looped 500 times and then failed: the task was
        // gone, nothing joined it, and every poll re-`touch`ed the entry out of
        // the reaper's reach, so `running` was permanent.
        let done = await_terminal(&registry, &run_id).await;
        assert_eq!(done["status"], "error", "{done}");
        assert_eq!(done["result"]["schema"], "car.do/1");
        assert_eq!(done["result"]["error"], "run_task_died", "{done}");
        // And it stays settled — a second poll must not re-decide.
        let again = poll(&registry, &run_id, 0).await;
        assert_eq!(again["status"], "error", "{again}");
    }

    #[tokio::test]
    async fn a_missing_task_is_a_protocol_error_not_a_refusal() {
        let dir = tempfile::tempdir().unwrap();
        let (state, _journal) = state();
        let registry = AssistantRunRegistry::for_test(
            state,
            RunBounds::default(),
            seam(dir.path(), Script::new(vec![])),
        );
        // The tool never ran, so this takes the JSON-RPC error channel — the
        // car#972 §5 split, inherited rather than reinvented.
        let e = registry
            .start(&json!({ "cwd": "." }))
            .await
            .expect_err("no task");
        assert!(!e.is_execution_error());
        assert!(e.message().contains("task"), "{}", e.message());
    }

    /// The counterpart to `car_mcp`'s `the_stdio_server_offers_no_assistant_tools`.
    /// One test alone cannot pin "daemon only"; these two together do.
    #[tokio::test]
    async fn the_daemon_server_advertises_all_three() {
        let (state, _journal) = state();
        let mut server = car_mcp::Server::new();
        register_assistant_tools(&mut server, state).expect("registers");

        let resp = server
            .handle(
                serde_json::from_value(json!({
                    "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {},
                }))
                .expect("request"),
            )
            .await
            .expect("response");
        let names: Vec<String> = resp.result.expect("result")["tools"]
            .as_array()
            .expect("array")
            .iter()
            .map(|t| t["name"].as_str().expect("name").to_string())
            .collect();
        for tool in ["assistant_start", "assistant_poll", "assistant_cancel"] {
            assert!(names.iter().any(|n| n == tool), "{tool} missing: {names:?}");
        }
    }
}