harness-rs-loop 0.0.71

//! ReAct agent loop with self-correction.
//!
//! Minimal v0.0.1 implementation:
//! - Applies guides once at the start.
//! - Sends `Context` (with `tools`) to the model.
//! - Dispatches each returned tool call via [`ToolRegistry`].
//! - Runs `Sensor::SelfCorrect` sensors after each action; auto-fix patches are
//!   applied directly to the world, blocking signals are fed back to the model.
//! - Stops when the model returns no tool calls, when `policy.max_iters` is
//!   hit, or when a stuck-detector ([`StuckPolicy`], [`MonotonyPolicy`]) says
//!   the run is spinning.

// ── crates that used to be their own ───────────────────────────────────────
// Ten of them, and not one carried a dependency a caller would want to avoid.
// The boundaries bought nothing and cost every release ten more publishes and
// every host ten more version pins; a crate boundary here has to name a
// dependency you might not want, and these did not.
pub mod blueprint;
pub mod compactor;
pub mod daemon;
pub mod experience;
pub mod hooks;
pub mod loop_engine;
pub mod orchestrator;
pub mod sandbox;
pub mod scheduler;
pub mod templates;

pub mod acceptance;
pub mod goal;
pub mod learning;
pub mod memory_layer;
pub mod meters;
#[cfg(feature = "otel")]
pub mod otel;
pub mod profile_guide;
pub mod recall_layer;
pub mod receipt;
pub mod registry;
pub mod seal;
pub use acceptance::{Acceptance, FilesExist, NonEmptyAnswer, Verdict};
pub use goal::{Goal, GoalStore, Phase, PhaseStatus};
pub use receipt::{Receipt, ReceiptBuilder};
pub use seal::{SealBreach, SealSet};
/// The token `AgentLoop::with_cancellation` takes, re-exported so a caller can
/// cancel a run without depending on `tokio-util` directly.
pub use tokio_util::sync::CancellationToken;
pub mod replay;
pub mod subagent;
pub mod telemetry;
pub mod timeline;
pub mod trajectory;
pub use trajectory::{
    GoldenDiff, Report, Rule, RuleSet, Step, Trajectory, TrajectoryCheck, Violation,
};

pub use learning::*;
pub use memory_layer::*;
pub mod boundary_guide;
pub use boundary_guide::*;
pub mod events;
pub use events::{HostEvent, NoticeGuide};
pub use profile_guide::*;
pub use recall_layer::*;
pub use registry::*;
pub use replay::*;
pub use subagent::*;
pub use telemetry::*;

use crate::compactor::{CALIBRATION_KEY, DefaultCompactor};
use crate::hooks::HookBus;
use harness_core::{
    Action, Block, CompactionStage, Compactor, Context, Event, Guide, HarnessError, HookOutcome,
    Model, ModelDelta, ModelOutput, ResponseFormat, Sensor, SessionSource, SignalSet, Stage, Task,
    ToolCall, ToolResult, Turn, TurnRole, World,
};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tracing::Instrument;
use tracing::field::Empty;

/// What a finished model call leaves behind: fields on its `chat` span, and
/// its duration and tokens as metrics.
fn finish_chat(
    span: &tracing::Span,
    info: &harness_core::ModelInfo,
    started: std::time::Instant,
    result: Option<Result<&ModelOutput, String>>,
) {
    let end = match &result {
        Some(Ok(out)) => {
            span.record("gen_ai.usage.input_tokens", out.usage.input_tokens)
                .record("gen_ai.usage.output_tokens", out.usage.output_tokens)
                .record(
                    "gen_ai.usage.cached_input_tokens",
                    out.usage.cached_input_tokens,
                )
                .record(
                    "gen_ai.response.finish_reasons",
                    format!("{:?}", out.stop_reason),
                )
                .record("tool_calls", out.tool_calls.len());
            meters::ModelCallEnd::Done(&out.usage)
        }
        Some(Err(e)) => {
            span.record("otel.status_code", "error")
                .record("otel.status_message", e.as_str());
            meters::ModelCallEnd::Failed
        }
        None => {
            span.record("otel.status_message", "cancelled");
            meters::ModelCallEnd::Cancelled
        }
    };
    meters::model_call(info, started.elapsed(), end);
}

/// A tool call's result and when it ran, by the World clock — the figures
/// behind [`Event::ToolTimed`].
struct Dispatched {
    result: ToolResult,
    started_ms: i64,
    finished_ms: i64,
}

/// The span one model call runs in. Named `chat {model}` for OTLP backends,
/// which group by span name; fields per the GenAI conventions.
fn chat_span(info: &harness_core::ModelInfo) -> tracing::Span {
    tracing::info_span!(
        target: "harness.telemetry",
        "chat",
        "otel.name" = %format!("chat {}", info.model),
        "otel.kind" = "client",
        "otel.status_code" = Empty,
        "otel.status_message" = Empty,
        "gen_ai.operation.name" = "chat",
        "gen_ai.provider.name" = %info.provider,
        "gen_ai.request.model" = %info.model,
        "gen_ai.usage.input_tokens" = Empty,
        "gen_ai.usage.output_tokens" = Empty,
        "gen_ai.usage.cached_input_tokens" = Empty,
        "gen_ai.response.finish_reasons" = Empty,
        tool_calls = Empty,
    )
}

/// Governs the loop's stuck-detector. When the model repeats the *same* tool
/// call (name + args) round after round without making progress, the loop first
/// nudges it to change approach, then terminates cleanly with [`Outcome::Stuck`]
/// rather than burning the rest of the budget spinning on a loop.
///
/// Enabled by default with conservative thresholds — the calls must be
/// *byte-identical*, so genuine "read the same file twice" work never trips it.
#[derive(Debug, Clone)]
pub struct StuckPolicy {
    pub enabled: bool,
    /// Consecutive identical tool-call rounds before injecting a "you are
    /// repeating yourself, change your approach" feedback signal.
    pub nudge_after: u32,
    /// Consecutive identical rounds before terminating with [`Outcome::Stuck`].
    pub abort_after: u32,
}

impl Default for StuckPolicy {
    fn default() -> Self {
        Self {
            enabled: true,
            nudge_after: 3,
            abort_after: 6,
        }
    }
}

/// A second stuck signal, orthogonal to [`StuckPolicy`]: the model calls *one*
/// tool round after round — a different argument every time, so nothing ever
/// repeats — and never moves on to anything else.
///
/// [`StuckPolicy`] cannot see this and must not be changed so that it can. It
/// keys on *byte-identical consecutive* rounds precisely so that genuine "read
/// the same file twice" work never trips it, and `graph_find("Xen")` followed
/// by `graph_find("KVM")` is correctly not a repeat — the request changed. Two
/// identical calls in a row and one tool worked for twenty rounds are different
/// shapes, and want different thresholds and different reasons.
///
/// Measured on a real run: asked which platforms a product integrates with, an
/// agent that could not retrieve the anchor entity began guessing candidates
/// and looking them up one at a time — `graph_find("Nomad")`,
/// `graph_find("Docker")`, `graph_find("VMware")`, `graph_find("Xen")`,
/// `graph_find("KVM")` — twelve calls to one tool across twenty rounds, each
/// with a different argument, none of them useful. Every individual call looked
/// like progress. The *shape* did not, and the shape is something the loop can
/// see without knowing anything about what `graph_find` means.
///
/// **Off by default, and it has to be.** The predicate is "one tool, many
/// rounds, nothing else", and there is a large class of healthy agents whose
/// every round is one tool: give an agent a shell and nothing else and
/// `run_shell` is the only name it will ever emit, thirty rounds running,
/// legitimately. No threshold rescues that — the false positive is unbounded in
/// N, not merely rare — so a default-on version of this guard would terminate
/// working agents, which is worse than the problem it fixes. A host that knows
/// its agent has several tools and is meant to use them can switch it on;
/// nobody else pays anything for its existence.
#[derive(Debug, Clone)]
pub struct MonotonyPolicy {
    /// Off by default. See the type docs for why no default-on threshold is
    /// safe for this predicate.
    pub enabled: bool,
    /// Consecutive single-tool rounds before injecting a "you are working one
    /// tool and nothing else" feedback signal.
    ///
    /// 8, argued rather than rounded. The honest single-tool runs are surveys —
    /// read five files, grep four paths — and a survey ends by using a
    /// *different* tool, which resets the count; reaching eight rounds of one
    /// name with nothing else touched is already past what surveying does. The
    /// cost of being wrong here is one sentence of feedback, not a terminated
    /// run, which is why the nudge can sit this close to ordinary work. This is
    /// also the half of the policy that does the work: in the measured case a
    /// nudge at round eight arrives with two thirds of the budget unspent.
    pub nudge_after: u32,
    /// Consecutive single-tool rounds before terminating with [`Outcome::Stuck`].
    ///
    /// 16 — twice the nudge, mirroring [`StuckPolicy`]'s own 3→6 ratio, and
    /// 2.6× its abort so the two can never race: a byte-identical spiral is
    /// always caught by the older policy first, under its own reason. Deliberately
    /// generous, because this end of the policy is fatal and its only job is to
    /// be a backstop for budgets long enough that the nudge was ignored.
    pub abort_after: u32,
}

impl Default for MonotonyPolicy {
    fn default() -> Self {
        Self {
            enabled: false,
            nudge_after: 8,
            abort_after: 16,
        }
    }
}

/// A ceiling on how much of one tool result reaches the context.
///
/// A single call can return more than the whole conversation: a lock file, a
/// `SELECT *`, an MCP tool the framework does not control. Measured on a real
/// run, "search these files for a word" cost 53,487 input tokens because one
/// `read_file` returned a lock file — the model then paid for it on every
/// subsequent turn, and compaction started throwing away real history to make
/// room. A per-result ceiling is the only place to stop that: the tools cannot
/// all be trusted (third-party MCP), and the compactor only runs after the
/// damage is in the context.
#[derive(Debug, Clone)]
pub struct ToolResultPolicy {
    /// Max serialized bytes of a single tool result. `None` disables the guard.
    /// Default ~24 KiB — roughly 6k tokens of English, generous for a file page
    /// or a query result, far below what blows a window.
    pub max_bytes: Option<usize>,
    /// Replace the payload of a read-only call that exactly repeats an earlier
    /// one in the same run, when nothing has modified the world in between.
    ///
    /// [`StuckPolicy`] only sees *consecutive* identical rounds. Reading a file
    /// at iteration 1 and again at iteration 5 is not that, and looks like
    /// progress — but the same bytes land in the context twice and the model
    /// learns nothing the second time. Measured on a real run, model wait was
    /// 36.7s against 3ms of tool execution: a repeat costs context, not time,
    /// so what is suppressed is the payload, not the call.
    ///
    /// Only `ToolRisk::ReadOnly` qualifies (`Network` is a separate risk, and an
    /// external endpoint may answer differently), and any non-read-only call
    /// clears the record — after a write, re-reading is the correct move.
    ///
    /// **Off by default, on the evidence.** Measured on the completion
    /// benchmark: ceilings alone solved 6/6 tasks for 160k effective tokens;
    /// adding repeat-suppression cut that to 32k — and lost a task. An agent
    /// working through a file larger than one page re-reads it because it cannot
    /// hold it, gets told it already has the answer, and does not: the content
    /// was paged away. Five times cheaper is not worth a task a framework could
    /// otherwise do, so this is opt-in for callers who would rather have the
    /// tokens. (Suppressing from the *third* identical call rather than the
    /// second would likely keep most of the saving without the failure — it
    /// needs measuring before it becomes the default.)
    pub dedupe_repeats: bool,
    /// When a result exceeds `max_bytes`, save the *full* payload to a file
    /// under `.harness/spill/` in the workspace and inline a bounded preview
    /// plus the path, instead of cutting the tail off and throwing it away.
    ///
    /// Truncation destroys information: the model is told "narrow your
    /// request", but the bytes it needed may be exactly the ones dropped, and
    /// the only recovery is re-running the call to be truncated again.
    /// Measured on the completion benchmark, that loop is visible as a single
    /// task paying 184k input tokens. Spilling keeps the ceiling — the context
    /// gets a preview, never the flood — while the whole result stays
    /// retrievable through the file tools the agent already has (`read_file`
    /// with offset/limit, `grep` on the spill path), because the spill lives
    /// inside the workspace jail. Borrowed from DeepSeek Harness's `spill`
    /// family (preview + retrieval locator), which is the same judgement.
    ///
    /// Costs nothing until it fires: the write happens only on the oversized
    /// path, which the default ceiling makes rare. When the write fails (e.g.
    /// read-only workspace) the guard falls back to plain truncation.
    pub spill: bool,
}

impl Default for ToolResultPolicy {
    fn default() -> Self {
        Self {
            max_bytes: Some(24 * 1024),
            dedupe_repeats: false,
            spill: true,
        }
    }
}

/// Governs *when* and *how far* the loop compacts context. Hysteresis: only
/// start compacting once usage crosses `high_water`, and stop as soon as it's
/// back under `target` — instead of running every stage above a threshold on
/// every turn. This avoids over-compacting (needlessly reaching the lossy,
/// main-model `AutoCompact` stage) and, because compaction rewrites history and
/// invalidates the provider prefix cache, avoids nibbling the context every
/// single turn.
#[derive(Debug, Clone)]
pub struct CompactPolicy {
    /// Start compacting when `used/window` exceeds this. Default 0.75.
    pub high_water: f32,
    /// Stop as soon as `used/window` is back at/under this. Default 0.55.
    pub target: f32,
}

impl Default for CompactPolicy {
    fn default() -> Self {
        Self {
            high_water: 0.75,
            target: 0.55,
        }
    }
}

