polyc_agent/

step.rs

//! The `TurnStep` seam: a small, ordered set of composable steps the turn loop
//! drives, each owning one cohesive slice of turn behavior.
//!
//! Slice 2 of #649 introduces the seam and proves it by migrating exactly one
//! behavior out of the turn-loop monolith — the forced closing completion. The
//! turn function's post-loop tail is now a driver over a list of
//! [`TurnStep`]s; later slices migrate the remaining stanzas behind the same
//! interface.
//!
//! A step reads and mutates the turn's working state through a [`TurnCtx`] and
//! reports back with a [`StepOutcome`] (keep going, pause for a human, or end
//! the turn early).

use std::collections::{HashMap, HashSet};

use async_trait::async_trait;
use futures::SinkExt as _;
use polyc_crypto::canon::canon_args;
use polyc_llm::{
    CompletionRequest, Content as LlmContent, LlmProvider, Message as LlmMessage, Role, StopReason,
    ToolSpec, Usage,
    request::ToolCall,
    turn::{collect_turn, collect_turn_observed},
};
use polyc_proto::proto::polychrome::agent::v1::Message;

use crate::{
    CallContext, CallDisposition, HandoffRequest, PendingApproval, ResolvedCall, RunTurnOptions,
    ToolExecutor, append_injected_notes, cap_tool_result, consume_decisions, forced_result,
    gate_decision, gate_missing, matching_decision_indexes, push_internal_note, push_reasoning,
    resolve_approved_call, run_and_redact, session_approves, splice_results_after, text_message,
    tool_result_message, untrusted_content_in_context,
};

/// The runtime-injected ground-truth note (`#743` change 1b) pushed after a
/// resume executes at least one previously-approved call: model-visible
/// (a System message in [`TurnCtx::messages`]) but never user-visible
/// (`internal_only` in [`TurnCtx::outputs`], via [`push_internal_note`]).
///
/// The resume's continuation text MUST still post — the codeless invite ack,
/// the demote confirmation, etc. are genuine narration, not a status guess —
/// so this does not suppress anything. It structurally corrects what the
/// model would otherwise have to infer from a bare tool result: that a person
/// already approved the call and it already ran, so the model's job now is to
/// report the outcome, not to describe (or re-describe) approval status. Same
/// mechanism as the `#67` approver-injected context: runtime-supplied ground
/// truth, not a prompt-level instruction the model could ignore as mere text.
///
/// The "who reports approval status" sentence is not restated here — the
/// note embeds [`polyc_llm::APPROVAL_STATUS_GROUND_RULE`] verbatim, the
/// same single source [`polyc_llm::GATED_TOOL_APPROVAL_NOTE`] embeds, so
/// the two moments the model hears the rule can never drift apart (#1141).
pub(crate) static RESUME_EXECUTED_GROUND_TRUTH_NOTE: std::sync::LazyLock<String> =
    std::sync::LazyLock::new(|| {
        format!(
            "A person approved this request and the tool has already run — the results above \
             are final. Tell the user what happened. {}",
            polyc_llm::APPROVAL_STATUS_GROUND_RULE
        )
    });

/// The turn's working state, threaded through each [`TurnStep`].
///
/// Owns what were locals in the turn function — the working transcript, the
/// accumulated wire outputs, the folded usage, the last stop reason, and the
/// loop-control flags — and borrows the turn's immutable inputs (the provider,
/// tool executor, model, and options) for the lifetime `'a` so a step can dial
/// the provider without re-plumbing them.
// Independent working flags a step reads/sets separately; folding them into an
// enum would force artificial combinations (a turn that executed tools also
// produced text, and either can coexist with a fired escape hatch).
#[allow(clippy::struct_excessive_bools)]
pub struct TurnCtx<'a, P, T>
where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    /// The LLM provider the turn dials.
    pub provider: &'a P,
    /// The executor that advertises and runs this turn's tools.
    pub tools: &'a T,
    /// The model identifier for provider requests.
    pub model: &'a str,
    /// The options the turn was invoked with (streaming channel, decisions).
    pub options: &'a RunTurnOptions,
    /// The working transcript driven through the loop and any post-steps.
    pub messages: Vec<LlmMessage>,
    /// The wire messages produced so far — assistant text and tool results.
    pub outputs: Vec<Message>,
    /// Usage folded across every provider call this turn has made.
    pub total_usage: Usage,
    /// Stop reason of the most recent provider step.
    pub last_stop: Option<StopReason>,
    /// Whether any tool ran this turn (resume pre-pass or the loop).
    pub executed_tools: bool,
    /// Whether the model ever emitted user-visible text this turn.
    pub produced_text: bool,
    /// Whether native search grounding was allowed for any step this turn
    /// made — see [`crate::TurnResult::grounded`] for why this is
    /// conservative (allowed, not necessarily used) and monotonic once set.
    pub grounded: bool,
    /// The pending handoff request, if the model asked to hand off.
    pub pending_handoff: Option<HandoffRequest>,
    /// STICKY/TERMINAL denials keyed to the tool *signature* (name + canonical
    /// args) rather than the provider call-id. Once a human denies an action,
    /// the model can re-emit the SAME logical call with a fresh call-id; a
    /// call-id-only check would re-pause and re-prompt for something already
    /// rejected. The resume pre-pass seeds this and the in-loop batch records
    /// into it, so a matching re-emit is auto-denied (synthetic result) without
    /// ever pausing again.
    pub denied_sigs: HashSet<(String, String)>,
    /// Occurrence-ordered decisions still awaiting one exact matching call.
    /// Arguments are canonicalized when the turn starts. A consumed entry is
    /// removed, so an identical later occurrence cannot inherit it.
    pub approval_decisions_remaining: Vec<crate::ApprovalDecision>,
    /// How many loop iterations have resolved a signature-matched terminal
    /// denial — the model retrying an action a human already denied. The first
    /// signed denial (by call-id, before any signature is recorded) does not
    /// count; only re-emits of an already-denied signature do. Persists across
    /// iterations so the stateless [`CircuitBreaker`] step can increment it and
    /// end the turn once it reaches `MAX_DENIAL_REPROMPTS`.
    pub denial_reprompts: usize,
    /// Whether the step that just resolved handled a signature-matched terminal
    /// denial. The loop republishes it onto the ctx each iteration before the
    /// [`CircuitBreaker`] step reads it; the step never touches the working
    /// state.
    pub saw_sig_match_denial: bool,
    /// Gated calls an unattended turn denied fail-closed (`#623`), accumulated
    /// across the loop iterations. Each entry is a call the capability gate would
    /// have escalated on a turn with [`RunTurnOptions::unattended`](crate::RunTurnOptions::unattended)
    /// set; the model saw a legible denial result and the call neither ran nor
    /// paused. The final
    /// [`TurnResult::unattended_denials`](crate::TurnResult::unattended_denials)
    /// carries them out for the control plane to audit. Always empty on an
    /// attended turn.
    pub unattended_denials: Vec<crate::UnattendedDenial>,
    /// Whether the fuzzy-match escape hatch (`#582`, invariant 9) has fired
    /// this turn. The hatch widens the advertised tool set at most ONCE per
    /// turn; once set, a later call naming an unadvertised tool resolves to
    /// the ordinary unknown-tool result again.
    pub escape_hatch_fired: bool,
    /// One entry per `__delegate_to` call dispatched this turn (`#872`),
    /// accumulated across the loop iterations. The final
    /// [`TurnResult::delegate_records`](crate::TurnResult::delegate_records)
    /// carries them out for the control plane to append as signed forensic
    /// events. Empty for every turn that never called `__delegate_to`.
    pub delegate_records: Vec<crate::DelegateRecord>,
    /// Questions from an `ask_question` call awaiting an answer (`#1660`).
    /// Populated only when the turn pauses on the question-pause phase
    /// (mirroring [`Self::pending_handoff`]) — a sibling pause path to the
    /// HITL approval gate, not a reuse of it. The final
    /// [`TurnResult::pending_questions`](crate::TurnResult::pending_questions)
    /// carries them out. Empty for every turn that never paused on a
    /// question.
    pub pending_questions: Vec<crate::question::PendingQuestion>,
}

