polyc-agent 0.1.3

//! The agent turn loop.
//!
//! Implements the standard function-calling loop: call the provider; while it
//! asks for tools, execute them and feed the results back; repeat until the
//! model ends its turn. Provider streaming chunks are folded into a turn via
//! [`polyc_llm::turn::collect_turn`]; the assistant/tool messages are
//! mapped to wire [`Message`]s for the control plane.

use async_trait::async_trait;
use buffa_types::google::protobuf::Struct;
use polyc_llm::request::ToolCall;
use polyc_llm::{
    CompletionRequest, Content as LlmContent, LlmProvider, Message as LlmMessage, Role, StopReason,
    ToolSpec, Usage,
    turn::{collect_turn, collect_turn_observed},
};
use polyc_proto::proto::polychrome::agent::v1::{
    Content, FunctionCallContent, FunctionResultContent, Message, TextContent, ToolCallContent,
    ToolResultContent, content, function_result_content, thought_summary_content,
    tool_call_content, tool_result_content,
};

pub mod handoff;
pub mod llm_summarizer;
pub mod participation;

pub use handoff::{
    DEFAULT_MAX_CARRY, HANDOFF_TOOL_NAME, HandoffRequest, handoff_tool_spec, parse_handoff_args,
};
pub use llm_summarizer::LlmSummarizer;
/// Re-export so callers can build a streaming channel without depending on
/// `polyc-llm` directly.
pub use polyc_llm::turn::TurnStreamEvent;

/// Map an `llm`-side [`StopReason`] to the wire enum value.
///
/// Mirrors the variants 1:1 against `harness_service.proto`; `None` (no
/// stop chunk observed in the stream) maps to the proto
/// `STOP_REASON_UNSPECIFIED` zero value via [`Default`] on the caller side.
#[must_use]
pub const fn llm_stop_to_wire_i32(stop: StopReason) -> i32 {
    use polyc_proto::proto::polychrome::harness::v1::StopReason as Wire;
    match stop {
        StopReason::EndTurn => Wire::STOP_REASON_END_TURN as i32,
        StopReason::ToolUse => Wire::STOP_REASON_TOOL_USE as i32,
        StopReason::MaxTokens => Wire::STOP_REASON_MAX_TOKENS as i32,
        StopReason::Refusal => Wire::STOP_REASON_REFUSAL as i32,
        StopReason::StopSequence => Wire::STOP_REASON_STOP_SEQUENCE as i32,
        // `StopReason` is marked `#[non_exhaustive]` upstream. Any future
        // variant maps to UNSPECIFIED on the wire until this match catches
        // up — losing it on the wire is preferable to a build break.
        _ => Wire::STOP_REASON_UNSPECIFIED as i32,
    }
}

/// Inverse of [`llm_stop_to_wire_i32`]: lift a wire enum value back.
///
/// Returns `None` for `STOP_REASON_UNSPECIFIED` or any unknown wire value
/// — the caller treats that as "no stop reason observed this turn",
/// matching the in-process [`TurnResult::stop`] semantics.
#[must_use]
pub const fn wire_to_llm_stop(wire: i32) -> Option<StopReason> {
    use polyc_proto::proto::polychrome::harness::v1::StopReason as Wire;
    match wire {
        x if x == Wire::STOP_REASON_END_TURN as i32 => Some(StopReason::EndTurn),
        x if x == Wire::STOP_REASON_TOOL_USE as i32 => Some(StopReason::ToolUse),
        x if x == Wire::STOP_REASON_MAX_TOKENS as i32 => Some(StopReason::MaxTokens),
        x if x == Wire::STOP_REASON_REFUSAL as i32 => Some(StopReason::Refusal),
        x if x == Wire::STOP_REASON_STOP_SEQUENCE as i32 => Some(StopReason::StopSequence),
        _ => None,
    }
}

/// Produces a textual summary of a transcript chunk that's about to be
/// dropped from the prompt window. Implementations can be deterministic
/// (the [`StubSummarizer`]) or LLM-backed (a follow-up).
///
/// Used by the control plane's *anchored iterative summarization* pass:
/// when the replayed history exceeds [`crate::SUMMARY_TRIGGER`], the
/// summarizer compresses the oldest segment and the result is persisted as
/// a `summary` event in the conversation's event log (durable, replayable).
/// Subsequent connects find the latest summary event and skip events at-or-
/// before its covered position, so the prompt is bounded indefinitely. The
/// "anchored" part means new summaries *merge* into the persistent state —
/// the next summarizer call sees the prior summary as context, keeping
/// detail across compactions rather than re-summarizing from scratch (per
/// Factory's evaluation across 36k engineering session messages).
#[async_trait]
pub trait Summarizer: Send + Sync {
    /// Summarize `transcript` (the about-to-be-dropped chunk), in the
    /// context of `prior_summary` (the persistent state from earlier
    /// compactions, empty on first compaction). Returns the new summary
    /// text that replaces `prior_summary` going forward.
    async fn summarize(&self, prior_summary: &str, transcript: &[LlmMessage]) -> String;
}

/// Trigger threshold for summarization.
///
/// When the reconstructed history exceeds this many messages, the control
/// plane summarizes the head into a `summary` event and drops the head from
/// the prompt. Picked higher than the prompt window so summarization stays
/// a rare, batched event rather than a per-turn cost.
pub const SUMMARY_TRIGGER: usize = 40;

/// Deterministic placeholder summarizer — formats a tiny excerpt of the
/// transcript so the data path is exercisable without a provider. Real
/// deployments swap in an LLM-backed summarizer (one-trait swap).
#[derive(Clone, Copy, Default)]
pub struct StubSummarizer;

#[async_trait]
impl Summarizer for StubSummarizer {
    async fn summarize(&self, prior_summary: &str, transcript: &[LlmMessage]) -> String {
        let head = transcript
            .iter()
            .take(2)
            .filter_map(|m| match m.content.first() {
                Some(LlmContent::Text(t)) => Some(format!("{:?}: {}", m.role, snippet(t, 80))),
                _ => None,
            })
            .collect::<Vec<_>>()
            .join("; ");
        let tail = transcript
            .iter()
            .rev()
            .take(2)
            .rev()
            .filter_map(|m| match m.content.first() {
                Some(LlmContent::Text(t)) => Some(format!("{:?}: {}", m.role, snippet(t, 80))),
                _ => None,
            })
            .collect::<Vec<_>>()
            .join("; ");
        let count = transcript.len();
        if prior_summary.is_empty() {
            format!("[summary: {count} prior messages — head: {head}; tail: {tail}]")
        } else {
            format!("{prior_summary}\n[+{count} messages: head: {head}; tail: {tail}]")
        }
    }
}

fn snippet(s: &str, max: usize) -> String {
    if s.len() <= max {
        return s.to_owned();
    }
    let mut end = max;
    while !s.is_char_boundary(end) && end > 0 {
        end -= 1;
    }
    format!("{}…", &s[..end])
}

/// Executes a tool call by name, returning a JSON result string. Also
/// advertises the tools it can execute so the provider knows what's callable.
#[async_trait]
pub trait ToolExecutor: Send + Sync {
    /// Specs for the tools this executor knows how to run. The default
    /// returns an empty list — the model won't be told about any tools, so it
    /// won't emit `tool_call`s. Real registries override this.
    fn specs(&self) -> Vec<ToolSpec> {
        Vec::new()
    }

