newt-core 0.8.0

//! A **non-blocking turn driver** around [`chat_complete`](super::chat_complete)
//! (issue #308 — the cowork foundation).
//!
//! [`chat_complete`] is `async` and, inside `run_chat`, is pumped by a blocking
//! `rustyline` REPL (`block_in_place` + `block_on`). A downstream `ratatui`
//! app cannot block its event loop like that: it must poll input, redraw, and
//! poll the turn's progress every frame. This module is that seam — a struct an
//! external `crossterm` event loop drives without ever calling `run_chat`:
//!
//! ```text
//! loop {                                 // the consumer's crossterm loop
//!     if event::poll(..) { /* read key, maybe driver.submit(line) */ }
//!     match driver.poll() {              // non-blocking
//!         TurnStatus::Completed { .. } => { /* redraw transcript */ }
//!         TurnStatus::Running          => { /* show a spinner */ }
//!         _ => {}
//!     }
//!     // turn driver.transcript() into rows via transcript_lines(.., width)
//!     terminal.draw(|f| draw(transcript_lines(driver.transcript(), f.area().width as usize)));
//! }
//! ```
//!
//! ## How it spans the async boundary
//!
//! [`ChatCtx`](super::ChatCtx) is borrow-heavy — most of its fields are
//! references with the caller's lifetime, several are `&mut dyn` session
//! handles (note sink, recall source, permission gate, …). Those trait objects
//! aren't `Send`, so `chat_complete`'s future isn't `Send` either and cannot go
//! through `tokio::spawn`. So the driver owns a [`TurnDriverConfig`] of plain
//! owned values and, on [`submit`](TurnDriver::submit), launches a **dedicated
//! OS thread** running its own current-thread tokio runtime. That thread builds
//! a **headless** `ChatCtx` from a clone of the config plus a snapshot of the
//! transcript, `block_on`s one turn against [`NoMcp`] (raced against a cancel
//! signal), and sends the result back over a [`oneshot`] channel. The
//! session-bound service fields (`note_sink`, `recall_source`, `summarizer`, …)
//! stay absent. Compression accounting and measured token calibration belong
//! to the driver session, shared across its worker threads, so an observed
//! under-count still tightens the next submitted turn's budget.
//!
//! Running on its own thread also means the driver does **not** require the
//! consumer to call from inside a tokio runtime — a plain `crossterm` loop can
//! own a `TurnDriver` with no async scaffolding of its own.
//!
//! ## Minimal by design
//!
//! `submit` → `poll` → `cancel`, plus `submit_observation` to fold a
//! [`ShellObservation`] into the next turn's context. That is the whole
//! surface; richer wiring (live MCP tools, a session note store) stays with the
//! downstream consumer, which can build a full `ChatCtx` itself if it needs
//! one.

use std::thread::JoinHandle;

use tokio::sync::oneshot;

use std::sync::Arc;

use super::observation::ShellObservation;
use super::{
    chat_complete, working_memory_head, ChatCtx, CodeSearch, CrewRunner, Embedder, NoMcp,
    PromptDisposition, ScratchpadStore, SemanticIndex,
};
use crate::{BackendKind, CompactionTriggerPolicy, MemMessage, TokenUsage};

/// Owned, cloneable retrieval handle for the headless driver — the `'static`
/// counterpart of the borrow-heavy [`CodeSearch`] `ChatCtx` field (#1280). A
/// crew-leaf / cowork consumer that has an embedder + a built session index
/// hands them in here (shared `Arc`s, so cloning the config per turn is cheap);
/// [`run_one_turn`] borrows them into the turn's `CodeSearch`. Absent ⇒ the
/// `code_search` tool is simply not advertised, exactly like today.
#[derive(Clone)]
pub struct HeadlessCodeSearch {
    /// The query embedder (the same trait the live TUI loop uses).
    pub embedder: Arc<dyn Embedder>,
    /// The session semantic index the embedder searches.
    pub index: Arc<dyn SemanticIndex>,
    /// Default number of chunks returned per search.
    pub top_k: usize,
}

impl std::fmt::Debug for HeadlessCodeSearch {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // The trait objects are not `Debug`; surface only the scalar knob so the
        // derived `Debug` on `TurnDriverConfig` still compiles.
        f.debug_struct("HeadlessCodeSearch")
            .field("top_k", &self.top_k)
            .finish_non_exhaustive()
    }
}

/// Owned, `'static` configuration for one [`TurnDriver`] — the cloneable
/// counterpart of the borrow-heavy [`ChatCtx`]. The driver clones it into each
/// spawned turn task so nothing borrows across the async boundary.
///
/// The fields mirror the inference-relevant subset of [`ChatCtx`]; the
/// session-bound `&mut` service handles (note sink, recall, permission gate,
/// summarizer) are intentionally absent. Runtime compression state is owned
/// separately by the driver and shared across its submitted turns.
#[derive(Debug, Clone)]
pub struct TurnDriverConfig {
    /// Shell guidance captured once at session construction.
    pub shell_dialect: &'static str,
    /// Resolved once; callers may supply a deterministic budget.
    pub command_budget: crate::RunCommandBudget,
    /// Shared accounted session and independent narration adjudicator.
    pub smart_harness: Option<Arc<super::smart_harness::SmartHarness>>,
    /// Inference endpoint base URL.
    pub url: String,
    /// Model name.
    pub model: String,
    /// Wire protocol of the backend.
    pub kind: BackendKind,
    /// Bearer token for authenticated OpenAI-compatible endpoints.
    pub api_key: Option<String>,
    /// Chat Completions extensions explicitly accepted by the endpoint.
    pub chat_completions_capability: crate::model_card::ChatCompletionsCapability,
    /// Responses declarations captured from the same active capability decision.
    pub responses_capability: crate::model_card::ResponsesCapability,
    /// The typed wire choice captured with this context, never reread mid-turn.
    pub openai_api: crate::OpenAiApi,
    /// Whether the active backend accepts replayed assistant reasoning.
    pub reasoning_replay_scope: crate::model_card::ReasoningReplayScope,
    /// Whether the active model streams a lone leading `</think>` closer, so
    /// the stream filter must start inside the reasoning block (#528).
    ///
    /// Resolved from the backend's capability layers (inline declaration,
    /// and a named `card =` binding via `ResolvedCapabilities`) — never from
    /// the model name.
    pub emits_leading_reasoning: bool,
    /// Absolute workspace path the tool loop runs against.
    pub workspace: String,
    /// Permission caveats enforced for this turn's tool calls.
    pub caveats: crate::caveats::Caveats,
    /// Initial tool-call round allowance; recent concrete progress may renew it.
    /// Explicit inference budgets and cancellation remain binding.
    pub max_tool_rounds: usize,
    /// Max narrate-then-stop rescue nudges per turn (see
    /// `[tui] narration_nudge_cap`); cowork default 1 keeps the historical
    /// one-shot rescue.
    pub narration_nudge_cap: usize,
    /// Rounds added per renewal backed by fresh progress; `0` makes
    /// `max_tool_rounds` a hard cap.
    pub workflow_grace_rounds: usize,
    /// Legacy line limit for pre-execution tool previews.
    pub tool_output_lines: usize,
    /// Full model context window. Sent as Ollama `options.num_ctx`; on the
    /// OpenAI-compatible path it stays local and provides the authoritative
    /// input ceiling after percentage and generation-output reserves.
    pub num_ctx: Option<u32>,
    /// TCP connect timeout.
    pub connect_timeout_secs: u64,
    /// Total timeout for single-response requests and idle timeout for streams.
    pub inference_timeout_secs: u64,
    /// Message count at which the in-flight conversation is trimmed mid-turn.
    pub mid_loop_trim_threshold: usize,
    /// Whether a known authoritative input ceiling suppresses a count-only
    /// compaction. Headless callers default to the same safe policy as the
    /// interactive harness.
    pub compaction_trigger_policy: CompactionTriggerPolicy,
    /// Token threshold that also triggers a mid-loop trim. `None` disables it.
    pub mid_loop_trim_tokens: Option<usize>,
    /// Highest input-token count the model has accepted (pre-send budget gate).
    pub max_ok_input: Option<u32>,
    /// Shell command run after every successful write for ground-truth feedback.
    pub build_check_cmd: Option<String>,
    /// Empirically safe context size used to detect likely overflow.
    pub safe_context: Option<u32>,
    /// #727: percent of `num_ctx` used as the input-token ceiling (default 80).
    pub input_ceiling_pct: u32,
    /// #727: remaining-budget percent below which the loop nudges (default 15).
    pub low_budget_pct: usize,
    /// Cheap request-token estimator configured under `[context.estimation]`.
    pub estimation: crate::tokens::TokenEstimation,
    /// Minimum summarizer input cap configured under `[context]`.
    pub summary_input_cap_floor_chars: usize,
    /// Context-management preset configured under `[context] manager`. Headless
    /// callers get the same preset the interactive harness resolves, so
    /// `manager = "append-only"` in a config file governs `newt solve` too —
    /// otherwise the setting would silently mean nothing off the TUI.
    pub context_manager: crate::ContextManager,
    /// #1280: optional semantic retrieval for the headless turn. `Some` ⇒ the
    /// `code_search` tool is advertised and executable against the supplied
    /// index; `None` (the default) ⇒ absent, preserving today's behavior.
    pub code_search: Option<HeadlessCodeSearch>,
    /// #2312: explicit output-token allowance for driven turns. `None` keeps
    /// the cognition table and wire defaults.
    pub output_allowance: Option<u32>,
    /// `[[model_tuning]] overflow_retry` for driven turns.
    pub overflow_retry: crate::config::OverflowRetry,
    /// #2313: explicit run-level call-count allowance for driven turns.
    /// `None` keeps dispatch unbudgeted (today's default behavior).
    pub run_allowance: Option<u32>,
}

