polyc-a2a 2026.8.3

//! The bridge between an A2A task and a polychrome turn.
//!
//! An A2A `message/send` runs one polychrome turn. [`TurnRunner`] abstracts
//! "run a turn and tell me how it ended" so the A2A state machine (the `rpc`
//! module's task engine) is testable with an in-process stub and wired in
//! production to the control plane via [`AgentDialerRunner`].
//!
//! The turn's streamed [`TurnEvent`]s fold into one [`TurnOutcome`]
//! ([`outcome_from_events`]). The load-bearing mapping is
//! [`TurnEvent::ApprovalPending`] → [`TurnOutcome::InputRequired`]: a turn that
//! paused on a human-in-the-loop approval gate becomes the A2A `input-required`
//! state, so an A2A peer learns it must supply a decision before the task can
//! finish.
//!
//! [`TurnRunner::run_turn_streaming`] surfaces that same event stream live
//! instead of folding it early — what backs the A2A `SendStreamingMessage`
//! transport (`crate::rpc`). [`TurnRunner::run_turn`] (the unary path) is
//! defined in terms of it (fold-then-return), so the two paths read the
//! identical stream and can't drift on what a turn actually did.

use std::{future::Future, pin::Pin};

use futures::{Stream, StreamExt as _};
use polyc_rpc_client::{
    AgentDialer, AssertedConversationVisibility, DialError, EdgeCredentials, IngressIdentity,
    TurnEvent, TurnIngress, user_message,
};

/// What this edge states about every turn's audience.
///
/// This surface talks to another agent over a signed request, and the turn's
/// answer is RETURNED to that peer — it rides back on the task's status
/// message and artifacts. So a room does exist, this edge cannot see who the
/// peer relays to, and it states "unknown".
///
/// Not "not applicable": that value says there is nothing to classify, needs
/// no grant to state, and is the disclosure-permitting reading. It belongs to
/// a surface whose output reaches nobody, like the scheduled trigger, whose
/// dial only logs the reply length.
const CONVERSATION_VISIBILITY: AssertedConversationVisibility =
    AssertedConversationVisibility::UNKNOWN;

/// The inputs one turn needs, resolved from an inbound A2A message.
#[derive(Debug, Clone)]
pub struct TurnRequest {
    /// The namespaced conversation id (`a2a:<context-id>`).
    pub conversation_id: String,
    /// A fresh, time-ordered execution id for this turn.
    pub exec_id: String,
    /// Stable identity of the authenticated peer message.
    pub source_identity: IngressIdentity,
    /// The user text extracted from the message's text parts.
    pub text: String,
}

/// Why State did not durably accept an A2A source message.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IngressReceiptError {
    /// Stable diagnostic suitable for a JSON-RPC error response.
    pub message: String,
    /// Whether redelivering the source message could plausibly succeed.
    pub retryable: bool,
    /// Whether State rejected the same source identity with changed content.
    pub content_conflict: bool,
}

/// Stable State identity returned after durable ingress.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IngressReceipt {
    /// State-derived dispatch identity for the source event.
    pub dispatch_id: String,
}

/// How a turn ended, in the terms the A2A task state machine maps from.
///
/// A transport failure is modelled as [`Self::Failed`] (an A2A `failed` task),
/// not a JSON-RPC error: the request was well-formed, the *task* could not
/// complete.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TurnOutcome {
    /// The turn produced a final assistant reply.
    Completed {
        /// The aggregated assistant answer text.
        text: String,
    },
    /// The turn paused on a human-approval gate — maps to A2A `input-required`.
    InputRequired {
        /// Turn that emitted this occurrence of `request_id`.
        turn_id: String,
        /// The approval `request_id` a decision must answer.
        request_id: String,
        /// The tool name awaiting approval.
        tool_name: String,
        /// A human-readable prompt describing what needs a decision.
        prompt: String,
        /// The short-lived signed capability (`#787`) that must be replayed
        /// unmodified on the eventual [`ApprovalResponder::respond`] call —
        /// carried off the originating [`TurnEvent::ApprovalPending`].
        resolve_token: String,
    },
    /// The turn failed (e.g. the control plane was unreachable).
    Failed {
        /// Operator/peer-facing failure summary.
        message: String,
    },
}