    /// Whether this executor advertises a tool named `name`.
    ///
    /// Used by composite/registry executors to route a call to its owning
    /// source without materialising every source's full [`Self::specs`] on the
    /// hot path. The default derives the answer from [`Self::specs`]; executors
    /// that cache or compute specs lazily should override with a cheaper check
    /// (e.g. a name lookup that avoids cloning the spec list).
    fn owns(&self, name: &str) -> bool {
        self.specs().iter().any(|s| s.name == name)
    }

    /// Whether `name` requires explicit human approval before [`Self::execute`]
    /// may run. The default is `false` — pure / read-only tools shouldn't
    /// trigger an approval gate. Override for sensitive tools (writes, code
    /// execution, network reach, anything with side effects).
    ///
    /// When this returns `true`, [`run_turn`] does NOT call [`Self::execute`].
    /// Instead it surfaces the unexecuted tool calls via
    /// [`TurnResult::pending_approvals`]; the caller is responsible for
    /// persisting an `approval_request` event, waiting for a (cryptographically
    /// signed) `approval_response`, and re-driving the loop on the next turn.
    fn needs_approval(&self, _name: &str) -> bool {
        false
    }

    /// Run `name` with JSON `args_json`; return a JSON result.
    async fn execute(&self, name: &str, args_json: &str) -> String;
}

/// Placeholder executor: advertises no tools and reports any call it
/// receives as unhandled (the model shouldn't call anything without specs,
/// but the guard keeps the loop progressing if it does).
#[derive(Clone, Copy, Default)]
pub struct StubTools;

#[async_trait]
impl ToolExecutor for StubTools {
    async fn execute(&self, name: &str, args_json: &str) -> String {
        format!(r#"{{"unhandled_tool":"{name}","args":{args_json}}}"#)
    }
}

/// Cap on provider↔tool round-trips, guarding against a runaway loop.
const MAX_STEPS: usize = 8;

/// Circuit-breaker bound (Anthropic-style) on how many times the model may
/// re-emit an action the human already denied before the turn is cut short.
///
/// A denial is keyed to the tool *signature* (name + args) — see `denied_sigs`
/// in [`run_turn_with`] — so a re-emitted denied call (same action, fresh
/// provider call-id) is auto-denied without re-prompting the human. But the
/// model can still burn the whole `MAX_STEPS` budget re-emitting it. Once this
/// many loop iterations have resolved a *signature-matched* terminal denial
/// (distinct from the first signed denial), the loop breaks so the turn ends
/// cleanly instead of looping the same dead-end.
const MAX_DENIAL_REPROMPTS: usize = 2;

/// Synthetic `tool_result` payload emitted for a tool call the human approver
/// denied. Mirrors the JSON shape a real executor would return so the model
/// reads it as an ordinary (failed) result and the function-calling loop closes
/// instead of re-pausing the turn forever.
const DENIAL_RESULT_JSON: &str = r#"{"approved":false,"error":"denied by human approver"}"#;

/// One tool call awaiting human-in-the-loop approval.
///
/// Emitted by [`run_turn`] when [`ToolExecutor::needs_approval`] returns
/// `true` for a tool the model wants to call. The caller surfaces these to
/// the human / approver, persists an `approval_request` event per entry, and
/// re-drives the loop once a matching `approval_response` event lands.
///
/// `id` matches the provider's tool-call id (so the assistant's tool-use
/// content block lines up with the eventual tool-result), and is also used as
/// the `request_id` on the wire `approval_request` event payload.
#[derive(Debug, Clone)]
pub struct PendingApproval {
    /// Provider-assigned tool-call id; also used as the approval `request_id`.
    pub id: String,
    /// Tool name, as advertised by [`ToolExecutor::specs`]. Raw machine
    /// identifier; the field of record for trust/audit (unchanged in the
    /// event log).
    pub name: String,
    /// Arguments as a JSON string (opaque at this layer).
    pub args_json: String,
    /// Human display label (MCP-style `title`) for the tool, carried from the
    /// harness wire for presentation in the approval prompt. May be empty when
    /// the harness produced no label; renderers derive one from
    /// [`name`](Self::name) then.
    pub title: String,
}

/// Output of one [`run_turn`] call.
///
/// Carries the wire messages produced (assistant text and tool results),
/// the aggregated usage across every provider call in the loop, and the
/// stop reason from the final step.
///
/// When [`Self::pending_approvals`] is non-empty the turn is *paused*:
/// the model asked for one or more sensitive tools, [`run_turn`] short-
/// circuited before executing them, and the caller must capture a
/// human-in-the-loop decision (idiomatic Temporal "signal" pattern) before
/// re-driving. The choice to surface this as a result field rather than an
/// out-of-band callback keeps `run_turn` pure (no side-effect handle), keeps
/// the durability boundary at the caller (the event log already gives us
/// replay), and lets the per-conversation Mutex / Lease release while we
/// wait — matching the durable-workflow pattern.
#[derive(Debug, Default, Clone)]
pub struct TurnResult {
    /// Wire messages — assistant text + tool result messages, in order.
    pub messages: Vec<Message>,
    /// Sum of `input_tokens` / `output_tokens` across every provider call
    /// this turn made (the function-calling loop may iterate multiple times).
    pub usage: Usage,
    /// Stop reason of the final provider step.
    pub stop: Option<StopReason>,
    /// Tool calls awaiting human approval. Empty in the common case; when
    /// non-empty, the turn paused before executing any tool in this batch.
    pub pending_approvals: Vec<PendingApproval>,
    /// Populated when the model emitted the reserved `__handoff_to` tool
    /// call. The loop suspends without executing any further tools and the
    /// caller (control plane) is expected to create a child conversation,
    /// emit a signed [`polyc_proto::proto::polychrome::handoff::v1::Handoff`]
    /// event into the parent's eventlog, and resume the parent's turn once a
    /// `HandoffReturn` lands.
    ///
    /// If multiple `__handoff_to` calls appear in the same tool batch (the
    /// model emitted two at once), only the first is honored — fan-out is a
    /// V2 concern and the wire shape doesn't model parallel children today.
    pub handoff: Option<HandoffRequest>,
}

/// Options for a single [`run_turn`] invocation.
///
/// A small builder-style struct rather than a long parameter list — keeps the
/// hot-path call sites readable (`RunTurnOptions::default()`) and gives the
/// HITL-resume path a typed slot for the approved-call-ids set without adding
/// a third positional `HashSet` argument every existing caller would have to
/// thread through.
#[derive(Debug, Default, Clone)]
pub struct RunTurnOptions {
    /// Provider-assigned tool-call ids the caller has previously gathered
    /// signed HITL approvals for. When [`ToolExecutor::needs_approval`]
    /// returns `true` for a tool call, the loop checks this set: if the
    /// call's id is present, the tool executes as normal; if absent, the
    /// loop pauses with a fresh [`PendingApproval`] as today.
    ///
    /// Used by the control plane → harness resume cycle: the control plane
    /// replays the conversation's event log, collects every verified
    /// `approval_response` that isn't yet answered by a matching `tool_result`
    /// message in the transcript, and passes the set here so the harness
    /// re-drives the function-calling loop with the previously-paused tools
    /// executed.
    ///
    /// Each entry is the signed `(request_id, tool_name, args_json)` tuple — the
    /// approval is bound to that exact call (#141), so a re-emitted same-id call
    /// with different args/tool does NOT inherit the approval (it re-pauses).
    pub approved_call_ids: std::collections::HashSet<(String, String, String)>,