impl TurnDriverConfig {
    /// Construct a config with sensible cowork defaults; only the endpoint, the
    /// model, the backend kind, and the workspace are required. Caveats default
    /// to [`Caveats::top`](crate::caveats::Caveats::top) — the downstream
    /// consumer is expected to narrow them.
    pub fn new(
        url: impl Into<String>,
        model: impl Into<String>,
        kind: BackendKind,
        workspace: impl Into<String>,
    ) -> Self {
        Self {
            command_budget: crate::RunCommandBudget::from_env(),
            shell_dialect: super::run_command_dialect_sentence(),
            smart_harness: None,
            url: url.into(),
            model: model.into(),
            kind,
            api_key: None,
            chat_completions_capability: Default::default(),
            responses_capability: Default::default(),
            openai_api: if super::responses_api_selected() {
                crate::OpenAiApi::Responses
            } else {
                crate::OpenAiApi::ChatCompletions
            },
            reasoning_replay_scope: crate::model_card::ReasoningReplayScope::Never,
            emits_leading_reasoning: false,
            workspace: workspace.into(),
            caveats: crate::caveats::Caveats::top(),
            max_tool_rounds: 40,
            narration_nudge_cap: 1,
            workflow_grace_rounds: 5,
            tool_output_lines: 20,
            num_ctx: None,
            connect_timeout_secs: 5,
            inference_timeout_secs: 120,
            mid_loop_trim_threshold: 40,
            compaction_trigger_policy: CompactionTriggerPolicy::HeadroomAware,
            mid_loop_trim_tokens: None,
            max_ok_input: None,
            build_check_cmd: None,
            safe_context: None,
            input_ceiling_pct: 80,
            low_budget_pct: 15,
            estimation: crate::tokens::TokenEstimation::default(),
            summary_input_cap_floor_chars: 8_192,
            context_manager: crate::ContextManager::default(),
            code_search: None,
            output_allowance: None,
            overflow_retry: Default::default(),
            run_allowance: None,
        }
    }

    /// #1280: enable semantic `code_search` for driven turns. The consumer owns
    /// the embedder + index (built once, shared via `Arc`); the driver borrows
    /// them into each turn's `ChatCtx`. Mirrors the live TUI loop, which offers
    /// the tool only when semantic is on.
    #[must_use]
    pub fn with_code_search(
        mut self,
        embedder: Arc<dyn Embedder>,
        index: Arc<dyn SemanticIndex>,
        top_k: usize,
    ) -> Self {
        self.code_search = Some(HeadlessCodeSearch {
            embedder,
            index,
            top_k,
        });
        self
    }
}

/// Which optional collaborators a turn's constructed [`ChatCtx`] carried
/// (#2314) — read off the context the loop ran with, by the same predicates
/// the tool catalog uses to advertise them, never off the config or the flags
/// that asked for them.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct InstantiatedFeatures {
    /// The `state_*` tools: the scratchpad flag AND a store.
    pub scratchpad: bool,
    /// The `code_search` tool: a searcher.
    pub code_search: bool,
    /// The `crew` / `compose_roster` tools: a crew runner.
    pub crew: bool,
}

impl InstantiatedFeatures {
    fn of(ctx: &ChatCtx<'_>) -> Self {
        Self {
            scratchpad: ctx.scratchpad && ctx.scratchpad_store.is_some(),
            code_search: ctx.code_search.is_some(),
            crew: ctx.crew_runner.is_some(),
        }
    }
}

/// The outcome of one completed turn.
#[derive(Debug, Clone)]
pub struct TurnOutcome {
    /// Operator-facing reply, including the harness report. Use
    /// `model_reply_for_history` when projecting assistant-authored history.
    pub reply: String,
    /// Token usage for the turn, when the backend reported it.
    pub usage: Option<TokenUsage>,
    /// Count of hallucinated (non-existent) tool calls the loop saw.
    pub hallucinations: u32,
    /// Per-tool-call trace of the turn (name, args-digest, ok, duration) — the
    /// trajectory a headless consumer (`newt solve` / Terminal-Bench) serializes
    /// to reason about what the agent actually did. Empty when the loop produced
    /// no tool calls.
    pub tool_events: Vec<crate::ToolEvent>,
    /// Why the turn ended (genuine completion vs round-cap vs empty), when the
    /// loop set it.
    pub end_reason: Option<crate::TurnEndReason>,
    /// The inference/loop error, when the turn failed part-way. `Some` means the
    /// turn did NOT complete cleanly — but `reply`/`usage` may be empty while
    /// `tool_events` still holds whatever the agent did *before* the error (an
    /// infrastructure failure must not erase the partial trajectory).
    pub error: Option<String>,
    /// Structural class of `error` (W0 #1511), read from the TYPED dispatch
    /// error chain — never from the message text. `Some(Harness)` when an
    /// error carried no dispatch classification at all (fail-closed: an
    /// unattributed failure is ours, not the model's). `None` iff `error` is.
    pub error_class: Option<crate::agentic::observability::ErrorClass>,
    /// The `model` field the backend reported in its response body, when it
    /// did — what was ACTUALLY served (the solve contract's `effective_model`
    /// source), as opposed to what we asked for.
    pub served_model: Option<String>,
    /// Per-round tool-call parse signals (ADR #1506 §5): recovered dialects
    /// and content-with-no-parseable-call rounds, in round order.
    pub parse_signals: Vec<crate::agentic::observability::ParseSignal>,
    /// Output-behavior signals, including bounded reasoning-overflow recovery.
    pub behavior_signals: Vec<crate::agentic::observability::BehaviorSignal>,
    /// The optional collaborators this turn's context actually carried.
    pub features: InstantiatedFeatures,
    /// Whether the loop already printed `reply` as it arrived. A host that
    /// shows answers must print an unstreamed reply itself (#2372).
    pub was_streamed: bool,
    /// `reply` is harness-written text, not the model's claim (#2372).
    pub harness_reply: bool,
    /// Semantic intent captured by this turn, independent of supported wire fields.
    pub semantic_cognition: Option<crate::role_profile::Cognition>,
    /// Captured admitted Responses declaration and actual wire effort.
    pub responses_capability: Option<crate::model_card::ResponsesCapability>,
    pub reasoning_effort: Option<crate::model_card::ReasoningEffort>,
    /// The output cap the turn's wire applied, and who enforced it (#2312).
    pub output_allowance: Option<crate::agentic::observability::OutputAllowance>,
    /// What this turn sent and generated, summed per inference attempt, with
    /// how many attempts reported no usage (#2313). `usage` above stays the
    /// context merge (largest prompt, generated tokens).
    pub attempts: Option<crate::attempts::UsageTotals>,
    /// The attempt ledger's chain lines, in observation order, for a trace
    /// (`newt solve --events`). The last line's id is the chain head. A boxed
    /// slice, not a `Vec`: the lines are final, and `TurnStatus` stays small.
    pub attempt_lines: Box<[crate::event_journal::JournalLine<crate::attempts::AttemptRecord>]>,
}

/// Non-blocking snapshot of the driver's state, returned by
/// [`TurnDriver::poll`].
#[derive(Debug)]
pub enum TurnStatus {
    /// No turn in flight (nothing submitted, or the last result was drained).
    Idle,
    /// A turn is running; the result is not ready yet.
    Running,
    /// A turn finished successfully. Returned **once**; the reply has already
    /// been appended to the transcript as an assistant message.
    Completed(Box<TurnOutcome>),
    /// A turn failed (transport error, dispatch error, …). Returned **once**.
    Failed(String),
}

/// Internal in-flight turn handle: the worker thread, the channel it reports
/// on, and the cancel signal that races its `block_on`.
struct InFlight {
    handle: JoinHandle<()>,
    rx: oneshot::Receiver<Result<TurnOutcome, String>>,
    cancel_tx: Option<oneshot::Sender<()>>,
}

/// Runtime choices captured for one driver independently of its source-stable
/// inference configuration. Each submitted turn clones this posture onto the
/// dedicated worker thread.
#[derive(Clone)]
struct HeadlessRuntimePosture {
    worktree_session: Arc<crate::worktree_adoption::WorktreeSession>,
    cognition: Option<crate::role_profile::Cognition>,
    tenacity: crate::tenacity::Tenacity,
    initiative: crate::initiative::Initiative,
    crew_runner: Option<Arc<dyn CrewRunner>>,
    scratchpad: Option<Arc<dyn ScratchpadStore>>,
    compress_state: Arc<tokio::sync::Mutex<super::CompressState>>,
}

impl HeadlessRuntimePosture {
    fn capture() -> Self {
        Self {
            worktree_session: Arc::default(),
            cognition: crate::cognition::effective_cognition(),
            tenacity: crate::tenacity::effective_tenacity(),
            initiative: crate::initiative::effective_initiative(),
            crew_runner: None,
            scratchpad: None,
            compress_state: Arc::new(tokio::sync::Mutex::new(super::CompressState::new())),
        }
    }
}

/// A pumpable, non-blocking driver for one agentic turn at a time.
///
/// Owns the running transcript and at most one in-flight turn. The external
/// event loop [`submit`](Self::submit)s input (or
/// [`submit_observation`](Self::submit_observation)s shell activity),
/// [`poll`](Self::poll)s for completion every frame, and may
/// [`cancel`](Self::cancel) the in-flight turn.
pub struct TurnDriver {
    config: TurnDriverConfig,
    runtime: HeadlessRuntimePosture,
    transcript: Vec<MemMessage>,
    in_flight: Option<InFlight>,
}

impl TurnDriver {
    /// Build a driver with an empty transcript.
    pub fn new(config: TurnDriverConfig) -> Self {
        Self {
            config,
            runtime: HeadlessRuntimePosture::capture(),
            transcript: Vec::new(),
            in_flight: None,
        }
    }

    /// Build a driver seeded with an existing transcript (e.g. a system prompt
    /// plus prior turns the consumer already assembled).
    pub fn with_transcript(config: TurnDriverConfig, transcript: Vec<MemMessage>) -> Self {
        Self {
            config,
            runtime: HeadlessRuntimePosture::capture(),
            transcript,
            in_flight: None,
        }
    }

    /// Explicit operator action; never exposed as a model tool.
    pub fn lift_worktree_restriction(&self) {
        self.runtime.worktree_session.lift();
    }

    /// The current transcript — every message the model has seen plus the
    /// replies it produced. Feed it to
    /// [`transcript_lines`](super::transcript_lines) for renderer-agnostic,
    /// width-wrapped rows, or render it however the consumer likes.
    pub fn transcript(&self) -> &[MemMessage] {
        &self.transcript
    }

    /// Whether a turn is currently in flight.
    pub fn is_running(&self) -> bool {
        self.in_flight.is_some()
    }

    /// Pin the semantic cognition used by every turn this driver submits.
    /// The default is captured from the process posture when the driver is
    /// constructed, before its dedicated turn thread is spawned.
    #[must_use]
    pub fn with_cognition(mut self, cognition: Option<crate::role_profile::Cognition>) -> Self {
        self.runtime.cognition = cognition;
        self
    }

    /// Pin the tenacity used by every resolver read in this driver's turns.
    /// The default is captured from the process posture at construction.
    #[must_use]
    pub fn with_tenacity(mut self, tenacity: crate::tenacity::Tenacity) -> Self {
        self.runtime.tenacity = tenacity;
        self
    }