/// Inline preview size for a spilled result. Deliberately smaller than the
/// ceiling: the point of spilling is that the context gets a *glimpse* and a
/// path, not four-fifths of the flood.
/// How often a streamed model call reports progress / a heartbeat.
const MODEL_PROGRESS_EVERY: Duration = Duration::from_secs(5);
const MODEL_HEARTBEAT_EVERY: Duration = Duration::from_secs(30);
/// Times one model turn is retried after a transport failure (dropped
/// connection, stream cut before a finish reason) before the error ends the run.
const MODEL_TRANSPORT_RETRIES: u32 = 2;
/// Turns in a row the model may be cut off by the output-token limit, with no
/// tool call to show for it, before the run is ended as [`Outcome::Stuck`].
/// Each one is sent back with an instruction to stop deliberating and act.
const MAX_TRUNCATED_TURNS: u32 = 2;
/// What a turn cut off by the output-token limit is told before it retries.
const TRUNCATED_TURN_NUDGE: &str = "Your previous turn hit the output token limit before \
     you produced an answer or a tool call, so nothing from it was used. Do not repeat that \
     reasoning. Act now: either call the next tool you need, or write the final answer \
     directly and keep it short.";

const SPILL_PREVIEW_BYTES: usize = 4 * 1024;

/// Told to the model when the step budget ran out and the loop stripped its
/// tools. See `force_final_synthesis`.
const BUDGET_SYNTHESIS_PROMPT: &str = "[system: budget exhausted] \
    You have run out of budget for tool calls. Write your final answer \
    NOW using only the tool results already in this conversation. Do not \
    request more tools. Mark facts you could not verify as UNKNOWN. \
    Include source URLs for every claim that is not UNKNOWN.";

/// Told to the model when [`MonotonyPolicy`] stopped the run. Deliberately
/// names the shape it was caught in: the failure mode this ends is a model
/// guessing arguments to one tool, and such a model has usually *found*
/// something it never got round to reporting. "Say what you could not
/// determine" is the instruction it needs; "mark it UNKNOWN" is the same
/// instruction the budget prompt gives for the same reason.
const MONOTONY_SYNTHESIS_PROMPT: &str = "[system: stopped — no progress] \
    You have called the same tool many rounds in a row without using any other \
    tool, so no further tool calls will be granted. Write your final answer NOW \
    using only the tool results already in this conversation. Do not request \
    more tools. State plainly which parts you could not determine instead of \
    guessing at them.";

/// First `n` bytes of `s`, cut on a char boundary.
fn head_of(s: &str, n: usize) -> &str {
    let mut end = n.min(s.len());
    while end > 0 && !s.is_char_boundary(end) {
        end -= 1;
    }
    &s[..end]
}

/// The string field that dominates an oversized result, if one does.
///
/// Most oversized results are one big string in a small envelope — a file
/// body, a command's stdout — and the right spill for those is the *raw text*:
/// multi-line, so `read_file`'s line paging and `grep`'s line matching work on
/// it. Spilling the serialized JSON instead would fold the whole payload onto
/// one escaped line that line-oriented tools can neither page nor match
/// (`read_file` pages by line; a 68 KB single-line file is unreachable past
/// the first 16 KB). Returns the dotted path and the string when one field is
/// ≥ 80% of the serialized size.
fn dominant_string(v: &serde_json::Value, total: usize) -> Option<(String, &str)> {
    fn walk<'a>(v: &'a serde_json::Value, path: &str, best: &mut Option<(String, &'a str)>) {
        match v {
            serde_json::Value::String(s) => {
                if best.as_ref().is_none_or(|(_, b)| s.len() > b.len()) {
                    *best = Some((path.to_string(), s.as_str()));
                }
            }
            serde_json::Value::Object(m) => {
                for (k, x) in m {
                    let p = if path.is_empty() {
                        k.clone()
                    } else {
                        format!("{path}.{k}")
                    };
                    walk(x, &p, best);
                }
            }
            serde_json::Value::Array(a) => {
                for (i, x) in a.iter().enumerate() {
                    walk(x, &format!("{path}[{i}]"), best);
                }
            }
            _ => {}
        }
    }
    let mut best = None;
    walk(v, "", &mut best);
    best.filter(|(_, s)| s.len() * 5 >= total * 4)
}

/// Replace the value at a `dominant_string` dotted path with `with`.
fn replace_at(v: &mut serde_json::Value, path: &str, with: serde_json::Value) {
    let mut cur = v;
    let mut rest = path;
    loop {
        // Next segment: `key`, `key[i]`, or a bare `[i]`.
        let (seg, tail) = match rest.find('.') {
            Some(dot) => (&rest[..dot], &rest[dot + 1..]),
            None => (rest, ""),
        };
        let (key, idx) = match seg.find('[') {
            Some(b) => (&seg[..b], seg[b + 1..seg.len() - 1].parse::<usize>().ok()),
            None => (seg, None),
        };
        if !key.is_empty() {
            match cur.get_mut(key) {
                Some(next) => cur = next,
                None => return,
            }
        }
        if let Some(i) = idx {
            match cur.get_mut(i) {
                Some(next) => cur = next,
                None => return,
            }
        }
        if tail.is_empty() {
            *cur = with;
            return;
        }
        rest = tail;
    }
}

/// Persist an oversized result under `.harness/spill/` in the workspace and
/// build the inline marker. `None` on any IO failure — the caller falls back
/// to truncation, because a guard that can error a run to protect a context
/// window has its priorities backwards.
fn spill_oversized(
    action: &Action,
    content: &serde_json::Value,
    serialized: &str,
    root: &std::path::Path,
) -> Option<serde_json::Value> {
    let dir = root.join(".harness").join("spill");
    std::fs::create_dir_all(&dir).ok()?;
    let id: String = action
        .call_id
        .chars()
        .filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_'))
        .take(48)
        .collect();
    let id = if id.is_empty() { "call".into() } else { id };

    let (rel, preview, meta) = match dominant_string(content, serialized.len()) {
        // One big string in a small envelope: spill the raw text, keep the
        // envelope inline with the big field swapped for a pointer.
        Some((field, text)) => {
            let rel = format!(".harness/spill/{id}-{}.txt", action.tool);
            std::fs::write(root.join(&rel), text).ok()?;
            let mut meta = content.clone();
            replace_at(
                &mut meta,
                &field,
                serde_json::Value::String(format!("[{} bytes spilled to {rel}]", text.len())),
            );
            (
                rel,
                head_of(text, SPILL_PREVIEW_BYTES).to_string(),
                Some(meta),
            )
        }
        // Structured payload (e.g. a long match list): spill pretty-printed,
        // one element per line, so line-oriented retrieval still works.
        None => {
            let rel = format!(".harness/spill/{id}-{}.json", action.tool);
            let pretty =
                serde_json::to_string_pretty(content).unwrap_or_else(|_| serialized.to_string());
            std::fs::write(root.join(&rel), &pretty).ok()?;
            (rel, head_of(&pretty, SPILL_PREVIEW_BYTES).to_string(), None)
        }
    };
    tracing::warn!(
        target: "harness.telemetry",
        event = "tool.result.spilled",
        "gen_ai.tool.name" = %action.tool,
        bytes = serialized.len(),
        path = %rel,
    );
    let mut marker = serde_json::json!({
        "spilled": true,
        "tool": action.tool,
        "bytes_total": serialized.len(),
        "path": rel,
        "preview": preview,
        "note": format!(
            "This result was {} bytes — too large to inline, so the FULL content was \
             saved to '{rel}' (workspace-relative). Nothing was lost. Retrieve exactly \
             the part you need: grep(path=\"{rel}\", pattern=...) or \
             read_file(path=\"{rel}\", offset=..., limit=...). Do not repeat the \
             original call — it will spill again.",
            serialized.len()
        ),
    });
    // The envelope metadata (paths, flags) rides along when it is small; a
    // pathological envelope that is itself oversized is dropped, not inlined.
    if let Some(meta) = meta
        && meta.to_string().len() <= SPILL_PREVIEW_BYTES
    {
        marker["meta"] = meta;
    }
    Some(marker)
}

/// A stable fingerprint of a round's tool calls (names + args, ignoring the
/// volatile call id). Two rounds with the same fingerprint asked for the exact
/// same actions — the signal the stuck-detector keys on.
fn tool_call_fingerprint(calls: &[ToolCall]) -> String {
    calls
        .iter()
        .map(|c| format!("{}({})", c.name, c.args))
        .collect::<Vec<_>>()
        .join("|")
}

/// The one tool a round asked for, when the whole round asked for exactly one
/// *name*. `None` when the round mixed names — reaching for a second tool is,
/// on its own, evidence the model is still moving, and resets
/// [`MonotonyPolicy`]'s count.
///
/// Several parallel calls to the same tool count as one round of that tool:
/// reading four files in one round is efficient work, not monotony, and the
/// policy's threshold is in *rounds* precisely so that batching is never
/// punished.
fn sole_tool(calls: &[ToolCall]) -> Option<&str> {
    let first = calls.first()?.name.as_str();
    calls.iter().all(|c| c.name == first).then_some(first)
}

/// Cheap identity for one tool result, for the stuck detector's "did anything
/// change" test. Not a checksum anyone depends on — only equality between two
/// consecutive rounds of the same process matters.
fn result_digest(tool: &str, content: &serde_json::Value) -> u64 {
    use std::hash::{Hash, Hasher};
    let mut h = std::collections::hash_map::DefaultHasher::new();
    tool.hash(&mut h);
    content.to_string().hash(&mut h);
    h.finish()
}

/// Where a run finished. Each variant is `#[non_exhaustive]` so new *fields*
/// don't break downstream matches — always include `..` when destructuring.
#[derive(Debug, Clone)]
pub enum Outcome {
    /// Model returned text with no tool calls (natural end).
    #[non_exhaustive]
    Done {
        text: Option<String>,
        iters: u32,
        tools_called: u32,
        usage: harness_core::Usage,
        /// What the acceptance checks said. `None` means nothing was asked —
        /// so "the model stopped" is all this outcome claims.
        verified: Option<Verdict>,
        /// The sealed contract as it stood before the model's first turn.
        ///
        /// Carried out of the run because a receipt has to say *what* was
        /// measured, not only that the measurement held. Empty when nothing
        /// was sealed.
        contract: crate::seal::SealSet,
        /// Set when a sealed acceptance contract changed during the run.
        ///
        /// Distinct from a plain failed `verified` because the two mean
        /// different things to a caller: a failed check is work not finished,
        /// this is the measuring instrument having been moved. A host may want
        /// to fail a build on the first and page someone on the second.
        seal_breach: Option<String>,
    },
    /// Policy budget exhausted before the model stopped requesting tools.
    /// Carries everything we know so the caller can recover partial work
    /// (saved notes, files written by tools, the last assistant text, etc.)
    /// instead of seeing a single bare "budget out" string.
    #[non_exhaustive]
    BudgetExhausted {
        iters: u32,
        last_text: Option<String>,
        tools_called: u32,
        usage: harness_core::Usage,
        /// True when `Task::deadline` stopped the run rather than `max_iters`.
        /// Both end the same way — a forced final answer — but "ran out of
        /// time" and "ran out of steps" call for different responses from the
        /// caller, and an unattended long run only ever hits the first.
        deadline_reached: bool,
    },
    /// The agent got stuck and the loop terminated early to save the rest of
    /// the budget. Carries partial work (last text, files already written by
    /// tools) like `BudgetExhausted`.
    ///
    /// Two independent detectors end here, and `reason` says which: it repeated
    /// the *same* tool call for `StuckPolicy::abort_after` rounds without
    /// progress, or it worked a *single* tool for `MonotonyPolicy::abort_after`
    /// rounds without touching anything else. Only the second forces a final
    /// synthesis before returning — see the abort site for why.
    #[non_exhaustive]
    Stuck {
        /// Human-readable reason, e.g. "repeated `read_file(...)` 6× without progress"
        /// or "called `graph_find` for 16 consecutive rounds without using any
        /// other tool".
        reason: String,
        /// How many consecutive rounds the detector counted before aborting.
        repeated: u32,
        iters: u32,
        last_text: Option<String>,
        tools_called: u32,
        usage: harness_core::Usage,
    },
    /// The caller stopped the run through [`AgentLoop::with_cancellation`]
    /// before the model reached a natural end. Carries partial work like
    /// `Stuck` does, because the user who pressed Ctrl-C still wants what was
    /// done. The loop makes **no** further model call after this — not even
    /// the forced final synthesis the other early exits perform.
    ///
    /// A tool that was mid-flight when the token fired may still finish its
    /// side effects — a file write on a blocking thread completes, a child
    /// process keeps running unless its runner kills it on drop — but its
    /// result is discarded before it reaches history. The dispatch is still
    /// counted in `tools_called` — the telemetry `run.end` line reconciles
    /// against that number.
    #[non_exhaustive]
    Cancelled {
        /// Iterations completed before the cancel. `0` means the token was
        /// already cancelled on entry.
        iters: u32,
        last_text: Option<String>,
        tools_called: u32,
        usage: harness_core::Usage,
    },
}