    /// Verified signed HITL *denials* as `(request_id, tool_name, args_json)`
    /// tuples (a verified `approval_response` with `approved == false`).
    ///
    /// A denial must RESOLVE the call, not leave it pending: when
    /// [`ToolExecutor::needs_approval`] returns `true` for a call whose
    /// `(id, name, args)` tuple is in this set, the loop emits a synthetic denial
    /// `tool_result` (carrying `{"approved":false,"error":"denied by human
    /// approver"}`) WITHOUT executing the tool and WITHOUT re-pausing. As with
    /// approvals the denial is bound to the exact call — the same id with
    /// different args is a new request, not an inherited denial.
    ///
    /// A call needing approval that is in neither [`Self::approved_call_ids`]
    /// nor this set still pends as before.
    pub denied_call_ids: std::collections::HashSet<(String, String, String)>,

    /// When set, the turn loop forwards each [`TurnStreamEvent`] (text delta,
    /// tool start) as it arrives, so a caller can stream partial output
    /// mid-turn (the harness forwards these over its bidi stream → control
    /// plane → Slack `chat.appendStream`). `None` keeps the buffered path:
    /// the full [`TurnResult`] is always returned regardless.
    pub stream_tx: Option<futures::channel::mpsc::UnboundedSender<TurnStreamEvent>>,

    /// When `true`, each request this turn sets [`CompletionRequest::web_search`]
    /// so the provider offers the model public-web grounding (Vertex Gemini maps
    /// it to the `googleSearch` tool). Only the answering loop sets this; the
    /// summarizer and classifier build their own requests and never enable it.
    pub web_search: bool,
}

tokio::task_local! {
    /// The id of the tool call currently being executed by [`run_turn_with`].
    /// Scoped only around each individual `tools.execute(..)` call.
    static CURRENT_TOOL_CALL_ID: String;
}

/// Returns the provider-assigned id of the tool call currently executing, when
/// called from within a [`run_turn_with`] tool execution; `None` outside that
/// scope.
///
/// The harness's payment-proxy tool reads this to correlate its mid-turn
/// `PaidFetchRequest` with the approved tool call (the control plane binds the
/// request to the matching signed `approval_response` before signing). Kept as
/// a task-local so [`ToolExecutor::execute`]'s signature stays unchanged.
#[must_use]
pub fn current_tool_call_id() -> Option<String> {
    CURRENT_TOOL_CALL_ID.try_with(String::clone).ok()
}

/// Runs `fut` with [`current_tool_call_id`] set to `id` for its duration.
///
/// `run_turn_with` already scopes this around each tool execution; this helper
/// is exposed for callers/tests that need to drive a tool body as if it were
/// executing tool call `id` (e.g. the harness payment-proxy tool's tests).
pub async fn with_tool_call_id<F>(id: String, fut: F) -> F::Output
where
    F: std::future::Future,
{
    CURRENT_TOOL_CALL_ID.scope(id, fut).await
}

/// Run one agent turn to completion with no caller-supplied options (the
/// common path; equivalent to `run_turn_with(..., RunTurnOptions::default())`).
///
/// # Errors
///
/// Propagates the provider's error.
pub async fn run_turn<P, T>(
    provider: &P,
    tools: &T,
    model: &str,
    messages: Vec<LlmMessage>,
) -> Result<TurnResult, P::Error>
where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    run_turn_with(provider, tools, model, messages, RunTurnOptions::default()).await
}

/// Single-pass HITL classification of one tool call in a batch (see the
/// classification step in [`run_turn_with`]). Computed once per call so the
/// pause decision and the resolve decision can't drift apart.
enum CallDisposition {
    /// Needs approval, but neither approved nor denied — must pause the batch.
    Pending,
    /// Needs approval and carries a signed/sticky denial — auto-denied (no
    /// pause). `sig_match` is true when the denial came from the sticky
    /// `denied_sigs` set (a re-emitted already-denied action) rather than the
    /// first call-id denial; only `sig_match` denials drive the circuit breaker.
    Denied { sig_match: bool },
    /// Approved, or never gated — execute it.
    Execute,
}