impl<P, T> TurnCtx<'_, P, T>
where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    /// Consumes the turn's working state into a [`crate::TurnResult`], moving
    /// every accumulated audit surface out in one place.
    ///
    /// The turn loop's return sites differ only in the approvals they surface
    /// and whether a handoff rides along; the transcript, folded usage, last
    /// stop reason, and the `#623` unattended-denial audit surface are always
    /// whatever the context accumulated. Owning
    /// that move here makes forgetting an audit surface at a return site
    /// impossible by construction.
    #[must_use]
    pub fn finish(
        self,
        pending_approvals: Vec<PendingApproval>,
        handoff: Option<HandoffRequest>,
    ) -> crate::TurnResult {
        // Per-turn prompt-cache effectiveness (#1299): every return site
        // funnels through here, so this fires exactly once per turn with
        // the fully folded `Usage` — including the cache counters the two
        // fold sites (`lib.rs`'s loop, `ForcedCompletion`) accumulate but
        // never otherwise leave the agent crate.
        crate::metrics::record_turn(&self.total_usage);
        crate::TurnResult {
            messages: self.outputs,
            usage: self.total_usage,
            stop: self.last_stop,
            pending_approvals,
            handoff,
            unattended_denials: self.unattended_denials,
            mid_stream_failure: None,
            delegate_records: self.delegate_records,
            grounded: self.grounded,
            pending_questions: self.pending_questions,
        }
    }

    /// Consumes the turn's working state into a [`crate::TurnResult`] that
    /// reports a mid-turn provider stream failure (`#798`), exactly like
    /// [`Self::finish`] but with [`crate::TurnResult::mid_stream_failure`] set
    /// and no pending approvals (a failed stream never paused for HITL).
    ///
    /// Whatever the loop already accumulated — executed tool results, produced
    /// text, folded usage — rides along on the returned [`crate::TurnResult`]
    /// instead of being discarded, which is the whole point: the caller can
    /// persist iterations `1..N-1`'s work AND fail the turn with a typed error,
    /// rather than losing both to a bare `Err` propagated via `?`.
    #[must_use]
    pub fn finish_failed(mut self, failure: crate::MidStreamFailure) -> crate::TurnResult {
        let handoff = self.pending_handoff.take();
        let mut result = self.finish(Vec::new(), handoff);
        result.mid_stream_failure = Some(failure);
        result
    }

    /// Fold one provider call's [`Usage`] into [`Self::total_usage`], via
    /// [`Usage`]'s [`AddAssign`](std::ops::AddAssign) impl — the single
    /// canonical field-by-field fold, never a `..Default::default()` spread
    /// (which would silently leave a newly-added field at zero instead of
    /// failing to compile; see `#1241`/`#1238`).
    ///
    /// The turn loop calls this once per provider call it makes (the main
    /// loop and [`ForcedCompletion`] are the two call sites, one provider
    /// call each), so `total_usage` always reflects every call the turn's
    /// tool-calling loop actually made, however many iterations that took.
    pub(crate) fn fold_usage(&mut self, delta: Usage) {
        self.total_usage += delta;
    }
}

/// What a [`TurnStep`] reports after running.
pub enum StepOutcome {
    /// Continue to the next step in the list.
    Continue,
    /// Pause the turn for human approval, surfacing the given calls.
    Pause(Vec<PendingApproval>),
    /// Pause the turn on one or more still-unanswered `ask_question`
    /// questions (`#1660`) — the question-pause SIBLING of [`Self::Pause`],
    /// not a reuse of it.
    PauseQuestions(Vec<crate::question::PendingQuestion>),
    /// End the step-driving phase early; skip any remaining steps.
    Done,
}

/// One cohesive slice of turn behavior the turn loop drives.
///
/// Each step reads and mutates the turn's working state through [`TurnCtx`] and
/// reports a [`StepOutcome`]. Generic over the provider `P` and tool executor
/// `T` so a step can dial the provider and use the turn's error type directly.
#[async_trait]
pub trait TurnStep<P, T>: Send + Sync
where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    /// Run this step against the turn context.
    ///
    /// # Errors
    ///
    /// Returns the provider's error type when the step fails in a way that
    /// should abort the turn. A step that is best-effort swallows its own
    /// provider failures and returns [`StepOutcome::Continue`] instead.
    async fn run(&self, ctx: &mut TurnCtx<'_, P, T>) -> Result<StepOutcome, P::Error>;
}

/// Last-resort reply when even the forced closing completion (below) comes
/// back with no text — e.g. a model stuck in a tool-calling groove that keeps
/// emitting `stop == ToolUse` with empty text even with no tools declared
/// (`#1317`). Used verbatim only when [`synthesize_forced_completion_fallback`]
/// finds nothing to name (no tool was ever called this turn); otherwise that
/// function's output is used instead. Honest about the outcome rather than
/// fabricating a summary of RESULTS: no machinery here could safely stand in
/// for the model's own words about what it found, so this states what
/// happened and what to do next, per the user-facing-copy rules (no
/// "sorry"/"please"/"unfortunately").
pub(crate) const FORCED_COMPLETION_FALLBACK_TEXT: &str =
    "I couldn't put together an answer to that — try asking again or rephrasing.";