/// The agent loop.
pub struct AgentLoop<M: Model> {
    pub model: M,
    pub tools: ToolRegistry,
    pub guides: Vec<Arc<dyn Guide>>,
    pub sensors: Vec<Arc<dyn Sensor>>,
    pub hooks: HookBus,
    pub compactor: Arc<dyn Compactor>,
    /// A deadline on each individual tool call. One hung call — a network tool
    /// on a dead endpoint, a shell command waiting on stdin — otherwise takes
    /// the whole run down with it, and the host's only recourse is a run-level
    /// timeout that throws away every turn of finished work (measured on the
    /// completion benchmark: a run that had already done the job was billed as
    /// a 0-token timeout). A per-call deadline converts the hang into an error
    /// *result* the model sees and can route around. `None` disables. The
    /// default is generous — 120s covers a slow build — because a false
    /// deadline on a legitimately long tool is worse than a late one.
    pub tool_timeout: Option<Duration>,
    /// Default response format applied to every run unless overridden by
    /// `run_typed`. See [`ResponseFormat`].
    pub response_format: ResponseFormat,
    /// When `true` (the default), the loop drives each model turn, including
    /// the forced final synthesis, via `Model::stream()` instead of
    /// `complete()`, firing `Event::ModelTokenDelta` for each text fragment.
    /// Tool-call deltas are still assembled inside the loop; only the terminal
    /// `ModelOutput` shape is observable downstream. `with_streaming(false)`
    /// is the explicit opt-out.
    pub streaming: bool,
    /// Stops the run from outside. Checked at the top of every iteration and
    /// raced against the model step and every tool dispatch, so a cancel
    /// lands within one await point rather than at the next iteration
    /// boundary. Defaults to a token nobody holds, which never fires.
    pub cancel: CancellationToken,
    /// Optional cross-session recall store. When set, the loop captures every
    /// turn and the `session_search` tool is registered. See `with_recall`.
    pub recall: Option<Arc<dyn harness_core::RecallStore>>,
    /// When true (and `recall` is set), a `RecallGuide` auto-injects top-k
    /// past context at session start.
    pub recall_auto_inject: bool,
    pub learning: Option<LearningConfig>,
    /// Loop-detection policy. Enabled by default — see [`StuckPolicy`].
    pub stuck: StuckPolicy,
    /// Second, independent loop-detection signal: one tool worked round after
    /// round with nothing else touched. Off by default — see [`MonotonyPolicy`].
    pub monotony: MonotonyPolicy,
    /// Ceiling on context tokens, overriding the model's own window. `None`
    /// (default) sizes the budget to `ModelInfo::context_window`.
    pub max_input_tokens: Option<u32>,
    /// Context-compaction hysteresis. See [`CompactPolicy`].
    pub compaction: CompactPolicy,
    /// Ceiling on a single tool result. See [`ToolResultPolicy`].
    pub tool_results: ToolResultPolicy,
    /// Conditions the run must satisfy before the loop reports success. The
    /// model stopping is evidence it *believes* it is finished; these say
    /// whether it is. See [`acceptance`].
    pub acceptance: Vec<Arc<dyn Acceptance>>,
    /// How many times a failed acceptance is handed back to the model before
    /// the loop gives up and reports what there is. One is usually enough: a
    /// model that ignores the first correction rarely takes the second.
    pub acceptance_retries: u32,
    /// System instruction injected into every run's `Context.system` (unless the
    /// built context already carries its own). Set via [`with_system`](Self::with_system).
    pub system: Vec<Block>,
    /// Named auxiliary models for side tasks — compaction, memory synthesis,
    /// judging, subagents — so the main conversation stays on one model (and
    /// keeps its provider prompt-cache prefix intact) while cheap or specialist
    /// work goes elsewhere. Populated via [`with_model_role`](Self::with_model_role),
    /// read via [`model_for`](Self::model_for). An unregistered role means
    /// "use the main model" — components fall back rather than fail.
    pub model_roles: std::collections::HashMap<String, Arc<dyn Model>>,
    /// True once the host installed its own compactor via
    /// [`with_compactor`](Self::with_compactor). Guards the `"compactor"`
    /// model-role convenience from overwriting an explicit choice.
    pub compactor_custom: bool,
}

impl AgentLoop<harness_core::DynModel> {
    /// Build a loop from a boxed model — what every model factory hands back
    /// (`ApiKind::build`, a router, anything stored behind a trait object).
    ///
    /// `Arc<dyn Model>` deliberately does not implement `Model` (see
    /// [`DynModel`](harness_core::DynModel) for why), so `AgentLoop::new` cannot
    /// take one. Without this constructor every caller writes the wrapper
    /// themselves, and the first thing a new user meets is a trait-bound error
    /// naming a type they have never heard of.
    ///
    /// ```ignore
    /// let model = ApiKind::OpenAI.build(base_url, model_id, key);
    /// let agent = AgentLoop::boxed(model).with_tool(Arc::new(ReadFile));
    /// ```
    pub fn boxed(model: Arc<dyn Model>) -> Self {
        Self::new(harness_core::DynModel(model))
    }
}

impl<M: Model> AgentLoop<M> {
    pub fn new(model: M) -> Self {
        Self {
            model,
            tools: ToolRegistry::new(),
            guides: Vec::new(),
            sensors: Vec::new(),
            hooks: HookBus::new(),
            compactor: Arc::new(DefaultCompactor::new()),
            tool_timeout: Some(Duration::from_secs(120)),
            response_format: ResponseFormat::Free,
            streaming: true,
            cancel: CancellationToken::new(),
            recall: None,
            recall_auto_inject: false,
            learning: None,
            stuck: StuckPolicy::default(),
            monotony: MonotonyPolicy::default(),
            max_input_tokens: None,
            compaction: CompactPolicy::default(),
            // On by default, because the failure it catches is invisible: a
            // turn that produced nothing is reported as a turn that finished.
            tool_results: ToolResultPolicy::default(),
            acceptance: vec![Arc::new(acceptance::NonEmptyAnswer)],
            acceptance_retries: 1,
            system: Vec::new(),
            model_roles: std::collections::HashMap::new(),
            compactor_custom: false,
        }
    }

    /// Register an auxiliary model under a role name — the loop's seam for
    /// "side tasks don't have to run on the main model".
    ///
    /// The main conversation stays pinned to one model, which keeps its
    /// provider prompt-cache prefix byte-stable; compaction summaries, memory
    /// synthesis, judging, and subagents can go to a cheaper or specialist
    /// model instead. Components read roles via [`model_for`](Self::model_for)
    /// and fall back to the main model when a role is unregistered.
    ///
    /// One role is wired automatically: registering `"compactor"` upgrades the
    /// default structural compactor to a [`ModelBackedCompactor`] on that
    /// model — unless the host already installed its own via
    /// [`with_compactor`](Self::with_compactor), which always wins regardless
    /// of call order.
    ///
    /// ```ignore
    /// let agent = AgentLoop::boxed(main)
    ///     .with_model_role("compactor", cheap.clone())
    ///     .with_model_role("judge", strong);
    /// ```
    pub fn with_model_role(mut self, role: impl Into<String>, model: Arc<dyn Model>) -> Self {
        let role = role.into();
        if role == "compactor" && !self.compactor_custom {
            self.compactor = Arc::new(crate::compactor::ModelBackedCompactor::new(model.clone()));
        }
        self.model_roles.insert(role, model);
        self
    }

    /// Look up an auxiliary model by role. `None` means "no model registered
    /// for this role — use the main model"; callers fall back rather than fail,
    /// so wiring stays optional everywhere.
    pub fn model_for(&self, role: &str) -> Option<Arc<dyn Model>> {
        self.model_roles.get(role).cloned()
    }

    /// Set a system instruction applied to every run (into `Context.system`) —
    /// e.g. "answer only via the governed tools; never claim you can't access
    /// data; never invent numbers". This is the first-class seam for a system
    /// prompt; small local models in particular need it to reliably call tools
    /// instead of refusing or hallucinating.
    pub fn with_system(mut self, text: impl Into<String>) -> Self {
        self.system = vec![Block::Text(text.into())];
        self
    }

    /// Cap context tokens below the model's own window.
    ///
    /// Compaction fires at a fraction of this, so it is also the only way to
    /// exercise the compaction path without first paying for a hundred
    /// thousand tokens of real conversation — which is how "does a long run
    /// survive being compacted" stayed untested.
    pub fn with_max_input_tokens(mut self, n: u32) -> Self {
        self.max_input_tokens = Some(n);
        self
    }

    /// Override the loop-detection policy (thresholds, or disable entirely).
    pub fn with_stuck_policy(mut self, policy: StuckPolicy) -> Self {
        self.stuck = policy;
        self
    }

    /// Opt in to (or retune) the single-tool monotony signal. Independent of
    /// [`with_stuck_policy`](Self::with_stuck_policy): both detectors run, with
    /// their own thresholds and their own `Outcome::Stuck` reason.
    ///
    /// ```ignore
    /// let agent = AgentLoop::boxed(model)
    ///     .with_monotony_policy(MonotonyPolicy { enabled: true, ..Default::default() });
    /// ```
    pub fn with_monotony_policy(mut self, policy: MonotonyPolicy) -> Self {
        self.monotony = policy;
        self
    }

    /// Override the compaction hysteresis policy. See [`CompactPolicy`].
    /// Set the ceiling on a single tool result. See [`ToolResultPolicy`].
    pub fn with_tool_result_policy(mut self, policy: ToolResultPolicy) -> Self {
        self.tool_results = policy;
        self
    }

    pub fn with_compact_policy(mut self, policy: CompactPolicy) -> Self {
        self.compaction = policy;
        self
    }

    /// Streaming is the default; pass `false` to opt out and use one
    /// non-streaming `Model::complete()` per turn. Streamed turns go through
    /// `Model::stream()`, and hooks subscribed to `Event::ModelTokenDelta` see
    /// each fragment as it arrives; the rest of the loop is unchanged.
    pub fn with_streaming(mut self, enable: bool) -> Self {
        self.streaming = enable;
        self
    }

    /// Hand the run a token the caller can cancel from anywhere.
    ///
    /// The token belongs to the **loop, not to one run**, and cancellation is
    /// **one-way**: once `cancel()` has been called, every later `run` (and
    /// every later `Session::turn`) on this loop returns `Outcome::Cancelled`
    /// immediately, with `iters: 0`. That is right for a loop built for one
    /// job and wrong for one that serves many turns. For per-turn cancellation
    /// give each turn its own token — either rebuild the loop, or keep a parent
    /// and hand out `parent.child_token()`, dropping the child when the turn
    /// ends so the parent stays live:
    ///
    /// ```ignore
    /// let token = CancellationToken::new();
    /// let agent = AgentLoop::new(model).with_cancellation(token.clone());
    /// // … later, from anywhere — this loop is now finished with, permanently:
    /// token.cancel();
    /// ```
    ///
    /// The field is public, so the simplest per-turn form is to assign
    /// `loop_.cancel = fresh_token` before each run, which is what
    /// `harness code` does.
    ///
    /// Cloning a `CancellationToken` shares it, so keep one and pass a clone.
    pub fn with_cancellation(mut self, token: CancellationToken) -> Self {
        self.cancel = token;
        self
    }

    /// Add a condition the run must satisfy before it can report success.
    pub fn with_acceptance(mut self, a: Arc<dyn Acceptance>) -> Self {
        self.acceptance.push(a);
        self
    }

    /// Replace the acceptance set outright (including the default).
    pub fn with_acceptance_set(mut self, set: Vec<Arc<dyn Acceptance>>) -> Self {
        self.acceptance = set;
        self
    }

    pub fn with_acceptance_retries(mut self, n: u32) -> Self {
        self.acceptance_retries = n;
        self
    }

    pub fn with_tool_timeout(mut self, t: Option<Duration>) -> Self {
        self.tool_timeout = t;
        self
    }

    pub fn with_compactor(mut self, c: Arc<dyn Compactor>) -> Self {
        self.compactor = c;
        // An explicit compactor always wins over the "compactor" model-role
        // convenience, in either call order.
        self.compactor_custom = true;
        self
    }

    pub fn with_tool(mut self, t: Arc<dyn harness_core::Tool>) -> Self {
        self.tools.insert(t);
        self
    }

    pub fn with_guide(mut self, g: Arc<dyn Guide>) -> Self {
        self.guides.push(g);
        self
    }

    pub fn with_sensor(mut self, s: Arc<dyn Sensor>) -> Self {
        self.sensors.push(s);
        self
    }

    pub fn with_hook(mut self, h: Arc<dyn harness_core::Hook>) -> Self {
        self.hooks.register(h);
        self
    }

    /// Pull in every `#[hook]`-registered hook.
    pub fn with_macro_hooks(mut self) -> Self {
        self.hooks = self.hooks.with_macro_hooks_take();
        self
    }

    /// Enable cross-session recall: capture every turn into `store` and
    /// register the `session_search` tool. Owner + session id are read from
    /// `world.profile.extra["recall_owner"|"recall_session"]` at run time.
    pub fn with_recall(mut self, store: Arc<dyn harness_core::RecallStore>) -> Self {
        self.tools
            .insert(Arc::new(crate::SessionSearchTool::new(store.clone())));
        self.recall = Some(store);
        self
    }

    /// Capture every turn into `store` **without** registering a search tool.
    ///
    /// Ingest and retrieval are separate concerns that [`with_recall`] happens
    /// to bundle, and the bundling is a trap: capture only ever happens when
    /// `self.recall` is set, so a host that wants a different search tool — one
    /// scoped per tenant, or one that copes with a language the backend's index
    /// does not — has no way to get the writes without also getting
    /// `session_search`, and ends up offering the model two overlapping tools
    /// to choose between.
    ///
    /// Use this, then register whichever retrieval tool suits the deployment.
    pub fn with_recall_ingest(mut self, store: Arc<dyn harness_core::RecallStore>) -> Self {
        self.recall = Some(store);
        self
    }

    /// After `with_recall`, also auto-inject top-k relevant past context at
    /// session start (off by default — tool-only is prompt-cache friendly).
    pub fn auto_inject(mut self) -> Self {
        self.recall_auto_inject = true;
        self
    }

    /// Enable the self-evolving learning loop: after a session that made
    /// `>= cfg.nudge_interval` tool calls, fork a review subagent (white-listed to
    /// `cfg.tools`) to update skills + memory from the transcript. Best-effort.
    pub fn with_learning_loop(mut self, cfg: LearningConfig) -> Self {
        self.learning = Some(cfg);
        self
    }