/// Run one agent turn to completion with caller-supplied [`RunTurnOptions`].
///
/// Used by the harness when resuming a previously-paused turn: the
/// `approved_call_ids` set lets the function-calling loop execute the
/// specific tool calls a human has signed off on while still pausing on any
/// other `needs_approval=true` calls that haven't been approved.
///
/// # Errors
///
/// Propagates the provider's error.
#[allow(clippy::too_many_lines)] // cohesive function-calling loop
#[tracing::instrument(skip_all, fields(model = model, input_messages = messages.len(), approved = options.approved_call_ids.len(), denied = options.denied_call_ids.len()))]
pub async fn run_turn_with<P, T>(
    provider: &P,
    tools: &T,
    model: &str,
    mut messages: Vec<LlmMessage>,
    options: RunTurnOptions,
) -> Result<TurnResult, P::Error>
where
    P: LlmProvider + ?Sized,
    T: ToolExecutor + ?Sized,
{
    let mut outputs = Vec::new();
    let mut total_usage = Usage::default();
    let mut last_stop: Option<StopReason> = None;
    let mut pending_handoff: Option<HandoffRequest> = None;
    // STICKY/TERMINAL DENIAL set, keyed to the tool *signature* (name +
    // args_json) 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; that new id isn't in `options.denied_call_ids`, so a
    // call-id-only check would re-pause and re-prompt the human for something
    // they already rejected. Recording the signature here makes the denial
    // stick across re-emits: a matching call is auto-denied (synthetic result)
    // without ever pausing again.
    let mut denied_sigs: std::collections::HashSet<(String, String)> =
        std::collections::HashSet::new();
    // Circuit-breaker counter: how many loop iterations have resolved a
    // signature-matched terminal denial (the model retrying an already-denied
    // action). The first signed denial — by call-id, before any signature is
    // recorded — does NOT count; only re-emits of an already-denied signature
    // do. When this reaches `MAX_DENIAL_REPROMPTS` the loop breaks.
    let mut denial_reprompts: usize = 0;
    for _ in 0..MAX_STEPS {
        // Advertise the tool specs FRESH each iteration. Re-reading `specs()` per
        // step lets an executor that DEFERS tools (progressive disclosure behind a
        // `search_tools` meta-tool) grow the advertised set within a turn: after
        // the model searches and the executor unlocks the matched tools, they
        // appear on the next request so the model can call them. A static registry
        // returns the same set each step, so its behavior is unchanged (just a
        // re-clone). The reserved `__handoff_to` primitive is always advertised;
        // a real registry that declares that name is short-circuited below.
        let mut tool_specs = tools.specs();
        if !tool_specs.iter().any(|s| s.name == HANDOFF_TOOL_NAME) {
            tool_specs.push(handoff_tool_spec());
        }
        let mut req = CompletionRequest::new(model);
        req.messages.clone_from(&messages);
        req.tools = tool_specs.clone();
        req.web_search = options.web_search;
        let stream = provider.complete(req).await?;
        let turn = if let Some(tx) = options.stream_tx.clone() {
            // Forward deltas live; an unbounded send never blocks the fold.
            collect_turn_observed(stream, move |ev| {
                let _ = tx.unbounded_send(ev);
            })
            .await?
        } else {
            collect_turn(stream).await?
        };
        total_usage.input_tokens += turn.usage.input_tokens;
        total_usage.output_tokens += turn.usage.output_tokens;
        last_stop = turn.stop;

        if !turn.text.is_empty() {
            outputs.push(text_message("model", &turn.text));
        }
        // Persist the assistant's tool calls *structurally* (not as text), so
        // eventlog replay reconstructs a real tool_use/tool_result pair —
        // carrying the provider signature — instead of a lossy `[tool_call:id]`
        // marker. These render as `ToolStarted` (ignored) downstream, never as
        // user-visible reply text.
        for tc in &turn.tool_calls {
            outputs.push(tool_call_message(tc));
        }

        // Reflect the assistant turn back onto the transcript.
        let mut assistant = LlmMessage::assistant(turn.text.clone());
        for tc in &turn.tool_calls {
            // Preserve the provider signature (e.g. a thinking model's thought
            // signature) so the next request — which carries this call in the
            // history — echoes it back; some providers reject the follow-up
            // otherwise.
            assistant.content.push(LlmContent::tool_use_signed(
                tc.id.clone(),
                tc.name.clone(),
                tc.args_json.clone(),
                tc.signature.clone(),
            ));
        }
        messages.push(assistant);

        // Execute tool calls whenever the model emitted any — don't gate on
        // `stop == ToolUse`. Providers can report a normal terminal stop
        // alongside tool calls (some stream the tool call and the end-of-turn
        // marker as separate events), and skipping execution there would
        // strand the turn with no output. BUT a hard stop (MaxTokens/Refusal)
        // means the output was truncated or refused — the tool call may be
        // incomplete (e.g. partial args JSON), so do NOT execute it.
        let wants_tools = !turn.tool_calls.is_empty()
            && !matches!(turn.stop, Some(StopReason::MaxTokens | StopReason::Refusal));
        if !wants_tools {
            break;
        }

        // Short-circuit (handoff): if any of the tool calls is the reserved
        // handoff name, suspend the turn immediately — do NOT execute the
        // companion tools in the batch, and do NOT feed any tool_results back
        // to the provider. The control plane sees `handoff = Some(..)` on the
        // returned `TurnResult` and takes over: it creates the child
        // conversation and writes the signed `Handoff` event. On the parent's
        // *next* turn the resumed transcript will include the `__handoff_to`
        // call + its `HandoffReturn`-derived result, so the function-calling
        // loop closes cleanly.
        if let Some(tc) = turn.tool_calls.iter().find(|t| t.name == HANDOFF_TOOL_NAME)
            && let Some(req) = handoff::parse_handoff_args(&tc.id, &tc.args_json, &messages)
        {
            pending_handoff = Some(req);
            break;
        }

        // HITL approval gate: if ANY tool in this batch needs human approval
        // *and* the caller hasn't already supplied a signed approval for it,
        // pause the entire batch — execute nothing, surface every still-
        // unapproved call so the caller can route them through approval
        // together. Atomicity matters: the model's prompt sees either all
        // results (after every approval lands) or no results (paused). Mixed
        // batches with some pre-executed read-only tools would force the
        // rest into a different batch on resume and confuse the model's
        // tool_use accounting.
        //
        // On a resumed turn the caller passes the set of previously-approved
        // call ids via `options.approved_call_ids` and the set of denied ids
        // via `options.denied_call_ids`. Tools whose id is approved execute as
        // normal; tools whose id is denied resolve to a synthetic denial
        // result (below) without executing; only tools that still need approval
        // but have neither a signed approval nor a signed denial cause the
        // pause.
        // SINGLE-PASS CLASSIFICATION. Classify every tool call in the batch
        // exactly once into one of three dispositions, then act on the batch
        // as a whole. This replaces the old `still_needs_approval` closure +
        // the inline `denied = ...` recomputation, which evaluated the same
        // predicates twice and drifted apart easily.
        //
        // A call is DENIED if its id carries a signed denial
        // (`options.denied_call_ids`) OR its signature is already in the
        // sticky `denied_sigs` set (the model re-emitted an already-denied
        // action with a fresh call-id). A denied call NEVER pauses — it
        // resolves to a synthetic denial result directly.
        let dispositions = turn
            .tool_calls
            .iter()
            .map(|tc| {
                let needs_approval = tools.needs_approval(&tc.name);
                let sig = (tc.name.clone(), tc.args_json.clone());
                let sig_denied = denied_sigs.contains(&sig);
                // The approval/denial is bound to the EXACT (id, name, args) tuple
                // the human signed (#141): a re-emitted same-id call with changed
                // args/tool matches neither set, so it re-pauses rather than
                // inheriting the prior verdict.
                let approval_key = (tc.id.clone(), tc.name.clone(), tc.args_json.clone());
                let is_denied = options.denied_call_ids.contains(&approval_key) || sig_denied;
                let is_approved = options.approved_call_ids.contains(&approval_key);
                if needs_approval && is_denied {
                    // A signature match means the model re-emitted an
                    // already-denied action; a call-id-only denial is the
                    // first signed denial (does not count toward the breaker).
                    CallDisposition::Denied {
                        sig_match: sig_denied,
                    }
                } else if needs_approval && !is_approved {
                    CallDisposition::Pending
                } else {
                    CallDisposition::Execute
                }
            })
            .collect::<Vec<_>>();

        // PAUSE the whole batch iff ANY call is Pending — preserving the
        // atomic-batch semantics (the model's prompt sees either all results
        // or none) and the existing `PendingApproval` surface. Denied calls
        // do NOT trigger a pause; they resolve below.
        let batch_needs_approval = dispositions
            .iter()
            .any(|d| matches!(d, CallDisposition::Pending));
        if batch_needs_approval {
            let pending = turn
                .tool_calls
                .iter()
                .zip(&dispositions)
                .filter(|(_, d)| matches!(d, CallDisposition::Pending))
                .map(|(tc, _)| {
                    // Carry the tool's curated display title (the MCP-style
                    // annotation) when its spec advertised one; empty otherwise
                    // (downstream derives a label from `name`). The raw `name`
                    // remains the audit identifier.
                    let title = tool_specs
                        .iter()
                        .find(|s| s.name == tc.name)
                        .and_then(|s| s.title.clone())
                        .unwrap_or_default();
                    PendingApproval {
                        id: tc.id.clone(),
                        name: tc.name.clone(),
                        args_json: tc.args_json.clone(),
                        title,
                    }
                })
                .collect::<Vec<_>>();
            return Ok(TurnResult {
                messages: outputs,
                usage: total_usage,
                stop: last_stop,
                pending_approvals: pending,
                handoff: None,
            });
        }

        // Resolve each tool call per its disposition. Denied calls get a
        // synthetic denial result (NOT executed) and record their signature in
        // `denied_sigs` so any later re-emit is auto-denied; every Execute call
        // runs concurrently via join_all (denials are instant). Results are
        // gathered in `turn.tool_calls` order so the next provider call sees
        // the same shape as a sequential loop.
        //
        // The denial result payload (`DENIAL_RESULT_JSON`) mirrors the JSON
        // shape an executor would return, so the model reads it as an ordinary
        // (failed) tool_result and the function-calling loop closes cleanly
        // instead of re-pausing.
        let mut saw_sig_match_denial = false;
        let tool_futures = turn
            .tool_calls
            .iter()
            .zip(&dispositions)
            .map(|(tc, disposition)| {
                let denied = matches!(disposition, CallDisposition::Denied { .. });
                if let CallDisposition::Denied { sig_match } = disposition {
                    // Make the denial sticky for this turn: future re-emits of
                    // the same action are auto-denied without re-prompting.
                    denied_sigs.insert((tc.name.clone(), tc.args_json.clone()));
                    if *sig_match {
                        saw_sig_match_denial = true;
                    }
                }
                let name = tc.name.clone();
                let args = tc.args_json.clone();
                let call_id = tc.id.clone();
                async move {
                    if denied {
                        DENIAL_RESULT_JSON.to_owned()
                    } else {
                        // Scope the call id as a task-local for the duration of
                        // this one execution, so a tool (e.g. the harness payment
                        // proxy) can correlate without an `execute` signature
                        // change. Same task as the tool body — task-local is
                        // visible inside `execute`.
                        CURRENT_TOOL_CALL_ID
                            .scope(call_id, tools.execute(&name, &args))
                            .await
                    }
                }
            })
            .collect::<Vec<_>>();
        let results = futures::future::join_all(tool_futures).await;
        for (tc, result) in turn.tool_calls.iter().zip(results) {
            // Structured tool result (not text) so replay reconstructs a real
            // tool_result keyed to its call id (pairs with the tool_call above).
            outputs.push(tool_result_message(&tc.id, &result));
            messages.push(LlmMessage {
                role: Role::Tool,
                content: vec![LlmContent::tool_result(tc.id.clone(), result, false)],
            });
        }

        // CIRCUIT BREAKER: if this step resolved 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 — break so the turn ends cleanly with the
        // last stop reason instead of burning the rest of `MAX_STEPS` looping
        // the same dead-end. tool_results for this step are already appended
        // above, so the transcript stays well-formed.
        if saw_sig_match_denial {
            denial_reprompts += 1;
            if denial_reprompts >= MAX_DENIAL_REPROMPTS {
                tracing::warn!(
                    denial_reprompts,
                    max = MAX_DENIAL_REPROMPTS,
                    "HITL circuit breaker: model re-emitted a denied action repeatedly; \
                     ending turn instead of re-prompting"
                );
                break;
            }
        }
    }
    Ok(TurnResult {
        messages: outputs,
        usage: total_usage,
        stop: last_stop,
        pending_approvals: Vec::new(),
        handoff: pending_handoff,
    })
}