    /// Pin the initiative (the read-only nudge and the plan-exit edit) used by
    /// every resolver read in this driver's turns. The default is captured
    /// from the process posture at construction.
    #[must_use]
    pub fn with_initiative(mut self, initiative: crate::initiative::Initiative) -> Self {
        self.runtime.initiative = initiative;
        self
    }

    /// Enable the real crew/team surface for driven turns. The runner is owned
    /// through an `Arc` because every submitted turn executes on a dedicated
    /// OS thread.
    #[must_use]
    pub fn with_crew_runner(mut self, runner: Arc<dyn CrewRunner>) -> Self {
        self.runtime.crew_runner = Some(runner);
        self
    }

    /// Opt driven turns into the scratchpad (#2314): the `state_*` tools are
    /// advertised and executed against `store`, and its `<state>` block leads
    /// each turn's message[0] exactly as the TUI's does. The store is the only
    /// state source — seed it before handing it in.
    #[must_use]
    pub fn with_scratchpad(mut self, store: Arc<dyn ScratchpadStore>) -> Self {
        self.runtime.scratchpad = Some(store);
        self
    }

    /// Whether driven turns will advertise and execute crew tools.
    #[must_use]
    pub fn has_crew_runner(&self) -> bool {
        self.runtime.crew_runner.is_some()
    }

    /// Append a [`ShellObservation`] to the transcript so it becomes part of the
    /// **next** turn's context. The observation is redacted and framed by
    /// [`ShellObservation::into_mem_message`] — credentials in shell output are
    /// scrubbed before they enter the transcript. This does **not** start a
    /// turn on its own; a real human message ([`submit`](Self::submit)) drives
    /// the loop, with the accumulated observations already in context.
    pub fn submit_observation(&mut self, obs: ShellObservation) {
        self.transcript.push(obs.into_mem_message());
    }

    /// Submit a human message and start a turn.
    ///
    /// Appends the message as a `User` turn, snapshots the transcript, and
    /// spawns a task that runs one [`chat_complete`] against the configured
    /// backend. Returns an error (without starting anything) if a turn is
    /// already in flight — the driver runs one turn at a time. Poll
    /// [`poll`](Self::poll) for the result.
    pub fn submit(&mut self, input: impl Into<String>) -> Result<(), TurnDriverError> {
        if self.in_flight.is_some() {
            return Err(TurnDriverError::Busy);
        }
        let task = input.into();
        self.transcript.push(MemMessage::user(task.clone()));
        self.spawn_turn(task);
        Ok(())
    }

    /// Launch the turn on a dedicated thread over a clone of the config + a
    /// snapshot of the transcript. The `task` string is the current user
    /// message, threaded into `ChatCtx.task` (the loop's nudges key on it).
    ///
    /// `chat_complete`'s future is `!Send` (its `ChatCtx` holds `&mut dyn`
    /// session handles), so it can't ride `tokio::spawn`. A current-thread
    /// runtime on its own OS thread keeps the non-`Send` future local; a
    /// cancel oneshot races the turn so [`cancel`](Self::cancel) returns the
    /// thread promptly instead of waiting out the inference timeout.
    fn spawn_turn(&mut self, task: String) {
        let config = self.config.clone();
        let runtime = self.runtime.clone();
        let messages = self.transcript.clone();
        let (tx, rx) = oneshot::channel();
        let (cancel_tx, cancel_rx) = oneshot::channel();
        let handle = std::thread::spawn(move || {
            // A current-thread runtime: no work-stealing, nothing to require
            // `Send`, and self-contained so the consumer needn't be in a
            // runtime itself.
            let rt = match tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
            {
                Ok(rt) => rt,
                Err(e) => {
                    let _ = tx.send(Err(format!("failed to start turn runtime: {e}")));
                    return;
                }
            };
            let result = rt.block_on(async move {
                tokio::select! {
                    biased;
                    // Cancellation wins the race when signalled.
                    _ = cancel_rx => Err("turn cancelled".to_string()),
                    out = run_one_turn(&config, &runtime, &messages, &task) => out,
                }
            });
            // The receiver may have been dropped (driver went away); fine.
            let _ = tx.send(result);
        });
        self.in_flight = Some(InFlight {
            handle,
            rx,
            cancel_tx: Some(cancel_tx),
        });
    }

    /// Non-blocking poll for the in-flight turn's status.
    ///
    /// - [`TurnStatus::Idle`] — nothing in flight.
    /// - [`TurnStatus::Running`] — in flight, not done.
    /// - [`TurnStatus::Completed`] — done; the reply has been appended to the
    ///   transcript as an assistant message. Returned exactly once.
    /// - [`TurnStatus::Failed`] — errored; nothing appended. Returned once.
    pub fn poll(&mut self) -> TurnStatus {
        let Some(in_flight) = self.in_flight.as_mut() else {
            return TurnStatus::Idle;
        };
        match in_flight.rx.try_recv() {
            Ok(result) => self.settle(Some(result)),
            Err(oneshot::error::TryRecvError::Empty) => TurnStatus::Running,
            Err(oneshot::error::TryRecvError::Closed) => self.settle(None),
        }
    }

    /// Await the in-flight turn's completion, then settle it exactly as
    /// [`poll`](Self::poll) would. For a consumer with nothing to draw between
    /// frames (headless runs, tests) this replaces a `poll` + sleep loop: the
    /// worker's result wakes this future, so no interval and no deadline is
    /// involved. [`TurnStatus::Idle`] when nothing is in flight; never
    /// [`TurnStatus::Running`].
    pub async fn wait(&mut self) -> TurnStatus {
        let Some(in_flight) = self.in_flight.as_mut() else {
            return TurnStatus::Idle;
        };
        let result = (&mut in_flight.rx).await.ok();
        self.settle(result)
    }

    /// Record the worker's final word. `None` means the worker ended without
    /// sending (it was aborted, or panicked).
    fn settle(&mut self, result: Option<Result<TurnOutcome, String>>) -> TurnStatus {
        self.in_flight = None;
        match result {
            Some(Ok(outcome)) => {
                self.transcript
                    .push(MemMessage::assistant(outcome.reply.clone()));
                TurnStatus::Completed(Box::new(outcome))
            }
            Some(Err(err)) => TurnStatus::Failed(err),
            None => TurnStatus::Failed("turn task ended without a result".to_string()),
        }
    }

    /// Cancel the in-flight turn, if any. Signals the worker thread to abandon
    /// the turn (winning the `select!` race against `chat_complete`) and joins
    /// it so the runtime tears down before returning. The user message that
    /// started the turn stays in the transcript — the consumer can pop it for a
    /// clean retry. No-op when idle.
    ///
    /// A turn already blocked deep inside a single HTTP dispatch returns when
    /// that dispatch resolves (or times out); the cancel signal is honored at
    /// the next `select!` poll point, which in practice is between dispatches.
    pub fn cancel(&mut self) {
        if let Some(mut in_flight) = self.in_flight.take() {
            if let Some(cancel_tx) = in_flight.cancel_tx.take() {
                let _ = cancel_tx.send(());
            }
            let _ = in_flight.handle.join();
        }
    }
}