    /// Set the default response format for all runs through this loop. See
    /// [`ResponseFormat`]. For typed deserialisation, prefer `run_typed::<T>()`.
    pub fn with_response_format(mut self, fmt: ResponseFormat) -> Self {
        self.response_format = fmt;
        self
    }

    /// Shortcut for `with_response_format(ResponseFormat::JsonSchema { name, schema })`.
    /// Accepts a raw `serde_json::Value` so callers can hand-roll the schema or
    /// pull it from `schemars::schema_for!(T)`.
    pub fn with_response_schema(self, name: impl Into<String>, schema: serde_json::Value) -> Self {
        self.with_response_format(ResponseFormat::JsonSchema {
            name: name.into(),
            schema,
        })
    }

    pub async fn run(&self, task: Task, world: &mut World) -> Result<Outcome, HarnessError> {
        let max = harness_core::Policy::default().max_iters;
        self.run_with_max_iters(task, world, max).await
    }

    /// Run, and hand back the evidence alongside the result.
    ///
    /// The [`Receipt`] is built here rather than by the caller because the loop
    /// already knows the two things a caller would otherwise have to restate —
    /// the task and the model — and restating them is how a receipt ends up
    /// describing a different run than the one that happened. `now_ms` stays a
    /// parameter: this crate does not read the clock, so a receipt is
    /// reproducible in a test.
    pub async fn run_receipted(
        &self,
        task: Task,
        world: &mut World,
        now_ms: i64,
    ) -> Result<(Outcome, Receipt), HarnessError> {
        let description = task.description.clone();
        let handle = self.model.info().handle;
        let outcome = self.run(task, world).await?;
        let receipt = ReceiptBuilder::new(description, handle, now_ms).build(&outcome);
        Ok((outcome, receipt))
    }

    /// Advance a [`Goal`] by one phase, recording the result durably.
    ///
    /// Returns `None` when every phase is done — so a resume loop is
    /// `while let Some(..) = loop_.run_goal(..).await?`.
    ///
    /// The phase is marked `Running` and **saved before the model starts**, so
    /// a process that dies mid-run leaves a goal that says where it was rather
    /// than one that looks untouched. It is saved again afterwards on both
    /// paths. That second save on the failure path is the whole reason this
    /// method exists: written out by hand at each call site it is four lines,
    /// and the failure branch is the one that gets forgotten — which loses
    /// exactly the run you most wanted a record of.
    pub async fn run_goal(
        &self,
        goal: &mut Goal,
        store: &GoalStore,
        world: &mut World,
        now_ms: i64,
    ) -> Result<Option<(Outcome, Receipt)>, HarnessError> {
        let Some(i) = goal.start_current(now_ms) else {
            return Ok(None);
        };
        let _ = store.save(goal);

        let task = Task {
            description: goal.brief(),
            source: None,
            deadline: None,
        };
        let result = self.run_receipted(task, world, now_ms).await;

        match &result {
            Ok((_, receipt)) => {
                if receipt.passed {
                    goal.finish(i, receipt.summary(), now_ms);
                } else {
                    goal.fail(i, receipt.summary(), now_ms);
                }
            }
            // A run that errored outright still happened, and the goal has to
            // say so or a resume will retry it as though it were untouched.
            Err(e) => goal.fail(i, format!("the run errored: {e}"), now_ms),
        }
        let _ = store.save(goal);

        result.map(Some)
    }

    pub async fn run_with_max_iters(
        &self,
        task: Task,
        world: &mut World,
        max_iters: u32,
    ) -> Result<Outcome, HarnessError> {
        self.run_with_seed_history(task, Vec::new(), world, max_iters)
            .await
    }

    /// Run the agent and deserialise the terminal reply into `T`.
    ///
    /// The schema for `T` is derived via `schemars::schema_for!(T)` and
    /// installed as `ResponseFormat::JsonSchema` for this run only — any
    /// pre-existing `self.response_format` is ignored. On success the
    /// returned `T` is parsed from `Outcome::Done.text` (or, on budget
    /// exhaustion, from `Outcome::BudgetExhausted.last_text`).
    ///
    /// Errors:
    /// - `HarnessError::Other` if the model returns no text at all
    /// - `HarnessError::Other` if `serde_json::from_str::<T>(text)` fails —
    ///   the original text is included in the message for debugging.
    pub async fn run_typed<T>(&self, task: Task, world: &mut World) -> Result<T, HarnessError>
    where
        T: serde::de::DeserializeOwned + schemars::JsonSchema + 'static,
    {
        let max = harness_core::Policy::default().max_iters;
        self.run_typed_with_max_iters::<T>(task, world, max).await
    }

    /// Like `run_typed` but with explicit `max_iters`.
    pub async fn run_typed_with_max_iters<T>(
        &self,
        task: Task,
        world: &mut World,
        max_iters: u32,
    ) -> Result<T, HarnessError>
    where
        T: serde::de::DeserializeOwned + schemars::JsonSchema + 'static,
    {
        let schema_root = schemars::schema_for!(T);
        let schema = serde_json::to_value(&schema_root)
            .map_err(|e| HarnessError::Other(format!("response schema: {e}")))?;
        let name = std::any::type_name::<T>()
            .rsplit("::")
            .next()
            .unwrap_or("response")
            .to_string();
        let fmt = ResponseFormat::JsonSchema { name, schema };
        let outcome = self
            .run_with_response_format(task, world, max_iters, fmt)
            .await?;
        let text = match outcome {
            Outcome::Done { text: Some(t), .. }
            | Outcome::BudgetExhausted {
                last_text: Some(t), ..
            }
            | Outcome::Stuck {
                last_text: Some(t), ..
            }
            | Outcome::Cancelled {
                last_text: Some(t), ..
            } => t,
            Outcome::Done { text: None, .. } => {
                return Err(HarnessError::Other(
                    "run_typed: model returned no text".into(),
                ));
            }
            Outcome::Stuck {
                last_text: None, ..
            } => {
                return Err(HarnessError::Other(
                    "run_typed: agent stuck with no text".into(),
                ));
            }
            Outcome::BudgetExhausted {
                last_text: None, ..
            } => {
                return Err(HarnessError::Other(
                    "run_typed: budget exhausted with no text".into(),
                ));
            }
            Outcome::Cancelled {
                last_text: None, ..
            } => {
                return Err(HarnessError::Other(
                    "run_typed: run cancelled with no text".into(),
                ));
            }
        };
        serde_json::from_str::<T>(&text).map_err(|e| {
            HarnessError::Other(format!(
                "run_typed: decode {} failed: {e} — raw text was: {text}",
                std::any::type_name::<T>()
            ))
        })
    }

    /// Run with a one-off `ResponseFormat` override (doesn't touch `self`).
    pub async fn run_with_response_format(
        &self,
        task: Task,
        world: &mut World,
        max_iters: u32,
        fmt: ResponseFormat,
    ) -> Result<Outcome, HarnessError> {
        // Borrow checker won't let us swap `self.response_format` because
        // `self` is `&`. Easiest workaround: hand-roll the same setup that
        // `run_with_seed_history` does, but with our `fmt`. We do this by
        // calling through a private helper.
        self.run_with_seed_history_and_format(task, Vec::new(), world, max_iters, Some(fmt))
            .await
    }

    async fn run_with_seed_history_and_format(
        &self,
        task: Task,
        seed: Vec<Turn>,
        world: &mut World,
        max_iters: u32,
        fmt_override: Option<ResponseFormat>,
    ) -> Result<Outcome, HarnessError> {
        let mut ctx = Context::new(task);
        ctx.policy.max_iters = max_iters;
        ctx.tools = self.tools.schemas();
        ctx.history = seed;
        ctx.response_format = fmt_override.unwrap_or_else(|| self.response_format.clone());
        self.run_built_context(ctx, world).await
    }

    /// Like `run_with_max_iters` but seeds `ctx.history` with `seed` **before**
    /// the current user task is appended. Use this for multi-turn REPLs so
    /// prior conversation lives in `ctx.history` (where the Compactor can see
    /// it) instead of being concatenated into `task.description` (where it
    /// previously bypassed compaction entirely — see audit #2).
    pub async fn run_with_seed_history(
        &self,
        task: Task,
        seed: Vec<Turn>,
        world: &mut World,
        max_iters: u32,
    ) -> Result<Outcome, HarnessError> {
        self.run_with_seed_and_metadata(task, seed, Default::default(), world, max_iters)
            .await
    }

    /// Like [`run_with_seed_history`](Self::run_with_seed_history) but also seeds
    /// `ctx.metadata` with per-request key/values. Hooks and a
    /// [`ModelRouter`](harness_models::ModelRouter) read this map — e.g.
    /// `audit.actor` / `audit.session` for the audit trail, or
    /// `router.keep_local` to pin a request to the local model. This is the
    /// entry point a serving layer uses to pass caller identity and routing
    /// flags into a single, shared, reused loop.
    pub async fn run_with_seed_and_metadata(
        &self,
        task: Task,
        seed: Vec<Turn>,
        metadata: std::collections::BTreeMap<String, serde_json::Value>,
        world: &mut World,
        max_iters: u32,
    ) -> Result<Outcome, HarnessError> {
        let mut ctx = Context::new(task);
        ctx.policy.max_iters = max_iters;
        ctx.tools = self.tools.schemas();
        ctx.history = seed;
        ctx.metadata = metadata;
        ctx.response_format = self.response_format.clone();
        self.run_built_context(ctx, world).await
    }