/// `#1317` "robust fix": when even the forced closing completion comes back
/// empty, synthesize the fallback reply from what was actually TRIED this
/// turn (never a third completion attempt — no further retry exists past
/// this) instead of the fully generic [`FORCED_COMPLETION_FALLBACK_TEXT`].
/// Deterministic and honest: it names which tools were called, never
/// fabricates what they found.
///
/// Collects each distinct tool name called anywhere in `messages` (first-seen
/// order), rendered through [`polyc_proto::humanize_tool_name`] rather than
/// the raw machine identifier — the same "never hand-write tool-name jargon
/// into user-facing copy" rule every edge's status text already follows.
/// Falls back to [`FORCED_COMPLETION_FALLBACK_TEXT`] verbatim when nothing was
/// ever called (e.g. the very first completion came back structurally empty,
/// with neither a tool call nor text).
fn synthesize_forced_completion_fallback(messages: &[LlmMessage]) -> String {
    let mut seen = HashSet::new();
    let mut names = Vec::new();
    for name in messages
        .iter()
        .filter(|m| m.role == Role::Assistant)
        .flat_map(|m| &m.content)
        .filter_map(|c| match c {
            LlmContent::ToolUse(call) => Some(call.name.as_str()),
            _ => None,
        })
    {
        if seen.insert(name) {
            names.push(polyc_proto::humanize_tool_name(name));
        }
    }
    if names.is_empty() {
        return FORCED_COMPLETION_FALLBACK_TEXT.to_owned();
    }
    format!(
        "I tried {} but couldn't put together an answer — try asking again, or rephrasing what \
         you need.",
        names.join(", ")
    )
}

/// `#1317` "cheap fix": collapse a trailing run of pure tool-call/tool-result
/// turns — the exact pattern that primes a model to keep emitting
/// `functionCall` instead of answering in text — into one terse text summary,
/// instead of cloning the raw transcript verbatim into the forced closing
/// completion's request. Only the TRAILING run collapses; everything before
/// it (the real conversation) is untouched.
///
/// A message counts as "pure tool" when every [`LlmContent`] block in it is a
/// [`LlmContent::ToolUse`] (an [`Role::Assistant`] turn) or a
/// [`LlmContent::ToolResult`] (a [`Role::Tool`] turn) — i.e. it carries no
/// text at all. Returns `messages` unchanged (cloned) when there is no such
/// trailing run to collapse.
fn collapse_trailing_tool_only_run(messages: &[LlmMessage]) -> Vec<LlmMessage> {
    let is_pure_tool_turn = |m: &LlmMessage| -> bool {
        !m.content.is_empty()
            && match m.role {
                Role::Assistant => m
                    .content
                    .iter()
                    .all(|c| matches!(c, LlmContent::ToolUse(_))),
                Role::Tool => m
                    .content
                    .iter()
                    .all(|c| matches!(c, LlmContent::ToolResult(_))),
                Role::User | Role::System | _ => false,
            }
    };
    let split = messages
        .iter()
        .rposition(|m| !is_pure_tool_turn(m))
        .map_or(0, |i| i + 1);
    if split == messages.len() {
        return messages.to_vec();
    }
    let mut collapsed = messages[..split].to_vec();
    let tool_names: Vec<&str> = messages[split..]
        .iter()
        .flat_map(|m| &m.content)
        .filter_map(|c| match c {
            LlmContent::ToolUse(call) => Some(call.name.as_str()),
            LlmContent::ToolResult(_) | LlmContent::Text(_) | LlmContent::Image(_) | _ => None,
        })
        .collect();
    let summary = if tool_names.is_empty() {
        "Earlier this turn, tool calls were made with no further progress. Answer directly \
         from what is already known instead of calling another tool."
            .to_owned()
    } else {
        format!(
            "Earlier this turn, these tools were called with no further progress: {}. \
             Answer directly from what is already known instead of calling another tool.",
            tool_names.join(", ")
        )
    };
    collapsed.push(LlmMessage {
        role: Role::System,
        content: vec![LlmContent::text(summary)],
    });
    collapsed
}

/// The forced closing completion.
///
/// Whenever a turn is about to end with no user-visible text, force one
/// final text answer. A pending one-way handoff deliberately stays silent.
pub struct ForcedCompletion;

#[async_trait]
impl<P, T> TurnStep<P, T> for ForcedCompletion
where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    async fn run(&self, ctx: &mut TurnCtx<'_, P, T>) -> Result<StepOutcome, P::Error> {
        // FALLBACK: the turn is about to end with no user-visible text, so
        // `outputs` carries only tool calls/results (or nothing at all) — the
        // edge would post nothing (the "agent produced no text" dead-end).
        // This covers every shape that can reach here with `produced_text ==
        // false`: the loop exhausting MAX_STEPS while still calling tools; a
        // resume whose pre-pass executed an approved call and then got an
        // empty continuation; a MAX_STEPS of 0 (a delegated worker's own step
        // budget can be configured to zero, meaning the loop body never ran at
        // all); AND a turn whose very FIRST completion came back with neither
        // a tool call nor any text (no `tool_use` for `executed_tools` to have
        // ever been set on) — native search grounding is exactly this shape,
        // since a failed/empty grounding attempt is invisible to
        // `executed_tools` (grounding is a request-level flag, never a
        // `tool_use` call). The old guard required `executed_tools`, which
        // covered the first three shapes but not the fourth — see the
        // `__delegate_to` "worker produced no answer" incident this fixes.
        // Force ONE final completion with tools disabled so the model must
        // answer in text, summarizing what it did or explaining it couldn't
        // proceed. An intentional handoff skips this completion because its
        // parent turn ends at the one-way transfer. Best-effort: a failure
        // here still yields fallback text rather than leaving the turn silent.
        if ctx.produced_text || ctx.pending_handoff.is_some() {
            return Ok(StepOutcome::Continue);
        }
        let mut req = CompletionRequest::new(ctx.model);
        // `#1317` "cheap fix": collapse a trailing tool-call-only run instead
        // of cloning the raw transcript verbatim — see
        // `collapse_trailing_tool_only_run`'s doc comment.
        req.messages = collapse_trailing_tool_only_run(&ctx.messages);
        // Removing tools is not enough: a model deep in a tool-calling groove
        // will keep emitting a functionCall (stop == ToolUse) and no text even
        // with no tools declared. Also disable web-search grounding (another
        // tool surface) and append an explicit instruction so the model writes a
        // plain-text final answer from what it already has.
        // A System instruction (folded into systemInstruction by the provider,
        // not the visible transcript) so the model follows it without echoing it
        // into the reply; a User message gets paraphrased back by thinking models.
        // Kept non-meta for the same reason.
        req.messages.push(LlmMessage {
            role: Role::System,
            content: vec![LlmContent::Text(
                "No tools are available for the remainder of this turn. Give the \
                 user a direct, plain-text answer using the information already \
                 gathered."
                    .to_owned(),
            )],
        });
        req.tools = Vec::new();
        req.web_search = false;
        // Best-effort closing completion: its output is discarded on any error,
        // so don't spend the retry budget's backoff here — a single attempt
        // keeps a wedged turn from also paying tens of seconds of backoff.
        let mut closing_text: Option<String> = None;
        if let Ok(stream) = ctx.provider.complete(req).await {
            let turn = if let Some(tx) = ctx.options.stream_tx.clone() {
                // Bounded (`#251`): see the matching forwarding site in
                // `lib.rs` — `tx` is cloned once here (not per event) and the
                // `.await`ed send applies real backpressure.
                let mut tx = tx;
                collect_turn_observed(stream, async move |ev| {
                    let _ = tx.send(ev).await;
                })
                .await
            } else {
                collect_turn(stream).await
            };
            if let Ok(turn) = turn {
                ctx.fold_usage(turn.usage);
                push_reasoning(&mut ctx.outputs, &turn.reasoning);
                ctx.last_stop = turn.stop;
                if !turn.text.is_empty() {
                    closing_text = Some(turn.text);
                }
            }
        }
        if let Some(text) = closing_text {
            ctx.outputs.push(text_message("model", &text));
            ctx.produced_text = true;
            tracing::info!("forced closing completion produced text; turn now yields a reply");
        } else {
            // `#1317`: the forced pass itself can also come back empty (a
            // model stuck in the same tool-calling groove even with no tools
            // declared, or the completion request failing outright) — log a
            // distinct WARN so this is greppable from harness logs alone,
            // then fall back to the honest static reply so the turn NEVER
            // drops silently.
            tracing::warn!(
                "forced closing completion also produced no text — falling back to a synthesized reply so the turn doesn't drop silently"
            );
            let fallback = synthesize_forced_completion_fallback(&ctx.messages);
            ctx.outputs.push(text_message("model", &fallback));
            ctx.produced_text = true;
        }
        Ok(StepOutcome::Continue)
    }
}