/// Convert an llm [`LlmMessage`] back to a wire [`Message`] for transmission
/// over `HarnessService`.
///
/// Lossy on multi-content messages (concatenates text content blocks);
/// non-text variants are skipped — the assumption is that the caller has
/// already normalized via [`wire_to_llm`] and windowing, so each llm
/// message carries a single text content block by construction.
#[must_use]
pub fn llm_to_wire(msg: &LlmMessage) -> Message {
    let role = match msg.role {
        Role::Assistant => "model",
        Role::Tool => "tool",
        Role::System => "system",
        // User and any future non-exhaustive variant map to wire "user".
        _ => "user",
    };
    let text = msg
        .content
        .iter()
        .filter_map(|c| match c {
            LlmContent::Text(s) => Some(s.as_str()),
            _ => None,
        })
        .collect::<Vec<_>>()
        .join("");
    text_message(role, &text)
}

/// Convert an owned wire [`Message`] into an llm [`LlmMessage`].
///
/// Preserves the role and reconstructs faithful content so a replayed
/// transcript carries the same tool and reasoning state the model emitted
/// originally — not lossy placeholders. Concretely:
/// - text survives verbatim;
/// - a tool call surfaces as [`LlmContent::tool_use`] carrying the real
///   function name and JSON-encoded arguments;
/// - a tool result surfaces as [`LlmContent::tool_result`] carrying the
///   JSON-encoded result payload keyed by its originating call id;
/// - model reasoning surfaces as text built from the thought summary parts.
///
/// Image / audio / document / video / confirmation variants surface as no
/// content (no fabrication). The inverse of [`text_message`]; both bridges
/// live here so the wire ↔ llm conversion has one canonical owner used by the
/// control plane (eventlog replay) and the harness (`HarnessService` input).
#[must_use]
pub fn wire_to_llm(msg: &Message) -> LlmMessage {
    let role = match msg.role.as_str() {
        "model" | "assistant" => Role::Assistant,
        "tool" | "function" => Role::Tool,
        "system" => Role::System,
        _ => Role::User,
    };
    let content = match msg.content.as_option().and_then(|c| c.r#type.as_ref()) {
        Some(content::Type::Text(t)) => vec![LlmContent::Text(t.text.clone())],
        Some(content::Type::ToolCall(tc)) => {
            // The function name and arguments live on the inner FunctionCall
            // oneof. Arguments are a structured `Struct` on the wire; serialize
            // it to the JSON-string `args_json` the llm layer expects. Fall
            // back to an empty name / `{}` args when either is absent so a
            // partial call still replays as a well-formed tool_use.
            let (name, args_json) = match tc.r#type.as_ref() {
                Some(tool_call_content::Type::FunctionCall(fc)) => {
                    let args_json = fc
                        .arguments
                        .as_option()
                        .and_then(|s| serde_json::to_string(s).ok())
                        .unwrap_or_else(|| "{}".to_owned());
                    (fc.name.clone(), args_json)
                }
                None => (String::new(), "{}".to_owned()),
            };
            // Recover the provider signature (stored as bytes on the wire) so
            // a replayed tool call still echoes it back on the next request.
            let signature = (!tc.signature.is_empty())
                .then(|| String::from_utf8_lossy(&tc.signature).into_owned());
            vec![LlmContent::tool_use_signed(
                tc.id.clone(),
                name,
                args_json,
                signature,
            )]
        }
        Some(content::Type::ToolResult(tr)) => {
            // The result payload is a structured `Struct` on the inner
            // FunctionResult oneof; serialize it to the JSON-string the llm
            // layer expects. Replayed results are observed history, never
            // errors, so `is_error` is false.
            let result_json = match tr.r#type.as_ref() {
                Some(tool_result_content::Type::FunctionResult(fr)) => match fr.result.as_ref() {
                    Some(function_result_content::Result::Response(resp)) => {
                        serde_json::to_string(resp).unwrap_or_else(|_| "{}".to_owned())
                    }
                    None => "{}".to_owned(),
                },
                None => "{}".to_owned(),
            };
            vec![LlmContent::tool_result(
                tr.call_id.clone(),
                result_json,
                false,
            )]
        }
        Some(content::Type::Thought(t)) => {
            // Reasoning has no top-level text; its `summary` repeated field
            // carries the text parts. Concatenate them so the reasoning
            // survives the round trip; skip entirely when empty rather than
            // emit a blank text block.
            let mut buf = String::new();
            for s in &t.summary {
                if let Some(thought_summary_content::Type::Text(text)) = s.r#type.as_ref()
                    && !text.text.is_empty()
                {
                    if !buf.is_empty() {
                        buf.push(' ');
                    }
                    buf.push_str(&text.text);
                }
            }
            if buf.is_empty() {
                Vec::new()
            } else {
                vec![LlmContent::Text(buf)]
            }
        }
        // Image / audio / document / video / confirmation: skip rather than
        // fabricate a misleading text representation.
        _ => Vec::new(),
    };
    LlmMessage { role, content }
}

/// Build a wire [`Message`] carrying a structured tool call.
///
/// Preserves the provider signature (e.g. a thinking model's thought
/// signature) as the `ToolCallContent.signature` bytes. Persisted to the event
/// log so replay reconstructs a real `tool_use` (paired with
/// [`tool_result_message`]) instead of a lossy text marker, and the signature
/// survives to be echoed back on the next request. Rendered as an (ignored)
/// tool-start downstream — never as user-visible reply text.
#[must_use]
pub fn tool_call_message(tc: &ToolCall) -> Message {
    let arguments = serde_json::from_str::<Struct>(&tc.args_json)
        .map(buffa::MessageField::some)
        .unwrap_or_default();
    Message {
        role: "model".to_owned(),
        content: buffa::MessageField::some(Content {
            r#type: Some(content::Type::ToolCall(Box::new(ToolCallContent {
                id: tc.id.clone(),
                signature: tc
                    .signature
                    .clone()
                    .map(String::into_bytes)
                    .unwrap_or_default(),
                r#type: Some(tool_call_content::Type::FunctionCall(Box::new(
                    FunctionCallContent {
                        name: tc.name.clone(),
                        arguments,
                        ..Default::default()
                    },
                ))),
                ..Default::default()
            }))),
            ..Default::default()
        }),
        internal_only: false,
        ..Default::default()
    }
}