    /// Start a persistent multi-turn [`Session`]. Each `turn` re-runs the loop
    /// against the accumulated history with a **stable prefix** (system +
    /// name-sorted tool schemas), so a provider's prefix cache (e.g. DeepSeek's,
    /// ~10% price on cache-hit tokens) hits across turns instead of paying full
    /// price to re-read the same bytes every round. For maximum hit rate, keep
    /// your guides' output stable (put per-turn volatile context in the message,
    /// not a recomputed system guide).
    pub fn session(&self) -> Session<'_, M> {
        Session {
            loop_: self,
            history: Vec::new(),
            max_iters: harness_core::Policy::default().max_iters,
        }
    }

    /// Inner ReAct loop on an already-prepared `Context`. Use the public
    /// `run*` methods unless you need to inject a non-standard `Context`
    /// (e.g. `run_with_response_format` does to apply a one-off
    /// `ResponseFormat` without mutating `self`).
    ///
    /// The run executes inside an `agent_run` span, and every model call and
    /// tool call inside a child span of it (`chat`, `execute_tool`). The loop
    /// owns these rather than a hook, because only the loop knows when a call
    /// really starts and ends: read-only tools are dispatched in parallel
    /// *before* their `PreToolUse` fires, so a hook timing them would see a
    /// zero-length call. Spans cost nothing without a subscriber; with an OTLP
    /// layer (feature `otel`) they are the trace waterfall.
    async fn run_built_context(
        &self,
        ctx: Context,
        world: &mut World,
    ) -> Result<Outcome, HarnessError> {
        let info = self.model.info();
        let span = tracing::info_span!(
            target: "harness.telemetry",
            "agent_run",
            "gen_ai.operation.name" = "invoke_agent",
            "gen_ai.provider.name" = %info.provider,
            "gen_ai.request.model" = %info.model,
            outcome = Empty,
            "otel.status_code" = Empty,
            "otel.status_message" = Empty,
        );
        let started = std::time::Instant::now();
        let r = self.run_in_span(ctx, world).instrument(span.clone()).await;
        let outcome = match &r {
            Ok(Outcome::Done { .. }) => "done",
            Ok(Outcome::BudgetExhausted { .. }) => "budget_exhausted",
            Ok(Outcome::Stuck { .. }) => "stuck",
            Ok(Outcome::Cancelled { .. }) => "cancelled",
            Err(e) => {
                span.record("otel.status_code", "error")
                    .record("otel.status_message", e.to_string());
                "error"
            }
        };
        span.record("outcome", outcome);
        meters::run(outcome, started.elapsed());
        r
    }

    async fn run_in_span(
        &self,
        mut ctx: Context,
        world: &mut World,
    ) -> Result<Outcome, HarnessError> {
        if ctx.system.is_empty() && !self.system.is_empty() {
            ctx.system = self.system.clone();
        }

        // Size the context budget to the model actually in use.
        //
        // Compaction fires at a fraction of `max_input_tokens`, so a value
        // unrelated to the model is wrong in one of two directions: too high and
        // the provider rejects the request before the compactor ever runs (a 32k
        // model under the 150k default would need 112,500 tokens to trigger, and
        // it cannot hold that many); too low and the loop discards history it
        // could have kept. Nothing read `ModelInfo::context_window` before this —
        // the framework's own defaults disagreed, 150,000 against 128,000.
        //
        // Only when the caller left the default in place; an explicit policy is
        // a decision and stays untouched. The output allowance is reserved,
        // because the window is shared between the prompt and the reply.
        if let Some(n) = self.max_input_tokens {
            // An explicit cap wins over both the default and the model window.
            ctx.policy.max_input_tokens = n.max(1);
        } else if ctx.policy.max_input_tokens == harness_core::Policy::default().max_input_tokens {
            let window = self.model.info().context_window;
            if window > 0 {
                // Reserve room for the reply, but never let the reservation eat
                // the window: an 8k model with the default 8k output allowance
                // would leave 0 tokens for input, and every turn — a 26-token
                // one included — would run all five compaction stages against a
                // budget of 1. Cap the reservation at a quarter of the window.
                let reserve = ctx.policy.max_output_tokens.min(window / 4);
                ctx.policy.max_input_tokens = window.saturating_sub(reserve).max(1);
            }
        }

        self.hooks.fire(
            &Event::SessionStart {
                source: SessionSource::Startup,
            },
            world,
        );

        // ── recall: resolve owner/session, ensure the session row ──
        let (recall_owner, recall_session) = if self.recall.is_some() {
            use std::sync::atomic::Ordering;
            let owner = crate::recall_owner(world);
            let session = world
                .profile
                .extra
                .get("recall_session")
                .and_then(|v| v.as_str())
                .map(|s| s.to_string())
                .unwrap_or_else(|| {
                    format!(
                        "sess-{}-{}",
                        world.clock.now_ms(),
                        RECALL_SEQ.fetch_add(1, Ordering::SeqCst)
                    )
                });
            if let Some(store) = &self.recall {
                let meta = harness_core::SessionMeta::new(&session, world.clock.now_ms());
                if let Err(e) = store.ensure_session(&owner, &session, &meta).await {
                    tracing::warn!(error = %e, "recall ensure_session failed");
                }
            }
            (owner, session)
        } else {
            (String::new(), String::new())
        };

        let recall_guide: Option<Arc<dyn Guide>> = if self.recall_auto_inject {
            if self.recall.is_none() {
                tracing::warn!(
                    "auto_inject() set but no recall store — call with_recall(store) first; skipping recall guide"
                );
                None
            } else {
                self.recall
                    .clone()
                    .map(|s| Arc::new(crate::RecallGuide::new(s)) as Arc<dyn Guide>)
            }
        } else {
            None
        };
        let all_guides: Vec<&Arc<dyn Guide>> =
            self.guides.iter().chain(recall_guide.iter()).collect();
        for g in &all_guides {
            if g.scope().matches(&ctx.task) {
                self.hooks.fire(&Event::PreGuide { guide: g.id() }, world);
                g.apply(&mut ctx, world).await?;
                self.hooks.fire(&Event::PostGuide { guide: g.id() }, world);
            }
        }

        ctx.history.push(Turn {
            role: TurnRole::User,
            blocks: vec![Block::Text(ctx.task.description.clone())],
        });

        if self.recall.is_some() {
            self.recall_append(
                &recall_owner,
                &recall_session,
                harness_core::RecallMessage::new(
                    "user",
                    ctx.task.description.clone(),
                    world.clock.now_ms(),
                ),
            )
            .await;
        }

        // Running totals — surface to caller even on BudgetExhausted.
        let mut tools_called: u32 = 0;
        let mut total_usage = harness_core::Usage::default();
        let mut last_text: Option<String> = None;

        // Stuck-detector state: the previous round's tool-call fingerprint and
        // how many consecutive rounds have repeated it.
        let mut last_fingerprint: Option<String> = None;
        let mut repeat_count: u32 = 0;
        // Monotony state: the tool that has been the round's only tool, and for
        // how many consecutive rounds. Separate counter from `repeat_count` on
        // purpose — the two detectors answer different questions.
        let mut monotone_tool: Option<String> = None;
        let mut monotone_rounds: u32 = 0;
        // Read-only calls already answered this run, cleared whenever anything
        // mutates the world. See `ToolResultPolicy::dedupe_repeats`.
        let mut answered: std::collections::HashSet<String> = std::collections::HashSet::new();
        // How many times the model has stopped mid-work with nothing to show.
        let mut acceptance_retries_left = self.acceptance_retries;
        // Consecutive turns cut off by the output-token limit with nothing to
        // act on. Such a turn is not an answer, however the loop ends.
        let mut truncated_turns: u32 = 0;

        // The contract as it stood before the model touched anything. Taken
        // now, not at verdict time, because the whole point is to compare
        // against a state the model has not had the chance to influence.
        // Set once a sealed file is found to have moved; makes the failure
        // terminal and carries the reason out to the caller.
        let mut seal_breached: Option<String> = None;
        let sealed_before: crate::seal::SealSet = {
            let paths: Vec<std::path::PathBuf> =
                self.acceptance.iter().flat_map(|a| a.seals()).collect();
            if paths.is_empty() {
                crate::seal::SealSet::default()
            } else {
                crate::seal::SealSet::capture(&world.repo.root, paths)
            }
        };

        // How far the loop actually got. Equal to `max_iters` on a normal
        // budget exhaustion, less when the wall clock stopped it first.
        let mut iters_done = ctx.policy.max_iters;
        let mut deadline_reached = false;

        for iter in 0..ctx.policy.max_iters {
            if self.cancel.is_cancelled() {
                tracing::info!(iter, "run cancelled by caller");
                self.hooks.fire(&Event::Cancelled, world);
                self.hooks.fire(&Event::SessionEnd, world);
                return Ok(Outcome::Cancelled {
                    iters: iter,
                    last_text,
                    tools_called,
                    usage: total_usage,
                });
            }

            // ── wall-clock budget ───────────────────────────────────────
            // `max_iters` bounds *steps*, which says nothing about elapsed
            // time: one iteration can be a 100ms read or a 20-minute build.
            // An unattended run is bounded in hours, not rounds, so a task
            // may carry a deadline — and reaching it goes through the same
            // forced-synthesis exit as a spent step budget, so the caller
            // gets the conclusion of the work rather than a dropped future.
            if let Some(at) = ctx.task.deadline
                && world.clock.now_ms() >= at
            {
                tracing::warn!(
                    iter,
                    deadline = at,
                    "task deadline reached — forcing a final answer"
                );
                iters_done = iter;
                deadline_reached = true;
                break;
            }
            self.hooks.fire(&Event::Heartbeat { iter }, world);

            // Compaction with hysteresis: only once over the high-water mark,
            // then escalate stage-by-stage, re-estimating after each, and stop
            // the moment we're back under target. Avoids over-compacting to the
            // lossy AutoCompact stage and avoids rewriting history (→ prefix
            // cache miss) every turn. Token estimate is calibrated against the
            // last real `input_tokens` via `CALIBRATION_KEY`.
            let mut budget = self.compactor.budget(&ctx);
            if budget.ratio() > self.compaction.high_water {
                for stage in CompactionStage::ALL {
                    if budget.ratio() <= self.compaction.target {
                        break;
                    }
                    self.hooks.fire(&Event::PreCompact { stage }, world);
                    let before = budget.used;
                    let turns_before = ctx.history.len();
                    self.compactor.compact(stage, &mut ctx).await?;
                    budget = self.compactor.budget(&ctx);
                    // Compaction used to leave no trace anywhere: not in the
                    // log, not in the transcript. That is a bad property for
                    // the one mechanism that rewrites the conversation, and it
                    // is how a compaction bug that killed long runs stayed
                    // invisible — the only symptom was a provider error with
                    // no hint of what had touched the history.
                    tracing::info!(
                        stage = ?stage,
                        tokens_before = before,
                        tokens_after = budget.used,
                        turns_before,
                        turns_after = ctx.history.len(),
                        "compacted context"
                    );
                    meters::compaction(stage, before, budget.used);
                    self.hooks.fire(
                        &Event::PostCompact {
                            stage,
                            before,
                            after: budget.used,
                        },
                        world,
                    );
                }
            }

            // Per-iteration guides — recall-style adapters that want to
            // refresh their injected context every turn (e.g. MemoryGuide
            // re-recalling against the latest user message). Default
            // `apply_before_iter` is a no-op, so this loop is cheap for
            // guides that don't override it.
            for g in &all_guides {
                if g.scope().matches(&ctx.task)
                    && let Err(e) = g.apply_before_iter(&mut ctx, world).await
                {
                    tracing::warn!(guide = %g.id(), error = %e, "apply_before_iter failed; continuing");
                }
            }

            self.hooks.fire(&Event::PreModel { ctx: &ctx }, world);
            let mut partial = String::new();
            let Some(out) = self.model_step(&ctx, world, &mut partial, iter).await? else {
                tracing::info!(iter, "run cancelled during model step");
                // What the stream delivered before the cancel is what the person
                // saw; it belongs to the outcome, not to the dropped future.
                if !partial.is_empty() {
                    last_text = Some(std::mem::take(&mut partial));
                }
                self.hooks.fire(&Event::Cancelled, world);
                self.hooks.fire(&Event::SessionEnd, world);
                return Ok(Outcome::Cancelled {
                    iters: iter + 1,
                    last_text,
                    tools_called,
                    usage: total_usage,
                });
            };
            self.hooks.fire(&Event::PostModel { out: &out }, world);

            // Calibrate the compactor against ground truth: `ctx` still holds
            // exactly what we just sent, so `budget().used` is the estimate for
            // it. Nudge the stored correction so estimate·correction ≈ the
            // model's real `input_tokens` next time. Self-correcting (converges
            // even as the raw estimate drifts); clamped so one odd turn can't
            // blow it up. This is what makes compaction fire at the right moment
            // for *this* model + language instead of a blind char heuristic.
            if out.usage.input_tokens > 0 {
                let used = self.compactor.budget(&ctx).used;
                if used > 0 {
                    let prev = ctx
                        .metadata
                        .get(CALIBRATION_KEY)
                        .and_then(|v| v.as_f64())
                        .filter(|f| f.is_finite() && *f > 0.0)
                        .unwrap_or(1.0);
                    let next =
                        (prev * out.usage.input_tokens as f64 / used as f64).clamp(0.1, 10.0);
                    ctx.metadata
                        .insert(CALIBRATION_KEY.into(), serde_json::json!(next));
                }
            }

            // Accumulate usage even if the run later exhausts budget.
            total_usage.input_tokens += out.usage.input_tokens;
            total_usage.output_tokens += out.usage.output_tokens;
            total_usage.cached_input_tokens += out.usage.cached_input_tokens;
            total_usage.cache_write_input_tokens += out.usage.cache_write_input_tokens;
            if let Some(t) = &out.text {
                last_text = Some(t.clone());
            }
            ctx.push_model_output(&out);

            if self.recall.is_some() {
                let calls = if out.tool_calls.is_empty() {
                    None
                } else {
                    serde_json::to_string(&out.tool_calls).ok()
                };
                let mut m = harness_core::RecallMessage::new(
                    "assistant",
                    out.text.clone().unwrap_or_default(),
                    world.clock.now_ms(),
                );
                m.tool_calls = calls;
                self.recall_append(&recall_owner, &recall_session, m).await;
            }

            if out.tool_calls.is_empty() && out.stop_reason == harness_core::StopReason::MaxTokens {
                // The provider cut the turn off. Whatever came out (usually
                // all reasoning, sometimes half an answer) is not the model
                // saying "done", so it must never become `Outcome::Done`.
                truncated_turns += 1;
                tracing::warn!(
                    truncated_turns,
                    max_output_tokens = ctx.policy.max_output_tokens,
                    "model turn hit the output token limit without a tool call"
                );
                // Drop the cut-off turn from the history. Nothing in it is
                // used, and keeping it costs the context a full output budget
                // of half-finished reasoning (8000 tokens on a 32k window),
                // which is what pushed a 9B run straight into compaction.
                if ctx
                    .history
                    .last()
                    .is_some_and(|t| matches!(t.role, TurnRole::Assistant))
                {
                    ctx.history.pop();
                }
                if truncated_turns <= MAX_TRUNCATED_TURNS && iter + 1 < ctx.policy.max_iters {
                    ctx.history.push(Turn {
                        role: TurnRole::User,
                        blocks: vec![Block::Text(TRUNCATED_TURN_NUDGE.into())],
                    });
                    continue;
                }
                let reason = format!(
                    "the model hit the output token limit ({} tokens) {truncated_turns} turn(s) \
                     in a row without producing an answer or a tool call",
                    ctx.policy.max_output_tokens
                );
                self.hooks.fire(&Event::SessionEnd, world);
                return Ok(Outcome::Stuck {
                    reason,
                    repeated: truncated_turns,
                    iters: iter + 1,
                    last_text: out.text.filter(|t| !t.trim().is_empty()),
                    tools_called,
                    usage: total_usage,
                });
            }
            truncated_turns = 0;

            if out.tool_calls.is_empty() {
                // The model stopping is its opinion that the work is done.
                // Before taking it as fact, run whatever the caller said
                // "done" actually means. A failure goes back as an instruction
                // and the loop carries on; the retry cap keeps a model that
                // ignores corrections from eating the budget.
                let mut verdict: Option<Verdict> = None;
                if !self.acceptance.is_empty() {
                    // Record this turn first — the checks read the transcript.
                    let mut probe = ctx.clone();
                    probe.history.push(Turn {
                        role: TurnRole::Assistant,
                        blocks: out
                            .text
                            .as_deref()
                            .filter(|t| !t.trim().is_empty())
                            .map(|t| vec![Block::Text(t.to_string())])
                            .unwrap_or_default(),
                    });
                    for check in &self.acceptance {
                        let v = check.check(&probe, world).await;
                        self.hooks.fire(
                            &Event::AcceptanceChecked {
                                name: check.name(),
                                passed: v.passed,
                                reason: &v.reason,
                            },
                            world,
                        );
                        if !v.passed {
                            tracing::info!(
                                check = check.name(),
                                reason = %v.reason,
                                "acceptance failed"
                            );
                            verdict = Some(v);
                            break;
                        }
                    }
                    // Everything passed. Say so explicitly: "checked, and it
                    // holds up" is a different claim from "nobody looked", and
                    // the host has to be able to tell them apart.
                    verdict = verdict.or_else(|| Some(Verdict::passed()));
                }

                // A pass is only worth the contract it was measured against.
                // Re-read the sealed files and refuse if any moved.
                if !sealed_before.is_empty() && verdict.as_ref().is_some_and(|v| v.passed) {
                    let paths: Vec<std::path::PathBuf> =
                        self.acceptance.iter().flat_map(|a| a.seals()).collect();
                    let now = crate::seal::SealSet::capture(&world.repo.root, paths);
                    let breaches = sealed_before.breaches(&now);
                    if !breaches.is_empty() {
                        let what = breaches
                            .iter()
                            .map(|b| b.describe())
                            .collect::<Vec<_>>()
                            .join("; ");
                        tracing::error!(
                            breaches = %what,
                            "acceptance contract changed during the run — refusing the pass"
                        );
                        self.hooks
                            .fire(&Event::SealBreached { detail: &what }, world);
                        // Deliberately NOT a retry. Every other acceptance
                        // failure is handed back as an instruction because the
                        // model can act on it; this one would be handing back
                        // "you edited the gate", to the party that edited it,
                        // with the file still writable. The run is over.
                        seal_breached = Some(what);
                        verdict = Some(Verdict::failed(
                            "the acceptance contract was modified during this run",
                        ));
                    }
                }

                if let Some(v) = verdict.clone().filter(|v| !v.passed)
                    && seal_breached.is_none()
                    && acceptance_retries_left > 0
                    && iter + 1 < ctx.policy.max_iters
                {
                    acceptance_retries_left -= 1;
                    ctx.history.push(Turn {
                        role: TurnRole::User,
                        blocks: vec![Block::Text(v.reason)],
                    });
                    continue;
                }

                self.hooks.fire(&Event::TaskCompleted, world);
                self.hooks.fire(&Event::SessionEnd, world);
                self.run_learning_review(&ctx, world, tools_called).await;
                // Thinking models (e.g. Qwen3 via Ollama) sometimes emit the
                // whole answer into the reasoning channel and leave `text`
                // empty. Fall back to the reasoning so the turn isn't blank —
                // but `verified` says whether anyone agreed it was done.
                let text = out
                    .text
                    .filter(|t| !t.trim().is_empty())
                    .or_else(|| out.reasoning.filter(|r| !r.trim().is_empty()));
                return Ok(Outcome::Done {
                    text,
                    iters: iter + 1,
                    tools_called,
                    usage: total_usage,
                    verified: verdict,
                    contract: sealed_before.clone(),
                    seal_breach: seal_breached,
                });
            }

            // Stuck detection runs *after* dispatch — see the block below the
            // tool loop, which needs this round's results to judge progress.

            // Parallel-safe prefetch: dispatch the *leading run* of read-only
            // tool calls concurrently (a mutating tool is a serial barrier).
            // The sequential loop below still processes every call in order —
            // hooks, sensors, and history stay ordered — only the dispatch IO
            // overlaps. Reads before any write are safe; anything at/after the
            // first mutating call runs on the normal path.
            // Digest of what each tool handed back this round, for the stuck
            // check below. Digests rather than the payloads: results are
            // already capped, but a per-round copy of every one of them is
            // still a cost the detector does not need to pay.
            let mut round_results: Vec<u64> = Vec::new();

            let mut prefetched: HashMap<String, Dispatched> = HashMap::new();
            {
                let lead: Vec<&_> = out
                    .tool_calls
                    .iter()
                    .take_while(|c| {
                        self.tools.risk(&c.name) == Some(harness_core::ToolRisk::ReadOnly)
                    })
                    .collect();
                if lead.len() > 1 {
                    let futs = lead.iter().map(|c| {
                        let mut w = world.clone();
                        let action = Action {
                            tool: c.name.clone(),
                            call_id: c.id.clone(),
                            args: c.args.clone(),
                        };
                        async move {
                            let r = self.dispatch_bounded(&action, &mut w).await;
                            (action.call_id, r)
                        }
                    });
                    let results = tokio::select! {
                        biased;
                        _ = self.cancel.cancelled() => Vec::new(),
                        rs = futures::future::join_all(futs) => rs,
                    };
                    for (id, r) in results {
                        prefetched.insert(id, r);
                    }
                }
            }

            for call in &out.tool_calls {
                let action = Action {
                    tool: call.name.clone(),
                    call_id: call.id.clone(),
                    args: call.args.clone(),
                };

                // PreToolUse hook can deny destructive actions
                if let HookOutcome::Deny { reason } = self
                    .hooks
                    .fire(&Event::PreToolUse { action: &action }, world)
                {
                    ctx.history.push(Turn {
                        role: TurnRole::Tool,
                        blocks: vec![Block::ToolResult {
                            call_id: action.call_id.clone(),
                            content: serde_json::json!({
                                "ok": false,
                                "denied_by_hook": reason,
                            }),
                        }],
                    });
                    if self.recall.is_some() {
                        self.recall_append(
                            &recall_owner,
                            &recall_session,
                            harness_core::RecallMessage::new(
                                "tool",
                                format!("[denied by hook] {reason}"),
                                world.clock.now_ms(),
                            )
                            .with_tool_name(action.tool.clone()),
                        )
                        .await;
                    }
                    continue;
                }

                // Use the concurrently-prefetched result if we have one;
                // otherwise dispatch now.
                let Dispatched {
                    result,
                    started_ms,
                    finished_ms,
                } = if let Some(r) = prefetched.remove(&action.call_id) {
                    r
                } else {
                    self.dispatch_bounded(&action, world).await
                };
                tools_called += 1;

                if self.cancel.is_cancelled() {
                    tracing::info!(iter, "run cancelled during tool dispatch");
                    self.hooks.fire(&Event::Cancelled, world);
                    self.hooks.fire(&Event::SessionEnd, world);
                    return Ok(Outcome::Cancelled {
                        iters: iter + 1,
                        last_text,
                        tools_called,
                        usage: total_usage,
                    });
                }

                // Decide the final payload *before* announcing the result, so
                // hooks, telemetry and the context all describe the same thing:
                // an audit that logs a 200 KB blob the model never saw is not an
                // audit of what happened.
                let result = ToolResult {
                    content: self.shape_result(&action, &result, &mut answered, &world.repo.root),
                    ..result
                };
                self.hooks.fire(
                    &Event::ToolTimed {
                        action: &action,
                        started_ms,
                        finished_ms,
                    },
                    world,
                );
                self.hooks.fire(
                    &Event::PostToolUse {
                        action: &action,
                        result: &result,
                    },
                    world,
                );

                round_results.push(result_digest(&action.tool, &result.content));

                ctx.history.push(Turn {
                    role: TurnRole::Tool,
                    blocks: vec![Block::ToolResult {
                        call_id: action.call_id.clone(),
                        content: result.content.clone(),
                    }],
                });

                if self.recall.is_some() {
                    let body = serde_json::to_string(&result.content).unwrap_or_default();
                    self.recall_append(
                        &recall_owner,
                        &recall_session,
                        harness_core::RecallMessage::new("tool", body, world.clock.now_ms())
                            .with_tool_name(action.tool.clone()),
                    )
                    .await;
                }

                // run self-correct sensors
                let mut all_signals = Vec::new();
                for s in &self.sensors {
                    if s.stage() != Stage::SelfCorrect {
                        continue;
                    }
                    self.hooks.fire(&Event::PreSensor { sensor: s.id() }, world);
                    let sigs = s.observe(&action, world).await.unwrap_or_else(|e| {
                        tracing::warn!(?e, "sensor failed");
                        Vec::new()
                    });
                    self.hooks.fire(
                        &Event::PostSensor {
                            sensor: s.id(),
                            signals: &sigs,
                        },
                        world,
                    );
                    all_signals.extend(sigs);
                }
                if !all_signals.is_empty() {
                    let bundle = SignalSet::new(all_signals);
                    let (patches, remaining) = bundle.partition_auto_fix();

                    // audit #7: each patch goes through PreAutoFix.
                    // Hooks can Deny (skip silently). Default safelist on
                    // RunCommand catches the obvious misuses with no hook.
                    let approved: Vec<harness_core::FixPatch> = patches.into_iter().filter(|p| {
                        if !is_default_safe_fix(p) {
                            tracing::warn!(?p, "auto-fix rejected by default safelist (use PreAutoFix hook to override)");
                            self.hooks.fire(&Event::PostAutoFix { patch: p, applied: false }, world);
                            return false;
                        }
                        match self.hooks.fire(&Event::PreAutoFix { patch: p }, world) {
                            HookOutcome::Deny { reason } => {
                                tracing::warn!(?p, %reason, "auto-fix denied by hook");
                                self.hooks.fire(&Event::PostAutoFix { patch: p, applied: false }, world);
                                false
                            }
                            _ => true,
                        }
                    }).collect();

                    let applied = apply_patches(&approved, world).await;
                    // Emit PostAutoFix for each approved patch with the application result.
                    for (i, p) in approved.iter().enumerate() {
                        self.hooks.fire(
                            &Event::PostAutoFix {
                                patch: p,
                                applied: i < applied.len(),
                            },
                            world,
                        );
                    }
                    if !applied.is_empty() {
                        ctx.push_feedback(vec![harness_core::Signal {
                            severity: harness_core::Severity::Hint,
                            origin: "auto-fix".into(),
                            message: format!(
                                "applied {} auto-fix patch(es): {applied:?}",
                                applied.len()
                            ),
                            agent_hint: Some(
                                "re-check the affected files before continuing".into(),
                            ),
                            auto_fix: None,
                            location: None,
                        }]);
                    }
                    if remaining.has_blocking() {
                        ctx.push_feedback(remaining.signals);
                    }
                }
            }

            // ── stuck detection ─────────────────────────────────────────
            // A round repeats when the request AND what came back were both
            // identical. The same call whose *result moved* is the model
            // watching something progress — polling a build, a background
            // job, a queue — not spinning, which is why this runs after
            // dispatch: before it, this round's results do not exist yet and
            // the check could only compare the request. Judging on the
            // request alone aborted runs that were making progress, and a
            // long unattended task is mostly made of such waits.
            //
            // A poll whose result never changes still counts, and still
            // aborts: burning model calls on a thing that is not moving is
            // exactly what this guard is for. What bounds a *legitimate*
            // long wait is `max_iters`, not this.
            if self.stuck.enabled {
                let fp = format!(
                    "{}#{}",
                    tool_call_fingerprint(&out.tool_calls),
                    round_results
                        .iter()
                        .map(u64::to_string)
                        .collect::<Vec<_>>()
                        .join(",")
                );
                if last_fingerprint.as_ref() == Some(&fp) {
                    repeat_count += 1;
                } else {
                    repeat_count = 1;
                    last_fingerprint = Some(fp);
                }

                if repeat_count >= self.stuck.abort_after {
                    let reason = format!(
                        "repeated the same tool call {repeat_count}× with an unchanging result"
                    );
                    tracing::warn!(repeated = repeat_count, "stuck: aborting run");
                    self.hooks.fire(&Event::SessionEnd, world);
                    return Ok(Outcome::Stuck {
                        reason,
                        repeated: repeat_count,
                        iters: iter + 1,
                        last_text,
                        tools_called,
                        usage: total_usage,
                    });
                }

                if repeat_count == self.stuck.nudge_after {
                    tracing::warn!(
                        repeated = repeat_count,
                        "stuck: nudging model to change approach"
                    );
                    ctx.push_feedback(vec![harness_core::Signal {
                        severity: harness_core::Severity::Warn,
                        origin: "stuck-detector".into(),
                        message: format!(
                            "You have issued the same tool call {repeat_count} rounds in a row \
                             and the result has not changed."
                        ),
                        agent_hint: Some(
                            "Stop repeating it. Inspect the actual tool result/error, try a \
                             different approach, or give your final answer with no tool call. \
                             If you are waiting on something, wait inside one call rather than \
                             polling in a tight loop."
                                .into(),
                        ),
                        auto_fix: None,
                        location: None,
                    }]);
                }
            }

            // ── monotony detection ──────────────────────────────────────
            // The second signal, and deliberately not part of the block above:
            // that one asks "did this round repeat?", this one asks "has the
            // model been working a single tool and nothing else?". A guessing
            // spiral answers no to the first and yes to the second, which is
            // why the byte-identical policy stays exactly as strict as it is.
            //
            // Off unless the host asked for it — see `MonotonyPolicy` for why
            // no threshold is safe by default. Independent of `stuck.enabled`,
            // so a host can run either, both, or neither.
            if self.monotony.enabled {
                match sole_tool(&out.tool_calls) {
                    // Same single tool as last round: the run continues.
                    Some(name) if monotone_tool.as_deref() == Some(name) => monotone_rounds += 1,
                    // A different single tool: a new run of one.
                    Some(name) => {
                        monotone_tool = Some(name.to_string());
                        monotone_rounds = 1;
                    }
                    // A mixed round. Reaching for a second tool is movement,
                    // and movement is what this detector is looking for.
                    None => {
                        monotone_tool = None;
                        monotone_rounds = 0;
                    }
                }

                if monotone_rounds >= self.monotony.abort_after {
                    let name = monotone_tool.clone().unwrap_or_default();
                    let reason = format!(
                        "called `{name}` for {monotone_rounds} consecutive rounds \
                         without using any other tool"
                    );
                    tracing::warn!(tool = %name, rounds = monotone_rounds, "monotony: aborting run");

                    // Force the final answer before returning. A model caught
                    // here is by definition one that will not stop calling
                    // tools on its own, so terminating it without asking for
                    // text hands the caller the same empty partial answer the
                    // detector exists to prevent — earlier, and therefore
                    // worse. Stripping the tools and asking once is the exit
                    // the model could not find; it costs one model call and no
                    // prompt bytes at all. (The budget-exhausted path has done
                    // this since it existed; this is the same exit, reached
                    // sooner.)
                    match self
                        .force_final_synthesis(
                            &mut ctx,
                            world,
                            &mut total_usage,
                            MONOTONY_SYNTHESIS_PROMPT,
                        )
                        .await
                    {
                        Synthesis::Text(Some(t)) => last_text = Some(t),
                        Synthesis::Text(None) => {}
                        // Stop means stop: the synthesis is a model call like
                        // any other, and the run owes the caller a `Cancelled`
                        // rather than the `Stuck` it was on its way to.
                        Synthesis::Cancelled => {
                            tracing::info!(iter, "run cancelled during final synthesis");
                            self.hooks.fire(&Event::Cancelled, world);
                            self.hooks.fire(&Event::SessionEnd, world);
                            return Ok(Outcome::Cancelled {
                                iters: iter + 1,
                                last_text,
                                tools_called,
                                usage: total_usage,
                            });
                        }
                    }

                    self.hooks.fire(&Event::SessionEnd, world);
                    return Ok(Outcome::Stuck {
                        reason,
                        repeated: monotone_rounds,
                        iters: iter + 1,
                        last_text,
                        tools_called,
                        usage: total_usage,
                    });
                }

                if monotone_rounds == self.monotony.nudge_after {
                    let name = monotone_tool.clone().unwrap_or_default();
                    tracing::warn!(
                        tool = %name,
                        rounds = monotone_rounds,
                        "monotony: nudging model to change approach"
                    );
                    ctx.push_feedback(vec![harness_core::Signal {
                        severity: harness_core::Severity::Warn,
                        origin: "monotony-detector".into(),
                        message: format!(
                            "You have called `{name}` for {monotone_rounds} rounds in a row \
                             and used no other tool. A different argument each time is not \
                             the same thing as making progress."
                        ),
                        agent_hint: Some(
                            "If you are guessing at arguments to see which one lands, stop: \
                             report what you did find and say plainly what you could not \
                             determine. Otherwise reach for a different tool, or give your \
                             final answer with no tool call."
                                .into(),
                        ),
                        auto_fix: None,
                        location: None,
                    }]);
                }
            }
        }
        // ── Budget exhausted ─────────────────────────────────────────
        // Force a final synthesis pass with tools DISABLED. Otherwise the
        // model often spins on tool calls right up to the budget cap and
        // never emits a text conclusion, leaving the caller with nothing
        // but `last_text` from some earlier intermediate turn (or None).
        //
        // The synthesis call is "free" — it costs one extra model call
        // beyond max_iters but doesn't count toward `iters`. The result
        // lands in `last_text` so callers display it as the answer.
        //
        // Signal to any observer (LiveProgressHook, SessionRecorder, custom
        // hooks) that we've used 100% of the budget and are about to force
        // synthesis. Pre-existing `BudgetWarning` event was unused; this is
        // its natural home.
        self.hooks.fire(&Event::BudgetWarning { ratio: 1.0 }, world);
        match self
            .force_final_synthesis(&mut ctx, world, &mut total_usage, BUDGET_SYNTHESIS_PROMPT)
            .await
        {
            Synthesis::Text(Some(t)) => last_text = Some(t),
            Synthesis::Text(None) => {}
            // A cancel that lands here has to win: this is the last model call
            // of the run, and it is made after the loop's last cancel check.
            // No learning review either — it is one more model call, and the
            // other cancel exits all skip it.
            Synthesis::Cancelled => {
                tracing::info!(iter = iters_done, "run cancelled during final synthesis");
                self.hooks.fire(&Event::Cancelled, world);
                self.hooks.fire(&Event::SessionEnd, world);
                return Ok(Outcome::Cancelled {
                    iters: iters_done,
                    last_text,
                    tools_called,
                    usage: total_usage,
                });
            }
        }

        self.hooks.fire(&Event::SessionEnd, world);
        self.run_learning_review(&ctx, world, tools_called).await;
        Ok(Outcome::BudgetExhausted {
            iters: iters_done,
            last_text,
            tools_called,
            usage: total_usage,
            deadline_reached,
        })
    }

    /// One model call, racing the run's cancellation token.
    ///
    /// `None` means the token fired first. Dropping the un-awaited future is
    /// what cancels the underlying HTTP request (reqwest aborts on drop) or
    /// the SSE stream, so the model stops generating rather than finishing
    /// into a void. `biased` so a token already cancelled wins a tie.
    ///
    /// `partial` collects streamed text so a cancel can hand it to the outcome.
    async fn model_step(
        &self,
        ctx: &Context,
        world: &mut World,
        partial: &mut String,
        iter: u32,
    ) -> Result<Option<ModelOutput>, HarnessError> {
        let info = self.model.info();
        let span = chat_span(&info);
        let started = std::time::Instant::now();
        let r = async {
            tokio::select! {
                biased;
                _ = self.cancel.cancelled() => Ok(None),
                r = self.model_call(ctx, world, partial, iter) => r.map(Some),
            }
        }
        .instrument(span.clone())
        .await;
        let end = match &r {
            Ok(Some(out)) => Some(Ok(out)),
            Ok(None) => None,
            Err(e) => Some(Err(e.to_string())),
        };
        finish_chat(&span, &info, started, end);
        r
    }

    /// One model request: streamed unless the loop was opted out with
    /// `with_streaming(false)`.
    async fn model_call(
        &self,
        ctx: &Context,
        world: &mut World,
        partial: &mut String,
        iter: u32,
    ) -> Result<ModelOutput, HarnessError> {
        let mut attempt: u32 = 0;
        loop {
            let r = if self.streaming {
                self.complete_via_stream(ctx, world, partial, iter).await
            } else {
                self.model
                    .complete(ctx)
                    .await
                    .map_err(harness_core::HarnessError::Model)
            };
            // A transport failure (connection dropped, stream cut before the
            // model finished) says nothing about the request itself, so the
            // turn is asked again from the start rather than failing the run.
            match r {
                Err(HarnessError::Model(harness_core::ModelError::Transport(msg)))
                    if attempt < MODEL_TRANSPORT_RETRIES =>
                {
                    attempt += 1;
                    tracing::warn!(
                        attempt,
                        error = %msg,
                        "model call failed in transport; retrying the turn"
                    );
                    partial.clear();
                    tokio::time::sleep(Duration::from_secs(u64::from(attempt) * 2)).await;
                }
                other => return other,
            }
        }
    }

    /// Drive `Model::stream()` and assemble the result into a `ModelOutput`,
    /// firing `Event::ModelTokenDelta` for each text fragment along the way.
    ///
    /// Adapters without a real `stream()` fall back to the default trait impl,
    /// which runs `complete()` and emits the whole reply as a single delta, so
    /// streaming is safe regardless of which provider the user picked. The
    /// assembly itself is [`harness_core::collect_stream_observed`], shared
    /// with the adapters' own `complete()`.
    ///
    /// While the stream runs, a `model_progress` custom event fires about every
    /// 5s and a `Heartbeat` about every 30s, both driven by arriving deltas.
    ///
    /// `sink` is where the text accumulates, owned by the caller so that a
    /// cancel — which drops this future mid-stream — does not drop the chunks
    /// already delivered along with it. A completed stream leaves it empty.
    ///
    /// Tool-call deltas need no such rescue: they fire no hook, and no dispatch
    /// happens until after the output is pushed, so a cancel mid-tool-call
    /// loses nothing anyone observed.
    async fn complete_via_stream(
        &self,
        ctx: &Context,
        world: &mut World,
        sink: &mut String,
        iter: u32,
    ) -> Result<ModelOutput, HarnessError> {
        let stream = self
            .model
            .stream(ctx)
            .await
            .map_err(harness_core::HarnessError::Model)?;
        let started = std::time::Instant::now();
        let mut last_progress = started;
        let mut last_heartbeat = started;
        let (mut chars, mut deltas) = (0usize, 0usize);
        let hooks = &self.hooks;
        let out = harness_core::collect_stream_observed(stream, |delta| {
            deltas += 1;
            match delta {
                ModelDelta::Text(t) if !t.is_empty() => {
                    chars += t.chars().count();
                    hooks.fire(&Event::ModelTokenDelta { text: t }, world);
                    sink.push_str(t);
                }
                ModelDelta::Reasoning(t) => chars += t.chars().count(),
                ModelDelta::ToolCallArgs { partial_json, .. } => {
                    chars += partial_json.chars().count()
                }
                _ => {}
            }
            // Driven by arriving deltas, so a heartbeat means "bytes are
            // still flowing", and its absence in a trace means stuck.
            let now = std::time::Instant::now();
            if now.duration_since(last_heartbeat) >= MODEL_HEARTBEAT_EVERY {
                last_heartbeat = now;
                hooks.fire(&Event::Heartbeat { iter }, world);
            }
            if now.duration_since(last_progress) >= MODEL_PROGRESS_EVERY {
                last_progress = now;
                let elapsed = now.duration_since(started).as_secs_f64();
                let data = serde_json::json!({
                    "elapsed_s": elapsed,
                    "chars": chars,
                    "deltas": deltas,
                    "chars_per_s": if elapsed > 0.0 { chars as f64 / elapsed } else { 0.0 },
                });
                hooks.fire(
                    &Event::Custom {
                        name: "model_progress",
                        data: &data,
                    },
                    world,
                );
            }
        })
        .await
        .map_err(harness_core::HarnessError::Model)?;
        // Completed: the output now owns the text; only a cancel needs the sink.
        sink.clear();
        Ok(out)
    }

    /// One tool call, under the per-call deadline and the run's cancellation
    /// token. A timeout becomes an error *result* — the model sees it and routes
    /// around it — never a hung run. A cancel drops the tool's future unpolled
    /// and returns a result the loop discards before it reaches history, so the
    /// model never sees that one at all. Errors are folded the same way: the
    /// loop's contract is that a tool call always produces a result turn.
    ///
    /// Dropping the future stops *this* side of the work. What it stops on the
    /// other side depends on the tool: an HTTP request aborts, a `spawn_blocking`
    /// file write runs to completion, and a child process is only killed if its
    /// runner asked for `kill_on_drop`.
    ///
    /// Runs in an `execute_tool` span, so parallel calls show up as overlapping
    /// bars in a trace and a failed one is marked as an error.
    async fn dispatch_bounded(&self, action: &Action, world: &mut World) -> Dispatched {
        let span = tracing::info_span!(
            target: "harness.telemetry",
            "execute_tool",
            "otel.name" = %format!("execute_tool {}", action.tool),
            "otel.status_code" = Empty,
            "otel.status_message" = Empty,
            "gen_ai.operation.name" = "execute_tool",
            "gen_ai.tool.name" = %action.tool,
            "gen_ai.tool.call.id" = %action.call_id,
            ok = Empty,
        );
        let clock = world.clock.clone();
        let started_ms = clock.now_ms();
        let started = std::time::Instant::now();
        let r = self
            .dispatch_bounded_in_span(action, world)
            .instrument(span.clone())
            .await;
        let finished_ms = clock.now_ms();
        meters::tool_call(&action.tool, r.ok, started.elapsed());
        span.record("ok", r.ok);
        if !r.ok {
            span.record("otel.status_code", "error");
            if let Some(msg) = r.content.get("error").and_then(|v| v.as_str()) {
                span.record("otel.status_message", msg);
            }
        }
        Dispatched {
            result: r,
            started_ms,
            finished_ms,
        }
    }

    async fn dispatch_bounded_in_span(&self, action: &Action, world: &mut World) -> ToolResult {
        let fut = self.tools.dispatch(action, world);
        let bounded = async {
            match self.tool_timeout {
                Some(deadline) => match tokio::time::timeout(deadline, fut).await {
                    Ok(r) => r,
                    Err(_) => {
                        tracing::warn!(
                            target: "harness.telemetry",
                            event = "tool.deadline",
                            "gen_ai.tool.name" = %action.tool,
                            seconds = deadline.as_secs(),
                        );
                        Ok(ToolResult {
                            ok: false,
                            content: serde_json::json!({
                                "error": format!(
                                    "tool call exceeded its {}s deadline and was cancelled; \
                                     the operation may be too broad — narrow it or try a \
                                     different approach",
                                    deadline.as_secs()
                                ),
                                "timeout": true,
                            }),
                            trace: None,
                        })
                    }
                },
                None => fut.await,
            }
        };
        // `biased` so that a token already cancelled wins even when the tool
        // future is also ready — the caller said stop, and a result produced
        // after that would be work the run is about to throw away anyway.
        let dispatched = tokio::select! {
            biased;
            _ = self.cancel.cancelled() => {
                tracing::info!("gen_ai.tool.name" = %action.tool, "tool call dropped: run cancelled");
                return ToolResult {
                    ok: false,
                    content: serde_json::json!({"error": "run cancelled", "cancelled": true}),
                    trace: None,
                };
            }
            r = bounded => r,
        };
        dispatched.unwrap_or_else(|e| ToolResult {
            ok: false,
            content: serde_json::json!({"error": e.to_string()}),
            trace: None,
        })
    }

    /// What a tool result contributes to the context: repeat suppression first,
    /// then the size ceiling. A repeat that is also oversized collapses to the
    /// pointer rather than to a truncated copy of what the model already holds.
    fn shape_result(
        &self,
        action: &Action,
        result: &ToolResult,
        answered: &mut std::collections::HashSet<String>,
        root: &std::path::Path,
    ) -> serde_json::Value {
        if !(self.tool_results.dedupe_repeats && result.ok) {
            return self.cap_result(action, &result.content, root);
        }
        match self.tools.risk(&action.tool) {
            Some(harness_core::ToolRisk::ReadOnly) => {
                let fp = format!("{}({})", action.tool, action.args);
                if answered.contains(&fp) {
                    tracing::info!(
                        target: "harness.telemetry",
                        event = "tool.result.repeat",
                        "gen_ai.tool.name" = %action.tool,
                    );
                    serde_json::json!({
                        "repeat_of_earlier_call": true,
                        "tool": action.tool,
                        "note": "You already made this exact call in this run and nothing has \
                                 changed the workspace since. The earlier result above still \
                                 stands — use it rather than asking again.",
                    })
                } else {
                    answered.insert(fp);
                    self.cap_result(action, &result.content, root)
                }
            }
            // A write invalidates every earlier read.
            _ => {
                answered.clear();
                self.cap_result(action, &result.content, root)
            }
        }
    }

    /// Enforce [`ToolResultPolicy`] on one result before it reaches the context.
    ///
    /// Over the ceiling, the guard prefers to *spill*: full payload to a file
    /// inside the workspace, bounded preview plus the path inline — nothing is
    /// lost, and the model retrieves slices with the file tools it already has.
    /// Only when spilling is off (or the write fails) does it fall back to
    /// destructive truncation: a byte-level cut handed back as a marker object
    /// rather than mangled JSON, saying how much was dropped and what to do
    /// instead.
    fn cap_result(
        &self,
        action: &Action,
        content: &serde_json::Value,
        root: &std::path::Path,
    ) -> serde_json::Value {
        let Some(max) = self.tool_results.max_bytes else {
            return content.clone();
        };
        let serialized = content.to_string();
        if serialized.len() <= max {
            return content.clone();
        }
        if self.tool_results.spill
            && let Some(marker) = spill_oversized(action, content, &serialized, root)
        {
            return marker;
        }
        // Cut on a char boundary so the kept head is valid UTF-8.
        let mut end = max;
        while end > 0 && !serialized.is_char_boundary(end) {
            end -= 1;
        }
        tracing::warn!(
            target: "harness.telemetry",
            event = "tool.result.truncated",
            "gen_ai.tool.name" = %action.tool,
            bytes = serialized.len(),
            max_bytes = max,
        );
        serde_json::json!({
            "truncated": true,
            "tool": action.tool,
            "bytes_total": serialized.len(),
            "bytes_kept": end,
            "head": serialized[..end],
            "note": format!(
                "This result was {} bytes and was cut to {} to protect the context window. \
                 Do not ask for it again unchanged — narrow it: request a smaller range, \
                 a filter, or a specific field.",
                serialized.len(), end
            ),
        })
    }

    async fn recall_append(&self, owner: &str, session: &str, msg: harness_core::RecallMessage) {
        if let Some(store) = &self.recall
            && let Err(e) = store.append(owner, session, &msg).await
        {
            tracing::warn!(error = %e, "recall append failed");
        }
    }

    /// Best-effort post-session review. Never affects the finished run.
    async fn run_learning_review(&self, ctx: &Context, world: &mut World, tools_called: u32) {
        let Some(cfg) = &self.learning else { return };
        if tools_called < cfg.nudge_interval {
            return;
        }
        let transcript = crate::render_transcript(&ctx.history, 12_000);
        let task = harness_core::Task {
            description: format!(
                "{}\n\n## Conversation transcript\n{}",
                cfg.review_prompt, transcript
            ),
            source: None,
            deadline: None,
        };
        let mut spec =
            crate::SubagentSpec::new("learning-review", task).with_max_iters(cfg.max_iters);
        for t in &cfg.tools {
            spec = spec.with_tool(t.clone());
        }
        let sub = crate::Subagent::new(harness_core::DynModel(cfg.review_model.clone()), spec);
        // Box::pin breaks the recursive async-future cycle: AgentLoop<M> →
        // run_learning_review → Subagent<DynModel>::run →
        // AgentLoop<Arc<dyn Model>>::run_built_context. Without pinning the
        // compiler rejects the infinite-sized future.
        if let Err(e) = Box::pin(sub.run(world)).await {
            tracing::warn!(error = %e, "learning review failed");
        }
    }

    /// One final model call with tools removed, asking it to write the
    /// best-effort conclusion from whatever it has already gathered.
    ///
    /// Errors from the model are swallowed — observability is best-effort
    /// here, and a transport blip during synthesis should not turn a
    /// near-complete run into a hard failure.
    ///
    /// The call is raced against the run's cancellation token like every
    /// other model call: a cancel before or during it produces no text,
    /// leaves the context as it was, and is reported as
    /// [`Synthesis::Cancelled`] so the caller can return `Outcome::Cancelled`
    /// instead of pretending the run ended on its own terms.
    ///
    /// `prompt` says *why* the loop stopped granting tool calls. Budget
    /// exhaustion is not the only way a run can end with the model still
    /// reaching for tools — a detected loop is another — and the model writes a
    /// better final answer when it is told which one happened.
    async fn force_final_synthesis(
        &self,
        ctx: &mut Context,
        world: &mut World,
        total_usage: &mut harness_core::Usage,
        prompt: &str,
    ) -> Synthesis {
        // Snapshot + clear tool schemas so the model has no choice but text.
        let saved_tools = std::mem::take(&mut ctx.tools);
        ctx.history.push(Turn {
            role: TurnRole::User,
            blocks: vec![Block::Text(prompt.into())],
        });

        self.hooks.fire(&Event::PreModel { ctx }, world);
        // `biased` with the token first means an already-cancelled token wins
        // the race outright, so no separate `is_cancelled()` pre-check is
        // needed here — the select is the check.
        let info = self.model.info();
        let span = chat_span(&info);
        let started = std::time::Instant::now();
        let mut partial = String::new();
        let iter = ctx.policy.max_iters;
        let result = tokio::select! {
            biased;
            _ = self.cancel.cancelled() => None,
            r = self
                .model_call(ctx, world, &mut partial, iter)
                .instrument(span.clone()) => Some(r),
        };
        finish_chat(
            &span,
            &info,
            started,
            result
                .as_ref()
                .map(|r| r.as_ref().map_err(|e| e.to_string())),
        );
        ctx.tools = saved_tools;

        let Some(result) = result else {
            // Drop the synthesis prompt pushed above. Nothing answered it, and
            // leaving it would hand the next reader of this context a dangling
            // user turn.
            ctx.history.pop();
            return Synthesis::Cancelled;
        };

        match result {
            Ok(out) => {
                self.hooks.fire(&Event::PostModel { out: &out }, world);
                total_usage.input_tokens += out.usage.input_tokens;
                total_usage.output_tokens += out.usage.output_tokens;
                total_usage.cached_input_tokens += out.usage.cached_input_tokens;
                total_usage.cache_write_input_tokens += out.usage.cache_write_input_tokens;
                ctx.push_model_output(&out);
                Synthesis::Text(out.text)
            }
            Err(_) => Synthesis::Text(None),
        }
    }
}

