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::{
    ApprovalOverride, CallContext, CallDisposition, HandoffRequest, PendingApproval, ResolvedCall,
    RunTurnOptions, ToolExecutor, append_injected_notes, cap_tool_result, forced_result,
    gate_decision, gate_missing, 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.
pub(crate) const RESUME_EXECUTED_GROUND_TRUTH_NOTE: &str = "A person approved this request and the tool has \
    already run — the results above are final. Tell the user what happened; do not describe \
    approval status, the system already showed it.";

/// 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,
    /// 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)>,
    /// Approvals still awaiting execution, keyed by the canonicalized
    /// `(id, name, args)` identity (#141). Seeded from
    /// [`RunTurnOptions::approved_call_ids`]; the resume pre-pass removes each
    /// entry it spends so neither the pre-pass nor the loop re-executes an
    /// approval the model re-emits. `args` is canonicalized through `canon_args`
    /// so a re-emit with reordered keys still matches by value.
    pub approved_remaining: HashSet<(String, String, String)>,
    /// 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,
    /// Gate clears a remembered passkey grant was solely responsible for
    /// (`#594`), accumulated across the turn's loop iterations. Each entry is an
    /// executed tool call that ran only because a grant kept a capability
    /// untrusted content in context would have revoked; the final
    /// [`TurnResult::grant_replays`](crate::TurnResult::grant_replays) carries
    /// them out for the control plane to audit.
    pub grant_replays: Vec<crate::GrantReplayClear>,
    /// 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, where no live grant covered the shape; 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>,
}

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 audit surfaces (`#594` grant replays, `#623`
    /// unattended denials) 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 {
        crate::TurnResult {
            messages: self.outputs,
            usage: self.total_usage,
            stop: self.last_stop,
            pending_approvals,
            handoff,
            grant_replays: self.grant_replays,
            unattended_denials: self.unattended_denials,
            mid_stream_failure: None,
            delegate_records: self.delegate_records,
        }
    }

    /// 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
    }
}

/// 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>),
    /// 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>;
}

/// The forced closing completion: when a turn executed tools but the model
/// never produced any user-visible text, force one final text answer so the
/// turn always yields a reply.
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 executed tools but the model never produced any
        // user-visible text, so `outputs` carries only tool calls/results — the
        // edge would post nothing (the "agent produced no text" dead-end). This
        // covers two shapes: the loop exhausting MAX_STEPS while still calling
        // tools, AND a resume whose pre-pass executed an approved call in one step
        // and then got an empty continuation (which breaks the loop far short of
        // MAX_STEPS, so the old `steps_used >= MAX_STEPS` guard let it fall through
        // silent — the approved action ran but the human saw no reply). Force ONE
        // final completion with tools disabled so the model must answer in text,
        // summarizing what it did or explaining it couldn't proceed. Skipped for an
        // intentional handoff (the parent resumes with the child's result).
        // Best-effort: a failure here leaves the turn as-is rather than erroring.
        if !(ctx.executed_tools && !ctx.produced_text && ctx.pending_handoff.is_none()) {
            return Ok(StepOutcome::Continue);
        }
        let mut req = CompletionRequest::new(ctx.model);
        req.messages.clone_from(&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.
        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.total_usage.input_tokens += turn.usage.input_tokens;
                ctx.total_usage.output_tokens += turn.usage.output_tokens;
                push_reasoning(&mut ctx.outputs, &turn.reasoning);
                if !turn.text.is_empty() {
                    ctx.outputs.push(text_message("model", &turn.text));
                }
                ctx.last_stop = turn.stop;
                tracing::info!(
                    "forced closing completion (tool loop produced no text); turn now yields a reply"
                );
            }
        }
        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::approved_call_ids`] / [`RunTurnOptions::denied_call_ids`].
/// 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
/// `approved_call_ids` 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],
    /// Approver edits (#67), keyed by the canonicalized approval identity.
    pub approved_overrides: &'a HashMap<(String, String, String), ApprovalOverride>,
    /// Verified signed denials as canonicalized `(id, name, args)` tuples.
    pub denied_call_ids: &'a HashSet<(String, String, String)>,
}

#[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.

        // Every tool_call id that already has a tool_result somewhere in the
        // transcript is "answered" and must not be re-executed.
        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();
        // Unanswered assistant tool_use blocks, paired with the index of the
        // message they live in so each synthesized result can be inserted
        // directly after its `tool_use` (preserving provider ordering).
        let mut unanswered: Vec<(usize, ToolCall)> = Vec::new();
        for (idx, m) in ctx.messages.iter().enumerate() {
            for c in &m.content {
                if let LlmContent::ToolUse(tc) = c
                    && !answered.contains(tc.id.as_str())
                {
                    unanswered.push((idx, tc.clone()));
                }
            }
        }

        // 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.approved_call_ids.is_empty() {
            let unanswered_ids: HashSet<&str> =
                unanswered.iter().map(|(_, tc)| tc.id.as_str()).collect();
            for (id, name, _args) in &ctx.options.approved_call_ids {
                if unanswered_ids.contains(id.as_str()) {
                    continue;
                }
                if answered.contains(id.as_str()) {
                    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.approved_call_ids.len(),
            denied = ctx.options.denied_call_ids.len(),
            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 dispositions: Vec<CallDisposition> = unanswered
            .iter()
            .map(|(_, tc)| {
                let gate = gate_decision(
                    ctx.tools,
                    ctx.options,
                    untrusted_in_context,
                    &tc.name,
                    &tc.args_json,
                );
                let key = (tc.id.clone(), tc.name.clone(), canon_args(&tc.args_json));
                let is_denied = self.denied_call_ids.contains(&key);
                // 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 `approved_remaining` entry still runs.
                let is_approved = ctx.approved_remaining.contains(&key)
                    || 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 {
                        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(),
                    })
                })
                .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()
            .map(|(_, tc)| {
                let key = (tc.id.clone(), tc.name.clone(), canon_args(&tc.args_json));
                resolve_approved_call(&tc.args_json, self.approved_overrides.get(&key))
            })
            .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 recorder = recorder.clone();
                async move {
                    if let Some(result) = forced {
                        result
                    } else {
                        run_and_redact(tools, recorder.as_ref(), call_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;

        // Mark each EXECUTED approval as spent so the loop below cannot re-execute
        // it if the model re-emits the same call. Only Execute consumes an
        // approval — a denial or a policy veto (#67) ran no tool.
        for ((_, tc), disposition) in unanswered.iter().zip(&dispositions) {
            if matches!(disposition, CallDisposition::Execute) {
                ctx.approved_remaining.remove(&(
                    tc.id.clone(),
                    tc.name.clone(),
                    canon_args(&tc.args_json),
                ));
            }
        }

        // 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,
            );
        }

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