/// The approval resume pre-pass: resolve the tool calls a human already decided.
///
/// Before the turn drives the model, execute the calls the human approved that
/// are dangling in the resumed transcript, resolve signed or denied calls to
/// synthetic results, splice those results in after the paused batch, and append
/// any approver-injected context notes.
///
/// On an approval resume the control plane replays the paused turn's assistant
/// `tool_use` (which has NO paired `tool_result` — the call was paused, never
/// executed) and forwards the signed decisions via
/// [`RunTurnOptions::approval_decisions`].
/// The function-calling loop only executes tool calls the *model emits this
/// turn*, so without this step an approval takes effect only if the model
/// happens to RE-EMIT the same call. Resolving the dangling calls
/// deterministically here makes an approval ALWAYS take effect, independent of
/// whether the model re-emits.
///
/// Classification and the #141 approval binding mirror the in-loop batch so the
/// two can't drift. It runs only on a resume: a fresh turn carries an empty
/// decision set, so this step is a no-op and the hot path is unchanged.
///
/// A forwarded signed decision must never resolve silently to nothing: whenever
/// `approval_decisions` is non-empty, [`Self::run`] logs a resolution summary and
/// `tracing::warn!`s individually for every approved tuple that matches no
/// unanswered call in the resumed transcript — distinguishing an already-
/// answered (harmless) re-forward from a call that is missing outright (a
/// projection loss upstream, e.g. one folded into a compaction summary). This
/// is purely observational: an approval that resolves to nothing still resolves
/// to nothing (re-pausing here would loop), but the loss is now loud instead of
/// surfacing only as an unexplained model refusal downstream.
///
/// Borrows the turn's read-only resume inputs — the pinned tool-spec set (for
/// pause-card titles), the approver edits, and the signed denials — while the
/// mutable working state (transcript, outputs, the sticky denial set, and the
/// remaining approvals) rides the [`TurnCtx`].
pub struct ResumePrePass<'a> {
    /// The turn's pinned tool-spec set, read once for pause-card titles.
    pub tool_specs: &'a [ToolSpec],
}