/// What the forced final synthesis produced.
enum Synthesis {
    /// The model answered (or errored — errors are swallowed, see the fn doc).
    Text(Option<String>),
    /// The run's token fired before or during the call; no text was produced
    /// and nothing was pushed to the context.
    Cancelled,
}

/// A persistent multi-turn conversation over one [`AgentLoop`].
///
/// Holds the append-only history and, on each [`turn`](Session::turn), re-runs
/// the loop against a **stable prefix** (system + name-sorted tool schemas).
/// That byte-stable prefix is what lets a provider's prefix cache hit across
/// turns — the difference between paying full price to re-read the same context
/// every round and paying ~10% for the cached bytes (DeepSeek).
pub struct Session<'a, M: Model> {
    loop_: &'a AgentLoop<M>,
    history: Vec<Turn>,
    max_iters: u32,
}

impl<'a, M: Model> Session<'a, M> {
    pub fn with_max_iters(mut self, n: u32) -> Self {
        self.max_iters = n;
        self
    }
    /// Preload prior turns (e.g. resumed from disk).
    pub fn with_seed(mut self, seed: Vec<Turn>) -> Self {
        self.history = seed;
        self
    }
    /// The accumulated conversation so far.
    pub fn history(&self) -> &[Turn] {
        &self.history
    }
    /// Start over (branch): drop the accumulated turns.
    pub fn reset(&mut self) {
        self.history.clear();
    }

