polyc-turn-runner 2026.8.3

//! The turn-running core, shared by the harness server and the in-process
//! replay transport.
//!
//! One agent turn runs the same way regardless of who drives it: decode the
//! inbound wire transcript, verify the HITL approval responses, resolve the
//! per-turn backend + model, assemble [`RunTurnOptions`], run the
//! function-calling loop ([`polyc_agent::run_turn_with`]) against an **injected**
//! provider + [`ToolExecutor`], and frame the result as the terminal `batch`
//! message the control plane persists.
//!
//! This crate owns that orchestration so the two callers cannot drift:
//!
//! - the harness server (`polyc_harness`) composes its proxy-augmented tool
//!   surface and streams live deltas, then calls the fine-grained pieces
//!   ([`verify_approval_responses`], [`run_turn_options`], [`run_turn_captured`],
//!   [`build_final_batch`], [`connect_error_from_turn`]) around its bidirectional
//!   pump;
//! - the control-plane in-process transport composes a buffered tool surface and
//!   calls the whole-turn entry [`run_wire_turn`] directly.
//!
//! The runner is generic over the injected [`ToolExecutor`]: the harness's
//! mid-turn proxies (payment / history / dispatch / peer) live in the caller, so
//! this crate never depends on them and stays a Component both Containers can
//! depend on inward.

use std::{collections::HashMap, sync::Arc};

use connectrpc::{ConnectError, ErrorCode};
use futures::channel::mpsc;
use polyc_agent::{
    DelegateRecord, DispatchRecorder, HandoffRequest, PendingApproval, RunTurnOptions,
    ToolExecutor, TurnResult, TurnStreamEvent, UnattendedDenial, llm_stop_to_wire_i32,
    retry::Clock, run_turn_with, wire_to_llm,
};
use polyc_llm::{
    CacheHint, DynProvider, LlmError, LlmErrorKind, LlmProvider, Message as LlmMessage, Usage,
    into_dyn, turn::StubProvider,
};
use polyc_proto::humanize_tool_name;
use polyc_proto::proto::polychrome::agent::v1::Message as WireMessage;
use polyc_proto::proto::polychrome::harness::v1::{
    ApprovalResponse as WireApprovalResponse, DelegateDescriptor as WireDelegateDescriptor,
    DelegateFact as WireDelegateFact, ExecutionLabel as WireExecutionLabel,
    HandoffRequest as WireHandoffRequest, HarnessMessage, PendingApproval as WirePendingApproval,
    PendingQuestion as WirePendingQuestion, QuestionAnswer as WireQuestionAnswer,
    QuestionOptionWire as WireQuestionOptionWire, StepOutcomeProposal as WireStepOutcomeProposal,
    StopReason as WireStopReason, TurnBatch as WireTurnBatch, TurnFailure as WireTurnFailure,
    TurnFailureKind as WireTurnFailureKind, TurnInput as WireTurnInput,
    UnattendedDenialFact as WireUnattendedDenialFact, Usage as WireUsage,
    harness_message::Frame as HarnessFrame,
};

/// D7 capability-bound admission for tool, connector, and peer broker calls.
pub mod broker;
pub mod execution;
pub mod model;

use execution::ExecutionLabel;