/// Run exactly one turn: assemble a headless [`ChatCtx`] from owned config +
/// messages and dispatch on the right wire protocol. Factored out of the spawn
/// closure so it is directly unit-testable against a wiremock backend.
async fn run_one_turn(
    config: &TurnDriverConfig,
    runtime: &HeadlessRuntimePosture,
    messages: &[MemMessage],
    task: &str,
) -> Result<TurnOutcome, String> {
    // Freeze every lazy cognition, tenacity, and initiative read
    // beneath `chat_complete` (workflow state and exit_plan_mode included) to
    // the posture captured by this driver. The current-thread runtime keeps
    // these RAII TLS guards on the dedicated turn thread for the entire future.
    let _cognition = crate::cognition::scoped_effective_cognition(runtime.cognition);
    let _tenacity = crate::tenacity::scoped_effective_tenacity(runtime.tenacity);
    let _initiative = crate::initiative::scoped_effective_initiative(runtime.initiative);
    // Capture the per-tool trajectory + end reason for the outcome. The ChatCtx
    // borrows these mutably; the borrows release when `ctx` is consumed by the
    // await below, after which they move into the TurnOutcome.
    let mut tool_events: Vec<crate::ToolEvent> = Vec::new();
    let mut end_reason: Option<crate::TurnEndReason> = None;
    // W0 (#1511): served-model + parse-signal observation for the solve
    // contract — same lend-per-turn pattern as `tool_events`.
    let mut solve_obs = crate::agentic::observability::SolveObservation::default();
    // Session disclosure filter: register the live provider secret so a tool
    // result or summary that echoes it is value-redacted before it reaches the
    // model (`disclosure-gate-live-path`). Must outlive `ctx`, which borrows it.
    let session_disclosure = crate::ocap::session_disclosure_filter(config.api_key.as_deref());
    // Also install it on this turn's thread so the memory / observation /
    // compaction / spill paths (which funnel through `redact_secrets`, not the
    // explicit `disclosure` param) value-filter against the same secret.
    let _disclosure_guard = crate::ocap::scoped_session_disclosure(session_disclosure.clone());
    // A driver permits one active turn. Holding the shared state through this
    // future keeps learned usage on success, errors, and cancellation alike;
    // dropping the future releases the guard before the next worker starts.
    let mut compress_state = runtime.compress_state.lock().await;
    // The working-memory head is ephemeral like the TUI's: it leads this turn's
    // message[0] and never enters the driver transcript.
    let messages: std::borrow::Cow<'_, [MemMessage]> =
        match working_memory_head(runtime.scratchpad.as_deref(), false) {
            None => messages.into(),
            Some(head) => {
                let mut turn = messages.to_vec();
                match turn.first_mut() {
                    Some(first) if first.role == crate::Role::System => {
                        first.content = format!("{head}\n\n{}", first.content);
                    }
                    _ => turn.insert(0, MemMessage::system(head)),
                }
                turn.into()
            }
        };
    // #2313: one attempt ledger per turn; its totals ride the outcome.
    let attempt_ledger = std::sync::Mutex::new(crate::attempts::AttemptLedger::default());
    // #2313: one run allowance per turn, built from the configured budget.
    // `None` (today's default for every existing caller) is unbudgeted,
    // bit-for-bit unchanged dispatch.
    let run_allowance = config
        .run_allowance
        .map(super::run_allowance::RunAllowance::new);
    let ctx = ChatCtx {
        shell_dialect: config.shell_dialect,
        command_budget: config.command_budget,
        run_allowance: run_allowance.as_ref(),
        verify_outcomes: crate::agentic::self_verify::outcomes_enabled(),
        round_cap_hit: None,
        smart_harness: config.smart_harness.as_deref(),
        rewrites_history: config.context_manager.rewrites_history(),
        url: &config.url,
        model: &config.model,
        kind: config.kind,
        api_key: config.api_key.as_deref(),
        // Headless driver: action nudges stay on (the interactive /nudge dial
        // is a TUI-session concern, #1162).
        action_nudges: true,
        // Headless callers preserve the historical execute-capable contract
        // unless they opt into the prompt-comprehension ingress in the TUI.
        prompt_disposition: PromptDisposition::Act,
        prompt_intake: None,
        messages: &messages,
        task,
        workspace: &config.workspace,
        default_command_cwd: None,
        worktree_session: Some(&runtime.worktree_session),
        // The driver is headless: the loop's inline progress prints are
        // suppressed so nothing fights the consumer's ratatui frame.
        color: false,
        // Cowork renders into the consumer's frame, not the scroller — no
        // markdown ANSI here (it would fight the host UI).
        markdown: false,
        // Headless: no tool-offload (26.3); scratchpad (26.4) only by opt-in.
        tool_offload: false,
        spill_store: None,
        disclosure: Some(&session_disclosure),
        completed_spill_renderer: None,
        compaction_store: None,
        scratchpad: runtime.scratchpad.is_some(),
        scratchpad_store: runtime.scratchpad.as_deref(),
        // #1280: borrow the consumer-supplied embedder + index into this turn's
        // searcher. `Some` only when the config carries retrieval — otherwise the
        // `code_search` tool is not advertised (unchanged headless behavior).
        code_search: config.code_search.as_ref().map(|cs| CodeSearch {
            embedder: cs.embedder.as_ref(),
            index: cs.index.as_ref(),
            top_k: cs.top_k,
            steer: None,
            status: None,
        }),
        // #1285: headless driver builds no symbol index yet (a follow-up wires
        // it from the leaf's gather) — the where_is tool degrades honestly.
        where_is: None,
        nav: None,
        // Headless driver: identity exposure (advertise the full authorized
        // catalog, unchanged). The `[tool_exposure]` controller is a TUI-session
        // concern; the driver keeps bit-for-bit behaviour.
        exposure: crate::agentic::tools::ExposureSettings::default(),
        experience_store: None,
        step_ledger: None,
        caveats: &config.caveats,
        // Headless cowork carries no persona surface (FR-1 part 2, #997). The
        // driver captured the effective cognition before spawning this thread,
        // preserving semantic intent independently of supported wire controls.
        persona_tools: None,
        cognition: super::projected_cognition(
            runtime.cognition,
            config.kind,
            config.openai_api,
            config.chat_completions_capability,
        ),
        chat_completions_capability: config.chat_completions_capability,
        responses_capability: config.responses_capability.clone(),
        openai_api: config.openai_api,
        output_allowance: config.output_allowance,
        overflow_retry: config.overflow_retry,
        attempt_ledger: Some(&attempt_ledger),
        reasoning_replay_scope: config.reasoning_replay_scope,
        emits_leading_reasoning: config.emits_leading_reasoning,
        max_tool_rounds: config.max_tool_rounds,
        narration_nudge_cap: config.narration_nudge_cap,
        workflow_grace_rounds: config.workflow_grace_rounds,
        tool_output_lines: config.tool_output_lines,
        debug: false,
        trace: false,
        num_ctx: config.num_ctx,
        // #727 tunables: headless driver keeps the historical defaults
        // (80% input ceiling, 15% low-budget nudge threshold).
        input_ceiling_pct: config.input_ceiling_pct,
        low_budget_pct: config.low_budget_pct,
        connect_timeout_secs: config.connect_timeout_secs,
        inference_timeout_secs: config.inference_timeout_secs,
        mid_loop_trim_threshold: config.mid_loop_trim_threshold,
        compaction_trigger_policy: config.compaction_trigger_policy,
        mid_loop_trim_tokens: config.mid_loop_trim_tokens,
        max_ok_input: config.max_ok_input,
        build_check_cmd: config.build_check_cmd.clone(),
        safe_context: config.safe_context,
        // Numbered context recovery is stateless and shared with the TUI.
        recover_cw_400: Some(super::recover_context_window_400),
        note_sink: None,
        note_nudge: None,
        recall_source: None,
        memory_source: None,
        summarizer: None,
        compress_state: Some(&mut compress_state),
        tool_events: Some(&mut tool_events),
        phantom_reaches: None,
        end_reason: Some(&mut end_reason),
        solve_obs: Some(&mut solve_obs),
        permission_gate: None,
        // Headless surfaces do not use the TUI capability cache. Same-request
        // usage still calibrates this driver's session through compress_state.
        on_round_usage: None,
        estimate_ratio: None,
        estimation: config.estimation,
        summary_input_cap_floor_chars: config.summary_input_cap_floor_chars,
        // #307: the headless driver carries no preset clamp — exec authority is
        // whatever `config.caveats` already grants, exactly like the ACP worker
        // / newt-eval callers. A consumer enforcing a mode clamps `caveats` itself.
        exec_floor: None,
        write_ledger: None,
        attribution: None,
        // The headless driver has no interactive keyboard to interrupt from;
        // a consumer that wants cancellation drives its own ChatCtx.
        cancel: None,
        live_tool_output: None,
        git_tool: None,
        crew_runner: runtime.crew_runner.as_deref(),
        operating_mode_control: None,
        plan_mode_control: None,
        plan_draft_sink: None,
        steering: None,
    };
    let features = InstantiatedFeatures::of(&ctx);
    // NoMcp: the cowork driver advertises only the built-in tools. A consumer
    // that wants live MCP tools assembles its own ChatCtx.
    let mut mcp = NoMcp;
    // ONE dispatch owner. `chat_complete` routes openai-vs-ollama AND, for an
    // openai backend, responses-vs-chat from the captured typed API. The headless driver goes through the SAME entry the
    // interactive TUI does, so the wire-format decision lives in exactly one
    // place and cannot drift. (This path used to fork its own `if kind==Openai`
    // branch that called `openai_chat_complete` directly, bypassing the
    // responses check — which is exactly why a responses-only model like
    // gpt-5.6-sol was mis-routed to /v1/chat/completions.)
    let dispatch = chat_complete(ctx, &mut mcp).await;
    let (attempts, attempt_lines) = {
        let ledger = attempt_ledger
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        (Some(ledger.totals()), ledger.lines().into())
    };
    // Both arms move `tool_events`/`end_reason` out (only one arm runs). On a
    // failed turn we still return `Ok` carrying the PARTIAL trajectory + the
    // error, rather than `Err` that discards what the agent already did — an
    // infrastructure failure (e.g. a context-window 500) must not misreport the
    // agent as having done nothing.
    match dispatch {
        Ok((reply, was_streamed, usage, hallucinations)) => Ok(TurnOutcome {
            reply,
            was_streamed,
            usage,
            hallucinations,
            tool_events,
            end_reason,
            error: None,
            error_class: None,
            served_model: solve_obs.served_model,
            parse_signals: solve_obs.parse_signals,
            behavior_signals: solve_obs.behavior_signals,
            features,
            semantic_cognition: runtime.cognition,
            responses_capability: solve_obs.responses_capability,
            reasoning_effort: solve_obs.reasoning_effort,
            output_allowance: solve_obs.output_allowance,
            harness_reply: solve_obs.harness_reply,
            attempts,
            attempt_lines,
        }),
        Err(e) => {
            // #2313 / U4a: an exhausted `--run-allowance` is a budget WALL, not a
            // failure. Type it like the round cap (a clean stop, `RunAllowance`),
            // and let the HARNESS write the notice: a model summary is a call the
            // allowance now refuses, and a summary before the round cap would break
            // BHV-ROUND-002.
            let exhausted = e.chain().any(|c| {
                c.downcast_ref::<super::run_allowance::RunAllowanceExhausted>()
                    .is_some()
            });
            Ok(TurnOutcome {
                reply: if exhausted {
                    RUN_ALLOWANCE_NOTICE.to_string()
                } else {
                    String::new()
                },
                was_streamed: false,
                usage: None,
                hallucinations: 0,
                tool_events,
                // F14: a loop that errored without typing its end (e.g. an
                // exhausted context-overflow recovery) ended in a backend/loop
                // error — `Failed`, never an untyped `None`. A reason the loop
                // did write (e.g. `Cancelled`) wins.
                end_reason: if exhausted {
                    Some(crate::TurnEndReason::RunAllowance)
                } else {
                    end_reason.or(Some(crate::TurnEndReason::Failed))
                },
                // W0 (#1511): the class is read from the TYPED chain; an error
                // with no dispatch classification is OURS — harness_error,
                // fail-closed, never a guess from the message text.
                error_class: (!exhausted).then(|| {
                    crate::agentic::observability::error_class(&e)
                        .unwrap_or(crate::agentic::observability::ErrorClass::Harness)
                }),
                error: (!exhausted).then(|| e.to_string()),
                served_model: solve_obs.served_model,
                parse_signals: solve_obs.parse_signals,
                behavior_signals: solve_obs.behavior_signals,
                features,
                semantic_cognition: runtime.cognition,
                responses_capability: solve_obs.responses_capability,
                reasoning_effort: solve_obs.reasoning_effort,
                output_allowance: solve_obs.output_allowance,
                harness_reply: solve_obs.harness_reply || exhausted,
                attempts,
                attempt_lines,
            })
        }
    }
}

/// The harness-written reply when `--run-allowance` runs out (no model call is
/// possible). Points the dispatcher at the state the run left behind.
const RUN_ALLOWANCE_NOTICE: &str = "The run allowance is exhausted: the whole-run inference-call \
budget was spent before the task finished. Work done so far is left in the workspace; the \
hand-back names what changed. Re-run with a larger --run-allowance to continue.";

/// Errors from driving a turn.
#[derive(Debug, thiserror::Error)]
pub enum TurnDriverError {
    /// A turn is already in flight; the driver runs one at a time.
    #[error("a turn is already in flight — poll() for it or cancel() before submitting another")]
    Busy,
}

#[cfg(test)]
#[path = "driver_calibration_tests.rs"]
mod calibration_tests;