#[async_trait]
impl<P, T> TurnStep<P, T> for ResumePrePass<'_>
where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    #[allow(clippy::too_many_lines)] // cohesive resume-resolution pass
    async fn run(&self, ctx: &mut TurnCtx<'_, P, T>) -> Result<StepOutcome, P::Error> {
        // RESUME: execute approved-but-unanswered tool calls ALREADY PRESENT in
        // the input transcript, before driving the model. When the model instead
        // reads its own dangling `tool_use` as already-done and narrates
        // completion (e.g. "OK, I've torn it down"), the approved action silently
        // never executes and the human's decision is lost — so resolve the
        // dangling calls deterministically here.
        //
        // #1154: this step used to short-circuit here whenever BOTH decision
        // sets were empty, on the assumption that "no approvals and no
        // denials" implies "a genuinely fresh turn, no dangling tool_use." A
        // resume whose only signed decision failed harness-side verification
        // (e.g. a dropped signed field) breaks that assumption: the wire
        // carried a real decision, it just verified to nothing, so this step
        // was skipped and the model was left narrating a dangling call it
        // never ran. There is no cheaper-but-safe proxy for "this is a fresh
        // turn" than actually checking for a dangling `tool_use` below, so the
        // scan always runs; a genuinely fresh turn still exits immediately at
        // the `unanswered.is_empty()` check just past it.

        // Match each result to ONE preceding same-id tool use. Provider ids can
        // repeat across turns, so a global answered-id set would let turn A's
        // result hide turn B's later occurrence.
        let mut calls: Vec<Option<(usize, ToolCall)>> = Vec::new();
        let mut open_by_id: HashMap<String, std::collections::VecDeque<usize>> = HashMap::new();
        let mut answered_ids: HashSet<String> = HashSet::new();
        for (idx, m) in ctx.messages.iter().enumerate() {
            for c in &m.content {
                match c {
                    LlmContent::ToolUse(tc) => {
                        let slot = calls.len();
                        calls.push(Some((idx, tc.clone())));
                        open_by_id.entry(tc.id.clone()).or_default().push_back(slot);
                    }
                    LlmContent::ToolResult(result) => {
                        answered_ids.insert(result.tool_call_id.clone());
                        if let Some(slot) = open_by_id
                            .get_mut(&result.tool_call_id)
                            .and_then(std::collections::VecDeque::pop_front)
                        {
                            calls[slot] = None;
                        }
                    }
                    _ => {}
                }
            }
        }
        let unanswered: Vec<(usize, ToolCall)> = calls.into_iter().flatten().collect();

        // Observability (hardening after the #699/#700 admin-invite silent-no-op:
        // an approved resume that resolved to nothing with no trace beyond a
        // model-generated refusal). A forwarded signed decision must never
        // resolve silently — WARN individually for every approved tuple that
        // matches no unanswered call in THIS resumed transcript, distinguishing
        // "already answered" (its id already carries a live `tool_result` — a
        // harmless re-forward of a spent decision) from "not found" (no call
        // anywhere in the transcript carries this id — the call itself is
        // missing, e.g. folded into a compaction summary or otherwise dropped
        // between pause and resume) so a silent loss is loud at the exact site
        // that would otherwise have swallowed it.
        if !ctx.options.approval_decisions.is_empty() {
            let unanswered_ids: HashSet<&str> =
                unanswered.iter().map(|(_, tc)| tc.id.as_str()).collect();
            for decision in &ctx.options.approval_decisions {
                let id = &decision.request_id;
                let name = &decision.tool_name;
                if unanswered_ids.contains(id.as_str()) {
                    continue;
                }
                if answered_ids.contains(id) {
                    tracing::info!(
                        request_id = %id,
                        tool = %name,
                        "approved call already answered on this resume; decision is a no-op re-forward"
                    );
                } else {
                    tracing::warn!(
                        request_id = %id,
                        tool = %name,
                        "approved call id matches no tool call in the resumed \
                         transcript; the signed decision cannot resolve to anything"
                    );
                }
            }
        }
        tracing::info!(
            approved = ctx
                .options
                .approval_decisions
                .iter()
                .filter(|decision| decision.approved)
                .count(),
            denied = ctx
                .options
                .approval_decisions
                .iter()
                .filter(|decision| !decision.approved)
                .count(),
            unanswered = unanswered.len(),
            "resume pre-pass: resolving forwarded decisions against the resumed transcript"
        );

        if unanswered.is_empty() {
            return Ok(StepOutcome::Continue);
        }

        // Taint state, evaluated against the resumed transcript: any untrusted
        // tool-result (a prior fetch) already in context, OR the durable seed the
        // control plane computed over the full event log (untrusted content that
        // compaction folded out of the projection, or a non-principal
        // participant's input — neither of which survives as a live `ToolResult`).
        let untrusted_in_context =
            untrusted_content_in_context(&ctx.messages) || ctx.options.untrusted_context_seed;
        // Classify exactly as the in-loop batch does (same #141 binding:
        // approval/denial bound to the exact (id, name, args) tuple).
        let unanswered_calls: Vec<ToolCall> =
            unanswered.iter().map(|(_, call)| call.clone()).collect();
        let matched_decisions =
            matching_decision_indexes(&unanswered_calls, &ctx.approval_decisions_remaining);
        let dispositions: Vec<CallDisposition> = unanswered
            .iter()
            .zip(&matched_decisions)
            .map(|((_, tc), decision_index)| {
                let gate = gate_decision(
                    ctx.tools,
                    ctx.options,
                    untrusted_in_context,
                    &tc.name,
                    &tc.args_json,
                );
                let decision = decision_index.map(|index| &ctx.approval_decisions_remaining[index]);
                let is_denied = decision.is_some_and(|decision| !decision.approved);
                // A remembered session approval ("don't ask again") satisfies the
                // gate only when its signed covered set includes everything this
                // call is currently missing (#595; see the in-loop site for the
                // rationale). An explicit occurrence decision still runs.
                let is_approved = decision.is_some_and(|decision| decision.approved)
                    || session_approves(ctx.options, ctx.tools, &tc.name, gate_missing(&gate));
                // No sticky-signature denial at pre-pass time (denied_sigs is
                // empty until the loop runs), so sig_match is always false. A
                // resume is by definition attended (a human answered an approval),
                // so `unattended` is false here — the #623 fail-closed denial only
                // arises on a fresh trigger-originated firing, never on resume.
                CallDisposition::classify(
                    gate,
                    CallContext {
                        approved: is_approved,
                        denied: is_denied,
                        ..CallContext::default()
                    },
                )
            })
            .collect();
        // Tally the four disposition classes in a SINGLE traversal rather than
        // one filter/count pass per class.
        let mut execute = 0usize;
        let mut denied = 0usize;
        let mut policy_denied = 0usize;
        let mut pending = 0usize;
        for disposition in &dispositions {
            match disposition {
                CallDisposition::Execute => execute += 1,
                CallDisposition::Denied { .. } => denied += 1,
                // #623: an unattended fail-closed denial is a non-HITL denial,
                // tallied with the policy/sandbox class (this count only feeds a
                // tracing line; an unattended turn never resumes, ADR 0003).
                CallDisposition::PolicyDenied { .. } | CallDisposition::UnattendedDenied { .. } => {
                    policy_denied += 1;
                }
                // #582 invariant 9: constructed only by the in-loop escape
                // hatch, never by `classify` — unreachable on a resume, but the
                // tally stays total so a future refactor can't miscount.
                CallDisposition::Recovered { .. } => {}
                CallDisposition::Pending { .. } => pending += 1,
            }
        }
        tracing::info!(
            execute,
            denied,
            policy_denied,
            pending,
            "resume pre-pass: classified every unanswered dangling call"
        );

        // A dangling call that still needs approval (neither approved nor denied)
        // must NOT be executed — re-pause the turn so the human is re-prompted,
        // exactly as a fresh gated call would.
        if dispositions
            .iter()
            .any(|d| matches!(d, CallDisposition::Pending { .. }))
        {
            let pending = unanswered
                .iter()
                .zip(&dispositions)
                .filter_map(|((_, tc), d)| {
                    let CallDisposition::Pending { reason, missing } = d else {
                        return None;
                    };
                    let title = self
                        .tool_specs
                        .iter()
                        .find(|s| s.name == tc.name)
                        .and_then(|s| s.title.clone())
                        .unwrap_or_default();
                    Some(PendingApproval {
                        occurrence_turn_id: tc.approval_turn_id.clone(),
                        id: tc.id.clone(),
                        name: tc.name.clone(),
                        args_json: tc.args_json.clone(),
                        title,
                        // Sandbox-unaware here; the harness stamps the mode onto
                        // the wire payload.
                        sandbox_mode: String::new(),
                        // The gate's reason carried on the disposition (empty for
                        // an ordinary intrinsic/sandbox gate).
                        reason: reason.clone(),
                        missing_capabilities: missing
                            .names()
                            .iter()
                            .map(|n| (*n).to_owned())
                            .collect(),
                        // Filled in later, control-plane side, for a
                        // `routine_delete` call (see the field's own doc).
                        computed_preview: String::new(),
                    })
                })
                .collect::<Vec<_>>();
            return Ok(StepOutcome::Pause(pending));
        }

        // Execute approved calls concurrently; denied calls resolve to the
        // synthetic denial payload (mirrors the in-loop resolution). Resolve each
        // paused call's approver edit (#67) once: the edited args to execute + any
        // context to inject. Aligned with `unanswered`.
        let pre_resolutions: Vec<ResolvedCall> = unanswered
            .iter()
            .zip(&matched_decisions)
            .map(|((_, tc), decision_index)| {
                let r#override = decision_index
                    .and_then(|index| ctx.approval_decisions_remaining[index].r#override.as_ref());
                resolve_approved_call(&tc.args_json, r#override)
            })
            .collect();
        // Resumed calls execute the args the human already approved; the dispatch
        // policy's INPUT mutations (#539) belong to a fresh dispatch, but
        // `post_dispatch` result redaction (#540) still applies to their output.
        let tools = ctx.tools;
        let recorder = ctx.options.dispatch_recorder.clone();
        let futures = unanswered
            .iter()
            .zip(&dispositions)
            .zip(&pre_resolutions)
            .map(|(((_, tc), disposition), resolved)| {
                if matches!(disposition, CallDisposition::Denied { .. }) {
                    // Sticky for the loop below: any re-emit of the same action is
                    // auto-denied without re-prompting.
                    ctx.denied_sigs
                        .insert((tc.name.clone(), canon_args(&tc.args_json)));
                }
                // A human denial OR a policy veto (#67) resolves to a synthetic
                // result instead of executing.
                let forced = forced_result(disposition);
                let name = tc.name.clone();
                let args = resolved.args_json.clone();
                let call_id = tc.id.clone();
                let approval_turn_id = tc.approval_turn_id.clone();
                let recorder = recorder.clone();
                async move {
                    if let Some(result) = forced {
                        result
                    } else {
                        run_and_redact(
                            tools,
                            recorder.as_ref(),
                            call_id,
                            approval_turn_id,
                            name,
                            args,
                        )
                        .await
                    }
                }
            })
            .collect::<Vec<_>>();
        let results = futures::future::join_all(futures).await;
        // The pre-pass resolved dangling calls (executed approvals and/or
        // synthesized denial results); either way the turn produced tool_results
        // that need narrating, so guarantee a closing reply.
        ctx.executed_tools = true;

        // Every matched human decision is now represented by exactly one tool
        // result (real execution or synthetic denial), so spend it once.
        consume_decisions(&mut ctx.approval_decisions_remaining, &matched_decisions);

        // Append each result to the persisted `outputs` (so a LATER resume sees
        // the call as answered) and into the transcript GROUPED after the paused
        // batch's last tool_use — never interleaved between two calls. A paused
        // batch can be parallel tool calls, and the function-calling contract
        // requires a turn's `functionCall`s to be followed by ALL their
        // `functionResponse`s together: a response spliced between two parallel
        // calls is rejected (the provider 400s, which would fail the re-drive and
        // strand the calls unanswered — poisoning the conversation). The in-loop
        // path groups the same way.
        let mut result_msgs = Vec::with_capacity(unanswered.len());
        for ((_, tc), result) in unanswered.iter().zip(results) {
            let result = cap_tool_result(&result);
            // Stamp ingestion-time provenance so the durable trifecta tag mirrors
            // the live-scan predicate: a first-party tool's result does not taint
            // context (see `output_msg_trust`).
            let first_party = !ctx.tools.ingests_untrusted_content(&tc.name);
            ctx.outputs
                .push(tool_result_message(&tc.id, &result, first_party));
            // #874 (headline fix): stamp the same verdict onto the in-memory
            // transcript, mirroring the main dispatch loop's fix. The static
            // per-tool-name check is sufficient HERE specifically: a
            // `__delegate_to` call is never gated (`gate_decision` returns
            // `Allow` unconditionally for it, the same seam this pre-pass and
            // the in-loop batch share), so it can never be paused and
            // therefore never appears in `unanswered` — this resume path
            // structurally never dispatches a delegate call, only ordinary
            // gated tools whose provenance IS the static per-name check.
            result_msgs.push(LlmMessage {
                role: Role::Tool,
                content: vec![LlmContent::tool_result(
                    tc.id.clone(),
                    result,
                    false,
                    first_party,
                )],
            });
        }
        // The paused batch is the tail of the transcript, so its results go after
        // its last call. `unanswered` is non-empty in this branch.
        let after = unanswered
            .iter()
            .map(|(idx, _)| *idx)
            .max()
            .unwrap_or(ctx.messages.len());
        ctx.messages = splice_results_after(std::mem::take(&mut ctx.messages), after, result_msgs);
        // #67: approver-injected context lands as internal-only system notes after
        // the spliced results (the paused batch is the transcript tail),
        // preserving the function-call ⇒ all-responses grouping.
        append_injected_notes(&mut ctx.outputs, &mut ctx.messages, &pre_resolutions);

        // `#743` change 1b: when at least one dangling call actually EXECUTED
        // this resume (as opposed to only denials/policy vetoes resolving),
        // tell the model — as runtime-injected ground truth, not a
        // suppressible instruction — that the results above are the final,
        // already-approved outcome. This is what makes the resume's
        // continuation narrate the real result instead of re-guessing
        // approval status from a bare tool result.
        if dispositions
            .iter()
            .any(|d| matches!(d, CallDisposition::Execute))
        {
            push_internal_note(
                &mut ctx.outputs,
                &mut ctx.messages,
                RESUME_EXECUTED_GROUND_TRUTH_NOTE.as_str(),
            );
        }

        Ok(StepOutcome::Continue)
    }
}

/// The runtime-injected ground-truth note pushed after a resume applies at
/// least one `ask_question` answer (`#1660`) — the question-pause SIBLING of
/// [`RESUME_EXECUTED_GROUND_TRUTH_NOTE`] above, not a reuse of it: a
/// clarifying-question answer is a decision, not a permission grant, so it
/// gets its own wording rather than borrowing the approval gate's.
pub(crate) static QUESTION_ANSWERED_GROUND_TRUTH_NOTE: &str = "The question(s) above have been resolved — each carries its own `state` \
     (answered/declined/auto_resolved) telling you exactly how. Read each one and act on it \
     directly; do not re-ask a question that already has a result here.";

/// The ephemeral reminder pushed alongside the invariant-I8 interim splice
/// below — model-visible only ([`TurnCtx::messages`]), never persisted
/// ([`TurnCtx::outputs`]) for the same reason the interim result itself
/// isn't: it describes a fact ("still open") that's only true at this
/// instant and would go stale the moment a real answer lands.
pub(crate) static QUESTION_STILL_PENDING_EPHEMERAL_NOTE: &str = "A question you asked earlier is still open — see the still_pending result above. That's \
     not an answer; it means nobody has responded yet. Handle the message below on its own \
     terms, and only bring the open question back up if it's still relevant once you have.";

/// True when nothing after message index `after` in `messages` is genuinely
/// new turn input — i.e. every trailing message is blank/whitespace-only
/// text, matching how the control plane's edge-facing
/// `new_inputs_are_blank` recognizes a pure resume redrive (`vec![user_message("")]`).
/// A non-text block (a real tool result, image, etc.) or any non-blank text
/// counts as real input. `crates/agent` can't depend on `crates/control-plane`
/// (the layer rule points inward), so this is the same semantics reimplemented
/// against [`LlmContent`] rather than shared code.
fn trailing_input_is_blank(messages: &[LlmMessage], after: usize) -> bool {
    messages
        .get(after.saturating_add(1)..)
        .unwrap_or(&[])
        .iter()
        .flat_map(|m| m.content.iter())
        .all(|c| matches!(c, LlmContent::Text(t) if t.trim().is_empty()))
}

/// One `tool_result` per dangling call in `resolved`, grouped after the
/// batch's last call and spliced into [`TurnCtx::messages`] — the shape
/// [`QuestionResumePrePass`]'s two splice sites share (the invariant-I8
/// transcript-only interim splice, and the real-answer splice once every
/// question has a verified answer): same `resolved.iter()` walk, same
/// `after` computation, same `tool_result` construction, differing only in
/// which JSON each call renders and whether the result is durable.
///
/// `durable: true` ALSO writes to [`TurnCtx::outputs`] — the real-answer
/// path, where a genuine signed answer must survive as part of the turn's
/// persisted transcript. `durable: false` writes to `ctx.messages` ONLY —
/// the I8 interim splice, which must never be persisted (see that call
/// site's own doc for why).
fn splice_question_results<P, T>(
    ctx: &mut TurnCtx<'_, P, T>,
    resolved: &[(
        usize,
        ToolCall,
        Vec<crate::question::QuestionItem>,
        Vec<crate::question::VerifiedAnswer>,
    )],
    durable: bool,
    mut result_json: impl FnMut(
        &ToolCall,
        &[crate::question::QuestionItem],
        &[crate::question::VerifiedAnswer],
    ) -> String,
) where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    let after = resolved
        .iter()
        .map(|(idx, ..)| *idx)
        .max()
        .unwrap_or(ctx.messages.len());
    let mut result_msgs = Vec::with_capacity(resolved.len());
    for (_, tc, items, answers) in resolved {
        let json = result_json(tc, items, answers);
        if durable {
            // `first_party: true` — the result is either a human's own
            // selection or the control plane's own signed auto-resolution,
            // never externally-fetched content, so it must not be treated
            // as untrusted-content-in-context.
            ctx.outputs.push(tool_result_message(&tc.id, &json, true));
        }
        result_msgs.push(LlmMessage {
            role: Role::Tool,
            content: vec![LlmContent::tool_result(tc.id.clone(), json, false, true)],
        });
    }
    ctx.messages = splice_results_after(std::mem::take(&mut ctx.messages), after, result_msgs);
}