    /// Send one user message. Runs the ReAct loop against the accumulated
    /// history, then appends this user turn + the assistant reply so the next
    /// turn extends the same cached prefix.
    pub async fn turn(
        &mut self,
        message: impl Into<String>,
        world: &mut World,
    ) -> Result<Outcome, HarnessError> {
        let message = message.into();
        let task = Task {
            description: message.clone(),
            source: None,
            deadline: None,
        };
        let outcome = self
            .loop_
            .run_with_seed_history(task, self.history.clone(), world, self.max_iters)
            .await?;
        let reply = match &outcome {
            Outcome::Done { text, .. } => text.clone().unwrap_or_default(),
            Outcome::BudgetExhausted { last_text, .. }
            | Outcome::Stuck { last_text, .. }
            | Outcome::Cancelled { last_text, .. } => last_text.clone().unwrap_or_default(),
        };
        self.history.push(Turn {
            role: TurnRole::User,
            blocks: vec![Block::Text(message)],
        });
        self.history.push(Turn {
            role: TurnRole::Assistant,
            blocks: vec![Block::Text(reply)],
        });
        Ok(outcome)
    }
}

/// Audit #7: default safelist for `FixPatch::RunCommand`.
///
/// Sensors emitting `RunCommand` patches would otherwise be a silent
/// arbitrary-code-execution channel. We restrict the *program* by name to a
/// short list of well-known, side-effect-bounded formatters/fixers. Anything
/// else returns false and the patch is rejected (write your own `PreAutoFix`
/// hook returning `HookOutcome::Allow` to widen the policy).
///
/// `ReplaceFile` and `UnifiedDiff` are not restricted here — they only touch
/// files inside the workspace and are covered by the symlink-safe path
/// resolution in `harness-tools-fs`.
pub fn is_default_safe_fix(patch: &harness_core::FixPatch) -> bool {
    use harness_core::FixPatch;
    match patch {
        FixPatch::ReplaceFile { .. } | FixPatch::UnifiedDiff { .. } => true,
        FixPatch::RunCommand { program, args, .. } => match program.as_str() {
            // Cargo subcommands proven side-effect-bounded.
            "cargo" => matches!(
                args.first().map(String::as_str),
                Some("fmt" | "clippy" | "fix"),
            ),
            "rustfmt" | "gofmt" | "prettier" | "ruff" | "black" => true,
            _ => false,
        },
        // Future FixPatch variants: deny by default — review and add to the list above.
        _ => false,
    }
}