#[cfg(test)]
mod tests {
    include!("driver_cognition_tests.rs");
    include!("driver_cognition_wire_tests.rs");
    use super::*;
    use crate::agentic::SessionSemanticIndex;
    use crate::caveats::{Caveats, CountBound, Scope};
    use crate::Role;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;
    use std::time::Duration;
    use wiremock::matchers::{method, path};
    use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};

    /// Ollama-shaped responder: a plain text answer (no tools), echoing back a
    /// fixed reply. Counts how many chat requests it served.
    struct PlainOllama {
        served: Arc<AtomicUsize>,
        reply: String,
    }

    impl Respond for PlainOllama {
        fn respond(&self, _req: &Request) -> ResponseTemplate {
            self.served.fetch_add(1, Ordering::SeqCst);
            ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "message": { "content": self.reply }
            }))
        }
    }

    fn cfg(url: &str) -> TurnDriverConfig {
        let mut c = TurnDriverConfig::new(url, "test-model", BackendKind::Ollama, ".");
        c.caveats = Caveats::top();
        c
    }

    #[test]
    fn constructors_capture_the_current_runtime_posture() {
        use crate::cognition::{set_cli_cognition, CognitionOverride};
        use crate::initiative::{set_cli_initiative, Initiative};
        use crate::role_profile::Cognition;
        use crate::tenacity::{set_cli_tenacity, Tenacity};
        use crate::test_guard::GlobalSettingsGuard;

        let _settings = GlobalSettingsGuard::acquire();
        set_cli_cognition(CognitionOverride::Set(Cognition::Rational));
        set_cli_tenacity(Tenacity::Relentless);
        set_cli_initiative(Initiative::Decisive);
        let empty = TurnDriver::new(cfg("http://placeholder"));

        set_cli_cognition(CognitionOverride::Off);
        set_cli_tenacity(Tenacity::Normal);
        set_cli_initiative(Initiative::Patient);
        let seeded = TurnDriver::with_transcript(
            cfg("http://placeholder"),
            vec![MemMessage::user("already started")],
        );

        assert_eq!(empty.runtime.cognition, Some(Cognition::Rational));
        assert_eq!(empty.runtime.tenacity, Tenacity::Relentless);
        assert_eq!(empty.runtime.initiative, Initiative::Decisive);
        assert_eq!(seeded.runtime.cognition, None);
        assert_eq!(seeded.runtime.tenacity, Tenacity::Normal);
        assert_eq!(seeded.runtime.initiative, Initiative::Patient);
    }

    /// Pump the driver to completion: one non-blocking `poll()` (the frame a
    /// crossterm loop would draw), then await the worker's result itself.
    ///
    /// No clock decides this. It used to be 600 polls at 10 ms, then a 30 s
    /// wall-clock deadline, and each failed by machine speed rather than by
    /// behaviour: the deadline panicked here under a 32-thread nextest run
    /// beside a build, where a turn that takes a second at idle took over a
    /// minute. The worker's oneshot is the event; the only time left is the
    /// shared hang guard, which a completing turn cannot reach.
    async fn pump_to_done(driver: &mut TurnDriver) -> TurnStatus {
        match driver.poll() {
            TurnStatus::Running => {}
            settled => return settled,
        }
        crate::test_guard::hang_guarded("the driven turn", driver.wait()).await
    }

    /// `poll` settles whatever the worker's oneshot carries, exactly once, and
    /// with no worker thread in the picture: the channel is the whole seam.
    /// A dropped sender (the worker died without a word) is a failure too.
    #[tokio::test]
    async fn poll_settles_the_oneshot_once() {
        let mut driver = TurnDriver::new(cfg("http://placeholder"));
        let (tx, rx) = oneshot::channel();
        driver.in_flight = Some(InFlight {
            handle: std::thread::spawn(|| {}),
            rx,
            cancel_tx: None,
        });
        assert!(matches!(driver.poll(), TurnStatus::Running));
        tx.send(Err("boom".to_string())).unwrap();
        assert!(matches!(driver.poll(), TurnStatus::Failed(e) if e == "boom"));
        assert!(matches!(driver.poll(), TurnStatus::Idle));

        let (tx, rx) = oneshot::channel::<Result<TurnOutcome, String>>();
        driver.in_flight = Some(InFlight {
            handle: std::thread::spawn(|| {}),
            rx,
            cancel_tx: None,
        });
        drop(tx);
        assert!(matches!(
            driver.wait().await,
            TurnStatus::Failed(e) if e == "turn task ended without a result"
        ));
        assert!(matches!(driver.wait().await, TurnStatus::Idle));
    }

    #[tokio::test]
    async fn driver_accounts_for_a_question_without_delivering_a_completion() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(PlainOllama {
                served: Arc::new(AtomicUsize::new(0)),
                reply: "Which repository?".into(),
            })
            .mount(&server)
            .await;
        let harness = Arc::new(
            super::super::smart_harness::SmartHarness::new(
                agent_harness::Session::new(crate::test_guard::unbudgeted_session_config())
                    .unwrap(),
                Arc::new(|_| Box::pin(async { Ok(("\"question\"".to_string(), None)) })),
                Default::default(),
            )
            .unwrap(),
        );
        let before = harness.head().unwrap();
        let mut config = cfg(&server.uri());
        config.smart_harness = Some(harness.clone());
        let mut driver = TurnDriver::new(config);
        driver.submit("Update the repository").unwrap();
        let TurnStatus::Completed(outcome) = pump_to_done(&mut driver).await else {
            panic!("the turn must finish by awaiting the operator");
        };
        assert_eq!(
            outcome.end_reason,
            Some(crate::TurnEndReason::AwaitingOperator)
        );
        assert_eq!(
            crate::agentic::model_explanation(&outcome.reply),
            "Which repository?"
        );
        assert_ne!(before, harness.head().unwrap());
    }

    /// THE acceptance test: a consumer drives one turn through the driver and
    /// gets the answer back — no `run_chat`, no blocking REPL, only the
    /// submit → poll surface.
    #[tokio::test]
    async fn driver_pumps_one_turn_and_yields_the_answer() {
        let server = MockServer::start().await;
        let served = Arc::new(AtomicUsize::new(0));
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(PlainOllama {
                served: served.clone(),
                reply: "the driver answered".into(),
            })
            .mount(&server)
            .await;

        let mut driver = TurnDriver::new(cfg(&server.uri()));
        // Idle before anything is submitted.
        assert!(matches!(driver.poll(), TurnStatus::Idle));
        assert!(!driver.is_running());

        driver
            .submit("what is 2 + 2?")
            .expect("submit starts a turn");
        assert!(driver.is_running());

        let status = pump_to_done(&mut driver).await;
        match status {
            TurnStatus::Completed(outcome) => {
                assert_eq!(
                    crate::agentic::model_explanation(&outcome.reply),
                    "the driver answered"
                );
            }
            other => panic!("expected Completed, got {other:?}"),
        }

        // The transcript now holds the user turn AND the assistant reply, with
        // no run_chat involved.
        let t = driver.transcript();
        assert_eq!(t.len(), 2);
        assert_eq!(t[0].role, Role::User);
        assert_eq!(t[0].content, "what is 2 + 2?");
        assert_eq!(t[1].role, Role::Assistant);
        assert_eq!(
            crate::agentic::model_explanation(&t[1].content),
            "the driver answered"
        );

        // The backend served the turn's generation; at least one request landed.
        assert!(served.load(Ordering::SeqCst) >= 1);
        // Drained — back to idle.
        assert!(matches!(driver.poll(), TurnStatus::Idle));
        assert!(!driver.is_running());
    }

    /// #1280: a stub embedder so the driver can carry a `code_search` handle.
    /// Advertisement is gated on `code_search.is_some()`, so `embed` need not be
    /// called for this test — a fixed vector keeps it total if it ever is.
    struct StubEmbedder;
    #[async_trait::async_trait]
    impl Embedder for StubEmbedder {
        async fn embed(&self, _text: &str) -> anyhow::Result<Vec<f32>> {
            Ok(vec![0.0_f32; 4])
        }
    }

    /// Ollama responder that records every chat request body, so a test can
    /// assert what tools the driver advertised.
    struct CapturingOllama {
        bodies: Arc<std::sync::Mutex<Vec<serde_json::Value>>>,
        reply: String,
    }
    impl Respond for CapturingOllama {
        fn respond(&self, req: &Request) -> ResponseTemplate {
            if let Ok(v) = serde_json::from_slice::<serde_json::Value>(&req.body) {
                self.bodies.lock().unwrap().push(v);
            }
            ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "message": { "content": self.reply }
            }))
        }
    }

    /// True if any recorded request advertised a tool named `code_search`.
    fn advertised_code_search(bodies: &[serde_json::Value]) -> bool {
        bodies.iter().any(|b| {
            b["tools"]
                .as_array()
                .is_some_and(|ts| ts.iter().any(|t| t["function"]["name"] == "code_search"))
        })
    }

    fn advertised_tool(bodies: &[serde_json::Value], name: &str) -> bool {
        bodies.iter().any(|body| {
            body["tools"].as_array().is_some_and(|tools| {
                tools
                    .iter()
                    .any(|tool| tool["function"]["name"].as_str() == Some(name))
            })
        })
    }

    struct StubCrewRunner;

    #[async_trait::async_trait]
    impl CrewRunner for StubCrewRunner {
        async fn dispatch(
            &self,
            _op: &str,
            _args: &serde_json::Value,
            _caveats: &Caveats,
        ) -> Result<String, String> {
            Ok("unused fixture crew".to_string())
        }
    }

    /// #2313 (b2): a headless turn owns a per-turn attempt ledger and returns
    /// its totals — what was sent, summed per attempt — beside the context
    /// merge in `usage`. Attempts equal the wire requests: the accepted answer
    /// is one generation (#2372).
    #[tokio::test]
    async fn a_headless_turn_reports_its_per_attempt_totals() {
        let body = [
            r#"{"choices":[{"delta":{"content":"the answer is 4"}}]}"#,
            r#"{"choices":[{"delta":{},"finish_reason":"stop"}]}"#,
            r#"{"choices":[],"usage":{"prompt_tokens":900,"completion_tokens":40}}"#,
            "[DONE]",
        ]
        .iter()
        .map(|frame| format!("data: {frame}\n\n"))
        .collect::<String>();
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/v1/chat/completions"))
            .respond_with(
                ResponseTemplate::new(200).set_body_raw(body.into_bytes(), "text/event-stream"),
            )
            .mount(&server)
            .await;
        // Pin the wire: `new` follows NEWT_OPENAI_API, which parallel tests set.
        let mut config = TurnDriverConfig::new(
            server.uri(),
            "test-model",
            BackendKind::Openai,
            "newt-core-test-workspace-that-does-not-exist",
        );
        config.openai_api = crate::OpenAiApi::ChatCompletions;
        let mut driver = TurnDriver::new(config);
        driver.submit("what is two plus two").expect("submit");
        let TurnStatus::Completed(outcome) = pump_to_done(&mut driver).await else {
            panic!("the headless turn did not complete");
        };
        let received = server.received_requests().await.expect("journal");
        assert!(received
            .iter()
            .all(|r| r.url.path() == "/v1/chat/completions"));
        assert_eq!(
            outcome.usage,
            Some(TokenUsage {
                input_tokens: 900,
                output_tokens: 40
            }),
            "usage stays the context merge"
        );
        assert_eq!(
            outcome.attempts,
            Some(crate::attempts::UsageTotals {
                attempts: u32::try_from(received.len()).unwrap(),
                usage_missing: 0,
                in_tokens: 900,
                out_tokens: 40,
                usage_complete: true,
            })
        );
        assert_eq!(received.len(), 1, "one generation for the accepted answer");
    }

    async fn drive_once_capturing(mut driver: TurnDriver) -> (Vec<serde_json::Value>, TurnOutcome) {
        let server = MockServer::start().await;
        let bodies = Arc::new(std::sync::Mutex::new(Vec::new()));
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(CapturingOllama {
                bodies: bodies.clone(),
                reply: "done".into(),
            })
            .mount(&server)
            .await;
        // Rebind the driver's URL to this server (the caller built it with a
        // placeholder so the code_search wiring is the only variable).
        driver.config.url = server.uri();
        driver.submit("find the retry backoff").expect("submit");
        let TurnStatus::Completed(outcome) = pump_to_done(&mut driver).await else {
            panic!("the captured turn did not complete");
        };
        let out = bodies.lock().unwrap().clone();
        (out, *outcome)
    }

    /// #1280: the headless driver advertises `code_search` **iff** the config
    /// carries a retrieval handle — the fix for the hardcoded `code_search: None`
    /// that left crew leaves with no semantic retrieval.
    #[tokio::test]
    async fn code_search_advertised_only_when_the_config_carries_retrieval() {
        // Without retrieval: the tool is absent (unchanged headless behavior).
        let (bare, outcome) =
            drive_once_capturing(TurnDriver::new(cfg("http://placeholder"))).await;
        assert!(
            !advertised_code_search(&bare),
            "no code_search tool without a configured index"
        );
        // #2314: every receipt read is paired with the wire body in the same
        // arm — a receipt claiming a tool the request never advertised fails.
        assert_eq!(outcome.features, InstantiatedFeatures::default());
        // No headless path instantiates the scratchpad: no state tools, no
        // injected `<state>` block, and the receipt says so.
        assert!(!advertised_tool(&bare, "state_set"));
        // `</state>`: tool descriptions name the opening tag; only a
        // rendered block closes it.
        assert!(!bare.iter().any(|b| b.to_string().contains("</state>")));

        // With a supplied embedder + index: the tool is advertised + executable.
        let index: Arc<dyn SemanticIndex> = Arc::new(SessionSemanticIndex::default());
        let (with, outcome) = drive_once_capturing(TurnDriver::new(
            cfg("http://placeholder").with_code_search(Arc::new(StubEmbedder), index, 3),
        ))
        .await;
        assert!(
            advertised_code_search(&with),
            "code_search advertised once the driver carries retrieval"
        );
        assert!(outcome.features.code_search && !outcome.features.crew);
    }

    /// #2312: the driver threads its configured allowance into the turn, and the
    /// Ollama wire sends an explicit one as `options.num_predict`, so the server
    /// enforces it. With none configured the body gains no `options` at all.
    #[tokio::test]
    async fn a_configured_output_allowance_is_sent_as_num_predict_on_ollama() {
        let (bare, outcome) =
            drive_once_capturing(TurnDriver::new(cfg("http://placeholder"))).await;
        assert_eq!(
            outcome.output_allowance, None,
            "nothing configured, nothing reserved"
        );

        let mut config = cfg("http://placeholder");
        config.output_allowance = Some(3_000);
        let (capped, outcome) = drive_once_capturing(TurnDriver::new(config)).await;
        assert_eq!(
            outcome.output_allowance,
            Some(crate::agentic::OutputAllowance {
                tokens: 3_000,
                enforced: crate::agentic::Enforcement::Server,
            })
        );
        for body in &bare {
            assert!(body.get("options").is_none(), "{body}");
        }
        for body in &capped {
            assert_eq!(
                body["options"],
                serde_json::json!({"num_predict": 3_000}),
                "{body}"
            );
        }
        let mut config = cfg("http://placeholder");
        config.num_ctx = Some(8_192);
        config.output_allowance = Some(3_000);
        let (windowed, _) = drive_once_capturing(TurnDriver::new(config)).await;
        for body in &windowed {
            assert_eq!(
                body["options"],
                serde_json::json!({"num_ctx": 8_192, "num_predict": 3_000}),
                "{body}"
            );
        }
        for body in bare.iter().chain(&capped) {
            assert!(body.get("max_tokens").is_none(), "{body}");
        }
    }

    /// Grounds the owned thread-crossing seam itself: a configured runner is
    /// what advertises both crew tools, and absence keeps both off the wire.
    #[tokio::test]
    async fn crew_advertised_only_when_driver_carries_a_runner() {
        let (bare, outcome) =
            drive_once_capturing(TurnDriver::new(cfg("http://placeholder"))).await;
        assert!(!advertised_tool(&bare, "crew"));
        assert!(!advertised_tool(&bare, "compose_roster"));
        assert!(!outcome.features.crew);

        let (with, outcome) = drive_once_capturing(
            TurnDriver::new(cfg("http://placeholder")).with_crew_runner(Arc::new(StubCrewRunner)),
        )
        .await;
        assert!(advertised_tool(&with, "crew"));
        assert!(advertised_tool(&with, "compose_roster"));
        assert!(outcome.features.crew && !outcome.features.code_search);
    }

    /// Ollama responder that calls one tool, then answers once its result is
    /// in the conversation. Records every request body.
    struct ToolCallingOllama {
        name: &'static str,
        arguments: serde_json::Value,
        bodies: Arc<std::sync::Mutex<Vec<serde_json::Value>>>,
    }

    impl Respond for ToolCallingOllama {
        fn respond(&self, request: &Request) -> ResponseTemplate {
            let body: serde_json::Value =
                serde_json::from_slice(&request.body).expect("chat request JSON");
            let has_result = body["messages"]
                .as_array()
                .is_some_and(|messages| messages.iter().any(|message| message["role"] == "tool"));
            self.bodies.lock().unwrap().push(body);
            if has_result {
                ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "message": { "content": "tool result reviewed" }
                }))
            } else {
                ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "message": {
                        "content": "",
                        "tool_calls": [{
                            "function": { "name": self.name, "arguments": self.arguments }
                        }]
                    }
                }))
            }
        }
    }

    /// #2314: an opted-in scratchpad is advertised, injected and executed, and
    /// the model sees the same head the TUI builds. Seeded, because an empty
    /// store injects no block and would pass the injection check vacuously.
    #[tokio::test]
    async fn an_opted_in_scratchpad_is_advertised_injected_and_executed() {
        use crate::agentic::{working_memory_head, SessionScratchpadStore};
        async fn drive(
            store: Arc<SessionScratchpadStore>,
        ) -> (Vec<serde_json::Value>, TurnOutcome) {
            let server = MockServer::start().await;
            let bodies = Arc::new(std::sync::Mutex::new(Vec::new()));
            Mock::given(method("POST"))
                .and(path("/api/chat"))
                .respond_with(ToolCallingOllama {
                    name: "state_set",
                    arguments: serde_json::json!({"key": "found", "value": "yes"}),
                    bodies: bodies.clone(),
                })
                .mount(&server)
                .await;
            let mut driver = TurnDriver::new(cfg(&server.uri())).with_scratchpad(store);
            driver.submit("remember what you find").expect("submit");
            let TurnStatus::Completed(outcome) = pump_to_done(&mut driver).await else {
                panic!("scratchpad turn did not complete");
            };
            assert!(
                driver
                    .transcript()
                    .iter()
                    .all(|m| m.role != crate::Role::System),
                "the head is ephemeral: it never enters the driver transcript"
            );
            let bodies = bodies.lock().unwrap().clone();
            (bodies, *outcome)
        }

        let store = Arc::new(SessionScratchpadStore::default());
        store.set("k", "v".to_string());
        let expected_head = working_memory_head(Some(store.as_ref()), false).unwrap();
        let (bodies, outcome) = drive(store.clone()).await;
        let first = &bodies[0];
        for tool in ["state_set", "state_get", "state_clear"] {
            assert!(advertised_tool(&bodies, tool), "{tool} advertised");
        }
        assert_eq!(first["messages"][0]["role"], "system");
        assert_eq!(first["messages"][0]["content"], expected_head.as_str());
        assert!(expected_head.contains("<state>\nk: v\n</state>"));
        assert!(outcome.features.scratchpad);
        assert_eq!(
            store.entries().get("found").map(String::as_str),
            Some("yes"),
            "the model's state_set executed against the supplied store"
        );

        // A second fresh store sees only its own seed, never the first run's.
        let fresh = Arc::new(SessionScratchpadStore::default());
        fresh.set("k", "v".to_string());
        let (bodies, _) = drive(fresh).await;
        assert_eq!(bodies[0]["messages"][0]["content"], expected_head.as_str());
    }

    /// #2314: when the transcript already opens with a system prompt, the head
    /// leads THAT message exactly as the TUI's does — head, blank line, prompt —
    /// rather than adding a system message or touching the stored transcript.
    #[tokio::test]
    async fn the_scratchpad_head_leads_an_existing_system_message() {
        use crate::agentic::{working_memory_head, SessionScratchpadStore};
        const PROMPT: &str = "You are the headless solver.\n\nKeep going.";
        async fn drive(store: Option<Arc<SessionScratchpadStore>>) -> Vec<serde_json::Value> {
            let server = MockServer::start().await;
            let bodies = Arc::new(std::sync::Mutex::new(Vec::new()));
            Mock::given(method("POST"))
                .and(path("/api/chat"))
                .respond_with(CapturingOllama {
                    bodies: bodies.clone(),
                    reply: "done".into(),
                })
                .mount(&server)
                .await;
            let mut driver =
                TurnDriver::with_transcript(cfg(&server.uri()), vec![MemMessage::system(PROMPT)]);
            if let Some(store) = store {
                driver = driver.with_scratchpad(store);
            }
            driver.submit("remember what you find").expect("submit");
            let TurnStatus::Completed(_) = pump_to_done(&mut driver).await else {
                panic!("turn did not complete");
            };
            assert_eq!(
                driver.transcript()[0],
                MemMessage::system(PROMPT),
                "the stored prompt survives intact; the head is per-turn only"
            );
            let first = bodies.lock().unwrap()[0].clone();
            first["messages"].as_array().expect("messages").clone()
        }
        let system_count = |messages: &[serde_json::Value]| {
            messages.iter().filter(|m| m["role"] == "system").count()
        };

        let store = Arc::new(SessionScratchpadStore::default());
        store.set("k", "v".to_string());
        let head = working_memory_head(Some(store.as_ref()), false).unwrap();
        let with = drive(Some(store)).await;
        let without = drive(None).await;

        assert_eq!(with[0]["role"], "system");
        assert_eq!(
            with[0]["content"].as_str(),
            Some(format!("{head}\n\n{PROMPT}").as_str())
        );
        assert_eq!(without[0]["content"].as_str(), Some(PROMPT));
        // The loop adds its own active-prompt card as a further system message,
        // so "one system message" is measured against the same run without the
        // head: the head must add none.
        assert_eq!(system_count(&with), system_count(&without));
    }

    #[derive(Clone, Debug, PartialEq, Eq)]
    struct CrewDispatch {
        op: String,
        args: serde_json::Value,
        caveats: Caveats,
        tenacity: crate::tenacity::Tenacity,
        initiative: crate::initiative::Initiative,
    }

    struct RecordingCrewRunner {
        dispatches: Arc<std::sync::Mutex<Vec<CrewDispatch>>>,
        drops: Arc<AtomicUsize>,
    }

    impl Drop for RecordingCrewRunner {
        fn drop(&mut self) {
            self.drops.fetch_add(1, Ordering::SeqCst);
        }
    }

    #[async_trait::async_trait]
    impl CrewRunner for RecordingCrewRunner {
        async fn dispatch(
            &self,
            op: &str,
            args: &serde_json::Value,
            caveats: &Caveats,
        ) -> Result<String, String> {
            self.dispatches.lock().unwrap().push(CrewDispatch {
                op: op.to_string(),
                args: args.clone(),
                caveats: caveats.clone(),
                tenacity: crate::tenacity::effective_tenacity(),
                initiative: crate::initiative::effective_initiative(),
            });
            Ok("crew ran: diff +3/-1; just check PASS".to_string())
        }
    }

    /// Grounds the full model → agent loop → crew seam with a real HTTP turn.
    /// The owned `Arc` must cross the dedicated driver thread without losing
    /// the model's arguments or attenuated session authority.
    #[tokio::test]
    async fn model_crew_call_dispatches_exact_args_caveats_and_captured_posture() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(ToolCallingOllama {
                name: "crew",
                arguments: serde_json::json!({
                    "task": "repair qualification harness",
                    "mode": "crew",
                    "crew": "nemotron-pair",
                    "verify": "just check"
                }),
                bodies: Arc::default(),
            })
            .mount(&server)
            .await;

        let expected_caveats = Caveats {
            fs_read: Scope::only(["/qualification/read".to_string()]),
            fs_write: Scope::only(["/qualification/write".to_string()]),
            exec: Scope::only(["just".to_string()]),
            net: Scope::only(["127.0.0.1".to_string()]),
            max_calls: CountBound::AtMost(9),
            valid_for_generation: Scope::only([42]),
        };
        let expected_args = serde_json::json!({
            "task": "repair qualification harness",
            "mode": "crew",
            "crew": "nemotron-pair",
            "verify": "just check"
        });
        let dispatches = Arc::new(std::sync::Mutex::new(Vec::new()));
        let drops = Arc::new(AtomicUsize::new(0));
        let runner = Arc::new(RecordingCrewRunner {
            dispatches: dispatches.clone(),
            drops: drops.clone(),
        });
        let weak_runner = Arc::downgrade(&runner);

        let _settings = crate::test_guard::GlobalSettingsGuard::acquire();
        crate::tenacity::set_cli_tenacity(crate::tenacity::Tenacity::Normal);
        crate::initiative::set_cli_initiative(crate::initiative::Initiative::Measured);

        let mut config = cfg(&server.uri());
        config.caveats = expected_caveats.clone();
        let mut driver = TurnDriver::new(config)
            .with_cognition(Some(crate::role_profile::Cognition::Rational))
            .with_tenacity(crate::tenacity::Tenacity::Relentless)
            .with_initiative(crate::initiative::Initiative::Eager)
            .with_crew_runner(runner.clone());
        crate::tenacity::set_cli_tenacity(crate::tenacity::Tenacity::Normal);
        crate::initiative::set_cli_initiative(crate::initiative::Initiative::Patient);
        drop(runner);
        assert!(
            weak_runner.upgrade().is_some(),
            "the driver owns the only remaining crew runner Arc"
        );

        driver.submit("field the approved crew").expect("submit");
        let TurnStatus::Completed(outcome) = pump_to_done(&mut driver).await else {
            panic!("crew turn did not complete")
        };
        assert_eq!(outcome.error, None);
        assert_eq!(
            crate::agentic::model_explanation(&outcome.reply),
            "tool result reviewed"
        );
        assert_eq!(
            dispatches.lock().unwrap().as_slice(),
            &[CrewDispatch {
                op: "crew".to_string(),
                args: expected_args,
                caveats: expected_caveats,
                tenacity: crate::tenacity::Tenacity::Relentless,
                initiative: crate::initiative::Initiative::Eager,
            }]
        );

        drop(driver);
        assert!(weak_runner.upgrade().is_none());
        assert_eq!(drops.load(Ordering::SeqCst), 1);
    }

    /// Integrated regression for headless numbered recovery: the declared 65K
    /// window is stale, vLLM reports its actual 32K full window, and
    /// meticulous reserves 16K output. The parser must therefore compact
    /// toward 16,768 input tokens; the old numberless fallback (80% of the
    /// current cap twice) leaves this
    /// fixture above the server threshold and cannot succeed.
    #[tokio::test]
    async fn headless_vllm_400_recovers_with_full_window_output_reserve() {
        const SERVER_MESSAGE_CHAR_LIMIT: usize = 16_768 * 4;
        struct VllmWindow {
            calls: Arc<AtomicUsize>,
            message_chars: Arc<std::sync::Mutex<Vec<usize>>>,
        }
        impl Respond for VllmWindow {
            fn respond(&self, request: &Request) -> ResponseTemplate {
                self.calls.fetch_add(1, Ordering::SeqCst);
                let body: serde_json::Value =
                    serde_json::from_slice(&request.body).expect("chat request JSON");
                assert_eq!(body["max_tokens"], 16_000);
                let chars = body["messages"]
                    .as_array()
                    .expect("messages array")
                    .iter()
                    .filter_map(|message| message["content"].as_str())
                    .map(str::len)
                    .sum();
                self.message_chars.lock().unwrap().push(chars);
                if chars > SERVER_MESSAGE_CHAR_LIMIT {
                    ResponseTemplate::new(400).set_body_string(
                        "This model's maximum context length is 32768 tokens. However, you requested 16000 output tokens and your prompt contains 20000 input tokens, for a total of 36000 tokens (20000 + 16000 = 36000 > 32768). Please reduce the length of the input prompt or the number of requested output tokens.",
                    )
                } else {
                    ResponseTemplate::new(200).set_body_json(serde_json::json!({
                        "model": "test-model",
                        "choices": [{
                            "message": {"role": "assistant", "content": "recovered"},
                            "finish_reason": "stop"
                        }],
                        "usage": {
                            "prompt_tokens": 16_000,
                            "completion_tokens": 1,
                            "total_tokens": 16_001
                        }
                    }))
                }
            }
        }

        let server = MockServer::start().await;
        let calls = Arc::new(AtomicUsize::new(0));
        let message_chars = Arc::new(std::sync::Mutex::new(Vec::new()));
        Mock::given(method("POST"))
            .and(path("/v1/chat/completions"))
            .respond_with(VllmWindow {
                calls: calls.clone(),
                message_chars: message_chars.clone(),
            })
            .mount(&server)
            .await;

        let mut config =
            TurnDriverConfig::new(server.uri(), "test-model", BackendKind::Openai, ".");
        // Pin the wire: `new` follows NEWT_OPENAI_API, which parallel tests set.
        config.openai_api = crate::OpenAiApi::ChatCompletions;
        config.chat_completions_capability = crate::model_card::ChatCompletionsCapability {
            cognition: Some(true),
            ..Default::default()
        };
        config.num_ctx = Some(65_536);
        config.safe_context = Some(52_428);
        config.max_ok_input = Some(52_428);

        let mut transcript = Vec::new();
        for index in 0..20 {
            let content = format!("history-{index:02} {}", "x".repeat(3_988));
            transcript.push(if index % 2 == 0 {
                MemMessage::user(content)
            } else {
                MemMessage::assistant(content)
            });
        }
        let estimated_history = config
            .estimation
            .tokens_for_chars(transcript.iter().map(|m| m.content.len()).sum());
        assert!(
            (16_769..31_702).contains(&estimated_history),
            "fixture must distinguish full-window recovery from two 80% fallbacks: {estimated_history}"
        );

        let mut driver = TurnDriver::with_transcript(config, transcript)
            .with_cognition(Some(crate::role_profile::Cognition::Meticulous));
        driver.submit("finish the current task").expect("submit");
        let TurnStatus::Completed(outcome) = pump_to_done(&mut driver).await else {
            panic!("headless turn did not complete after numbered recovery")
        };
        assert_eq!(outcome.error, None, "numbered recovery must be clean");
        assert_eq!(
            crate::agentic::model_explanation(&outcome.reply),
            "recovered"
        );
        assert!(calls.load(Ordering::SeqCst) >= 2);
        let observed = message_chars.lock().unwrap();
        assert!(
            observed[0] > SERVER_MESSAGE_CHAR_LIMIT,
            "first request must overflow"
        );
        assert!(
            *observed.last().unwrap() <= SERVER_MESSAGE_CHAR_LIMIT,
            "retry must compact under the 16,768-token input cap: {observed:?}"
        );
    }

    /// A shell observation feeds the NEXT turn's context, redacted. We prove the
    /// observation reaches the backend's request body (so the model sees it) and
    /// that a secret in it never does.
    #[tokio::test]
    async fn observation_is_in_context_for_the_next_turn_and_redacted() {
        let server = MockServer::start().await;
        // Capture the request body the model would see.
        let seen = Arc::new(std::sync::Mutex::new(String::new()));
        struct Capture {
            seen: Arc<std::sync::Mutex<String>>,
        }
        impl Respond for Capture {
            fn respond(&self, req: &Request) -> ResponseTemplate {
                *self.seen.lock().unwrap() = String::from_utf8_lossy(&req.body).into_owned();
                ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "message": { "content": "ack" }
                }))
            }
        }
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(Capture { seen: seen.clone() })
            .mount(&server)
            .await;

        let mut driver = TurnDriver::new(cfg(&server.uri()));
        driver.submit_observation(ShellObservation::new(
            "zsh",
            "$ cat creds\nsecret_key=wJalrXUtnFEMIabcdEFGHIJKLMNOPbPxRfiCY",
        ));
        // Observation alone does not start a turn.
        assert!(!driver.is_running());
        assert_eq!(driver.transcript().len(), 1);

        driver.submit("did you see my shell?").expect("submit");
        let _ = pump_to_done(&mut driver).await;

        let body = seen.lock().unwrap().clone();
        // The observation framing reached the model …
        assert!(
            body.contains("shell observation"),
            "observation missing from request body: {body}"
        );
        // … but the secret value did not.
        assert!(
            !body.contains("wJalrXUtnFEMIabcdEFGHIJKLMNOPbPxRfiCY"),
            "secret leaked into the request body: {body}"
        );
    }

    /// Submitting while a turn is in flight is rejected — one turn at a time.
    #[tokio::test]
    async fn second_submit_while_running_is_busy() {
        let server = MockServer::start().await;
        // No delay is needed: `submit` marks the turn in flight before it
        // returns, so the second submit below is rejected regardless of how
        // fast the server answers.
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({ "message": { "content": "slow" } })),
            )
            .mount(&server)
            .await;

        let mut driver = TurnDriver::new(cfg(&server.uri()));
        driver.submit("first").expect("first submit");
        let err = driver
            .submit("second")
            .expect_err("second must be rejected");
        assert!(matches!(err, TurnDriverError::Busy));
        // Let it finish so the test doesn't leak the task.
        let _ = pump_to_done(&mut driver).await;
    }

    /// W0 (#1511): the served model + parse signals flow from the loop to the
    /// `TurnOutcome`. The backend body carries `model` (what it actually
    /// served); the reply is content with no tool call, so round 0 records a
    /// `no_parseable_tool_call` signal. Clean turn ⇒ no error class.
    #[tokio::test]
    async fn outcome_carries_served_model_and_parse_signals() {
        use crate::agentic::observability::ParseSignal;
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "model": "served-by-backend",
                "message": { "content": "plain prose, no tool call" }
            })))
            .mount(&server)
            .await;

        let mut driver = TurnDriver::new(cfg(&server.uri()));
        driver.submit("do a thing").expect("submit");
        let status = pump_to_done(&mut driver).await;
        let TurnStatus::Completed(o) = status else {
            panic!("expected Completed, got {status:?}");
        };
        assert_eq!(o.error, None);
        assert_eq!(o.error_class, None, "clean turn carries no error class");
        assert_eq!(
            o.served_model.as_deref(),
            Some("served-by-backend"),
            "effective model comes from the response body, not our request"
        );
        assert!(
            o.parse_signals
                .contains(&ParseSignal::NoParseableToolCall { round: 0 }),
            "content with no parseable call is signalled: {:?}",
            o.parse_signals
        );
    }

    /// U4a: a run allowance smaller than the work ends the turn as a TYPED,
    /// clean stop (like the round cap), not an untyped `harness_error`: the
    /// reason is `RunAllowance`, there is no error, and the harness — not a
    /// model call the allowance now refuses — writes the reply. The partial
    /// trajectory survives.
    #[tokio::test]
    async fn an_exhausted_run_allowance_ends_the_turn_typed_with_a_harness_notice() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "model": "m",
                "message": {
                    "content": "",
                    "tool_calls": [{"function": {"name": "list_dir", "arguments": {"path": "."}}}]
                }
            })))
            .mount(&server)
            .await;

        let mut config = cfg(&server.uri());
        config.run_allowance = Some(2);
        let mut driver = TurnDriver::new(config);
        driver.submit("keep listing").expect("submit");
        let TurnStatus::Completed(o) = pump_to_done(&mut driver).await else {
            panic!("an exhausted allowance is a completed turn, not a failure");
        };
        assert_eq!(o.error, None, "typed stop, not an error");
        assert_eq!(o.error_class, None);
        assert_eq!(o.end_reason, Some(crate::TurnEndReason::RunAllowance));
        assert!(o.harness_reply, "the notice is harness-written");
        assert!(o.reply.contains("run allowance"), "notice: {}", o.reply);
        assert!(!o.tool_events.is_empty(), "the partial trajectory is kept");
    }

    /// W0 (#1511): a non-success HTTP status is a `model_error` STRUCTURALLY —
    /// the class rides the typed `DispatchError` through the anyhow chain to
    /// the outcome; nothing string-matches the message. (404 is non-retryable,
    /// so the test does not sit through the backoff schedule.)
    #[tokio::test]
    async fn http_error_status_classifies_as_model_error() {
        use crate::agentic::observability::ErrorClass;
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(ResponseTemplate::new(404).set_body_string("no such model"))
            .mount(&server)
            .await;

        let mut driver = TurnDriver::new(cfg(&server.uri()));
        driver.submit("do a thing").expect("submit");
        let status = pump_to_done(&mut driver).await;
        let TurnStatus::Completed(o) = status else {
            panic!("expected Completed-with-error, got {status:?}");
        };
        let err = o
            .error
            .expect("the failed dispatch is carried on the outcome");
        assert!(err.contains("Ollama 404"), "historical wording kept: {err}");
        assert_eq!(o.error_class, Some(ErrorClass::Model));
    }

    /// #2318: a 2xx OpenAI reply whose tool call has no id was answered by the
    /// model, so the turn ends `model_error` on either route the defect takes:
    /// strict decoding of a stream (a `DispatchError` attached as context, which
    /// the outcome used to miss and file as `harness_error`), or the batch
    /// validator's `CorrelationImpossible` arm for a complete JSON reply (the
    /// arm Anthropic and Responses share). Nothing runs. An id-less batch, streamed
    /// or complete, is re-asked twice (P0 U3) before the third ends the turn; only a
    /// stream whose id is not a string is undecodable and is not retried.
    #[tokio::test]
    async fn a_tool_call_without_an_id_classifies_as_model_error() {
        use crate::agentic::observability::ErrorClass;
        let stream_with = |id: &str| {
            [
                format!(r#"{{"choices":[{{"delta":{{"tool_calls":[{{"index":0,{id}"type":"function","function":{{"name":"read_file","arguments":"{{}}"}}}}]}}}}]}}"#),
                r#"{"choices":[{"delta":{},"finish_reason":"tool_calls"}]}"#.to_string(),
                "[DONE]".to_string(),
            ]
            .iter()
            .map(|frame| format!("data: {frame}\n\n"))
            .collect::<String>()
        };
        let stream = stream_with("");
        let unreadable = stream_with(r#""id":7,"#);
        let json = serde_json::json!({"choices": [{"message": {"role": "assistant", "content": null,
            "tool_calls": [{"type": "function", "function": {"name": "read_file", "arguments": "{}"}}]},
            "finish_reason": "tool_calls"}]});
        for (name, reply, message, expected_posts) in [
            (
                "strict stream, id absent (re-asked, then aborted)",
                ResponseTemplate::new(200).set_body_raw(stream.into_bytes(), "text/event-stream"),
                "malformed provider output",
                3,
            ),
            (
                "strict stream, non-string id (undecodable, one POST)",
                ResponseTemplate::new(200)
                    .set_body_raw(unreadable.into_bytes(), "text/event-stream"),
                "invalid tool-call ID",
                1,
            ),
            (
                "complete JSON",
                ResponseTemplate::new(200).set_body_json(json),
                "malformed provider output",
                3,
            ),
        ] {
            let server = MockServer::start().await;
            Mock::given(method("POST"))
                .and(path("/v1/chat/completions"))
                .respond_with(reply)
                .mount(&server)
                .await;

            let mut config = cfg(&server.uri());
            config.kind = BackendKind::Openai;
            // Pin the wire: the constructor default reads `NEWT_OPENAI_API`,
            // which a concurrent test may hold at `responses` (a 404 here).
            config.openai_api = crate::OpenAiApi::ChatCompletions;
            let mut driver = TurnDriver::new(config);
            driver.submit("do a thing").expect("submit");
            let status = pump_to_done(&mut driver).await;
            let TurnStatus::Completed(o) = status else {
                panic!("{name}: expected Completed-with-error, got {status:?}");
            };
            let err = o.error.expect("the rejection is carried on the outcome");
            assert!(err.contains(message), "{name}: {err}");
            assert_eq!(o.error_class, Some(ErrorClass::Model), "{name}");
            assert_eq!(o.end_reason, Some(crate::TurnEndReason::Failed), "{name}");
            // Nothing ran; the re-asked rounds leave only not-ok markers (P0 U3).
            assert!(
                o.tool_events
                    .iter()
                    .all(|e| e.tool == "(rejected tool-call batch)" && !e.ok),
                "{name}: nothing ran"
            );
            let posts = server.received_requests().await.expect("journal");
            assert_eq!(
                posts.len(),
                expected_posts,
                "{name}: POSTs before the abort"
            );
        }
    }

    /// Cancel aborts the in-flight turn and returns the driver to idle.
    #[tokio::test]
    async fn cancel_aborts_the_in_flight_turn() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/api/chat"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_delay(Duration::from_secs(30))
                    .set_body_json(serde_json::json!({ "message": { "content": "never" } })),
            )
            .mount(&server)
            .await;

        let mut driver = TurnDriver::new(cfg(&server.uri()));
        driver.submit("start a slow turn").expect("submit");
        assert!(driver.is_running());
        driver.cancel();
        assert!(!driver.is_running());
        assert!(matches!(driver.poll(), TurnStatus::Idle));
        // The user message that started it survives for a clean retry.
        assert_eq!(driver.transcript().len(), 1);
        assert_eq!(driver.transcript()[0].content, "start a slow turn");
    }
}