/// One item of a turn's live event stream.
///
/// In the terms the A2A `SendStreamingMessage` transport maps from:
/// incremental assistant text (→ a `TaskArtifactUpdateEvent` chunk), or the
/// turn's terminal [`TurnOutcome`] (→ the closing `TaskStatusUpdateEvent`).
/// Exactly one [`Self::Outcome`] item ends the stream — mirroring how exactly
/// one [`TurnOutcome`] resolves [`TurnRunner::run_turn`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TurnStreamEvent {
    /// State acknowledged the source event durably.
    DurablyReceived,
    /// Incremental assistant answer text.
    TextDelta(String),
    /// The turn's terminal outcome — the last item of the stream.
    Outcome(TurnOutcome),
}

/// Runs one turn for an A2A task. Object-safe (the server holds an
/// `Arc<dyn TurnRunner>`) so the stub and the live dialer are interchangeable.
pub trait TurnRunner: Send + Sync {
    /// Receive one source message durably without dispatching its turn.
    ///
    /// The default is an in-process test double's acknowledgement. The live
    /// runner overrides it with State's `ReceiveIngress` RPC.
    fn receive_ingress<'a>(
        &'a self,
        _req: TurnRequest,
    ) -> Pin<Box<dyn Future<Output = Result<IngressReceipt, IngressReceiptError>> + Send + 'a>>
    {
        Box::pin(async {
            Ok(IngressReceipt {
                dispatch_id: "test-dispatch".to_owned(),
            })
        })
    }

    /// Run the turn described by `req` and resolve to its [`TurnOutcome`].
    fn run_turn<'a>(
        &'a self,
        req: TurnRequest,
    ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>>;

    /// Run the turn described by `req`, streaming each [`TurnStreamEvent`] as
    /// it happens rather than folding to one [`TurnOutcome`] — what backs the
    /// A2A `SendStreamingMessage` transport.
    ///
    /// The default implementation runs [`Self::run_turn`] to completion and
    /// yields its single [`TurnOutcome`] as the stream's only item — the
    /// correct (if non-streaming) behavior for any [`TurnRunner`] that has
    /// nothing finer-grained to report (e.g. the test stubs). A runner backed
    /// by a genuinely streamed turn (see [`AgentDialerRunner`]) overrides this
    /// to surface its live event stream instead of folding it early.
    fn run_turn_streaming<'a>(
        &'a self,
        req: TurnRequest,
    ) -> Pin<Box<dyn Stream<Item = TurnStreamEvent> + Send + 'a>> {
        Box::pin(async_stream::stream! {
            yield TurnStreamEvent::DurablyReceived;
            yield TurnStreamEvent::Outcome(self.run_turn(req).await);
        })
    }
}