/// Wrap one typed frame payload into a labeled [`HarnessMessage`].
///
/// Every outbound frame is exactly one `oneof frame` member; this is the single
/// construction site so the wrapping (and the `frame:` field name) lives in one
/// place. `HarnessFrame: From<each inner>` is buffa-generated.
///
/// `label` is the fenced Execution protocol label (`#1565`, D4). It is a
/// parameter rather than an option so a new frame cannot leave this crate
/// unlabeled: the receiving side refuses a frame no fence orders. An
/// [`execution::ExecutionSession`] mints the label for the frame's class.
#[must_use]
pub fn frame(label: &ExecutionLabel, payload: impl Into<HarnessFrame>) -> HarnessMessage {
    HarnessMessage {
        frame: Some(payload.into()),
        label: buffa::MessageField::some(WireExecutionLabel::from(label)),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

/// One registered backend plus the model to use when a turn names none.
///
/// The default model is a **per-provider** property: an OpenAI-compatible
/// backend pointed at Ollama defaults to e.g. `llama3.2`, while Vertex defaults
/// to a `gemini-…` id — a single global default would send one provider the
/// other's model id (a runtime 404).
#[derive(Clone)]
pub struct RegisteredProvider {
    /// The type-erased backend.
    pub provider: Arc<DynProvider>,
    /// Model used when a turn routed to this backend leaves `model` empty.
    pub default_model: String,
}

/// The backend + model a turn resolved to, for dispatch and logging.
pub struct Resolved {
    /// The erased backend to run the turn against.
    pub provider: Arc<DynProvider>,
    /// The registry key actually used (after any unknown-selector fallback).
    pub provider_name: String,
    /// The model id to put on the `CompletionRequest`.
    pub model: String,
}

/// A set of provider backends keyed by selector name, plus the default backend
/// used when a turn names no `provider`.
///
/// Built once at startup (e.g. from `POLYCHROME_PROVIDERS`) and shared across
/// requests. Each backend is type-erased to [`DynProvider`] so a single map
/// holds heterogeneous backends (Vertex, OpenAI-compatible, stub) and the
/// dispatch site is provider-agnostic. [`ProviderSet::new`] owns the registry's
/// invariants (a `stub` entry always exists; `default_provider` always names a
/// registered backend), so the resolve path never has to mask a gap.
#[derive(Clone)]
pub struct ProviderSet {
    /// Selector name (`"vertex"`, `"openai"`, `"stub"`, …) → registered backend.
    providers: Arc<HashMap<String, RegisteredProvider>>,
    /// Provider selector used when a turn leaves `provider` empty. Guaranteed
    /// by `new` to be a key of `providers`.
    default_provider: String,
}

impl ProviderSet {
    /// Build a provider set, enforcing the registry invariants in one place: a
    /// `stub` backend is always present, and `default_provider` is coerced to a
    /// registered key (`stub` if the requested default is missing).
    #[must_use]
    pub fn new(
        mut providers: HashMap<String, RegisteredProvider>,
        default_provider: impl Into<String>,
    ) -> Self {
        providers
            .entry("stub".to_owned())
            .or_insert_with(|| RegisteredProvider {
                provider: into_dyn(StubProvider),
                default_model: "stub".to_owned(),
            });
        let requested_default = default_provider.into();
        let default_provider = if providers.contains_key(&requested_default) {
            requested_default
        } else {
            tracing::error!(
                requested = %requested_default,
                "configured default provider is not registered; falling back to stub"
            );
            "stub".to_owned()
        };
        Self {
            providers: Arc::new(providers),
            default_provider,
        }
    }

    /// Builds a set with exactly one backend and no stub.
    ///
    /// Execution resolves no provider of its own since #1565 D6: every model
    /// call leaves the pod over the Control stream. A delegate must therefore
    /// resolve to the SAME brokered backend the parent turn runs on, which is
    /// what `#870` requires.
    ///
    /// This deliberately does not insert the `stub` backend [`Self::new`]
    /// guarantees. A stub here would be reachable by a delegate that names it,
    /// and it would answer with canned text instead of refusing — the failure
    /// this constructor exists to remove.
    #[must_use]
    pub fn proxy_only(
        provider_name: impl Into<String>,
        provider: Arc<DynProvider>,
        default_model: impl Into<String>,
    ) -> Self {
        let provider_name = provider_name.into();
        let mut providers = HashMap::new();
        providers.insert(
            provider_name.clone(),
            RegisteredProvider {
                provider,
                default_model: default_model.into(),
            },
        );
        Self {
            providers: Arc::new(providers),
            default_provider: provider_name,
        }
    }

    /// The backend registered under `name`, with its configured default model —
    /// e.g. `backend("openai")` to reach the self-hosted endpoint for a startup
    /// preflight. `None` if no backend is registered under that selector. The
    /// returned model is the backend's configured default; a turn may override it
    /// via the per-turn model selector.
    #[must_use]
    pub fn backend(&self, name: &str) -> Option<(Arc<DynProvider>, String)> {
        self.providers
            .get(name)
            .map(|reg| (Arc::clone(&reg.provider), reg.default_model.clone()))
    }

    /// Resolve the per-turn `provider`/`model` selectors to a concrete backend
    /// and model.
    ///
    /// An empty `provider` uses the default; an **unknown** provider is a
    /// misconfiguration (typo, or a backend that failed to register), so it is
    /// logged loudly and falls back to the default rather than silently serving
    /// the wrong backend. The model defaults to the resolved backend's own
    /// `default_model`.
    ///
    /// # Panics
    ///
    /// Never in practice: [`ProviderSet::new`] guarantees `default_provider`
    /// names a registered backend, so the fallback lookup is always present.
    #[must_use]
    pub fn resolve(&self, provider: &str, model: &str) -> Resolved {
        let requested = if provider.is_empty() {
            self.default_provider.as_str()
        } else {
            provider
        };
        let (name, reg) = self.providers.get(requested).map_or_else(
            || {
                tracing::warn!(
                    requested = %requested,
                    default = %self.default_provider,
                    registered = ?self.providers.keys().collect::<Vec<_>>(),
                    "unknown provider selector; falling back to the default provider"
                );
                // `new` guarantees `default_provider` is a registered key.
                let reg = self
                    .providers
                    .get(self.default_provider.as_str())
                    .expect("default_provider is a registered key (ProviderSet::new invariant)");
                (self.default_provider.as_str(), reg)
            },
            |reg| (requested, reg),
        );
        let model = if model.is_empty() {
            reg.default_model.clone()
        } else {
            model.to_owned()
        };
        Resolved {
            provider: reg.provider.clone(),
            provider_name: name.to_owned(),
            model,
        }
    }
}

impl Default for ProviderSet {
    /// A stub-only set — keeps CI / integration tests running with no
    /// credentials and no real backend.
    fn default() -> Self {
        Self::new(HashMap::new(), "stub")
    }
}

/// Verified HITL decisions extracted from a turn's inbound `approval_responses`.
///
/// The ordered sequence carries only entries whose signature verified. The
/// turn loop executes approved occurrences, resolves denied occurrences to a
/// synthetic denial result, and re-pauses on any gated call with no decision.
#[derive(Debug, Default)]
pub struct VerifiedApprovals {
    /// Verified one-shot decisions in durable approval-request order.
    decisions: Vec<polyc_agent::ApprovalDecision>,
    /// Verified, session-scoped ("don't ask again") approvals whose signed
    /// `caller` matches THIS turn's caller: tool name → the union of the
    /// signed covered capability sets (`#595`). Per-tool, not per-args (a
    /// model rarely repeats identical args); the gate additionally honors a
    /// grant only when its covered set includes the call's currently missing
    /// capabilities, and only for idempotent tools.
    session_approved: HashMap<String, polyc_capability::CapabilitySet>,
}

impl VerifiedApprovals {
    /// The number of one-shot APPROVED `(request_id, tool, args)` tuples — for
    /// the served path's turn-start log line.
    #[must_use]
    pub fn approved_count(&self) -> usize {
        self.decisions
            .iter()
            .filter(|decision| decision.approved)
            .count()
    }

    /// The number of one-shot DENIED tuples — for the turn-start log line.
    #[must_use]
    pub fn denied_count(&self) -> usize {
        self.decisions
            .iter()
            .filter(|decision| !decision.approved)
            .count()
    }

    /// The number of caller-scoped session ("don't ask again") grants — for the
    /// turn-start log line.
    #[must_use]
    pub fn session_approved_count(&self) -> usize {
        self.session_approved.len()
    }
}

/// Verify every inbound `approval_responses` entry and bucket the verified
/// tuples into occurrence-ordered decisions.
///
/// The verified grants also include the session-scoped ones that apply to
/// `current_caller`. The signature binds the full identity (tool name, args,
/// session flag, and the caller the memory is scoped to), so an approval/denial
/// applies only to that exact call and a session approval only to the same
/// caller. Bad signatures are logged and dropped — silent acceptance would let a
/// compromised control plane bypass the HITL gate, exactly the trust boundary
/// the harness pod owes the rest of the system.
#[must_use]
pub fn verify_approval_responses(
    responses: &[WireApprovalResponse],
    current_caller: &str,
    current_sandbox_mode: &str,
) -> VerifiedApprovals {
    let mut out = VerifiedApprovals::default();
    for resp in responses {
        if !polyc_crypto::approval::verify_wire_response(
            &resp.request_id,
            &resp.tool_name,
            &resp.args_json,
            &resp.modified_args_json,
            resp.approved,
            resp.approved_for_session,
            &resp.covered_capabilities,
            &resp.caller,
            &resp.approver,
            &resp.sandbox_mode,
            &resp.reason,
            &resp.injected_context,
            &resp.conversation_id,
            &resp.nonce,
            &resp.turn_id,
            resp.routine_grant,
            &resp.tool_descriptor_hash,
            &resp.grant_scope,
            &resp.signer_pk_hex,
            &resp.signature_hex,
        ) {
            tracing::warn!(
                request_id = %resp.request_id,
                "rejected approval_response: signature failed to verify"
            );
            continue;
        }
        if resp.approved {
            // A session grant applies only to its bound caller (per-user) and to
            // the tool regardless of args; the shared predicate keeps the
            // per-user rule identical to the control plane's in-process path. It
            // is additionally bound to the sandbox mode it was granted under, so
            // a grant from one mode never auto-approves a call running under a
            // different (e.g. more-privileged) mode.
            if polyc_crypto::approval::is_session_grant_for(
                resp.approved,
                resp.approved_for_session,
                &resp.caller,
                current_caller,
            ) && resp.sandbox_mode == current_sandbox_mode
            {
                // #595: the grant is keyed (caller, tool, covered
                // capabilities). Parse the signed covered set fail-closed
                // (unknown names cover nothing) and UNION across grants for
                // the same tool — each name was individually signed.
                let (covered, unknown) = polyc_capability::CapabilitySet::from_names(
                    resp.covered_capabilities.iter().map(String::as_str),
                );
                if !unknown.is_empty() {
                    tracing::warn!(
                        tool = %resp.tool_name,
                        ?unknown,
                        "session grant carries unknown covered-capability names; ignoring them"
                    );
                }
                let entry = out
                    .session_approved
                    .entry(resp.tool_name.clone())
                    .or_default();
                *entry = entry.union(covered);
            }
            // Re-add to the exact-call one-shot set ONLY if the original call
            // hasn't executed yet. A session grant is re-forwarded past
            // execution for its per-tool memory above; without this guard its
            // spent (id, tool, args) tuple would re-authorize — and a model
            // re-emitting the same call would re-run the original side effect.
            if !resp.already_executed {
                out.decisions
                    .push(polyc_agent::ApprovalDecision::from(resp));
            }
        } else {
            tracing::info!(
                request_id = %resp.request_id,
                "approval_response verified but denied"
            );
            out.decisions
                .push(polyc_agent::ApprovalDecision::from(resp));
        }
    }
    out
}

/// Verify every inbound `routine_tool_grants` entry and bucket the
/// verified grants into a [`polyc_agent::RoutineGrantSet`].
///
/// The control plane forwards these RAW — it never pre-trusts a grant record
/// (see `TurnInput.routine_tool_grants`'s doc) — so this is the sole trust
/// boundary a grant crosses before the gate honors it. A candidate is
/// admitted only when ALL hold, mirroring [`verify_approval_responses`]'s own
/// session-grant trust check exactly (the harness pod owes the rest of the
/// system the same boundary here it owes there):
///
/// * its signature verifies (`polyc_crypto::approval::verify_wire_response`,
///   reconstructing the canonical from the grant's own `routine_grant` marker,
///   `tool_descriptor_hash`, and `grant_scope`) — conversation binding is
///   covered by this same signature and was already checked against the fire
///   conversation by the control plane's collecting fold before the record
///   ever reached the wire, exactly as an ordinary session grant's
///   `conversation_id` is;
/// * [`polyc_crypto::approval::is_session_grant_for`] holds for
///   `current_caller` — a grant is bound to the turn's own beneficiary
///   exactly like a session ("don't ask again") approval, so a grant minted
///   for one caller can never auto-approve a different one;
/// * its signed `sandbox_mode` matches `current_sandbox_mode` — a grant
///   minted under one mode never auto-approves a call running under a
///   different (e.g. more-privileged) mode.
///
/// A malformed or unverifiable candidate is dropped and logged, never
/// silently accepted. `grant_scope == "tool"` buckets into
/// [`polyc_agent::routine_grant::RoutineGrantSet::per_tool`], keyed by the
/// grant's own `tool_name`; any other non-empty `grant_scope` (`"blanket_below_high"`
/// / `"blanket_all"`) buckets into
/// [`polyc_agent::routine_grant::RoutineGrantSet::blanket`] — the control
/// plane's collecting fold keeps at most one blanket record live, so the last
/// one seen here wins if more than one somehow arrives.
///
/// `expected_fire_conversation` binds every record to THIS turn's own fire
/// conversation. A grant's signature alone proves it was minted for some
/// fire conversation, never which one — a caller and sandbox mode can repeat
/// across routines owned by the same person, so a valid grant copied from
/// routine A's conversation must not admit routine B's dispatch.
#[must_use]
pub fn verify_routine_tool_grants(
    grants: &[WireApprovalResponse],
    current_caller: &str,
    current_sandbox_mode: &str,
    expected_fire_conversation: &str,
) -> polyc_agent::RoutineGrantSet {
    let mut out = polyc_agent::RoutineGrantSet::default();
    for resp in grants {
        if !resp.routine_grant {
            continue;
        }
        if !polyc_crypto::approval::verify_wire_response(
            &resp.request_id,
            &resp.tool_name,
            &resp.args_json,
            &resp.modified_args_json,
            resp.approved,
            resp.approved_for_session,
            &resp.covered_capabilities,
            &resp.caller,
            &resp.approver,
            &resp.sandbox_mode,
            &resp.reason,
            &resp.injected_context,
            &resp.conversation_id,
            &resp.nonce,
            &resp.turn_id,
            resp.routine_grant,
            &resp.tool_descriptor_hash,
            &resp.grant_scope,
            &resp.signer_pk_hex,
            &resp.signature_hex,
        ) {
            tracing::warn!(
                tool = %resp.tool_name,
                "rejected routine_tool_grant: signature failed to verify"
            );
            continue;
        }
        if resp.conversation_id != expected_fire_conversation {
            tracing::warn!(
                tool = %resp.tool_name,
                "rejected routine_tool_grant: not bound to this turn's fire conversation"
            );
            continue;
        }
        if !polyc_crypto::approval::is_session_grant_for(
            resp.approved,
            resp.approved_for_session,
            &resp.caller,
            current_caller,
        ) || resp.sandbox_mode != current_sandbox_mode
        {
            tracing::warn!(
                tool = %resp.tool_name,
                "rejected routine_tool_grant: not bound to this turn's caller or sandbox mode"
            );
            continue;
        }
        let (covered, unknown) = polyc_capability::CapabilitySet::from_names(
            resp.covered_capabilities.iter().map(String::as_str),
        );
        if !unknown.is_empty() {
            tracing::warn!(
                tool = %resp.tool_name,
                ?unknown,
                "routine_tool_grant carries unknown covered-capability names; ignoring them"
            );
        }
        match resp.grant_scope.as_str() {
            "tool" => {
                out.per_tool.insert(
                    resp.tool_name.clone(),
                    polyc_agent::routine_grant::PerToolGrant {
                        covered,
                        descriptor_hash: resp.tool_descriptor_hash.clone(),
                    },
                );
            }
            "blanket_below_high" | "blanket_all" => {
                out.blanket = Some(polyc_agent::routine_grant::BlanketGrant {
                    include_high: resp.grant_scope == "blanket_all",
                });
            }
            // A closed set: any other value — including an empty scope, a
            // version-skewed name, or a corrupted-but-still-signed one — is
            // no grant at all rather than a widened blanket by default.
            _ => {
                tracing::warn!(
                    tool = %resp.tool_name,
                    grant_scope = %resp.grant_scope,
                    "rejected routine_tool_grant: unrecognized grant_scope"
                );
            }
        }
    }
    out
}

/// Verify every wire `question_response` the control plane forwarded on a
/// resumed turn (`#1660`), dropping any that fails to verify.
///
/// The question-pause SIBLING of [`verify_approval_responses`], not a reuse
/// of it: a `question_response` carries no session-grant/override concept
/// (an answer applies to exactly one paused question, never remembered
/// across calls), so this is a straight per-entry signature check + state
/// mapping. Silent acceptance would let a compromised control plane forge a
/// user's answer — the trust boundary this exists for, exactly like
/// [`verify_approval_responses`]'s.
#[must_use]
pub fn verify_question_answers(
    answers: &[WireQuestionAnswer],
) -> Vec<polyc_agent::question::VerifiedAnswer> {
    use polyc_agent::question::AnswerState;
    use std::str::FromStr as _;

    let mut out = Vec::new();
    for a in answers {
        // The single shared string<->AnswerState conversion (mirrors
        // `harness_dialer.rs`'s own `From<&QuestionAnswerRecord>` call into
        // it) rather than a second, independently hand-rolled if/else chain.
        let Ok(state) = AnswerState::from_str(&a.state) else {
            tracing::warn!(
                call_id = %a.call_id,
                index = a.index,
                state = %a.state,
                "rejected question_response: unrecognized state"
            );
            continue;
        };
        // A decline signs no selection (`None`); answered/auto-resolved sign
        // the chosen index — reconstructing the SAME `Option` shape the
        // signer used is required for the canonical to match byte-for-byte.
        let selected_index = (!matches!(state, AnswerState::Declined)).then_some(a.selected_index);
        if !polyc_crypto::question::verify_wire_answer(
            &a.turn_id,
            &a.call_id,
            a.index,
            &a.question_args_json,
            &a.state,
            selected_index,
            &a.selected_label,
            &a.answered_by,
            &a.conversation_id,
            &a.nonce,
            &a.signer_pk_hex,
            &a.signature_hex,
        ) {
            tracing::warn!(
                call_id = %a.call_id,
                index = a.index,
                "rejected question_response: signature failed to verify"
            );
            continue;
        }
        out.push(polyc_agent::question::VerifiedAnswer {
            turn_id: a.turn_id.clone(),
            call_id: a.call_id.clone(),
            index: a.index,
            state,
            selected_index,
            selected_label: a.selected_label.clone(),
            answered_by: a.answered_by.clone(),
        });
    }
    out
}

/// The harness's current sandbox mode, as the CANONICAL string.
///
/// Resolves via [`polyc_tools::current_sandbox_mode`] — so the value the
/// session-approval binding stamps + compares is the mode that is actually
/// ENFORCED (`SandboxMode::from_env`, unset → `workspace-write`), not the raw
/// env string. One shared resolver, used at both ends of the binding (stamp +
/// gate) and by the control plane, so they cannot drift.
#[must_use]
pub fn current_sandbox_mode() -> String {
    polyc_tools::current_sandbox_mode()
}

/// Resolve the wire display title for a pending approval.
///
/// The agent ([`polyc_agent::run_turn_with`]) is the single source of the
/// curated title: it populates [`PendingApproval::title`](polyc_agent::PendingApproval::title)
/// from the tool's [`ToolSpec`](polyc_llm::ToolSpec) annotation, leaving it
/// empty when the spec advertised none. The harness forwards that value
/// verbatim and only applies the [`humanize_tool_name`] fallback when it is
/// empty, so the wire `title` is always non-empty (curated title, else a
/// humanized form of the raw `name`).
#[must_use]
pub fn wire_title(agent_title: &str, name: &str) -> String {
    if agent_title.is_empty() {
        humanize_tool_name(name)
    } else {
        agent_title.to_owned()
    }
}

/// Map a failed turn onto a Connect error with the most accurate code.
///
/// The provider classifies its own failure into a typed [`LlmErrorKind`]
/// (carried through type erasure by `polyc_llm::BoxError`), captured by
/// [`run_turn_captured`]; this maps that kind onto the Connect status code so
/// the control plane can tell retryable (rate-limit / timeout / unavailable)
/// from terminal (auth / bad-request) failures instead of a catch-all
/// `internal`. `msg` carries the original error text for diagnostics.
#[must_use]
pub fn connect_error_from_turn(kind: LlmErrorKind, msg: String) -> ConnectError {
    match kind {
        LlmErrorKind::RateLimit => ConnectError::resource_exhausted(msg),
        LlmErrorKind::Timeout => ConnectError::deadline_exceeded(msg),
        LlmErrorKind::Unavailable => ConnectError::unavailable(msg),
        LlmErrorKind::Auth => ConnectError::new(ErrorCode::Unauthenticated, msg),
        LlmErrorKind::BadRequest => ConnectError::invalid_argument(msg),
        // `Unknown` rather than `Unavailable`: the latter invites a retry,
        // and this attempt may already have applied. The durable sibling below
        // cannot express this kind at all.
        LlmErrorKind::Ambiguous => ConnectError::new(ErrorCode::Unknown, msg),
        LlmErrorKind::Other => ConnectError::internal(msg),
    }
}

/// Classifies a failed turn's [`LlmErrorKind`] into what a caller may record.
///
/// The durable sibling of [`connect_error_from_turn`]'s Connect-status mapping.
/// It classifies the same failure the same way on both surfaces. One kind has
/// no durable form: see [`CapturedTurnError`].
#[must_use]
pub fn classify_turn_error(kind: LlmErrorKind, message: String) -> CapturedTurnError {
    let kind = match kind {
        LlmErrorKind::RateLimit => WireTurnFailureKind::RateLimit,
        LlmErrorKind::Timeout => WireTurnFailureKind::Timeout,
        LlmErrorKind::Unavailable => WireTurnFailureKind::Unavailable,
        LlmErrorKind::Auth => WireTurnFailureKind::Auth,
        LlmErrorKind::BadRequest => WireTurnFailureKind::BadRequest,
        LlmErrorKind::Other => WireTurnFailureKind::Other,
        // The one kind with no durable form.
        //
        // A `TurnFailure` lands beside an unconditional `turn_complete` marker.
        // That marker alone says this turn's final batch landed. A durable
        // ambiguous failure would therefore assert that a turn completed
        // because its outcome is unknown.
        LlmErrorKind::Ambiguous => return CapturedTurnError::Ambiguous,
    };
    CapturedTurnError::Failed(WireTurnFailure {
        kind: kind.into(),
        message,
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    })
}

/// What a captured turn returns when it does not produce a result.
///
/// The two arms are not two kinds of failure. [`Self::Failed`] is a fact a
/// caller may persist. [`Self::Ambiguous`] is the absence of one: the attempt
/// may or may not have applied, so no caller may record a definite outcome and
/// none may retry.
#[derive(Debug, Clone)]
pub enum CapturedTurnError {
    /// A failure that is safe to persist as a durable terminal outcome.
    Failed(WireTurnFailure),
    /// The attempt's outcome is not known.
    Ambiguous,
}

impl CapturedTurnError {
    /// Projects this error onto a Connect status.
    ///
    /// The ephemeral surface carries ambiguity; the durable one cannot. A
    /// caller that only ends an RPC uses this. A caller that persists an
    /// outcome matches on the variant instead, so it cannot record a definite
    /// result for an indefinite one.
    #[must_use]
    pub fn connect_error(&self) -> ConnectError {
        match self {
            Self::Ambiguous => ConnectError::new(
                ErrorCode::Unknown,
                "the model attempt's outcome is not known, so this turn records none",
            ),
            Self::Failed(failure) => {
                let message = failure.message.clone();
                match failure.kind.as_known() {
                    Some(WireTurnFailureKind::RateLimit) => {
                        ConnectError::resource_exhausted(message)
                    }
                    Some(WireTurnFailureKind::Timeout) => ConnectError::deadline_exceeded(message),
                    Some(WireTurnFailureKind::Unavailable) => ConnectError::unavailable(message),
                    Some(WireTurnFailureKind::Auth) => {
                        ConnectError::new(ErrorCode::Unauthenticated, message)
                    }
                    Some(WireTurnFailureKind::BadRequest) => {
                        ConnectError::invalid_argument(message)
                    }
                    _ => ConnectError::internal(message),
                }
            }
        }
    }
}

/// Run a turn, capturing the provider error's typed [`LlmErrorKind`] alongside
/// its message.
///
/// Classifying here lets the spawned turn task hold one return type across
/// provider arms. It also means only this crate decides whether a provider
/// failure has a durable form at all.
///
/// `#798`: `run_turn_with` no longer propagates a mid-stream provider failure
/// as `Err` — it returns `Ok` with [`TurnResult::mid_stream_failure`] set, so
/// the already-executed iterations ride along instead of vanishing. The served
/// harness wire protocol has no field yet to carry a partial transcript
/// alongside a terminal failure (a `batch` and a `Failed` frame are mutually
/// exclusive — see `HarnessFrame`), so this function translates a mid-stream
/// failure back into the `Err` shape every caller already handles, preserving
/// their behavior byte-for-byte. The recovery this unlocks — persisting the
/// partial transcript instead of discarding it — is wired at the
/// control-plane in-process turn path (`polychrome-control-plane`'s
/// `execute_turn_streamed`), which calls `run_turn_with` directly rather than
/// through this function. Carrying the partial transcript through the wire
/// protocol too is a follow-up (needs a `WireTurnFailure` schema change).
///
/// # Errors
///
/// Returns a [`CapturedTurnError`] when the turn loop's underlying
/// `run_turn_with` fails outright, or when it succeeds but reports a mid-stream
/// failure. An ambiguous provider outcome returns
/// [`CapturedTurnError::Ambiguous`], which a caller may not persist as a
/// failure and may not retry.
pub async fn run_turn_captured(
    provider: &(impl LlmProvider + ?Sized),
    tools: &(impl ToolExecutor + ?Sized),
    model: &str,
    input: Vec<LlmMessage>,
    options: RunTurnOptions,
) -> Result<TurnResult, CapturedTurnError> {
    let turn = run_turn_with(provider, tools, model, input, options)
        .await
        .map_err(|e| classify_turn_error(e.kind(), e.to_string()))?;
    if let Some(failure) = turn.mid_stream_failure {
        return Err(classify_turn_error(failure.kind, failure.message));
    }
    Ok(turn)
}

/// Runs a captured turn under the bounded authority admitted from Execution.
///
/// The grant's capability set is a ceiling for the whole turn, not an
/// admission check: nothing inside the loop can widen it.
///
/// # Errors
///
/// Returns what [`run_turn_captured`] returns. The grant narrows what the turn
/// may do; it does not add a failure of its own.
pub async fn run_turn_captured_under_grant(
    provider: &(impl LlmProvider + ?Sized),
    tools: &(impl ToolExecutor + ?Sized),
    model: &str,
    input: Vec<LlmMessage>,
    options: RunTurnOptions,
    granted: polyc_capability::CapabilitySet,
) -> Result<TurnResult, CapturedTurnError> {
    polyc_agent::with_execution_capabilities(
        granted,
        run_turn_captured(provider, tools, model, input, options),
    )
    .await
}

/// The per-turn boolean policy toggles the control plane resolved for one
/// turn.
///
/// Read off the opening wire frame by [`TurnToggles::from_wire`] — one
/// carrier instead of adjacent positional `bool`s that could be silently
/// transposed and still compile.
// Independent per-turn policy flags, mirroring RunTurnOptions' own fields;
// an enum would force artificial combinations.
#[allow(clippy::struct_excessive_bools)]
#[derive(Clone, Copy, Debug, Default)]
pub struct TurnToggles {
    /// #301: escalate a sandbox-denied call to a human (an unsandboxed retry)
    /// instead of running it and returning a flat denial. Per-caller.
    pub escalate_sandbox_denials: bool,
    /// #369: the durable lethal-trifecta untrusted-content seed, OR-ed into
    /// the agent gate's structural check.
    pub untrusted_context: bool,
    /// #623: the turn runs unattended — an escalating gate denies fail-closed
    /// instead of pausing.
    pub unattended: bool,
    /// #582 invariant 9: the fuzzy-match escape hatch. When set, a call
    /// naming an unadvertised tool triggers ONE fuzzy in-turn re-present of
    /// the closest catalog matches; false keeps the unknown-tool path
    /// byte-for-byte today's.
    pub escape_hatch: bool,
    /// This dispatch is a routine's fire conversation (scheduled firing, or
    /// the future attended setup rehearsal) — the harness excludes every
    /// `ToolSpec.interactive` built-in from the advertised surface.
    pub fire_dispatch: bool,
    /// The deployment-global `POLYCHROME_APPROVAL_MODE=approve-all-dangerous`
    /// mode, resolved control-plane-side. On a fire dispatch, merges an
    /// effective approve-all blanket into the turn's routine tool grants
    /// (see [`run_turn_options`]). For ephemeral test rigs only.
    pub deployment_approve_all_dangerous: bool,
}

impl TurnToggles {
    /// Reads the toggles off one turn's opening wire frame — the ONE
    /// derivation both transports (the served `connect` handler and the
    /// buffered [`run_wire_turn`]) share, so they cannot drift. The
    /// dedicated wire fields map straight across; `escape_hatch` rides the
    /// wire retrieval config, resolved control-plane-side and never a
    /// harness constant (inert unless the gated executor actually hides
    /// tools to recover).
    #[must_use]
    pub fn from_wire(turn_input: &WireTurnInput) -> Self {
        Self {
            escalate_sandbox_denials: turn_input.escalate_sandbox_denials,
            untrusted_context: turn_input.untrusted_context,
            unattended: turn_input.unattended,
            escape_hatch: turn_input
                .retrieval
                .as_option()
                .is_some_and(|cfg| cfg.escape_hatch),
            fire_dispatch: turn_input.fire_dispatch,
            deployment_approve_all_dangerous: turn_input.deployment_approve_all_dangerous,
        }
    }
}

/// Resolve wire `DelegateDescriptor`s (#870) into native, ready-to-run
/// [`polyc_agent::DelegateDescriptor`]s.
///
/// Each wire descriptor already carries the control plane's fully resolved
/// policy (provider/model selectors, connector labels, built-in tool names,
/// step budget) — this function's only job is to turn those selectors into
/// what the agent loop actually needs to run the worker's nested turn:
///
///   * `providers.resolve(..)` picks the CONCRETE backend for the worker
///     (never the same generic-Provider-instance trick the tool executor
///     uses — a delegation genuinely may run on a different backend than the
///     orchestrator);
///   * the worker's advertised tool-spec set is filtered OUT of the SAME
///     already-composed `tools` the orchestrator's own turn runs against
///     (same dialed connectors, same sandboxed built-ins) — "concurrent
///     workers will eventually share the parent's sandbox" (#874) starts
///     here. A built-in name matches by exact name; a connector name matches
///     by its `<label>__` prefix.
///
/// Called by both the harness's streamed `connect` handler (after it
/// composes the turn's own tool executor) and [`run_wire_turn`] (which
/// already holds both `providers` and `tools`), so the two transports derive
/// identical worker configurations from the identical wire input.
#[must_use]
pub fn resolve_delegate_descriptors(
    providers: &ProviderSet,
    tools: &(impl ToolExecutor + ?Sized),
    wire: &[WireDelegateDescriptor],
) -> Vec<polyc_agent::DelegateDescriptor> {
    let all_specs = tools.specs();
    wire.iter()
        .map(|d| {
            let resolved = providers.resolve(&d.provider, &d.model);
            let tool_specs: Vec<_> = all_specs
                .iter()
                .filter(|s| {
                    d.builtin_tools.iter().any(|name| name == &s.name)
                        || d.tools_enabled.iter().any(|label| {
                            s.name
                                .strip_prefix(label.as_str())
                                .and_then(|rest| rest.strip_prefix("__"))
                                .is_some()
                        })
                })
                .cloned()
                .collect();
            polyc_agent::DelegateDescriptor {
                agent_id: d.agent_id.clone(),
                instructions: (!d.instructions.is_empty()).then(|| d.instructions.clone()),
                provider: resolved.provider,
                provider_name: resolved.provider_name,
                model: resolved.model,
                tool_specs,
                // A malformed (zero) wire budget still runs the worker rather
                // than refusing outright — `polyc_agent::run_turn_with`'s own
                // `MAX_STEPS` loop bound already treats 0 as "no iterations",
                // which degrades to the forced-closing-completion safety net
                // rather than panicking.
                max_steps: usize::try_from(d.max_steps).unwrap_or(usize::MAX),
                // Mirrors the parent turn's own `#1226` scoping (see this
                // function's `native_search_allowed` sibling below): grounding
                // has no `ToolSpec`, so it can't ride `tool_specs` above like
                // every other built-in — it's carried explicitly off the
                // wire descriptor's own `builtin_tools` instead.
                native_search_allowed: d
                    .builtin_tools
                    .iter()
                    .any(|n| n == polyc_tools::web::NATIVE_SEARCH_GROUNDING),
                // `#2295`: carried straight off the wire. The control plane
                // already applied the target agent's declaration and its
                // defaults, so a zero/empty ceiling here means the operator
                // never opted in — and `ShareInCeiling::admits_anything`
                // reads that as closed.
                share_in: polyc_agent::delegate::ShareInCeiling {
                    allow: d.share_in_allow.clone(),
                    max_files: usize::try_from(d.share_in_max_files).unwrap_or(usize::MAX),
                    max_bytes: d.share_in_max_bytes,
                },
            }
        })
        .collect()
}

/// Assemble the [`RunTurnOptions`] a turn runs under from the verified approvals
/// and the per-turn control-plane inputs.
///
/// Shared by the streaming harness `connect` handler and the socket-free
/// [`run_wire_turn`] entry so the security-load-bearing approval wiring — the
/// occurrence-ordered one-shot decisions, the `#67` approver overrides, and
/// the caller-scoped session grants — is assembled in ONE place and the two
/// paths cannot drift. `stream_tx`, `dispatch_recorder`, and `clock` are the only
/// per-path inputs: the streaming path forwards live deltas and signs dispatch
/// mutations over its bidirectional stream, so it passes both proxies; the
/// buffered path passes `None` for each (a dispatch mutation is then inert /
/// fail-closed, exactly as it is on the served path when the signer is
/// unreachable). `clock` is `None` on the production and served paths (the real
/// wall clock) and `Some(pinned)` only on the replay path. Every other field is
/// a policy decision the control plane already made, identical across
/// transports.
#[must_use]
#[allow(clippy::too_many_arguments)] // each is a distinct per-turn policy input the CP resolved
#[allow(clippy::implicit_hasher)] // the CP always builds these maps with the default hasher
pub fn run_turn_options(
    approvals: VerifiedApprovals,
    // Verified routine-fire tool grants for this turn (see
    // `verify_routine_tool_grants`) — the routine-grant SIBLING of
    // `approvals` above. `deployment_approve_all_dangerous` below may widen
    // a copy of this before it lands on the returned `RunTurnOptions`.
    // `approvals` above, assembled here for the same reason: every caller
    // already has its verified grants in hand at call time, and this keeps
    // the security-load-bearing wiring in the one place both transports
    // share.
    routine_tool_grants: polyc_agent::RoutineGrantSet,
    toggles: TurnToggles,
    prompt_cache_key: String,
    // `#68`: `TurnInput.step_budget` — an edge-authored `IngressDirective`'s
    // upper bound on this turn's step budget. `0` means the control plane set
    // no cap (the wire's "unset" convention); a positive value can only LOWER
    // the resolved budget, never raise it — see the `max_steps` field below.
    step_budget: u32,
    stream_tx: Option<mpsc::Sender<TurnStreamEvent>>,
    dispatch_recorder: Option<Arc<dyn DispatchRecorder>>,
    clock: Option<Arc<dyn Clock + Send + Sync>>,
    // Issue #1226: whether the resolved agent's `builtinTools` names
    // `polyc_tools::web::NATIVE_SEARCH_GROUNDING` — the caller resolves this
    // the same way it resolves `builtin_allow` for `build_tool_executor`, so
    // the scoping decision can never drift between the two. This is the grant
    // only; the per-step gate in the answering loop still requires
    // `ArbitraryEgress` to survive taint before actually turning grounding on.
    native_search_allowed: bool,
    // `#1660`: verified `question_response`s the control plane forwarded for
    // this resumed turn — the question-pause SIBLING of `approvals` above.
    // Assembled here (not left for callers to bolt on afterward, unlike
    // `delegate_descriptors`) because it is equally security-load-bearing
    // (I3): every caller of this function already has its verified answers
    // in hand at call time, unlike delegation's composed-tool-executor
    // dependency.
    question_answers: Vec<polyc_agent::question::VerifiedAnswer>,
    // `#1323`: `TurnInput.turn_start_unix_ms` — the control plane's frozen
    // dispatch clock, forwarded so a delegated worker's nested turn can
    // render its own turn-start stamp from this SAME value. `0` means the
    // control plane didn't set this — the same "0 means unset" convention
    // `step_budget` above still uses (`delegate_fanout_cap` and
    // `delegate_turn_call_budget` moved to real `optional` presence, #1680).
    turn_start_unix_ms: u64,
) -> RunTurnOptions {
    // `POLYCHROME_APPROVAL_MODE=approve-all-dangerous` (test rigs only):
    // merge an effective approve-all blanket into the fire's grants at
    // dispatch — a deployment-global override, not a routine-local grant,
    // so it never rides the signed grant wire and never applies outside a
    // fire dispatch. Widens (never narrows) whatever grant already exists;
    // an ordinary deployment leaves `routine_tool_grants` byte-for-byte
    // unaffected (`deployment_approve_all_dangerous` is false).
    let routine_tool_grants = if toggles.fire_dispatch && toggles.deployment_approve_all_dangerous {
        polyc_agent::RoutineGrantSet {
            blanket: Some(polyc_agent::routine_grant::BlanketGrant { include_high: true }),
            ..routine_tool_grants
        }
    } else {
        routine_tool_grants
    };
    RunTurnOptions {
        approval_decisions: approvals.decisions,
        // Caller-scoped session approvals ("don't ask again"). The agent
        // additionally requires the tool to be idempotent before honoring one.
        session_approved_tools: approvals.session_approved,
        // Verified routine-fire tool grants, consulted only when
        // `unattended` (below) is true.
        routine_tool_grants,
        stream_tx,
        // `#68`: an edge-authored `IngressDirective.budget_cap` can only LOWER
        // this turn's resolved step budget, never raise it — `0` (the wire's
        // "not set" convention) leaves `max_steps` unset, so the ordinary
        // resolution (`POLYCHROME_AGENT_MAX_STEPS`, then
        // `polyc_agent::DEFAULT_MAX_STEPS`) applies exactly as before `#68`.
        // A positive cap is clamped against that SAME baseline
        // (`polyc_agent::resolve_default_max_steps`) rather than applied
        // unconditionally, so a cap larger than the baseline is a no-op —
        // `effective = min(cap, baseline)`.
        max_steps: (step_budget > 0).then(|| {
            let cap = usize::try_from(step_budget).unwrap_or(usize::MAX);
            cap.min(polyc_agent::resolve_default_max_steps())
        }),
        // #1226: scoping only — whether this agent is granted the native
        // grounding primitive at all. The per-step taint/capability gate in
        // the answering loop decides whether it's actually on for a given
        // step.
        native_search_allowed,
        // The four per-turn policy toggles ride one carrier off the wire —
        // see [`TurnToggles`] for each flag's contract.
        escalate_sandbox_denials: toggles.escalate_sandbox_denials,
        untrusted_context_seed: toggles.untrusted_context,
        unattended: toggles.unattended,
        escape_hatch: toggles.escape_hatch,
        fire_dispatch: toggles.fire_dispatch,
        dispatch_recorder,
        // #629: a non-empty routing key marks the turn's stable prefix (system
        // text + the once-per-turn tool set) as cacheable under that key — the
        // conversation id, so a conversation's turns share a provider cache. Empty
        // ⇒ no caching requested.
        cache_hint: CacheHint::from_key(prompt_cache_key),
        // #656: production and the served/in-process paths wire the real clock
        // (`None`); only a replay pins a virtual clock to the recorded dispatch
        // time so a turn whose output depends on it reproduces (INV-11).
        clock,
        // #870: not resolved here — `run_turn_options`'s callers don't yet
        // hold the composed tool executor needed to resolve delegate
        // descriptors (the harness's streamed path composes it AFTER this
        // call). Callers that support delegation set this field on the
        // returned value themselves (see `resolve_delegate_descriptors`);
        // every other caller leaves it empty, so a turn with no delegation
        // support at its call site is byte-for-byte unaffected.
        delegate_descriptors: Vec::new(),
        // #874: not resolved here either, for the same reason as
        // `delegate_descriptors` above — a fan-out cap is only meaningful
        // once a caller has decided whether it advertises `__delegate_to`
        // at all. Callers set this alongside `delegate_descriptors` (from
        // the wire's `TurnInput.delegate_fanout_cap`, itself `Option<u32>`
        // with real presence — #1680); every other caller leaves it `None`,
        // so `polyc_agent`'s own default applies.
        delegate_max_fanout: None,
        // #874: same reasoning and wiring as `delegate_max_fanout` — set by
        // callers alongside it (from `TurnInput.delegate_turn_call_budget`);
        // every other caller leaves it `None`.
        delegate_turn_budget: None,
        // This function only ever builds options for a top-level/orchestrator
        // turn — a delegated worker's own nested turn is built separately,
        // entirely inside `polyc_agent::run_delegate_call`, never through
        // this path.
        is_delegated_worker: false,
        question_answers,
        // `#1323`: `0` (the wire's "unset" convention) leaves this `None`,
        // so a delegated worker this turn dispatches renders no turn-start
        // stamp at all — never a placeholder, mirroring
        // `turn_start_block`'s own "say nothing rather than guess" rule.
        turn_start_unix_ms: (turn_start_unix_ms != 0).then_some(turn_start_unix_ms),
    }
}

/// Build the wire-facing [`WirePendingApproval`] for `p`. Neither
/// [`PendingApproval`] nor `WirePendingApproval` is local to this crate, so a
/// `From` impl would violate the orphan rule — this named fn plays the same
/// role: every field is named explicitly, no `..Default::default()` spread,
/// so a field added to either side without updating it fails to compile
/// instead of silently dropping a value on the wire.
fn wire_pending_approval(p: PendingApproval) -> WirePendingApproval {
    let title = wire_title(&p.title, &p.name);
    WirePendingApproval {
        turn_id: p.occurrence_turn_id.unwrap_or_default(),
        id: p.id,
        tool_name: p.name,
        args_json: p.args_json,
        title,
        // Stamp the harness's current sandbox mode so the control plane
        // can bind a remembered approval to the mode it was granted under.
        sandbox_mode: current_sandbox_mode(),
        // Forward the containment-escalation reason (empty for an
        // ordinary gate) so it reaches the approval card.
        reason: p.reason,
        // #595: the capability shortfall that paused the call, so the
        // control plane records it on the approval_request and a
        // "don't ask again" response signs it as its covered set.
        missing_capabilities: p.missing_capabilities,
        // Forward the descriptor hash, full required-capability set,
        // and fire-dispatch marker so the control plane can persist them on
        // the durable approval_request and mint a per-tool grant on approval.
        tool_descriptor_hash: p.tool_descriptor_hash,
        required_capabilities: p.required_capabilities,
        fire_dispatch: p.fire_dispatch,
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

/// Build the wire-facing [`WirePendingQuestion`] for `p` (`#1660`). Same
/// exhaustive-fields convention as [`wire_pending_approval`] — the
/// question-pause SIBLING of it, not a reuse. `args_json` is `p`'s bound raw
/// `ask_question` call arguments, forwarded verbatim so the control plane's
/// durable `question_request` and any later signed answer bind to it.
fn wire_pending_question(p: polyc_agent::question::PendingQuestion) -> WirePendingQuestion {
    WirePendingQuestion {
        // Empty on a live pause: the agent never learns the running turn id,
        // so the control plane fills the occurrence in at wire-build time.
        turn_id: p.occurrence_turn_id.unwrap_or_default(),
        call_id: p.call_id,
        index: p.index,
        header: p.item.header,
        question: p.item.question,
        options: p
            .item
            .options
            .into_iter()
            .map(|o| WireQuestionOptionWire {
                label: o.label,
                description: o.description,
                recommended: o.recommended,
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })
            .collect(),
        args_json: p.args_json,
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

/// Build the wire-facing [`WireUnattendedDenialFact`] for `d`. Same
/// exhaustive-fields convention as [`wire_pending_approval`].
fn wire_unattended_denial_fact(d: UnattendedDenial) -> WireUnattendedDenialFact {
    WireUnattendedDenialFact {
        tool: d.tool,
        args_json: d.args_json,
        missing_capabilities: d.missing_capabilities,
        reason: d.reason,
        // The tool's descriptor hash and full required-capability set at
        // denial time — presentation/audit data, not a trust decision (the
        // control plane re-derives every gate decision itself). `fire_dispatch`
        // is NOT sent: mirrors `PendingApproval`'s own wire boundary, where
        // the control plane stamps it from its own authoritative dial fact
        // instead of trusting the harness's echo.
        tool_descriptor_hash: d.tool_descriptor_hash,
        required_capabilities: d.required_capabilities,
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

/// Build the wire-facing [`WireDelegateFact`] for `d`. Same exhaustive-fields
/// convention as [`wire_pending_approval`].
fn wire_delegate_fact(d: DelegateRecord) -> WireDelegateFact {
    WireDelegateFact {
        sub_agent_id: d.sub_agent_id,
        target_agent_id: d.target_agent_id,
        task: d.task,
        resolved_provider: d.resolved_provider,
        resolved_model: d.resolved_model,
        input_tokens: d.usage.input_tokens,
        output_tokens: d.usage.output_tokens,
        succeeded: d.succeeded,
        error: d.error,
        first_party: d.first_party,
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

/// Build the wire-facing [`WireHandoffRequest`] for `h`. Same
/// exhaustive-fields convention as [`wire_pending_approval`].
fn wire_handoff_request(h: HandoffRequest) -> WireHandoffRequest {
    WireHandoffRequest {
        child_agent_id: h.child_agent_id,
        reason: h.reason,
        max_carry: u32::try_from(h.max_carry).unwrap_or(u32::MAX),
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

/// Build the wire-facing [`WireUsage`] for `u`. Same exhaustive-fields
/// convention as [`wire_pending_approval`].
fn wire_usage(u: Usage) -> WireUsage {
    WireUsage {
        input_tokens: u.input_tokens,
        output_tokens: u.output_tokens,
        cache_read_input_tokens: u.cache_read_input_tokens,
        cache_creation_input_tokens: u.cache_creation_input_tokens,
        __buffa_unknown_fields: buffa::UnknownFields::default(),
    }
}

/// Maps accepted outcome bodies onto a State-backed step proposal.
///
/// The envelope label names the stable step identity. Control must commit this
/// proposal and return a receipt before terminal facts may be sent. Bodies
/// therefore occur exactly once in durable storage.
#[must_use]
pub fn build_step_outcome_proposal(
    label: &ExecutionLabel,
    messages: Vec<WireMessage>,
) -> HarnessMessage {
    frame(
        label,
        WireStepOutcomeProposal {
            messages,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        },
    )
}

/// Maps a completed [`TurnResult`] onto the terminal facts `batch` frame.
///
/// Accepted outcome bodies are sent in [`build_step_outcome_proposal`] and are
/// intentionally absent here. `usage` is always populated; `carried_context`
/// is intentionally not transmitted (the control plane re-derives it from
/// `max_carry`).
#[must_use]
pub fn build_final_batch(label: &ExecutionLabel, turn: TurnResult) -> HarnessMessage {
    tracing::info!(
        output_messages = turn.messages.len(),
        input_tokens = turn.usage.input_tokens,
        output_tokens = turn.usage.output_tokens,
        cache_read_input_tokens = turn.usage.cache_read_input_tokens,
        cache_creation_input_tokens = turn.usage.cache_creation_input_tokens,
        pending_approvals = turn.pending_approvals.len(),
        handoff_requested = turn.handoff.is_some(),
        stop = ?turn.stop,
        "harness turn complete"
    );
    let wire_pending: Vec<WirePendingApproval> = turn
        .pending_approvals
        .into_iter()
        .map(wire_pending_approval)
        .collect();
    // #1660: carry each question from an `ask_question` call out to the
    // control plane so it can persist a `question_request` per question,
    // inside this SAME atomic turn batch — the question-pause SIBLING of
    // `wire_pending` above. Empty for every turn that never called
    // `ask_question`.
    let wire_pending_questions: Vec<WirePendingQuestion> = turn
        .pending_questions
        .into_iter()
        .map(wire_pending_question)
        .collect();
    // #623: carry each unattended fail-closed denial out to the control plane so
    // it can append the durable, signed audit event recording what was attempted
    // and why it did not run. Empty for every attended turn.
    let wire_unattended_denials: Vec<WireUnattendedDenialFact> = turn
        .unattended_denials
        .into_iter()
        .map(wire_unattended_denial_fact)
        .collect();
    // #872: carry each `__delegate_to` call's forensic record out to the
    // control plane so it can append the signed `subagent_spawn`/
    // `subagent_result` pair plus a `subagent_model_call` determinism record,
    // all inside this turn's own atomic commit batch. Empty for every turn
    // that never called `__delegate_to`.
    let wire_delegate_records: Vec<WireDelegateFact> = turn
        .delegate_records
        .into_iter()
        .map(wire_delegate_fact)
        .collect();
    let wire_routine_grant_drift = turn.routine_grant_drift;
    let wire_handoff = turn.handoff.map(wire_handoff_request);
    let stop_reason_i32 = turn.stop.map_or(
        WireStopReason::STOP_REASON_UNSPECIFIED as i32,
        llm_stop_to_wire_i32,
    );
    frame(
        label,
        WireTurnBatch {
            messages: Vec::new(),
            usage: buffa::MessageField::some(wire_usage(turn.usage)),
            pending_approvals: wire_pending,
            pending_questions: wire_pending_questions,
            handoff: wire_handoff.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
            stop_reason: buffa::EnumValue::from(stop_reason_i32),
            unattended_denials: wire_unattended_denials,
            delegate_records: wire_delegate_records,
            // The fire-abort classification — see
            // `TurnBatch.unattended_denial_aborted`'s doc.
            unattended_denial_aborted: turn.fire_stopped,
            routine_grant_drift: wire_routine_grant_drift,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        },
    )
}

/// Decode a wire turn transcript into the [`LlmMessage`]s the loop consumes.
///
/// Skips any message that maps to empty LLM content so phantom empty-content
/// turns (e.g. a display-only reasoning thought) never accumulate in history.
#[must_use]
pub fn decode_turn_messages(messages: &[WireMessage]) -> Vec<LlmMessage> {
    let mut input: Vec<LlmMessage> = Vec::new();
    for msg in messages {
        let m = wire_to_llm(msg);
        if !m.content.is_empty() {
            input.push(m);
        }
    }
    input
}

/// Run one turn end to end against an injected provider set and tool executor.
///
/// Returns the frames the served harness path would emit for the same input,
/// in the same order.
///
/// Under Execution protocol 2 that is one step-outcome proposal carrying the
/// accepted bodies, followed by the terminal facts `batch`. A turn that
/// produced no body returns the batch alone. The caller must commit each
/// proposal through the State step-commit door before it treats the terminal
/// facts as durable — the same contract the networked receive loop follows.
///
/// This is the buffered entry both in-process callers share (the harness's own
/// `run_turn_in_process` and the control plane's replay transport). It runs the
/// SAME orchestration the served path runs — [`decode_turn_messages`],
/// [`verify_approval_responses`], [`ProviderSet::resolve`], [`run_turn_options`],
/// the [`run_turn_captured`] loop, and [`build_final_batch`] — so the simulated
/// path exercises production code, not a lookalike, and a provider failure
/// classifies through [`connect_error_from_turn`] exactly as it does on the wire.
///
/// It differs from the served `connect` path only where the socket would be: the
/// turn is buffered (no live `delta` frames — the terminal batch carries the full
/// transcript) and it takes no stream/dispatch proxies (the caller composed the
/// tool surface without them). `clock` is `None` on production and `Some(pinned)`
/// only on the replay path, where it pins the turn's clock to the recorded
/// dispatch time (INV-11).
///
/// # Errors
///
/// Returns a [`ConnectError`] whose code classifies the provider failure (via
/// [`connect_error_from_turn`]) — the same mapping the served path returns — so
/// the caller's transient-vs-fatal handling is identical across transports.
///
/// Returns [`ErrorCode::InvalidArgument`] when the turn input carries no
/// fenced Execution grant, or one this build cannot admit. That is fail-closed:
/// a buffered turn runs under the same fence a served turn does.
pub async fn run_wire_turn(
    providers: &ProviderSet,
    request: &HarnessMessage,
    expected_audience: &execution::ExecutionAudience,
    tools: &(impl ToolExecutor + ?Sized),
    clock: Option<Arc<dyn Clock + Send + Sync>>,
) -> Result<Vec<HarnessMessage>, ConnectError> {
    // #1565 D4: admit the fenced Execution grant before any work runs. The
    // buffered path admits the same whole opening envelope as the served path.
    let Some(HarnessFrame::Input(turn_input)) = request.frame.as_ref() else {
        return Err(ConnectError::invalid_argument(
            "a buffered Execution turn must open with an input frame",
        ));
    };
    let grant = execution::required_grant(turn_input.execution.as_option())
        .map_err(|error| execution::connect_error_from_execution(&error))?;
    let mut session =
        execution::ExecutionSession::admit(&grant, expected_audience, execution::now())
            .map_err(|error| execution::connect_error_from_execution(&error))?;
    session
        .check_opening_envelope(
            &execution::required_label(request.label.as_option())
                .map_err(|error| execution::connect_error_from_execution(&error))?,
        )
        .map_err(|error| execution::connect_error_from_execution(&error))?;
    // Decode the inbound transcript exactly as `connect` does.
    let input = decode_turn_messages(&turn_input.messages);
    // Verify + bucket the inbound approvals against this turn's caller and the
    // harness's current sandbox mode — the same trust check the served path runs
    // before it will execute a previously-paused tool.
    let approvals = verify_approval_responses(
        &turn_input.approval_responses,
        &turn_input.caller,
        &current_sandbox_mode(),
    );
    // Verify + bucket the inbound routine-fire tool grants — the grant
    // SIBLING of `approvals` above.
    let routine_tool_grants = verify_routine_tool_grants(
        &turn_input.routine_tool_grants,
        &turn_input.caller,
        &current_sandbox_mode(),
        grant.conversation().as_str(),
    );
    // Resolve the per-turn backend + model from the wire selectors against the
    // configured ProviderSet (empty selectors → defaults).
    let resolved = providers.resolve(&turn_input.provider, &turn_input.model);
    // No live delta sink and no dispatch-recorder stream: the buffered turn has
    // no control plane to sign mutations over. A pinned clock threads only on the
    // replay path.
    // #1226: mirrors the served `connect` handler's `builtin_allow` resolution
    // exactly (`turn_input.scope_builtin_tools.then_some(turn_input.builtin_tools)`)
    // so the two paths can't drift on whether this agent is scoped for native
    // search grounding.
    let native_search_allowed = !turn_input.scope_builtin_tools
        || turn_input
            .builtin_tools
            .iter()
            .any(|n| n == polyc_tools::web::NATIVE_SEARCH_GROUNDING);
    // #1660: verify the same wire question_response entries the served path
    // does, before this turn's question-pause resume can trust any of them.
    let question_answers = verify_question_answers(&turn_input.question_answers);
    let mut options = run_turn_options(
        approvals,
        routine_tool_grants,
        TurnToggles::from_wire(turn_input),
        turn_input.prompt_cache_key.clone(),
        turn_input.step_budget,
        None,
        None,
        clock,
        native_search_allowed,
        question_answers,
        turn_input.turn_start_unix_ms,
    );
    // #870: resolve this turn's `__delegate_to` targets from the SAME
    // `providers`/`tools` this call already holds. Empty for every turn
    // whose agent declared none, so the agent loop advertises no delegate
    // tool and the turn is byte-for-byte unchanged.
    options.delegate_descriptors =
        resolve_delegate_descriptors(providers, tools, &turn_input.delegate_descriptors);
    // #874: this turn's resolved fan-out width cap and total delegate-call
    // budget, if the control plane set them. Both wire fields carry real
    // `optional` presence (#1680), so this is a direct pass-through — absent
    // stays `None` (harness applies its own default) and an explicit `0`
    // survives as `Some(0)` instead of being reinterpreted as unset.
    options.delegate_max_fanout = turn_input.delegate_fanout_cap;
    options.delegate_turn_budget = turn_input.delegate_turn_call_budget;
    let mut turn = run_turn_captured_under_grant(
        resolved.provider.as_ref(),
        tools,
        &resolved.model,
        input,
        options,
        grant.capabilities().granted(),
    )
    .await
    .map_err(|captured| captured.connect_error())?;

    // Labels are stamped per frame sent, exactly as the served path stamps
    // them, so a buffered turn's step numbering matches a served turn's for
    // the same shape.
    let mut frames = Vec::with_capacity(2);
    let bodies = std::mem::take(&mut turn.messages);
    if !bodies.is_empty() {
        let proposal_label = session
            .stamp_proposal()
            .map_err(|error| execution::connect_error_from_execution(&error))?;
        frames.push(build_step_outcome_proposal(&proposal_label, bodies));
    }
    let batch_label = session
        .stamp_proposal()
        .map_err(|error| execution::connect_error_from_execution(&error))?;
    frames.push(build_final_batch(&batch_label, turn));
    Ok(frames)
}

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

    /// #870 + #1565 D6: a delegate resolves to the SAME backend the parent
    /// turn runs on, and a stub is not reachable by naming one.
    ///
    /// Execution resolves no provider of its own now, so the delegate registry
    /// is built from the turn's brokered proxy. The previous code passed the
    /// injected registry, which `ProviderSet::new` guarantees contains a
    /// `stub`; every delegate resolved against it and answered with canned
    /// text without reaching the broker.
    #[test]
    fn a_proxy_only_set_resolves_every_delegate_to_the_one_brokered_backend() {
        let set = super::ProviderSet::proxy_only(
            "brokered",
            polyc_llm::into_dyn(polyc_llm::turn::StubProvider),
            "approved-model",
        );

        // The named backend resolves to itself.
        let named = set.resolve("brokered", "");
        assert_eq!(named.provider_name, "brokered");
        assert_eq!(named.model, "approved-model");

        // An unnamed selector falls to the same one.
        assert_eq!(set.resolve("", "").provider_name, "brokered");

        // Naming `stub` explicitly does NOT reach a stub backend: there is
        // none to reach, so it falls back to the brokered default.
        assert_eq!(
            set.resolve("stub", "").provider_name,
            "brokered",
            "a delegate must not be able to name its way onto a stub backend"
        );
        assert!(
            set.backend("stub").is_none(),
            "no stub backend may be registered in a proxy-only set"
        );

        // The contrast that makes the assertions above meaningful: the general
        // constructor does register a stub, which is what the old delegate
        // path resolved against.
        let general = super::ProviderSet::new(std::collections::HashMap::new(), "brokered");
        assert!(general.backend("stub").is_some());
        assert_eq!(general.resolve("stub", "").provider_name, "stub");
    }

    use super::{
        CapturedTurnError, ProviderSet, RegisteredProvider, TurnToggles, VerifiedApprovals,
        WireTurnFailureKind, classify_turn_error, connect_error_from_turn, run_turn_options,
        wire_title,
    };
    use connectrpc::ErrorCode;
    use polyc_llm::{LlmErrorKind, into_dyn, turn::StubProvider};
    use std::collections::HashMap;

    fn reg(default_model: &str) -> RegisteredProvider {
        RegisteredProvider {
            provider: into_dyn(StubProvider),
            default_model: default_model.to_owned(),
        }
    }

    fn provider_set(entries: &[(&str, &str)], default_provider: &str) -> ProviderSet {
        let map: HashMap<String, RegisteredProvider> = entries
            .iter()
            .map(|(name, model)| ((*name).to_owned(), reg(model)))
            .collect();
        ProviderSet::new(map, default_provider)
    }

    #[test]
    fn resolve_uses_named_provider_and_its_default_model() {
        let set = provider_set(&[("vertex", "gemini"), ("openai", "llama3.2")], "vertex");
        let r = set.resolve("openai", "");
        assert_eq!(r.provider_name, "openai");
        assert_eq!(r.model, "llama3.2");
        let r = set.resolve("openai", "mixtral");
        assert_eq!(r.model, "mixtral");
    }

    #[test]
    fn resolve_empty_provider_uses_default() {
        let set = provider_set(&[("openai", "llama3.2")], "openai");
        let r = set.resolve("", "");
        assert_eq!(r.provider_name, "openai");
        assert_eq!(r.model, "llama3.2");
    }

    #[test]
    fn resolve_unknown_provider_falls_back_to_default() {
        let set = provider_set(&[("openai", "llama3.2")], "openai");
        let r = set.resolve("openia", "");
        assert_eq!(r.provider_name, "openai");
        assert_eq!(r.model, "llama3.2");
    }

    #[test]
    fn turn_error_kind_maps_to_connect_code() {
        let cases = [
            (LlmErrorKind::RateLimit, ErrorCode::ResourceExhausted),
            (LlmErrorKind::Timeout, ErrorCode::DeadlineExceeded),
            (LlmErrorKind::Unavailable, ErrorCode::Unavailable),
            (LlmErrorKind::Auth, ErrorCode::Unauthenticated),
            (LlmErrorKind::BadRequest, ErrorCode::InvalidArgument),
            (LlmErrorKind::Other, ErrorCode::Internal),
        ];
        for (kind, expected) in cases {
            assert_eq!(
                connect_error_from_turn(kind, "boom".to_owned()).code,
                expected
            );
        }
    }

    /// `resolve_delegate_descriptors` derives each worker's
    /// `native_search_allowed` from the wire descriptor's own `builtin_tools`
    /// (`#1226`'s scoping, mirrored onto delegation) — never hardcoded, since
    /// grounding has no `ToolSpec` to ride `tool_specs` alongside every other
    /// built-in.
    #[test]
    fn resolve_delegate_descriptors_derives_native_search_allowed_from_builtin_tools() {
        use super::WireDelegateDescriptor;

        let providers = provider_set(&[("vertex", "gemini-x")], "vertex");
        let tools = polyc_tools::ToolRegistry::scoped(std::iter::empty());

        let grounded = WireDelegateDescriptor {
            agent_id: "researcher".to_owned(),
            provider: "vertex".to_owned(),
            model: "gemini-x".to_owned(),
            builtin_tools: vec![
                "web_fetch".to_owned(),
                polyc_tools::web::NATIVE_SEARCH_GROUNDING.to_owned(),
            ],
            max_steps: 4,
            ..WireDelegateDescriptor::default()
        };
        let ungrounded = WireDelegateDescriptor {
            agent_id: "researcher".to_owned(),
            provider: "vertex".to_owned(),
            model: "gemini-x".to_owned(),
            builtin_tools: vec!["web_fetch".to_owned()],
            max_steps: 4,
            ..WireDelegateDescriptor::default()
        };

        let out = super::resolve_delegate_descriptors(&providers, &tools, &[grounded, ungrounded]);
        assert_eq!(out.len(), 2);
        assert!(
            out[0].native_search_allowed,
            "a descriptor naming web_search_grounding in builtin_tools must resolve to \
             native_search_allowed: true"
        );
        assert!(
            !out[1].native_search_allowed,
            "a descriptor NOT naming web_search_grounding must resolve to \
             native_search_allowed: false — never hardcoded true, or every worker would \
             inherit a capability its own agent manifest never granted"
        );
    }

    /// [`classify_turn_error`] classifies the same [`LlmErrorKind`] the
    /// same way [`connect_error_from_turn`] does, mirrored onto the wire's
    /// [`WireTurnFailureKind`] instead of a Connect status code (`#756`).
    #[test]
    fn every_definite_kind_classifies_to_a_durable_failure() {
        let cases = [
            (LlmErrorKind::RateLimit, WireTurnFailureKind::RateLimit),
            (LlmErrorKind::Timeout, WireTurnFailureKind::Timeout),
            (LlmErrorKind::Unavailable, WireTurnFailureKind::Unavailable),
            (LlmErrorKind::Auth, WireTurnFailureKind::Auth),
            (LlmErrorKind::BadRequest, WireTurnFailureKind::BadRequest),
            (LlmErrorKind::Other, WireTurnFailureKind::Other),
        ];
        for (kind, expected) in cases {
            let CapturedTurnError::Failed(failure) = classify_turn_error(kind, "boom".to_owned())
            else {
                panic!("kind {kind:?} is definite and must carry a durable failure");
            };
            assert_eq!(
                failure.kind, expected,
                "kind {kind:?} should map to {expected:?}"
            );
            assert_eq!(failure.message, "boom");
        }
    }

    /// An ambiguous outcome has no durable form, by construction.
    ///
    /// A `TurnFailure` is persisted beside an unconditional `turn_complete`
    /// marker, and `turn_complete` alone is what says the turn's final batch
    /// landed. A durable ambiguous failure would therefore claim a turn
    /// completed because nobody knows what it did.
    #[test]
    fn an_ambiguous_outcome_carries_no_durable_failure() {
        let captured = classify_turn_error(LlmErrorKind::Ambiguous, "boom".to_owned());
        assert!(
            matches!(captured, CapturedTurnError::Ambiguous),
            "an ambiguous kind may not become a persistable failure"
        );
        // It still ends the call, on the surface that may carry it.
        assert_eq!(captured.connect_error().code, ErrorCode::Unknown);
    }

    /// Build a minimal [`super::RunTurnOptions`] under `step_budget`, holding
    /// every other `run_turn_options` input at its byte-for-byte default.
    fn options_with_step_budget(step_budget: u32) -> polyc_agent::RunTurnOptions {
        run_turn_options(
            VerifiedApprovals::default(),
            polyc_agent::RoutineGrantSet::default(),
            TurnToggles::default(),
            String::new(),
            step_budget,
            None,
            None,
            None,
            true,
            Vec::new(),
            0,
        )
    }

    /// The deployment-global `POLYCHROME_APPROVAL_MODE=approve-all-dangerous`
    /// override merges an effective `blanket_all` grant into a fire
    /// dispatch's routine tool grants, with NO routine-local grant present —
    /// the PRD's deployment-mode interaction requirement. An ordinary
    /// (non-fire) turn, or a fire dispatch under the ordinary Human mode,
    /// must never gain a blanket from this override.
    #[test]
    fn deployment_approve_all_dangerous_merges_a_blanket_all_grant_on_a_fire_dispatch() {
        let opts = run_turn_options(
            VerifiedApprovals::default(),
            polyc_agent::RoutineGrantSet::default(),
            TurnToggles {
                fire_dispatch: true,
                deployment_approve_all_dangerous: true,
                ..TurnToggles::default()
            },
            String::new(),
            0,
            None,
            None,
            None,
            true,
            Vec::new(),
            0,
        );
        assert!(
            opts.routine_tool_grants
                .blanket
                .is_some_and(|b| b.include_high),
            "the deployment override must merge a blanket_all-shaped grant"
        );
    }

    /// The SAME override on an ordinary (non-fire) turn must never merge a
    /// blanket — it only ever widens a fire dispatch's grants.
    #[test]
    fn deployment_approve_all_dangerous_is_inert_off_a_fire_dispatch() {
        let opts = run_turn_options(
            VerifiedApprovals::default(),
            polyc_agent::RoutineGrantSet::default(),
            TurnToggles {
                fire_dispatch: false,
                deployment_approve_all_dangerous: true,
                ..TurnToggles::default()
            },
            String::new(),
            0,
            None,
            None,
            None,
            true,
            Vec::new(),
            0,
        );
        assert!(
            opts.routine_tool_grants.blanket.is_none(),
            "the override must never apply outside a fire dispatch"
        );
    }

    /// `#68`: an `IngressDirective.budget_cap` BELOW the deployment baseline
    /// lowers the turn's `max_steps` to exactly the cap.
    #[test]
    fn run_turn_options_step_budget_cap_below_base_lowers_max_steps() {
        let base = polyc_agent::resolve_default_max_steps();
        // `base` is always >= 1 (the crate default is 8), so `base - 1` is a
        // genuine, strictly-lower cap even under a deployment override.
        let cap = u32::try_from(base.saturating_sub(1).max(1)).unwrap_or(1);
        let options = options_with_step_budget(cap);
        assert_eq!(
            options.max_steps,
            Some(cap as usize),
            "a cap below the baseline lowers max_steps to exactly the cap"
        );
    }

    /// `#68`: an `IngressDirective.budget_cap` ABOVE the deployment baseline
    /// is a no-op — a cap can only LOWER the resolved budget, never raise it.
    #[test]
    fn run_turn_options_step_budget_cap_above_base_does_not_raise_max_steps() {
        let base = polyc_agent::resolve_default_max_steps();
        let cap = u32::try_from(base).unwrap_or(u32::MAX).saturating_add(100);
        let options = options_with_step_budget(cap);
        assert_eq!(
            options.max_steps,
            Some(base),
            "a cap above the baseline never raises the budget past it"
        );
    }

    /// `#68`: `step_budget == 0` (the wire's "not set" convention) leaves
    /// `max_steps` unset — the ordinary `POLYCHROME_AGENT_MAX_STEPS` /
    /// `DEFAULT_MAX_STEPS` resolution applies exactly as before `#68`.
    #[test]
    fn run_turn_options_zero_step_budget_leaves_max_steps_unset() {
        let options = options_with_step_budget(0);
        assert_eq!(
            options.max_steps, None,
            "unset ⇒ None, so the agent loop's own resolution applies"
        );
    }

    #[test]
    fn new_inserts_stub_and_coerces_missing_default() {
        let set = provider_set(&[("openai", "llama3.2")], "ghost");
        let r = set.resolve("", "");
        assert_eq!(r.provider_name, "stub");
        assert_eq!(r.model, "stub");
        assert_eq!(set.resolve("stub", "").model, "stub");
    }

    #[test]
    fn wire_title_prefers_curated_then_humanizes() {
        assert_eq!(
            wire_title("Pay for & fetch a web page", "paid_fetch"),
            "Pay for & fetch a web page"
        );
        assert_eq!(wire_title("", "delete_file"), "Delete file");
        assert_eq!(wire_title("", "send_message"), "Send message");
    }

    // ── approval verification ──────────────────────────────────────────────

    use super::{WireApprovalResponse, verify_approval_responses};

    /// Build a signed wire approval response (mirrors what the control plane
    /// forwards) for use in the session-filtering tests below.
    fn wire_response(
        request_id: &str,
        tool: &str,
        args: &str,
        approved_for_session: bool,
        caller: &str,
        sandbox_mode: &str,
    ) -> WireApprovalResponse {
        use polyc_crypto::approval::{ApprovalSigner, response_payload};
        let signer = ApprovalSigner::from_seed(9);
        let (payload, _sig, _pk) = response_payload(
            request_id,
            tool,
            args,
            "",
            true,
            approved_for_session,
            &[],
            caller,
            "",
            sandbox_mode,
            "ok",
            "",
            "conv-h",
            "nonce-h",
            "00000000-0000-0000-0000-000000000001",
            &signer,
        );
        let d = polyc_crypto::approval::decode_response_full(&payload).expect("decode");
        WireApprovalResponse {
            request_id: d.request_id,
            tool_name: d.tool_name,
            args_json: d.args_json,
            modified_args_json: d.modified_args_json,
            injected_context: d.injected_context,
            approved: d.approved,
            approved_for_session: d.approved_for_session,
            caller: d.caller,
            approver: d.approver,
            sandbox_mode: d.sandbox_mode,
            reason: d.reason,
            conversation_id: d.conversation_id,
            nonce: d.nonce,
            covered_capabilities: d.covered_capabilities,
            signer_pk_hex: d.signer_pk_hex,
            signature_hex: d.signature_hex,
            turn_id: d.turn_id,
            routine_grant: d.routine_grant,
            tool_descriptor_hash: d.tool_descriptor_hash,
            grant_scope: d.grant_scope,
            // Not a signed field: a freshly minted test response was never
            // forwarded past execution, so it is always fresh.
            already_executed: false,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    // Test mode used by the session tests (both ends agree on the same string).
    const M: &str = "workspace-write";

    #[test]
    fn executed_session_grant_keeps_tool_memory_but_not_the_one_shot() {
        let mut resp = wire_response("call-1", "grep", "{}", true, "A", M);
        resp.already_executed = true;
        let v = verify_approval_responses(&[resp], "A", M);
        assert!(
            v.session_approved.contains_key("grep"),
            "per-tool session memory survives execution"
        );
        assert!(
            v.decisions.is_empty(),
            "the spent one-shot tuple is NOT re-armed for execution"
        );
    }

    #[test]
    fn session_grant_applies_only_to_its_bound_caller() {
        let responses = vec![wire_response("call-1", "grep", "{}", true, "A", M)];
        let for_a = verify_approval_responses(&responses, "A", M);
        assert!(for_a.session_approved.contains_key("grep"));
        assert_eq!(for_a.decisions.len(), 1);
        let for_b = verify_approval_responses(&responses, "B", M);
        assert!(
            for_b.session_approved.is_empty(),
            "A's session grant must NOT auto-approve B"
        );
        assert_eq!(for_b.decisions.len(), 1);
    }

    #[test]
    fn session_grant_is_bound_to_its_sandbox_mode() {
        let responses = vec![wire_response("call-1", "grep", "{}", true, "A", M)];
        assert!(
            verify_approval_responses(&responses, "A", M)
                .session_approved
                .contains_key("grep")
        );
        let other = verify_approval_responses(&responses, "A", "danger-full-access");
        assert!(
            other.session_approved.is_empty(),
            "a grant from one mode must not auto-approve under another"
        );
        assert_eq!(other.decisions.len(), 1);
    }

    #[test]
    fn non_session_approval_yields_no_session_memory() {
        let responses = vec![wire_response("call-1", "grep", "{}", false, "A", M)];
        let v = verify_approval_responses(&responses, "A", M);
        assert!(v.session_approved.is_empty());
        assert_eq!(v.decisions.len(), 1);
    }

    #[test]
    fn tampered_caller_fails_verification_and_is_dropped() {
        let mut resp = wire_response("call-1", "grep", "{}", true, "A", M);
        // Re-scope the grant to the attacker AFTER signing: the signature no
        // longer covers `caller`, so the whole response is rejected.
        resp.caller = "ATTACKER".to_owned();
        let v = verify_approval_responses(&[resp], "ATTACKER", M);
        assert!(v.decisions.is_empty(), "a tampered response is dropped");
        assert!(v.session_approved.is_empty());
    }

    // ── routine tool grant verification ─────────────────────────────────────

    use super::verify_routine_tool_grants;

    /// Build a signed routine-fire tool grant (mirrors what the control
    /// plane forwards on `TurnInput.routine_tool_grants`).
    fn wire_grant(
        tool: &str,
        caller: &str,
        sandbox_mode: &str,
        descriptor_hash: &str,
        grant_scope: &str,
    ) -> WireApprovalResponse {
        use polyc_crypto::approval::{ApprovalSigner, routine_grant_payload};
        let signer = ApprovalSigner::from_seed(9);
        let (payload, _sig, _pk) = routine_grant_payload(
            "call-1",
            tool,
            "{}",
            "",
            true,
            caller,
            "",
            sandbox_mode,
            "owner approved during setup",
            &["mutate-external".to_owned()],
            "conv-h",
            "nonce-h",
            "00000000-0000-0000-0000-000000000001",
            descriptor_hash,
            grant_scope,
            &signer,
        );
        let d = polyc_crypto::approval::decode_response_full(&payload).expect("decode");
        WireApprovalResponse {
            request_id: d.request_id,
            tool_name: d.tool_name,
            args_json: d.args_json,
            modified_args_json: d.modified_args_json,
            injected_context: d.injected_context,
            approved: d.approved,
            approved_for_session: d.approved_for_session,
            caller: d.caller,
            approver: d.approver,
            sandbox_mode: d.sandbox_mode,
            reason: d.reason,
            conversation_id: d.conversation_id,
            nonce: d.nonce,
            covered_capabilities: d.covered_capabilities,
            signer_pk_hex: d.signer_pk_hex,
            signature_hex: d.signature_hex,
            turn_id: d.turn_id,
            routine_grant: d.routine_grant,
            tool_descriptor_hash: d.tool_descriptor_hash,
            grant_scope: d.grant_scope,
            // Not a signed field: a freshly minted test grant was never
            // forwarded past execution, so it is always fresh.
            already_executed: false,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    #[test]
    fn a_matching_per_tool_grant_verifies_and_buckets_by_tool_name() {
        let grant = wire_grant("send_message", "A", M, "sha256:abc", "tool");
        let out = verify_routine_tool_grants(&[grant], "A", M, "conv-h");
        let entry = out
            .per_tool
            .get("send_message")
            .expect("the grant buckets under its tool name");
        assert_eq!(entry.descriptor_hash, "sha256:abc");
        assert!(out.blanket.is_none());
    }

    #[test]
    fn a_blanket_grant_verifies_and_buckets_as_blanket() {
        let grant = wire_grant("", "A", M, "", "blanket_all");
        let out = verify_routine_tool_grants(&[grant], "A", M, "conv-h");
        assert!(out.per_tool.is_empty());
        assert!(
            out.blanket.is_some_and(|b| b.include_high),
            "blanket_all sets include_high"
        );
    }

    #[test]
    fn a_grant_for_a_different_caller_is_dropped() {
        let grant = wire_grant("send_message", "A", M, "sha256:abc", "tool");
        let out = verify_routine_tool_grants(&[grant], "B", M, "conv-h");
        assert!(
            out.per_tool.is_empty(),
            "A's grant must not apply to a turn whose caller is B"
        );
    }

    #[test]
    fn a_grant_under_a_different_sandbox_mode_is_dropped() {
        let grant = wire_grant("send_message", "A", M, "sha256:abc", "tool");
        let out = verify_routine_tool_grants(&[grant], "A", "danger-full-access", "conv-h");
        assert!(
            out.per_tool.is_empty(),
            "a grant minted under one mode must not apply under another"
        );
    }

    #[test]
    fn a_tampered_grant_fails_verification_and_is_dropped() {
        let mut grant = wire_grant("send_message", "A", M, "sha256:abc", "tool");
        grant.tool_descriptor_hash = "sha256:evil".to_owned();
        let out = verify_routine_tool_grants(&[grant], "A", M, "conv-h");
        assert!(out.per_tool.is_empty(), "a tampered grant is dropped");
    }

    #[test]
    fn a_response_with_no_routine_grant_marker_is_ignored() {
        // An ordinary session approval, forwarded on `routine_tool_grants` by
        // mistake, carries no `routine_grant` marker and must not bucket.
        let resp = wire_response("call-1", "grep", "{}", true, "A", M);
        let out = verify_routine_tool_grants(&[resp], "A", M, "conv-h");
        assert!(out.per_tool.is_empty());
        assert!(out.blanket.is_none());
    }

    /// A correctly signed grant minted for one fire conversation must not
    /// apply when this turn's own fire conversation is a different one —
    /// the same caller and sandbox mode can repeat across two routines
    /// owned by the same person, so the conversation binding is the only
    /// thing that separates them.
    #[test]
    fn a_grant_for_a_different_fire_conversation_is_dropped() {
        // `wire_grant` signs conversation_id "conv-h" (routine A's fire
        // conversation); present it against routine B's.
        let grant = wire_grant("send_message", "A", M, "sha256:abc", "tool");
        let out = verify_routine_tool_grants(&[grant], "A", M, "conv-b");
        assert!(
            out.per_tool.is_empty(),
            "a grant signed for conversation A must not apply to conversation B"
        );
    }

    /// A signed grant with a `grant_scope` outside the closed set
    /// (`"tool"` / `"blanket_below_high"` / `"blanket_all"`) must yield no
    /// grant at all — never widen into a blanket by falling through the
    /// non-empty-string check.
    #[test]
    fn a_signed_unknown_grant_scope_yields_neither_per_tool_nor_blanket() {
        let grant = wire_grant("send_message", "A", M, "sha256:abc", "tools");
        let out = verify_routine_tool_grants(&[grant], "A", M, "conv-h");
        assert!(
            out.per_tool.is_empty(),
            "an unrecognized scope must not bucket as a per-tool grant"
        );
        assert!(
            out.blanket.is_none(),
            "an unrecognized scope must not widen into a blanket grant"
        );
    }

    // ── question-answer verification ────────────────────────────────────────

    use super::{WireQuestionAnswer, verify_question_answers};

    const TEST_TURN: &str = "018f47f0-5f70-7cc5-98df-123456789abc";

    fn wire_question_answer(
        call_id: &str,
        index: u32,
        state: &str,
        selected_index: u32,
        selected_label: &str,
        answered_by: &str,
        signer: &polyc_crypto::approval::ApprovalSigner,
    ) -> WireQuestionAnswer {
        let signed_selected_index =
            (state != polyc_crypto::question::DECLINED_STATE).then_some(selected_index);
        let (payload, sig, pk) = polyc_crypto::question::answer_payload(
            TEST_TURN,
            call_id,
            index,
            r#"{"questions":[]}"#,
            state,
            signed_selected_index,
            selected_label,
            answered_by,
            "conv-q",
            "nonce-q",
            signer,
        );
        let verified =
            polyc_crypto::question::verify_signed_answer(&payload).expect("payload verifies");
        WireQuestionAnswer {
            turn_id: TEST_TURN.to_owned(),
            call_id: verified.call_id,
            index: verified.index,
            question_args_json: verified.question_args_json,
            state: verified.state,
            selected_index: verified.selected_index.unwrap_or_default(),
            selected_label: verified.selected_label,
            answered_by: verified.answered_by,
            conversation_id: verified.conversation_id,
            nonce: verified.nonce,
            signer_pk_hex: polyc_crypto::hex::lower(&pk),
            signature_hex: polyc_crypto::hex::lower(&sig),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }

    /// Invariant I3: a well-formed, signed `question_response` verifies and
    /// maps to a [`polyc_agent::question::VerifiedAnswer`] with every field
    /// carried through.
    #[test]
    fn verify_question_answers_accepts_a_well_formed_signed_answer() {
        let signer = polyc_crypto::approval::ApprovalSigner::from_seed(1);
        let wire = wire_question_answer(
            "call-1",
            0,
            polyc_crypto::question::ANSWERED_STATE,
            1,
            "Production",
            "slack:T1:U9",
            &signer,
        );
        let out = verify_question_answers(&[wire]);
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].call_id, "call-1");
        assert_eq!(out[0].index, 0);
        assert_eq!(out[0].state, polyc_agent::question::AnswerState::Answered);
        assert_eq!(out[0].selected_index, Some(1));
        assert_eq!(out[0].selected_label, "Production");
    }

    /// Invariant I3: a tampered field invalidates the signature — the
    /// answer must be dropped, never trusted.
    #[test]
    fn verify_question_answers_drops_a_tampered_answer() {
        let signer = polyc_crypto::approval::ApprovalSigner::from_seed(1);
        let mut wire = wire_question_answer(
            "call-1",
            0,
            polyc_crypto::question::ANSWERED_STATE,
            1,
            "Production",
            "slack:T1:U9",
            &signer,
        );
        // Tamper with the selection AFTER signing.
        wire.selected_index = 0;
        wire.selected_label = "Staging".to_owned();
        assert!(
            verify_question_answers(&[wire]).is_empty(),
            "a tampered answer must never verify"
        );
    }

    /// Invariant I3: a signature from an untrusted/mismatched signer must
    /// never verify against a different signer's key.
    #[test]
    fn verify_question_answers_drops_an_answer_signed_by_a_different_key() {
        let signer = polyc_crypto::approval::ApprovalSigner::from_seed(1);
        let other = polyc_crypto::approval::ApprovalSigner::from_seed(2);
        let mut wire = wire_question_answer(
            "call-1",
            0,
            polyc_crypto::question::ANSWERED_STATE,
            1,
            "Production",
            "slack:T1:U9",
            &signer,
        );
        // Swap in a different (unrelated) public key while keeping the
        // original signature bytes — the signature no longer verifies
        // against the swapped key.
        wire.signer_pk_hex = polyc_crypto::hex::lower(&other.public_key_bytes());
        assert!(verify_question_answers(&[wire]).is_empty());
    }

    /// An unrecognized `state` string is dropped rather than silently
    /// coerced into one of the three known states.
    #[test]
    fn verify_question_answers_drops_an_unrecognized_state() {
        let signer = polyc_crypto::approval::ApprovalSigner::from_seed(1);
        let wire = wire_question_answer(
            "call-1",
            0,
            "maybe",
            1,
            "Production",
            "slack:T1:U9",
            &signer,
        );
        assert!(verify_question_answers(&[wire]).is_empty());
    }

    /// A decline carries no selection, even though the wire's plain
    /// `selected_index`/`selected_label` fields are present — the verified
    /// answer's `selected_index` must be `None`.
    #[test]
    fn verify_question_answers_declined_carries_no_selection() {
        let signer = polyc_crypto::approval::ApprovalSigner::from_seed(1);
        let wire = wire_question_answer(
            "call-1",
            0,
            polyc_crypto::question::DECLINED_STATE,
            0,
            "",
            "slack:T1:U9",
            &signer,
        );
        let out = verify_question_answers(&[wire]);
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].state, polyc_agent::question::AnswerState::Declined);
        assert_eq!(out[0].selected_index, None);
    }
}