/// Build a wire [`Message`] carrying a structured tool result keyed to its
/// originating `call_id`.
///
/// The inverse of the tool-result arm of [`wire_to_llm`]; persisted so replay
/// reconstructs a real `tool_result`.
#[must_use]
pub fn tool_result_message(call_id: &str, result_json: &str) -> Message {
    let response = serde_json::from_str::<Struct>(result_json)
        .ok()
        .map(|s| function_result_content::Result::Response(Box::new(s)));
    Message {
        role: "tool".to_owned(),
        content: buffa::MessageField::some(Content {
            r#type: Some(content::Type::ToolResult(Box::new(ToolResultContent {
                call_id: call_id.to_owned(),
                r#type: Some(tool_result_content::Type::FunctionResult(Box::new(
                    FunctionResultContent {
                        result: response,
                        ..Default::default()
                    },
                ))),
                ..Default::default()
            }))),
            ..Default::default()
        }),
        internal_only: false,
        ..Default::default()
    }
}

/// Build a wire [`Message`] carrying a single text content block.
///
/// Shared by the turn loop and by the control plane's eventlog write path; one
/// owner of the wire-message construction prevents the two from drifting.
#[must_use]
pub fn text_message(role: &str, text: &str) -> Message {
    Message {
        role: role.to_owned(),
        content: buffa::MessageField::some(Content {
            r#type: Some(content::Type::Text(Box::new(TextContent {
                text: text.to_owned(),
                ..Default::default()
            }))),
            ..Default::default()
        }),
        internal_only: false,
        ..Default::default()
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use futures::{StreamExt, stream};
    use polyc_llm::{Chunk, CompletionRequest, LlmProvider, error::DummyError, turn::StubProvider};
    use std::sync::atomic::{AtomicUsize, Ordering};

    use super::*;

    #[tokio::test]
    async fn stub_turn_yields_one_assistant_message() {
        let out = run_turn(
            &StubProvider,
            &StubTools,
            "stub",
            vec![LlmMessage::user("hi")],
        )
        .await
        .expect("turn");
        assert_eq!(out.messages.len(), 1);
        assert_eq!(out.messages[0].role, "model");
        assert!(out.pending_approvals.is_empty());
    }

    /// Provider that emits a single tool_call on the first complete() and
    /// EndTurn on subsequent calls. Lets us drive a deterministic two-step
    /// function-calling loop in tests.
    struct ScriptedToolCallProvider {
        calls: AtomicUsize,
    }

    #[async_trait]
    impl LlmProvider for ScriptedToolCallProvider {
        type Error = DummyError;

        async fn complete(
            &self,
            _req: CompletionRequest,
        ) -> Result<futures::stream::BoxStream<'static, Result<Chunk, Self::Error>>, Self::Error>
        {
            let n = self.calls.fetch_add(1, Ordering::SeqCst);
            let chunks = if n == 0 {
                vec![
                    Ok(Chunk::tool_call_start("call-1", "dangerous_tool")),
                    Ok(Chunk::tool_call_args_delta("call-1", r#"{"rm":"-rf"}"#)),
                    Ok(Chunk::tool_call_end("call-1")),
                    Ok(Chunk::Stop(polyc_llm::StopReason::ToolUse)),
                ]
            } else {
                vec![
                    Ok(Chunk::text_delta("done")),
                    Ok(Chunk::Stop(polyc_llm::StopReason::EndTurn)),
                ]
            };
            Ok(stream::iter(chunks).boxed())
        }
    }

    /// Tracking executor: records every execute() call and declares
    /// `dangerous_tool` as needing approval. Used to prove that a needs-
    /// approval batch is NEVER executed by `run_turn`.
    #[derive(Default)]
    struct ApprovalGatedTools {
        executed: std::sync::Mutex<Vec<String>>,
    }

    #[async_trait]
    impl ToolExecutor for ApprovalGatedTools {
        fn needs_approval(&self, name: &str) -> bool {
            name == "dangerous_tool"
        }
        async fn execute(&self, name: &str, args_json: &str) -> String {
            self.executed.lock().unwrap().push(name.to_owned());
            format!(r#"{{"ran":"{name}","args":{args_json}}}"#)
        }
    }

    #[tokio::test]
    async fn needs_approval_tool_pauses_with_pending_approval() {
        let provider = ScriptedToolCallProvider {
            calls: AtomicUsize::new(0),
        };
        let tools = ApprovalGatedTools::default();
        let out = run_turn(&provider, &tools, "scripted", vec![LlmMessage::user("hi")])
            .await
            .expect("turn");
        assert_eq!(
            out.pending_approvals.len(),
            1,
            "needs_approval tool short-circuits the loop"
        );
        let pa = &out.pending_approvals[0];
        assert_eq!(pa.id, "call-1");
        assert_eq!(pa.name, "dangerous_tool");
        assert_eq!(pa.args_json, r#"{"rm":"-rf"}"#);
        assert!(
            tools.executed.lock().unwrap().is_empty(),
            "execute() must not be called when needs_approval=true"
        );
    }

    #[tokio::test]
    async fn pending_approval_default_is_empty() {
        // The common path: a tool-less turn returns an empty pending list so
        // callers can use the field unconditionally.
        let out = run_turn(
            &StubProvider,
            &StubTools,
            "stub",
            vec![LlmMessage::user("hi")],
        )
        .await
        .expect("turn");
        assert!(out.pending_approvals.is_empty());
    }

    /// Read-only tool that does NOT need approval. Used to prove a non-
    /// sensitive batch still executes through the normal path.
    #[derive(Default)]
    struct ReadOnlyTools;

    #[async_trait]
    impl ToolExecutor for ReadOnlyTools {
        async fn execute(&self, _name: &str, _args_json: &str) -> String {
            r#"{"result":"ok"}"#.to_owned()
        }
    }

    /// Scripted provider that emits a single benign tool_call then ends.
    struct ScriptedBenignProvider {
        calls: AtomicUsize,
    }

    #[async_trait]
    impl LlmProvider for ScriptedBenignProvider {
        type Error = DummyError;

        async fn complete(
            &self,
            _req: CompletionRequest,
        ) -> Result<futures::stream::BoxStream<'static, Result<Chunk, Self::Error>>, Self::Error>
        {
            let n = self.calls.fetch_add(1, Ordering::SeqCst);
            let chunks = if n == 0 {
                vec![
                    Ok(Chunk::tool_call_start("call-1", "read_only")),
                    Ok(Chunk::tool_call_args_delta("call-1", "{}")),
                    Ok(Chunk::tool_call_end("call-1")),
                    Ok(Chunk::Stop(polyc_llm::StopReason::ToolUse)),
                ]
            } else {
                vec![
                    Ok(Chunk::text_delta("done")),
                    Ok(Chunk::Stop(polyc_llm::StopReason::EndTurn)),
                ]
            };
            Ok(stream::iter(chunks).boxed())
        }
    }

    #[tokio::test]
    async fn previously_approved_tool_executes_on_resume() {
        // Drive `run_turn_with` with the same scripted provider + gated tool
        // executor as the pause test, but populate `approved_call_ids` with
        // the call id the harness would carry on a resumed turn. The tool
        // must execute (executor.executed records the call) and no
        // pending_approvals must be surfaced.
        let provider = ScriptedToolCallProvider {
            calls: AtomicUsize::new(0),
        };
        let tools = ApprovalGatedTools::default();
        let mut approved = std::collections::HashSet::new();
        approved.insert((
            "call-1".to_owned(),
            "dangerous_tool".to_owned(),
            r#"{"rm":"-rf"}"#.to_owned(),
        ));
        let out = run_turn_with(
            &provider,
            &tools,
            "scripted",
            vec![LlmMessage::user("hi")],
            RunTurnOptions {
                approved_call_ids: approved,
                ..Default::default()
            },
        )
        .await
        .expect("turn");
        assert!(
            out.pending_approvals.is_empty(),
            "approved call must NOT re-pause the loop"
        );
        let executed = tools.executed.lock().unwrap().clone();
        assert_eq!(
            executed,
            vec!["dangerous_tool".to_owned()],
            "tool executes after approval lands"
        );
    }

    /// #141 regression: an approval bound to one (id, name, args) tuple must NOT
    /// authorize a same-id call with DIFFERENT args. The gate re-pauses instead
    /// of inheriting the approval.
    #[tokio::test]
    async fn approval_does_not_inherit_across_changed_args() {
        let provider = ScriptedToolCallProvider {
            calls: AtomicUsize::new(0),
        };
        let tools = ApprovalGatedTools::default();
        // Approve the SAME call-id + tool but DIFFERENT args than the scripted
        // call actually emits (`{"rm":"-rf"}`).
        let mut approved = std::collections::HashSet::new();
        approved.insert((
            "call-1".to_owned(),
            "dangerous_tool".to_owned(),
            r#"{"rm":"/tmp/safe"}"#.to_owned(),
        ));
        let out = run_turn_with(
            &provider,
            &tools,
            "scripted",
            vec![LlmMessage::user("hi")],
            RunTurnOptions {
                approved_call_ids: approved,
                ..Default::default()
            },
        )
        .await
        .expect("turn");
        assert_eq!(
            out.pending_approvals.len(),
            1,
            "an approval for different args must NOT authorize this call — it re-pauses"
        );
        assert!(
            tools.executed.lock().unwrap().is_empty(),
            "the tool must NOT execute under a mismatched-args approval"
        );
    }

    #[tokio::test]
    async fn denied_tool_resolves_without_executing_or_repausing() {
        // The denial path: the same scripted provider + gated tool executor as
        // the pause test, but the call id lands in `denied_call_ids` (a verified
        // approval_response with approved=false). The loop must NOT re-pause and
        // must NOT execute the tool; instead it emits a synthetic denial
        // tool_result so the model sees a result and the turn closes.
        let provider = ScriptedToolCallProvider {
            calls: AtomicUsize::new(0),
        };
        let tools = ApprovalGatedTools::default();
        let mut denied = std::collections::HashSet::new();
        denied.insert((
            "call-1".to_owned(),
            "dangerous_tool".to_owned(),
            r#"{"rm":"-rf"}"#.to_owned(),
        ));
        let out = run_turn_with(
            &provider,
            &tools,
            "scripted",
            vec![LlmMessage::user("hi")],
            RunTurnOptions {
                denied_call_ids: denied,
                ..Default::default()
            },
        )
        .await
        .expect("turn");
        assert!(
            out.pending_approvals.is_empty(),
            "denied call must NOT re-pause the loop"
        );
        // The FIRST signed denial (by call-id) must NOT trip the circuit
        // breaker: it records the signature, resolves the call, and lets the
        // model continue. Here the scripted provider ends the turn naturally on
        // its second call — so it was driven exactly twice (the breaker did not
        // cut it short on step 0).
        assert_eq!(
            provider.calls.load(Ordering::SeqCst),
            2,
            "first signed denial must not trip the breaker; model ends the turn itself"
        );
        assert!(
            tools.executed.lock().unwrap().is_empty(),
            "execute() must not be called for a denied call"
        );
        // A tool-result message must exist for the denied call, carrying the
        // denial payload (so the model gets a result, not a hang).
        let denial = out
            .messages
            .iter()
            .find(|m| {
                m.role == "tool"
                    && matches!(
                        m.content.as_option().and_then(|c| c.r#type.as_ref()),
                        Some(content::Type::ToolResult(tr)) if tr.call_id == "call-1"
                    )
            })
            .expect("denied call must produce a tool_result message");
        // Round-trip the wire message back to llm form and assert the payload
        // is the denial JSON (not an executed result).
        let llm = wire_to_llm(denial);
        match &llm.content[0] {
            LlmContent::ToolResult(tr) => {
                let parsed: serde_json::Value =
                    serde_json::from_str(&tr.result_json).expect("denial result is valid json");
                assert_eq!(
                    parsed.get("approved"),
                    Some(&serde_json::Value::Bool(false)),
                    "denial result must carry approved=false"
                );
                assert!(
                    parsed.get("error").is_some(),
                    "denial result must carry an error explanation"
                );
            }
            other => panic!("expected ToolResult, got {other:?}"),
        }
    }

    /// Scripted provider that re-emits the SAME logical tool call
    /// (`dangerous_tool` with identical args) on every step, each time under a
    /// fresh provider call-id (`call-1`, `call-2`, ...). Models the deny →
    /// re-emit loop: a denial keyed only to the call-id would never stick, so
    /// the signature-based sticky denial + circuit breaker must catch it.
    /// Records how many times the provider was driven so a test can assert the
    /// breaker bounded the loop well below `MAX_STEPS`.
    struct ReEmittingDeniedProvider {
        calls: AtomicUsize,
    }

    #[async_trait]
    impl LlmProvider for ReEmittingDeniedProvider {
        type Error = DummyError;

        async fn complete(
            &self,
            _req: CompletionRequest,
        ) -> Result<futures::stream::BoxStream<'static, Result<Chunk, Self::Error>>, Self::Error>
        {
            let n = self.calls.fetch_add(1, Ordering::SeqCst);
            // Fresh call-id each step; identical name + args (the signature).
            let id = format!("call-{}", n + 1);
            let chunks = vec![
                Ok(Chunk::tool_call_start(&id, "dangerous_tool")),
                Ok(Chunk::tool_call_args_delta(&id, r#"{"rm":"-rf"}"#)),
                Ok(Chunk::tool_call_end(&id)),
                Ok(Chunk::Stop(polyc_llm::StopReason::ToolUse)),
            ];
            Ok(stream::iter(chunks).boxed())
        }
    }

    #[tokio::test]
    async fn reemitted_denied_signature_is_auto_denied_and_circuit_breaker_bounds_loop() {
        // The deny → re-emit → re-prompt loop the fix targets. Step 0's call
        // (`call-1`) carries a signed denial via `denied_call_ids`, recording
        // its (name, args) signature. The model then re-emits the SAME action
        // with fresh call-ids on each later step. Those re-emits must be
        // AUTO-DENIED by signature — never surfaced as a PendingApproval and
        // never executed — and the circuit breaker must end the turn well
        // before MAX_STEPS.
        let provider = ReEmittingDeniedProvider {
            calls: AtomicUsize::new(0),
        };
        let tools = ApprovalGatedTools::default();
        let mut denied = std::collections::HashSet::new();
        denied.insert((
            "call-1".to_owned(),
            "dangerous_tool".to_owned(),
            r#"{"rm":"-rf"}"#.to_owned(),
        ));
        let out = run_turn_with(
            &provider,
            &tools,
            "scripted",
            vec![LlmMessage::user("hi")],
            RunTurnOptions {
                denied_call_ids: denied,
                ..Default::default()
            },
        )
        .await
        .expect("turn");

        // No PendingApproval: the re-emitted denied signature must NOT
        // re-prompt the human for an already-denied action.
        assert!(
            out.pending_approvals.is_empty(),
            "re-emitted denied signature must auto-deny, not re-prompt"
        );
        // Never executed — every step resolved to a synthetic denial.
        assert!(
            tools.executed.lock().unwrap().is_empty(),
            "auto-denied calls must never execute"
        );
        // Every step produced a denial tool_result for its (fresh) call-id.
        let denial_results = out
            .messages
            .iter()
            .filter(|m| {
                m.role == "tool"
                    && matches!(
                        m.content.as_option().and_then(|c| c.r#type.as_ref()),
                        Some(content::Type::ToolResult(_))
                    )
            })
            .count();
        assert!(
            denial_results >= 1,
            "each auto-denied call must still produce a tool_result"
        );
        // Circuit breaker bounded the loop: the provider was driven at most
        // `MAX_DENIAL_REPROMPTS + 1` times (step 0's first signed denial does
        // not count toward the breaker; the next two signature re-emits trip
        // it) — strictly fewer than MAX_STEPS.
        let driven = provider.calls.load(Ordering::SeqCst);
        assert!(
            driven <= MAX_DENIAL_REPROMPTS + 1,
            "circuit breaker must bound re-prompts: driven={driven} > {}",
            MAX_DENIAL_REPROMPTS + 1
        );
        assert!(
            driven < MAX_STEPS,
            "circuit breaker must end the turn before burning MAX_STEPS"
        );
    }

    #[tokio::test]
    async fn read_only_batch_runs_through_without_approval_pause() {
        let provider = ScriptedBenignProvider {
            calls: AtomicUsize::new(0),
        };
        let tools = ReadOnlyTools;
        let out = run_turn(&provider, &tools, "scripted", vec![LlmMessage::user("hi")])
            .await
            .expect("turn");
        assert!(
            out.pending_approvals.is_empty(),
            "no approval needed for read-only tools"
        );
        // One assistant text + one tool-result + final assistant text.
        // The exact count depends on whether the model emitted text on step 0
        // — here it did not, so we expect [tool-result, final-text].
        assert!(out.messages.iter().any(|m| m.role == "tool"));
    }

    #[test]
    fn wire_to_llm_preserves_tool_call_and_result() {
        use buffa::MessageField;
        use buffa_types::google::protobuf::Struct;
        use polyc_proto::proto::polychrome::agent::v1::{
            FunctionCallContent, FunctionResultContent, ToolCallContent, ToolResultContent,
        };

        fn wire(role: &str, ty: content::Type) -> Message {
            Message {
                role: role.to_owned(),
                content: MessageField::some(Content {
                    r#type: Some(ty),
                    ..Default::default()
                }),
                internal_only: false,
                ..Default::default()
            }
        }

        // Assistant tool call carrying a real function name + structured args.
        let args: Struct = serde_json::from_str(r#"{"query":"rust"}"#).expect("args struct");
        let call = wire(
            "model",
            content::Type::ToolCall(Box::new(ToolCallContent {
                id: "call_1".to_owned(),
                r#type: Some(tool_call_content::Type::FunctionCall(Box::new(
                    FunctionCallContent {
                        name: "search".to_owned(),
                        arguments: MessageField::some(args),
                        ..Default::default()
                    },
                ))),
                ..Default::default()
            })),
        );

        let llm_call = wire_to_llm(&call);
        assert_eq!(llm_call.role, Role::Assistant);
        assert_eq!(llm_call.content.len(), 1);
        match &llm_call.content[0] {
            LlmContent::ToolUse(tc) => {
                assert_eq!(tc.id, "call_1");
                assert_eq!(tc.name, "search", "function name must survive");
                let parsed: serde_json::Value =
                    serde_json::from_str(&tc.args_json).expect("args_json is valid json");
                assert_eq!(
                    parsed,
                    serde_json::json!({ "query": "rust" }),
                    "args must survive, not a placeholder"
                );
            }
            other => panic!("expected ToolUse, got {other:?}"),
        }

        // Tool result carrying a real structured payload.
        let resp: Struct = serde_json::from_str(r#"{"answer":42}"#).expect("resp struct");
        let result = wire(
            "tool",
            content::Type::ToolResult(Box::new(ToolResultContent {
                call_id: "call_1".to_owned(),
                r#type: Some(tool_result_content::Type::FunctionResult(Box::new(
                    FunctionResultContent {
                        name: "search".to_owned(),
                        result: Some(function_result_content::Result::Response(Box::new(resp))),
                        ..Default::default()
                    },
                ))),
                ..Default::default()
            })),
        );

        let llm_result = wire_to_llm(&result);
        assert_eq!(llm_result.role, Role::Tool);
        assert_eq!(llm_result.content.len(), 1);
        match &llm_result.content[0] {
            LlmContent::ToolResult(tr) => {
                assert_eq!(tr.tool_call_id, "call_1", "call id must correlate");
                assert!(!tr.is_error);
                let parsed: serde_json::Value =
                    serde_json::from_str(&tr.result_json).expect("result_json is valid json");
                // `google.protobuf.Struct` numbers are doubles, so `42`
                // round-trips as `42.0`; the payload itself is preserved.
                assert_eq!(
                    parsed,
                    serde_json::json!({ "answer": 42.0 }),
                    "result payload must survive, not a placeholder"
                );
            }
            other => panic!("expected ToolResult, got {other:?}"),
        }
    }

    #[test]
    fn tool_call_message_round_trips_through_wire_to_llm_with_signature() {
        // The persist→replay round-trip: build the structured wire message we
        // persist, decode it back, and assert the call (name, args, id) AND the
        // provider signature all survive.
        let tc = ToolCall {
            id: "call-7".to_owned(),
            name: "search".to_owned(),
            args_json: r#"{"query":"rust"}"#.to_owned(),
            signature: Some("sig-abc123".to_owned()),
        };
        let wire = tool_call_message(&tc);
        assert_eq!(wire.role, "model");
        let back = wire_to_llm(&wire);
        match &back.content[0] {
            LlmContent::ToolUse(rt) => {
                assert_eq!(rt.id, "call-7");
                assert_eq!(rt.name, "search");
                let parsed: serde_json::Value = serde_json::from_str(&rt.args_json).unwrap();
                assert_eq!(parsed, serde_json::json!({ "query": "rust" }));
                assert_eq!(
                    rt.signature.as_deref(),
                    Some("sig-abc123"),
                    "thought signature must survive the wire round-trip"
                );
            }
            other => panic!("expected ToolUse, got {other:?}"),
        }
    }

    #[test]
    fn tool_result_message_round_trips_through_wire_to_llm() {
        let wire = tool_result_message("call-7", r#"{"answer":42}"#);
        assert_eq!(wire.role, "tool");
        let back = wire_to_llm(&wire);
        match &back.content[0] {
            LlmContent::ToolResult(tr) => {
                assert_eq!(tr.tool_call_id, "call-7");
                let parsed: serde_json::Value = serde_json::from_str(&tr.result_json).unwrap();
                assert_eq!(parsed, serde_json::json!({ "answer": 42.0 }));
            }
            other => panic!("expected ToolResult, got {other:?}"),
        }
    }
}