/// Fold a turn's streamed events into a single [`TurnOutcome`].
///
/// A durable failure wins first: if any [`TurnEvent::TurnFailed`] (`#756`) is
/// present the turn is [`TurnOutcome::Failed`] — the SAME outcome a
/// dial/stream-level error already maps to, so a turn that ended without a
/// `batch` is never folded into an empty `Completed { text: "" }` the way an
/// unmatched wildcard would silently do.
/// Otherwise an approval pause wins: if any [`TurnEvent::ApprovalPending`] is
/// present the turn is [`TurnOutcome::InputRequired`] (a turn cannot both
/// finish and stay paused). Otherwise the assistant [`TurnEvent::TextDelta`]s
/// are concatenated into [`TurnOutcome::Completed`]. Pure, so the mapping is
/// unit-tested without a live stream.
#[must_use]
pub fn outcome_from_events(events: &[TurnEvent]) -> TurnOutcome {
    if let Some(message) = events.iter().find_map(|e| match e {
        TurnEvent::TurnFailed { message, .. } => Some(message.clone()),
        _ => None,
    }) {
        return TurnOutcome::Failed { message };
    }

    if let Some(pending) = events.iter().find_map(|e| match e {
        TurnEvent::ApprovalPending {
            turn_id,
            request_id,
            tool_name,
            title,
            args_json,
            reason,
            resolve_token,
            // The computed-preview enrichment (`#1496`) is a chat-edge
            // concern (Slack/Telegram/Discord/email); the A2A protocol
            // surface has no card to enrich, so it's unread here.
            preview: _,
            // Same reasoning as `preview` above: the standing-grant notice is
            // chat-card copy, not a field the A2A protocol surface reads.
            fire_dispatch: _,
        } => Some((
            turn_id,
            request_id,
            tool_name,
            title,
            args_json,
            reason,
            resolve_token,
        )),
        _ => None,
    }) {
        let (turn_id, request_id, tool_name, title, args_json, reason, resolve_token) = pending;
        let label = if title.is_empty() { tool_name } else { title };
        // Surface the gate's reason to the A2A peer when present (e.g. the
        // lethal-trifecta override) so the `input-required` prompt explains WHY
        // a human is needed; empty for an ordinary approval.
        let prompt = if reason.is_empty() {
            format!("Approval required to run `{label}` with arguments {args_json}")
        } else {
            format!("Approval required to run `{label}` with arguments {args_json} — {reason}")
        };
        return TurnOutcome::InputRequired {
            turn_id: turn_id.clone(),
            request_id: request_id.clone(),
            tool_name: tool_name.clone(),
            prompt,
            resolve_token: resolve_token.clone(),
        };
    }

    let text = events
        .iter()
        .filter_map(|e| match e {
            TurnEvent::TextDelta(t) => Some(t.as_str()),
            _ => None,
        })
        .collect::<Vec<_>>()
        .join("");
    TurnOutcome::Completed { text }
}

/// A [`TurnRunner`] for an edge brought up without a control-plane address.
///
/// The Agent Card endpoint serves regardless of control-plane wiring, so an
/// edge with no `agent_addr` still discloses its identity to peers. Every
/// `message/send` task, however, fails closed with a clear reason — naming the
/// unset address so an operator can fix it — rather than dialing an endpoint
/// that was never configured.
#[derive(Clone, Copy, Debug, Default)]
pub struct UnconfiguredRunner;

/// The failure text every [`UnconfiguredRunner`] turn returns. Names the unset
/// variable so the cause is actionable from the task's `failed` status alone.
const UNCONFIGURED_MESSAGE: &str = "control-plane address is unset \
     (POLYCHROME_AGENT_ADDR); this edge serves its Agent Card but cannot run \
     message/send tasks until it is set";

impl TurnRunner for UnconfiguredRunner {
    fn receive_ingress<'a>(
        &'a self,
        _req: TurnRequest,
    ) -> Pin<Box<dyn Future<Output = Result<IngressReceipt, IngressReceiptError>> + Send + 'a>>
    {
        Box::pin(async {
            Err(IngressReceiptError {
                message: UNCONFIGURED_MESSAGE.to_owned(),
                retryable: false,
                content_conflict: false,
            })
        })
    }

    fn run_turn<'a>(
        &'a self,
        _req: TurnRequest,
    ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
        Box::pin(async {
            TurnOutcome::Failed {
                message: UNCONFIGURED_MESSAGE.to_owned(),
            }
        })
    }
}

/// A [`TurnRunner`] backed by the live control plane.
///
/// It opens a streaming turn over the public `AgentService` and folds the
/// result with [`outcome_from_events`]. A dial/stream failure becomes
/// [`TurnOutcome::Failed`].
#[derive(Clone)]
pub struct AgentDialerRunner {
    dialer: AgentDialer,
}