/// Resolves dangling `ask_question` calls once their answers have arrived
/// (`#1660`).
///
/// The RESUME half of the question-pause phase — the question-pause SIBLING
/// of [`ResumePrePass`], not a reuse of it. The pause/emit half (recognizing
/// a fresh `ask_question` call and short-circuiting the batch) lives in the
/// in-loop QUESTION-PAUSE PHASE (`run_turn_with`) because `CollectedTurn`/
/// `tool_calls` are loop-body locals, not fields on [`TurnCtx`] — the same
/// seam constraint [`ResumePrePass`]'s own module doc notes for the approval
/// gate. This step only ever RESOLVES calls already dangling in the input
/// transcript.
///
/// Atomicity mirrors [`ResumePrePass`]: every dangling `ask_question` call's
/// disposition is computed FIRST; if ANY question across ANY of them is
/// still missing a [`crate::RunTurnOptions::question_answers`] entry, the
/// WHOLE resume re-pauses — UNLESS the dispatch also carries genuinely new
/// turn input (invariant I8), in which case the whole batch instead gets a
/// transcript-only "still pending" interim splice and the turn continues
/// (see the branch below for why: a provider requires every `tool_use` in
/// one assistant turn to receive a `tool_result` together, so a partial
/// splice — real answers for some calls, nothing for others — would corrupt
/// the transcript either way, real or interim). Only once every question in
/// every dangling call has a verified answer does this step build the
/// durable three-state result JSON (invariant I4) and splice it into both
/// [`TurnCtx::messages`] and [`TurnCtx::outputs`].
pub struct QuestionResumePrePass;