/// Monotonic counter for `.harness-patch-*.diff` temp filenames — millisecond
/// resolution alone collides under parallel agent runs.
static PATCH_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Monotonic counter for fallback recall session ids (no `uuid` dep).
static RECALL_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Apply auto-fix patches; return short descriptions of those that succeeded.
///
/// Made `pub` (was `pub(crate)`) so integration tests can call it directly.
pub async fn apply_patches(patches: &[harness_core::FixPatch], world: &mut World) -> Vec<String> {
    use harness_core::FixPatch;
    let mut applied = Vec::new();
    for p in patches {
        match p {
            FixPatch::ReplaceFile { path, content } => {
                let abs = world.repo.root.join(path);
                if let Some(parent) = abs.parent() {
                    let _ = tokio::fs::create_dir_all(parent).await;
                }
                if tokio::fs::write(&abs, content).await.is_ok() {
                    applied.push(format!("replaced {}", path.display()));
                }
            }
            FixPatch::UnifiedDiff { diff } => {
                if try_apply_diff(world, diff).await {
                    applied.push("unified diff applied".into());
                }
            }
            FixPatch::RunCommand { program, args, cwd } => {
                let cwd_ref = cwd.as_deref().unwrap_or(world.repo.root.as_path());
                let args_ref: Vec<&str> = args.iter().map(String::as_str).collect();
                if let Ok(out) = world.runner.exec(program, &args_ref, Some(cwd_ref)).await
                    && out.status == 0
                {
                    applied.push(format!("ran `{program} {}`", args.join(" ")));
                }
            }
            // FixPatch is `#[non_exhaustive]`; unknown variants are skipped.
            _ => tracing::warn!("apply_patches: unknown FixPatch variant — skipped"),
        }
    }
    applied
}

/// Write `diff` to a unique temp file and try `patch -p1` first, then `-p0`.
/// Returns whether either succeeded. The `-p1`-then-`-p0` order matches the
/// reality that most agent-emitted diffs are git-style (need `-p1`) but some
/// hand-rolled diffs use repo-relative paths (need `-p0`).
async fn try_apply_diff(world: &mut World, diff: &str) -> bool {
    use std::sync::atomic::Ordering;
    use tokio::io::AsyncWriteExt;

    let seq = PATCH_SEQ.fetch_add(1, Ordering::SeqCst);
    let pid = std::process::id();
    let now = world.clock.now_ms();
    let tmp = world
        .repo
        .root
        .join(format!(".harness-patch-{pid}-{now}-{seq}.diff"));

    let mut f = match tokio::fs::File::create(&tmp).await {
        Ok(f) => f,
        Err(e) => {
            tracing::warn!(error=%e, path=%tmp.display(), "could not create patch tempfile");
            return false;
        }
    };
    if let Err(e) = f.write_all(diff.as_bytes()).await {
        tracing::warn!(error=%e, "could not write patch tempfile");
        let _ = tokio::fs::remove_file(&tmp).await;
        return false;
    }
    drop(f);

    let tmp_str = tmp.to_string_lossy().to_string();
    let mut applied = false;
    for strip in ["-p1", "-p0"] {
        match world
            .runner
            .exec(
                "patch",
                &[strip, "--silent", "-i", tmp_str.as_str()],
                Some(world.repo.root.as_path()),
            )
            .await
        {
            Ok(out) if out.status == 0 => {
                tracing::info!(strip, "patch applied");
                applied = true;
                break;
            }
            Ok(out) => {
                tracing::debug!(strip, stderr=%out.stderr, "patch failed; trying next strip level");
            }
            Err(e) => {
                tracing::warn!(error=%e, "patch command not available");
                break; // patch tool missing — no point trying other strip
            }
        }
    }
    let _ = tokio::fs::remove_file(&tmp).await;
    applied
}