impl AgentDialerRunner {
    /// Build a runner dialing the control plane at `addr`
    /// (`http://host:port`).
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` is not a valid URI.
    pub fn new(addr: &str) -> Result<Self, DialError> {
        Ok(Self {
            dialer: AgentDialer::new(addr)?,
        })
    }

    /// Build a runner dialing the control plane at `addr`, authenticated with
    /// `creds` — every turn it dials carries the edge's bearer header and a
    /// signed `AssertedAttribution` envelope (see
    /// [`AgentDialer::with_credentials`]).
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` is not a valid URI, or
    /// [`DialError::InvalidBearer`] if `creds`' bearer can't be encoded as an
    /// HTTP header value.
    pub fn with_credentials(addr: &str, creds: EdgeCredentials) -> Result<Self, DialError> {
        Ok(Self {
            dialer: AgentDialer::with_credentials(addr, creds)?,
        })
    }
}

/// Classifies a durable-receive refusal for the JSON-RPC reply.
///
/// A same-identity/different-content refusal reaches this edge as
/// ALREADY-EXISTS. It was `invalid_argument` until `d50919731` corrected the
/// control plane's mapper, and this side was not updated with it — so every
/// real conflict classified as "not a conflict" and reached the peer as an
/// internal error (`-32603`) instead of a bad-parameter one (`-32602`).
///
/// This is a free function so a test can reach it. The closure it came from
/// could only be driven through a live dial, which is exactly how the drift
/// survived: the control plane's own
/// `a_digest_conflict_reaches_the_compat_route_as_a_spent_key` exists because
/// a hand-built error once hid this same mismatch. That test pins the mapper
/// end (a `DigestConflict` becomes ALREADY-EXISTS); the cases below pin this
/// end (ALREADY-EXISTS means a conflict). Neither half is sufficient alone.
fn classify_ingress_error(err: &polyc_rpc_client::DialError) -> IngressReceiptError {
    let content_conflict = err.is_already_exists();
    IngressReceiptError {
        retryable: err.is_retryable(),
        content_conflict,
        // A conflict's own text names both digests, and `is_already_exists`
        // says why that must not reach whoever presented the key: telling a
        // matching retry apart from a differing one is an equality oracle over
        // the stored request (`#2618`). Reading the predicate and then echoing
        // the text it warns about would defeat it, so the conflict branch
        // carries fixed copy — the same thing the local branch already says.
        // The control plane redacts the identical error the same way, in
        // `key_spent_response`.
        message: if content_conflict {
            "source event was already received with different content".to_owned()
        } else {
            err.to_string()
        },
    }
}