#[async_trait]
impl<P, T> TurnStep<P, T> for QuestionResumePrePass
where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    #[allow(clippy::too_many_lines)] // cohesive resume-resolution pass, mirrors ResumePrePass::run
    async fn run(&self, ctx: &mut TurnCtx<'_, P, T>) -> Result<StepOutcome, P::Error> {
        let answered: HashSet<&str> = ctx
            .messages
            .iter()
            .flat_map(|m| m.content.iter())
            .filter_map(|c| match c {
                LlmContent::ToolResult(tr) => Some(tr.tool_call_id.as_str()),
                _ => None,
            })
            .collect();
        let unanswered: Vec<(usize, ToolCall)> = ctx
            .messages
            .iter()
            .enumerate()
            .flat_map(|(idx, m)| m.content.iter().map(move |c| (idx, c)))
            .filter_map(|(idx, c)| match c {
                LlmContent::ToolUse(tc)
                    if tc.name == crate::question::ASK_QUESTION_TOOL_NAME
                        && !answered.contains(tc.id.as_str()) =>
                {
                    Some((idx, tc.clone()))
                }
                _ => None,
            })
            .collect();

        if unanswered.is_empty() {
            return Ok(StepOutcome::Continue);
        }

        // Parse each dangling call's questions once. A parse failure here is
        // unreachable in production: the exact same `args_json` bytes
        // already passed I5 validation before the call could ever pause
        // (the QUESTION-PAUSE PHASE never pauses a malformed call). Fail
        // defensively rather than panic — an empty item list contributes no
        // questions to resolve or re-pause, so a hypothetical future bug
        // here degrades to "this call is silently skipped" rather than a
        // crashed turn.
        let calls: Vec<(usize, ToolCall, Vec<crate::question::QuestionItem>)> = unanswered
            .into_iter()
            .map(|(idx, tc)| {
                let items = crate::question::parse_ask_question_args(&tc.args_json)
                    .inspect_err(|err| {
                        tracing::error!(
                            call_id = %tc.id,
                            %err,
                            "dangling ask_question call failed to re-parse on resume \
                             (unreachable: already validated before pausing)"
                        );
                    })
                    .unwrap_or_default();
                (idx, tc, items)
            })
            .collect();

        // First pass: match every question to a verified answer, or record it
        // as still missing. Nothing is mutated yet — the atomicity rule above.
        let mut missing: Vec<crate::question::PendingQuestion> = Vec::new();
        let mut resolved: Vec<(
            usize,
            ToolCall,
            Vec<crate::question::QuestionItem>,
            Vec<crate::question::VerifiedAnswer>,
        )> = Vec::with_capacity(calls.len());
        for (idx, tc, items) in calls {
            let mut matched = Vec::with_capacity(items.len());
            for (i, item) in items.iter().enumerate() {
                let index = u32::try_from(i).unwrap_or(u32::MAX);
                // Occurrence match (`#2523`): a provider re-mints a
                // tool-call id across turns, so an answer names the turn its
                // question was asked in as well as `(call_id, index)`. The
                // comparison is strict equality against the dangling call's
                // own stamped turn — the control plane re-binds a durable
                // answer that predates occurrence identity to the turn on its
                // event kind before forwarding it, so a historical answer
                // arrives here already carrying the turn it must match.
                let occurrence = tc.approval_turn_id.as_deref().unwrap_or_default();
                if let Some(answer) = ctx
                    .options
                    .question_answers
                    .iter()
                    .find(|a| a.turn_id == occurrence && a.call_id == tc.id && a.index == index)
                {
                    matched.push(answer.clone());
                } else {
                    missing.push(crate::question::PendingQuestion {
                        occurrence_turn_id: tc.approval_turn_id.clone(),
                        call_id: tc.id.clone(),
                        index,
                        item: item.clone(),
                        args_json: tc.args_json.clone(),
                    });
                }
            }
            resolved.push((idx, tc, items, matched));
        }

        if !missing.is_empty() {
            // #1662 follow-up / invariant I8: a hard re-pause here is only
            // correct when this dispatch carries no genuinely new turn
            // input — a blank redrive (the ordinary "resume after answering
            // elsewhere" shape) or a true no-op. When the caller appended
            // real new content after the dangling call (an unrelated
            // message the model hasn't seen yet), re-pausing identically
            // would silently swallow it: the pre-loop gate would return
            // before the model is ever dialed this turn, and the edge would
            // just re-render the byte-identical pending-question notice —
            // exactly the incident this invariant fixes (a user's follow-up
            // in the same Slack thread never reached the model; see #1659's
            // tracking issue for the root cause).
            let after = resolved
                .iter()
                .map(|(idx, ..)| *idx)
                .max()
                .unwrap_or(ctx.messages.len());
            if trailing_input_is_blank(&ctx.messages, after) {
                return Ok(StepOutcome::PauseQuestions(missing));
            }

            // Transcript-only interim result for every dangling call in this
            // batch — `durable: false` writes it to `ctx.messages` ONLY,
            // never `ctx.outputs`. `TurnCtx::finish` persists `ctx.outputs`
            // verbatim as the durable transcript every future resume
            // rebuilds from (`crates/agent/src/step.rs`'s own `finish`) —
            // writing this there would make the call look answered forever,
            // permanently losing the real question. Left out of
            // `ctx.outputs`, the very next dispatch re-parses the same
            // still-dangling call and re-splices fresh, so this never goes
            // stale and never blocks the real signed answer from resolving
            // it later exactly as today.
            splice_question_results(ctx, &resolved, false, |_, items, _| {
                crate::question::question_still_pending_json(items)
            });
            // Ephemeral reminder, transcript-only for the same reason as the
            // interim results above — appended after the user's new message
            // so it's the last thing the model reads before replying.
            ctx.messages.push(LlmMessage {
                role: Role::System,
                content: vec![LlmContent::text(
                    QUESTION_STILL_PENDING_EPHEMERAL_NOTE.to_owned(),
                )],
            });
            return Ok(StepOutcome::Continue);
        }

        // Every question in every dangling call now has a verified answer —
        // build the three-state result (I4) and splice it in durably
        // (mirrors `ResumePrePass`).
        ctx.executed_tools = true;
        splice_question_results(ctx, &resolved, true, |_, items, answers| {
            crate::question::question_call_result_json(items, answers)
        });

        push_internal_note(
            &mut ctx.outputs,
            &mut ctx.messages,
            QUESTION_ANSWERED_GROUND_TRUTH_NOTE,
        );

        Ok(StepOutcome::Continue)
    }
}