impl TurnRunner for AgentDialerRunner {
    fn receive_ingress<'a>(
        &'a self,
        req: TurnRequest,
    ) -> Pin<Box<dyn Future<Output = Result<IngressReceipt, IngressReceiptError>> + Send + 'a>>
    {
        Box::pin(async move {
            self.dialer
                .receive_ingress(
                    TurnIngress::new(
                        &req.conversation_id,
                        &req.exec_id,
                        req.source_identity,
                        crate::rpc::claimed_namespace(),
                        // The id is `scoped_id(kind, peer_id, raw)` over the
                        // server-authenticated peer, so this edge derives it
                        // from its own inbound event rather than taking a
                        // caller's word for it (`#2646`).
                        polyc_rpc_client::ConversationIdOrigin::AuthenticatedSource,
                        vec![user_message(&req.text)],
                    )
                    .with_conversation_visibility(CONVERSATION_VISIBILITY),
                )
                .await
                .map(|received| IngressReceipt {
                    dispatch_id: received.dispatch_id().to_owned(),
                })
                .map_err(|err| classify_ingress_error(&err))
        })
    }

    /// Folds [`Self::run_turn_streaming`]'s event stream into one
    /// [`TurnOutcome`] — the unary path stays a fold over the SAME stream the
    /// streaming path surfaces directly, so the two can't drift on what the
    /// control plane actually said.
    fn run_turn<'a>(
        &'a self,
        req: TurnRequest,
    ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
        Box::pin(async move {
            let mut stream = self.run_turn_streaming(req);
            let mut outcome = None;
            while let Some(event) = stream.next().await {
                if let TurnStreamEvent::Outcome(o) = event {
                    outcome = Some(o);
                }
            }
            // `run_turn_streaming` always yields exactly one `Outcome` before
            // ending (its own contract); this default only fires if that
            // contract were ever violated, so fail closed rather than panic.
            outcome.unwrap_or_else(|| TurnOutcome::Failed {
                message: "turn stream ended without a terminal outcome".to_owned(),
            })
        })
    }

    fn run_turn_streaming<'a>(
        &'a self,
        req: TurnRequest,
    ) -> Pin<Box<dyn Stream<Item = TurnStreamEvent> + Send + 'a>> {
        Box::pin(async_stream::stream! {
            let stream = match self
                .dialer
                .receive_ingress(
                    TurnIngress::new(
                        &req.conversation_id,
                        &req.exec_id,
                        req.source_identity,
                        crate::rpc::claimed_namespace(),
                        // The id is `scoped_id(kind, peer_id, raw)` over the
                        // server-authenticated peer, so this edge derives it
                        // from its own inbound event rather than taking a
                        // caller's word for it (`#2646`).
                        polyc_rpc_client::ConversationIdOrigin::AuthenticatedSource,
                        vec![user_message(&req.text)],
                    )
                    .with_conversation_visibility(CONVERSATION_VISIBILITY),
                )
                .await
            {
                Ok(received) => {
                    yield TurnStreamEvent::DurablyReceived;
                    match self.dialer.attach_ingress(&received).await {
                        Ok(stream) => stream,
                        Err(err) => {
                            yield TurnStreamEvent::Outcome(TurnOutcome::Failed {
                                message: err.to_string(),
                            });
                            return;
                        }
                    }
                }
                Err(err) => {
                    yield TurnStreamEvent::Outcome(TurnOutcome::Failed {
                        message: err.to_string(),
                    });
                    return;
                }
            };
            futures::pin_mut!(stream);
            let mut events = Vec::new();
            loop {
                match stream.next().await {
                    Some(Ok(event)) => {
                        // `outcome_from_events` needs the full transcript to
                        // apply its failure/pending precedence (`#756`), so
                        // every event is retained even though only
                        // `TextDelta` is surfaced live here.
                        if let TurnEvent::TextDelta(ref text) = event {
                            yield TurnStreamEvent::TextDelta(text.clone());
                        }
                        events.push(event);
                    }
                    Some(Err(err)) => {
                        yield TurnStreamEvent::Outcome(TurnOutcome::Failed {
                            message: err.to_string(),
                        });
                        return;
                    }
                    None => break,
                }
            }
            yield TurnStreamEvent::Outcome(outcome_from_events(&events));
        })
    }
}

/// Object-safe abstraction over submitting a human's approval decision.
///
/// For an A2A `input-required` task's continuation (`#792`), mirroring
/// [`TurnRunner`]'s stub/live split so `crate::rpc::handle`'s continuation
/// logic is testable without a live control-plane dial.
pub trait ApprovalResponder: Send + Sync {
    /// Submit `approved`'s decision for `request_id`, scoped to
    /// `conversation_id`. `resolve_token` is the short-lived signed
    /// capability (`#787`) carried unmodified off the originating
    /// [`TurnEvent::ApprovalPending`] / [`TurnOutcome::InputRequired`] —
    /// required: the control plane rejects a decision whose token is
    /// missing, expired, or bound to a different request or conversation.
    /// Returns whether a decision was newly persisted — `false` is the
    /// idempotent no-op (already answered / unknown `request_id`) — or a
    /// diagnostic string on a transport failure.
    fn respond<'a>(
        &'a self,
        turn_id: &'a str,
        request_id: &'a str,
        approved: bool,
        reason: &'a str,
        conversation_id: &'a str,
        resolve_token: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<bool, String>> + Send + 'a>>;
}

/// The failure text every [`UnconfiguredApprovalResponder`] decision returns.
/// Names the unset variable so the cause is actionable from the error alone,
/// matching [`UNCONFIGURED_MESSAGE`].
const UNCONFIGURED_APPROVAL_MESSAGE: &str = "control-plane address is unset \
     (POLYCHROME_AGENT_ADDR); this edge cannot submit approval decisions until \
     it is set";

/// An [`ApprovalResponder`] for an edge brought up without a control-plane
/// address — mirrors [`UnconfiguredRunner`]: fails closed rather than dialing
/// an endpoint that was never configured.
#[derive(Clone, Copy, Debug, Default)]
pub struct UnconfiguredApprovalResponder;

impl ApprovalResponder for UnconfiguredApprovalResponder {
    fn respond<'a>(
        &'a self,
        _turn_id: &'a str,
        _request_id: &'a str,
        _approved: bool,
        _reason: &'a str,
        _conversation_id: &'a str,
        _resolve_token: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<bool, String>> + Send + 'a>> {
        Box::pin(async { Err(UNCONFIGURED_APPROVAL_MESSAGE.to_owned()) })
    }
}

/// An [`ApprovalResponder`] backed by the live control plane's
/// `ApprovalService`.
#[derive(Clone)]
pub struct ApprovalDialerResponder {
    dialer: polyc_rpc_client::ApprovalDialer,
}

impl ApprovalDialerResponder {
    /// Build a responder dialing the control plane at `addr`
    /// (`http://host:port`) — the SAME endpoint [`AgentDialerRunner`] dials;
    /// `AgentService` and `ApprovalService` are served on one Connect port.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` is not a valid URI.
    pub fn new(addr: &str) -> Result<Self, DialError> {
        Ok(Self {
            dialer: polyc_rpc_client::ApprovalDialer::new(addr)?,
        })
    }

    /// Build a responder dialing the control plane at `addr`, authenticated
    /// with `bearer` — every decision it submits carries the edge's bearer
    /// header (see [`polyc_rpc_client::ApprovalDialer::with_bearer`]). No
    /// envelope is signed: `ApprovalService` calls carry no `AgentStart`.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` is not a valid URI, or
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value.
    pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            dialer: polyc_rpc_client::ApprovalDialer::with_bearer(addr, bearer)?,
        })
    }
}