/// The denied-action circuit breaker.
///
/// When the model re-emits an action a human already denied, the turn loop
/// auto-denies it (a synthetic result, never executed) and republishes the "saw
/// a signature-matching denial" signal onto the ctx. This step counts each such
/// re-emit and, once the model has done it `MAX_DENIAL_REPROMPTS` times, reports
/// [`StepOutcome::Done`] so the turn ends cleanly with the last stop reason
/// instead of burning the rest of the step budget looping the same dead-end.
pub struct CircuitBreaker;

#[async_trait]
impl<P, T> TurnStep<P, T> for CircuitBreaker
where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    async fn run(&self, ctx: &mut TurnCtx<'_, P, T>) -> Result<StepOutcome, P::Error> {
        // CIRCUIT BREAKER: if the step that just resolved handled a re-emitted
        // denied signature (the model retried an already-denied action), count
        // it. Once the model has done this `MAX_DENIAL_REPROMPTS` times, stop
        // giving it another chance — end the turn so it closes cleanly with the
        // last stop reason instead of burning the rest of the step budget
        // looping the same dead-end. The tool_results for the step are already
        // appended by the loop, so the transcript stays well-formed.
        if ctx.saw_sig_match_denial {
            ctx.denial_reprompts += 1;
            if ctx.denial_reprompts >= crate::MAX_DENIAL_REPROMPTS {
                tracing::warn!(
                    denial_reprompts = ctx.denial_reprompts,
                    max = crate::MAX_DENIAL_REPROMPTS,
                    "HITL circuit breaker: model re-emitted a denied action repeatedly; \
                     ending turn instead of re-prompting"
                );
                return Ok(StepOutcome::Done);
            }
        }
        Ok(StepOutcome::Continue)
    }
}