impl ApprovalResponder for ApprovalDialerResponder {
    fn respond<'a>(
        &'a self,
        turn_id: &'a str,
        request_id: &'a str,
        approved: bool,
        reason: &'a str,
        conversation_id: &'a str,
        resolve_token: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<bool, String>> + Send + 'a>> {
        Box::pin(async move {
            let choice = if approved {
                polyc_rpc_client::ApprovalChoice::Approve
            } else {
                polyc_rpc_client::ApprovalChoice::Deny
            };
            self.dialer
                .respond(
                    turn_id,
                    request_id,
                    choice,
                    reason,
                    conversation_id,
                    "",
                    "",
                    resolve_token,
                    None,
                )
                .await
                .map(|outcome| outcome.persisted)
                .map_err(|err| err.to_string())
        })
    }
}

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

    const TEST_TURN: &str = "00000000-0000-0000-0000-000000000001";
    use polyc_rpc_client::TurnFailureKind;

    /// A digest conflict is classified as a content conflict, not an internal
    /// error.
    ///
    /// The control plane maps `StateError::DigestConflict` to ALREADY-EXISTS
    /// (`d50919731`, pinned there by
    /// `a_digest_conflict_reaches_the_compat_route_as_a_spent_key`). This side
    /// read `invalid_argument` and was never updated, so a peer that reused a
    /// `messageId` with new content was told the server had broken
    /// (`-32603`) rather than that its parameters were bad (`-32602`).
    #[test]
    fn an_already_exists_refusal_is_a_content_conflict() {
        let classified = classify_ingress_error(&polyc_rpc_client::DialError::Connect(
            connectrpc::ConnectError::already_exists("recorded digest differs from presented"),
        ));
        assert!(
            classified.content_conflict,
            "a digest conflict must reach the peer as a bad parameter, not an internal error"
        );
    }

    /// A conflict's digests never reach the peer.
    ///
    /// `is_already_exists` exists as a predicate rather than an accessor
    /// precisely so a caller cannot echo the text behind it: separating a
    /// matching retry from a differing one is an equality oracle over the
    /// stored request (`#2618`). Reading that predicate and then forwarding
    /// `err.to_string()` — which names both digests — would hand the peer the
    /// oracle the predicate withholds.
    ///
    /// The refusal is namespaced per authenticated peer, so the reach is
    /// limited to that peer's own earlier message. This still asserts the
    /// digests are absent, because the leak is a contract violation
    /// independent of how far it reaches, and nothing else would catch it.
    #[test]
    fn a_conflict_does_not_echo_the_digests_to_the_peer() {
        let recorded = "1".repeat(64);
        let presented = "2".repeat(64);
        let classified = classify_ingress_error(&polyc_rpc_client::DialError::Connect(
            connectrpc::ConnectError::already_exists(format!(
                "command receive:whatever was recorded with digest {recorded}, not {presented}"
            )),
        ));
        assert!(
            classified.content_conflict,
            "precondition: this is the conflict branch"
        );
        assert!(
            !classified.message.contains(&recorded) && !classified.message.contains(&presented),
            "the digests must not reach the peer, got: {}",
            classified.message
        );
    }

    /// The control for the case above: the code the mapper NO LONGER emits for
    /// a digest conflict must not be read as one. Without this, the classifier
    /// could match every error and the case above would still pass.
    #[test]
    fn an_invalid_argument_refusal_is_not_a_content_conflict() {
        let classified = classify_ingress_error(&polyc_rpc_client::DialError::Connect(
            connectrpc::ConnectError::invalid_argument("malformed ingress"),
        ));
        assert!(
            !classified.content_conflict,
            "only ALREADY-EXISTS carries the same-identity/changed-content meaning"
        );
    }

    /// This surface has no room and no human audience, so it states "not
    /// applicable". Stating "unknown" would claim a room it does not have.
    #[test]
    fn this_edge_states_no_conversation_audience() {
        assert_eq!(
            CONVERSATION_VISIBILITY,
            AssertedConversationVisibility::UNKNOWN
        );
    }

    #[test]
    fn text_deltas_fold_to_completed() {
        let events = vec![
            TurnEvent::ToolStarted {
                name: "search".to_owned(),
            },
            TurnEvent::TextDelta("Hello, ".to_owned()),
            TurnEvent::TextDelta("world.".to_owned()),
            TurnEvent::Done,
        ];
        assert_eq!(
            outcome_from_events(&events),
            TurnOutcome::Completed {
                text: "Hello, world.".to_owned()
            }
        );
    }

    #[test]
    fn approval_pending_folds_to_input_required() {
        let events = vec![
            TurnEvent::TextDelta("working on it".to_owned()),
            TurnEvent::ApprovalPending {
                turn_id: TEST_TURN.to_owned(),
                request_id: "call-1".to_owned(),
                tool_name: "delete_repo".to_owned(),
                title: "Delete repository".to_owned(),
                args_json: r#"{"repo":"x"}"#.to_owned(),
                reason: "untrusted content in context".to_owned(),
                resolve_token: String::new(),
                preview: None,
                fire_dispatch: false,
            },
            TurnEvent::Done,
        ];
        match outcome_from_events(&events) {
            TurnOutcome::InputRequired {
                request_id,
                tool_name,
                prompt,
                ..
            } => {
                assert_eq!(request_id, "call-1");
                assert_eq!(tool_name, "delete_repo");
                assert!(prompt.contains("Delete repository"));
                assert!(prompt.contains(r#"{"repo":"x"}"#));
                assert!(
                    prompt.contains("untrusted content in context"),
                    "the gate reason must surface in the input-required prompt"
                );
            }
            other => panic!("expected InputRequired, got {other:?}"),
        }
    }

    /// A durable `TurnFailed` (`#756`) must fold to `Failed`, never a silent
    /// empty `Completed` — the exact bug an unmatched wildcard would
    /// reintroduce.
    #[test]
    fn turn_failed_folds_to_failed_not_a_silent_empty_completion() {
        let events = vec![
            TurnEvent::TurnFailed {
                kind: TurnFailureKind::RateLimit,
                message: "provider returned 429".to_owned(),
            },
            TurnEvent::Done,
        ];
        assert_eq!(
            outcome_from_events(&events),
            TurnOutcome::Failed {
                message: "provider returned 429".to_owned()
            }
        );
    }

    /// A failure wins even alongside other content — it's the SAME
    /// precedence the dial/stream-error path already has (a turn that failed
    /// durably never reports as merely paused or completed).
    #[test]
    fn turn_failed_wins_over_an_approval_pending_in_the_same_batch() {
        let events = vec![
            TurnEvent::ApprovalPending {
                turn_id: TEST_TURN.to_owned(),
                request_id: "call-1".to_owned(),
                tool_name: "delete_repo".to_owned(),
                title: String::new(),
                args_json: "{}".to_owned(),
                reason: String::new(),
                resolve_token: String::new(),
                preview: None,
                fire_dispatch: false,
            },
            TurnEvent::TurnFailed {
                kind: TurnFailureKind::Other,
                message: "durable failure".to_owned(),
            },
            TurnEvent::Done,
        ];
        assert_eq!(
            outcome_from_events(&events),
            TurnOutcome::Failed {
                message: "durable failure".to_owned()
            }
        );
    }

    #[test]
    fn empty_stream_is_empty_completed() {
        assert_eq!(
            outcome_from_events(&[]),
            TurnOutcome::Completed {
                text: String::new()
            }
        );
    }

    #[tokio::test]
    async fn unconfigured_runner_fails_naming_the_unset_address() {
        let outcome = UnconfiguredRunner
            .run_turn(TurnRequest {
                conversation_id: "a2a:ctx".to_owned(),
                exec_id: "exec".to_owned(),
                source_identity: IngressIdentity::reported("a2a:test-peer", "m1").unwrap(),
                text: "hello".to_owned(),
            })
            .await;
        match outcome {
            TurnOutcome::Failed { message } => assert!(
                message.contains("POLYCHROME_AGENT_ADDR"),
                "message must name the unset variable: {message}"
            ),
            other => panic!("expected Failed, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn unconfigured_approval_responder_fails_naming_the_unset_address() {
        let err = UnconfiguredApprovalResponder
            .respond(TEST_TURN, "req-1", true, "a2a:peer", "a2a:ctx", "tok-1")
            .await
            .expect_err("must fail closed");
        assert!(
            err.contains("POLYCHROME_AGENT_ADDR"),
            "message must name the unset variable: {err}"
        );
    }

    /// A stub carrying only a fixed [`TurnOutcome`] — exercises the trait's
    /// DEFAULT `run_turn_streaming`, proving it folds to exactly one
    /// [`TurnStreamEvent::Outcome`] matching `run_turn` (`#371`).
    struct FixedOutcomeRunner(TurnOutcome);
    impl TurnRunner for FixedOutcomeRunner {
        fn run_turn<'a>(
            &'a self,
            _req: TurnRequest,
        ) -> Pin<Box<dyn Future<Output = TurnOutcome> + Send + 'a>> {
            let outcome = self.0.clone();
            Box::pin(async move { outcome })
        }
    }

    #[tokio::test]
    async fn default_run_turn_streaming_yields_exactly_one_outcome() {
        let runner = FixedOutcomeRunner(TurnOutcome::Completed {
            text: "42".to_owned(),
        });
        let req = TurnRequest {
            conversation_id: "a2a:ctx".to_owned(),
            exec_id: "exec".to_owned(),
            source_identity: IngressIdentity::reported("a2a:test-peer", "m1").unwrap(),
            text: "hi".to_owned(),
        };
        let events: Vec<TurnStreamEvent> = runner.run_turn_streaming(req).collect().await;
        assert_eq!(
            events,
            vec![
                TurnStreamEvent::DurablyReceived,
                TurnStreamEvent::Outcome(TurnOutcome::Completed {
                    text: "42".to_owned()
                })
            ],
            "a runner with no finer-grained stream must acknowledge durability before its outcome"
        );
    }
}