polyc-rpc-client 2026.7.0

//! Thin connectrpc client over the generated
//! [`polyc_proto::proto::polychrome::agent::v1::AgentServiceClient`].
//!
//! Both the operator CLI (`polychrome send`) and the reference-edge receiver dial the
//! external-facing `AgentService` with the same Connect-streaming dance: build
//! a connect-rust client, send one [`AgentRequest`], drain the
//! [`AgentResponse`](polyc_proto::proto::polychrome::agent::v1::AgentResponse)
//! stream, and render each non-empty content block to user-visible text. This
//! crate is the single home for that logic so the two call sites can't drift.
//!
//! RPC-client concerns deliberately live here rather than in
//! `polyc-agent` (the turn-loop crate, the wrong layer for a transport
//! client).
//!
//! For v1 the whole turn is collected into a single string and returned at
//! end-of-turn; incremental streaming (e.g. Slack `chat.appendStream`) is a
//! follow-up that can build on the same client.

#![forbid(unsafe_code)]
#![warn(missing_docs)]

pub mod edge;
pub use edge::{EdgeAdapter, build_attribution};

use std::sync::Arc;

use connectrpc::client::{ClientConfig, HttpClient};
use futures::Stream;
use polyc_agent::text_message;
use polyc_proto::proto::polychrome::agent::v1::{
    AgentEnd, AgentRequest, AgentServiceClient, AgentStart, ClassifyRequest,
    CompactionReason as WireCompactionReason, ContextCompacted, InterruptRequest, Message,
    ParticipantMessage, Verdict, agent_response, content, tool_call_content,
};
use polyc_proto::proto::polychrome::approval::v1::{
    ApprovalResponseRequest, ApprovalServiceClient, ListPendingRequest,
};
use polyc_proto::proto::polychrome::ops::v1::{
    AckRequest, DecideRequest, NotificationServiceClient, OperatorMailboxServiceClient,
    PollPendingRequest, decide_reply, ops_action_view,
};
use polyc_proto::proto::polychrome::persona::v1::{
    AdminInviteRequest, AutoLinkOutcome, AutoLinkRequest, CompleteLinkRequest, DescribeRequest,
    LinkOutcome, PersonaServiceClient, StartDeepLinkRequest, StartLinkRequest,
};

/// Errors an agent dial can produce.
#[derive(Debug, thiserror::Error)]
pub enum DialError {
    /// Could not parse the configured agent address as a URI.
    #[error("invalid agent address {addr:?}: {source}")]
    InvalidAddress {
        /// The address string that failed to parse.
        addr: String,
        /// The underlying URI parse error.
        #[source]
        source: http::uri::InvalidUri,
    },
    /// Building the TLS client for an `https://` endpoint failed (e.g. no
    /// process-default crypto provider). The dial fails closed rather than
    /// silently downgrading to plaintext.
    #[error("tls setup failed for agent address: {0}")]
    Tls(String),
    /// Connect-level error from the `AgentService` stream.
    #[error(transparent)]
    Connect(#[from] connectrpc::ConnectError),
}

impl DialError {
    /// The Connect error code, when this is a transport-level Connect error
    /// (`None` for local address/TLS-setup failures).
    #[must_use]
    pub const fn code(&self) -> Option<connectrpc::ErrorCode> {
        match self {
            Self::Connect(e) => Some(e.code),
            Self::InvalidAddress { .. } | Self::Tls(_) => None,
        }
    }

    /// Whether retrying the call could plausibly succeed. Only transient
    /// transport conditions are retryable; terminal codes
    /// (`InvalidArgument`, `Unauthenticated`, `NotFound`, …) and local
    /// address/TLS failures are not. Edges use this to decide whether to ask
    /// the source platform to redeliver (retryable) or to drop / 4xx
    /// (terminal — redelivery would loop forever).
    #[must_use]
    pub const fn is_retryable(&self) -> bool {
        matches!(
            self.code(),
            Some(
                connectrpc::ErrorCode::Unavailable
                    | connectrpc::ErrorCode::DeadlineExceeded
                    | connectrpc::ErrorCode::ResourceExhausted
                    | connectrpc::ErrorCode::Aborted
            )
        )
    }
}

/// Default per-call deadline for an agent turn (emitted as `Connect-Timeout-Ms`
/// on every dial). Turns can be long (tool loops, slow providers), so this is
/// generous; edges add their own outer `tokio::time::timeout` for defence.
const AGENT_DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);

/// Default per-call deadline for the short control-plane RPCs (approval
/// responses, persona lookups/ceremonies). These never run a turn.
const CONTROL_DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

/// Bound the TCP connect phase of every control-plane dial so a dead or
/// terminating peer fails fast (surfaced as `Unavailable`) instead of
/// blackholing the SYN for the kernel `tcp_syn_retries` (~130s). Bounds only
/// `connect(2)`; DNS and TLS handshake are covered by the per-call deadline.
/// Mirrors the `CONNECT_TIMEOUT` const in the llm-vertex / llm-openai clients.
const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

/// Build the Connect HTTP transport for `uri`, honoring its scheme: `https`
/// dials over TLS (OS trust store); anything else (incl. a scheme-less
/// `host:port`) uses plaintext. An `https` address is **never** silently
/// downgraded — a TLS build failure surfaces as [`DialError::Tls`].
fn http_client_for(uri: &http::Uri) -> Result<HttpClient, DialError> {
    if uri.scheme_str() == Some("https") {
        use rustls_platform_verifier::ConfigVerifierExt;
        let tls = rustls::ClientConfig::with_platform_verifier()
            .map_err(|e| DialError::Tls(e.to_string()))?;
        Ok(HttpClient::builder()
            .connect_timeout(CONNECT_TIMEOUT)
            .with_tls(std::sync::Arc::new(tls)))
    } else {
        Ok(HttpClient::builder()
            .connect_timeout(CONNECT_TIMEOUT)
            .plaintext())
    }
}

/// Per-call options carrying the active span's W3C `traceparent`, so the
/// control-plane handler re-parents on this turn's span instead of starting a
/// fresh trace. Cheap when no propagator is installed (the global getter is a
/// no-op and no header is set). The per-call deadline comes from the client's
/// `with_default_timeout`, so it need not be repeated here.
fn traced_options() -> connectrpc::client::CallOptions {
    let mut headers = http::HeaderMap::new();
    polyc_runtime::propagation::inject_current_span_into(&mut headers);
    connectrpc::client::CallOptions::default()
        .with_headers(headers.into_iter().filter_map(|(n, v)| n.map(|n| (n, v))))
}

/// Why a turn's prompt was auto-compacted before it ran. The buffered analog
/// of [`WireCompactionReason`], lifted to a closed Rust enum so surfaces match
/// on it without touching the wire crate.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CompactionReason {
    /// Older turns were folded into an anchored-iterative summary
    /// (message-count trigger). The summary survives as the prompt's preamble.
    Summarized,
    /// Older tool-output payloads were trimmed to fit the token budget
    /// (char/token-budget trigger). No messages were dropped, only their bulk.
    ///
    /// NOTE: the control plane no longer EMITS this reason — the retroactive
    /// tool-output truncation layer was removed in favor of token-budget
    /// summarization (the per-call tool-output cap now bounds individual tool
    /// results). This variant is retained as a never-emitted-on-happy-path
    /// decode path so any old persisted/wire value still maps (the enum stays
    /// exhaustive) and an unknown future wire reason has a quiet fallback.
    Truncated,
}

impl CompactionReason {
    /// Canonical one-line headline (with a leading glyph) for a pre-turn
    /// compaction notice, shared by every edge so the user-facing wording can't
    /// drift between Slack, Telegram, and the rest. `summarized_messages` is used
    /// only by [`Self::Summarized`]. Returns PLAIN text — no markup — so each
    /// surface applies its own emphasis/escaping (Slack `_…_` mrkdwn, Telegram
    /// `*…*`, …) without double-formatting.
    #[must_use]
    pub fn notice_headline(self, summarized_messages: u32) -> String {
        match self {
            Self::Summarized => {
                let plural = if summarized_messages == 1 { "" } else { "s" };
                format!(
                    "🧠 Summarized {summarized_messages} earlier message{plural} to keep the \
                     conversation manageable"
                )
            }
            Self::Truncated => {
                "✂️ Trimmed earlier tool output to keep the conversation manageable".to_owned()
            }
        }
    }
}

/// Incremental event emitted while a turn streams from the control plane.
///
/// Unlike [`AgentDialer::run_turn`], which folds the whole turn into one
/// string, the streaming API surfaces each meaningful step as it arrives so a
/// live surface (e.g. Slack `chat.appendStream`) can update in place.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TurnEvent {
    /// The turn's context was auto-compacted before it ran. Emitted ONCE, as
    /// the first event of the turn (before any [`TurnEvent::TextDelta`]), so a
    /// live surface can render a brief notice rather than letting earlier work
    /// silently vanish from the model's view. The full transcript is retained
    /// server-side; only this turn's prompt was compacted.
    ContextCompacted {
        /// Whether the context was summarized or truncated.
        reason: CompactionReason,
        /// Earlier messages folded into the summary (0 when truncated).
        summarized_messages: u32,
        /// Short preview of the surviving summary; empty when truncated.
        summary_preview: String,
    },
    /// Incremental assistant answer text (model/assistant role).
    TextDelta(String),
    /// A tool call has started, named for a user-visible "thinking step".
    ToolStarted {
        /// The tool/function name, or the call id when the name is absent.
        name: String,
    },
    /// The turn paused before executing a tool that requires human approval.
    /// Emitted (one per pending call) from the terminal `AgentEnd` just before
    /// [`TurnEvent::Done`]. The caller submits a decision via
    /// `ApprovalService.Respond` (the THIN path) and re-drives the turn.
    ApprovalPending {
        /// Tool-call id == the approval `request_id` to answer.
        request_id: String,
        /// The tool/function name awaiting approval. Raw machine identifier;
        /// the field of record for trust/audit.
        tool_name: String,
        /// Human display label (MCP-style `title`) for the tool, for rendering
        /// in the approval prompt. May be empty; the surface then derives one
        /// from `tool_name`.
        title: String,
        /// Arguments JSON for the call.
        args_json: String,
        /// Why this call is gated, when the pause is an OVERRIDE of a call that
        /// would not otherwise need approval. Empty for an ordinary gated call;
        /// non-empty only for the lethal-trifecta / Rule-of-Two containment
        /// override — rendered on the approval card so the approver sees that
        /// untrusted content is in context and this is an outbound call.
        reason: String,
    },
    /// The turn suspended to delegate to a sub-agent: the model invoked the
    /// reserved `__handoff_to` primitive. Surfaced (once) from the terminal
    /// `AgentEnd.handoff` just before [`TurnEvent::Done`]. The child
    /// conversation runs independently; the parent resumes when the child
    /// returns, so an edge can render "delegating…" rather than going silent.
    HandoffStarted {
        /// The child agent / planner chosen; empty selects the parent's
        /// default planner.
        child_agent_id: String,
        /// Free-form reason captured for operator visibility.
        reason: String,
    },
    /// Terminal event: the turn has ended and no further events follow.
    Done,
}

/// Reusable handle for dialing the polychrome control plane.
#[derive(Clone)]
pub struct AgentDialer {
    client: Arc<AgentServiceClient<HttpClient>>,
}

impl AgentDialer {
    /// Build a dialer pointed at `addr` (expects `http://host:port`).
    ///
    /// # Errors
    ///
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI.
    pub fn new(addr: &str) -> Result<Self, DialError> {
        let uri = addr
            .parse::<http::Uri>()
            .map_err(|source| DialError::InvalidAddress {
                addr: addr.to_owned(),
                source,
            })?;
        let http = http_client_for(&uri)?;
        let config = ClientConfig::new(uri).with_default_timeout(AGENT_DIAL_TIMEOUT);
        let client = AgentServiceClient::new(http, config);
        Ok(Self {
            client: Arc::new(client),
        })
    }

    /// Build an [`ApprovalDialer`] for the SAME control-plane endpoint.
    /// `AgentService` and `ApprovalService` are served on one Connect port, so
    /// an approval client reuses the agent address — callers that already hold
    /// an `AgentDialer` don't need to thread a second address.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI.
    pub fn approval_dialer(addr: &str) -> Result<ApprovalDialer, DialError> {
        ApprovalDialer::new(addr)
    }

    /// Run one turn against the control plane and collect the aggregated
    /// text response.
    ///
    /// `conversation_id` is the stable id for the conversation (the Slack
    /// adapter derives it from the thread; the CLI takes it as an argument).
    /// `user_text` is the user message with any bot mention already stripped.
    /// Returns the concatenated assistant text across every batch in the
    /// response stream, or `Ok(String::new())` if the turn produced no text.
    /// Non-text content variants (tool calls, tool results, thoughts) are
    /// rendered as bracketed placeholders.
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for any transport/stream/encoding
    /// error from the `AgentService` call.
    pub async fn run_turn(
        &self,
        conversation_id: &str,
        exec_id: &str,
        user_text: &str,
    ) -> Result<String, DialError> {
        self.run_turn_with(conversation_id, exec_id, user_text, Attribution::default())
            .await
    }

    /// Like [`run_turn`](Self::run_turn) but attributes the turn to a caller
    /// (and participants). Non-streaming edges that resolve an identity via
    /// [`EdgeAdapter::caller`] use this so their turns populate
    /// `AgentStart.caller` (persona attribution) — the buffered analog of
    /// [`run_turn_streaming_messages_with`](Self::run_turn_streaming_messages_with).
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for any transport/stream/encoding error.
    pub async fn run_turn_with(
        &self,
        conversation_id: &str,
        exec_id: &str,
        user_text: &str,
        attribution: Attribution,
    ) -> Result<String, DialError> {
        let request = build_request(
            conversation_id,
            exec_id,
            vec![text_message("user", user_text)],
            None,
            attribution,
        );

        let mut stream = self
            .client
            .connect_with_options(request, traced_options())
            .await?;
        // The assistant's prose answer. Tool calls, tool results, and thoughts
        // are intermediate scaffolding — they're aggregated separately and only
        // surfaced when the turn produced no text at all (e.g. a turn that ends
        // on a tool_call / approval pause), so a normal answer reads cleanly
        // instead of "[tool_call:…]\n[tool_result:…]\nIt is 3pm."
        let mut text_parts: Vec<String> = Vec::new();
        let mut scaffolding: Vec<String> = Vec::new();
        while let Some(view) = stream.message().await? {
            let response = view.to_owned_message();
            match response.r#type {
                Some(agent_response::Type::Outputs(outputs)) => {
                    for msg in outputs.messages {
                        // The turn stream carries every message the turn
                        // produced — including the tool-execution result, which
                        // the control plane emits as a `tool`-role Text block
                        // (the raw tool output, e.g. `{"now":"…"}`). Only the
                        // `model`/`assistant` text is the user-facing answer;
                        // tool-role text is intermediate data, not the reply.
                        let is_assistant = matches!(msg.role.as_str(), "model" | "assistant");
                        let Some(content_block) = msg.content.into_option() else {
                            continue;
                        };
                        match content_block.r#type {
                            Some(content::Type::Text(t)) if is_assistant && !t.text.is_empty() => {
                                text_parts.push(t.text);
                            }
                            // Non-assistant text (tool result echo, user) is not
                            // the reply — drop it entirely (not even fallback).
                            Some(content::Type::Text(_)) => {}
                            // Tool calls/results render as scaffolding fallback;
                            // reasoning is excluded by `render_content` (returns
                            // None) so raw chain-of-thought never becomes the
                            // reply on a tool-only / approval-pause turn.
                            other => {
                                if let Some(rendered) = render_content(other) {
                                    scaffolding.push(rendered);
                                }
                            }
                        }
                    }
                }
                // An explicit AgentEnd ends the turn. The stream closing
                // without one is an implicit end-of-turn; we still return
                // whatever was aggregated.
                Some(agent_response::Type::End(_)) => break,
                // A pre-turn compaction notice carries no answer text, and an
                // empty envelope is defensive (the wire allows it). The buffered
                // API only aggregates the reply, so skip both.
                Some(agent_response::Type::Compacted(_)) | None => {}
            }
        }
        let reply = if text_parts.is_empty() {
            scaffolding.join("\n")
        } else {
            text_parts.join("\n")
        };
        Ok(reply)
    }

    /// Run one turn against the control plane and stream each meaningful step
    /// as a [`TurnEvent`], for live surfaces that update in place.
    ///
    /// The request is built identically to [`AgentDialer::run_turn`]; see that
    /// method for the meaning of `conversation_id`, `exec_id`, and
    /// `user_text`. The returned stream yields:
    ///
    /// - [`TurnEvent::TextDelta`] for each model/assistant-role text block,
    /// - [`TurnEvent::ToolStarted`] when a tool call begins, and
    /// - [`TurnEvent::Done`] once at end-of-turn, after which the stream ends.
    ///
    /// Tool-role result echoes and empty/non-textual blocks produce no event.
    ///
    /// # Errors
    ///
    /// The outer `Result` carries a [`DialError::Connect`] if opening the
    /// stream fails. Each item is a `Result` so per-message transport/decode
    /// errors surface inline without tearing down the whole stream type.
    pub async fn run_turn_streaming(
        &self,
        conversation_id: &str,
        exec_id: &str,
        user_text: &str,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
        self.run_turn_streaming_messages(
            conversation_id,
            exec_id,
            vec![text_message("user", user_text)],
        )
        .await
    }

    /// Like [`Self::run_turn_streaming`] but takes a pre-built (e.g. attributed
    /// multi-party) message list as the turn input.
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for any transport/stream/encoding error.
    pub async fn run_turn_streaming_messages(
        &self,
        conversation_id: &str,
        exec_id: &str,
        messages: Vec<Message>,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
        self.run_turn_streaming_messages_with(
            conversation_id,
            exec_id,
            messages,
            None,
            Attribution::default(),
        )
        .await
    }

    /// The full-fidelity variant: one method carries everything a turn's
    /// request can — a settled inbound [`PaymentReceipt`] (the control plane
    /// persists a signed `payment_receipt` event in the turn's atomic batch)
    /// and the caller [`Attribution`] (resolved to durable personas and
    /// recorded as `caller`/`participant` events). One method rather than a
    /// matrix of variants, so a paid *and* attributed edge can't silently
    /// drop one of the two.
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for any transport/stream/encoding error.
    pub async fn run_turn_streaming_messages_with(
        &self,
        conversation_id: &str,
        exec_id: &str,
        messages: Vec<Message>,
        payment_receipt: Option<PaymentReceipt>,
        attribution: Attribution,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
        let request = build_request(
            conversation_id,
            exec_id,
            messages,
            payment_receipt,
            attribution,
        );
        self.run_turn_streaming_request(request).await
    }

    /// Open the connect stream for one pre-built request and project the
    /// response envelopes into [`TurnEvent`]s — the single transport body
    /// every streaming variant shares.
    async fn run_turn_streaming_request(
        &self,
        request: AgentRequest,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
        let mut stream = self
            .client
            .connect_with_options(request, traced_options())
            .await?;
        Ok(async_stream::try_stream! {
            while let Some(view) = stream.message().await? {
                let response = view.to_owned_message();
                match response.r#type {
                    // A pre-turn compaction notice, surfaced before any text so
                    // the live surface can flag it ahead of the answer.
                    Some(agent_response::Type::Compacted(c)) => {
                        yield event_from_compacted(*c);
                    }
                    Some(agent_response::Type::Outputs(outputs)) => {
                        for msg in outputs.messages {
                            if let Some(event) = message_to_event(msg) {
                                yield event;
                            }
                        }
                    }
                    Some(agent_response::Type::End(end)) => {
                        // The terminal envelope's extensions (pending approvals,
                        // a sub-agent handoff) surface before the terminal Done
                        // so a live surface can prompt / show "delegating…".
                        for event in events_from_end(*end) {
                            yield event;
                        }
                        return;
                    }
                    // Empty envelope — defensive; ignore (mirrors `run_turn`).
                    None => {}
                }
            }
            // Stream closed without an explicit AgentEnd: still signal a
            // terminal event so the caller can finalise the live surface.
            yield TurnEvent::Done;
        })
    }

    /// Ask the control plane's participation gate whether to reply to the
    /// latest message of a (multi-party) thread. Returns `true` only on
    /// `respond`; `notify` / `ignore` map to `false` (stay silent). The gate
    /// runs a cheap classifier model server-side and never runs a turn.
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn should_respond(
        &self,
        conversation_id: &str,
        bot_name: &str,
        transcript: Vec<ParticipantMessage>,
    ) -> Result<bool, DialError> {
        let request = ClassifyRequest {
            conversation_id: conversation_id.to_owned(),
            bot_name: bot_name.to_owned(),
            transcript,
            ..Default::default()
        };
        let resp = self
            .client
            .classify_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(resp.verdict.to_i32() == Verdict::VERDICT_RESPOND as i32)
    }

    /// Cancel the conversation's in-flight turn without dropping a Connect
    /// stream. Returns `true` if a running turn was found and signalled to
    /// cancel; `false` is the idempotent no-op (no turn running on the replica
    /// that served this call).
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn interrupt(&self, conversation_id: &str) -> Result<bool, DialError> {
        let request = InterruptRequest {
            conversation_id: conversation_id.to_owned(),
            ..Default::default()
        };
        let resp = self
            .client
            .interrupt_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(resp.interrupted)
    }
}

/// A human's decision on a pending approval, as the edges present it.
///
/// One enum instead of a widening tuple of booleans, so an edge maps its button
/// once and its accessors ([`Self::approved`], [`Self::approved_for_session`],
/// [`Self::is_abort`]) yield the flags [`ApprovalDialer::respond`] takes — the
/// two can't drift.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApprovalChoice {
    /// Run this one call; ask again next time.
    Approve,
    /// Run it AND remember the approval for the caller's session ("don't ask
    /// again") — honored only for idempotent tools.
    ApproveForSession,
    /// Decline this call; the turn continues (a synthetic denial result).
    Deny,
    /// Decline this call AND stop the turn — the edge does not re-drive.
    Abort,
    /// Send the call back without approving or denying it (#67): the pause is
    /// recorded as deferred but stays open for a later decision.
    Defer,
}

impl ApprovalChoice {
    /// Whether the call was approved (approve or approve-for-session).
    #[must_use]
    pub const fn approved(self) -> bool {
        matches!(self, Self::Approve | Self::ApproveForSession)
    }

    /// Whether the approval is remembered for the session.
    #[must_use]
    pub const fn approved_for_session(self) -> bool {
        matches!(self, Self::ApproveForSession)
    }

    /// Whether the turn should stop (no re-drive).
    #[must_use]
    pub const fn is_abort(self) -> bool {
        matches!(self, Self::Abort)
    }

    /// Whether the call was deferred ("send back", #67) — neither approved nor
    /// denied; recorded but left pending.
    #[must_use]
    pub const fn is_defer(self) -> bool {
        matches!(self, Self::Defer)
    }
}

/// The one line of copy shown after a human decides a pending approval.
///
/// Shared by every edge so the identical state reads identically everywhere
/// (Slack replaces the approval card with this; Telegram edits the prompt
/// message to this). `label` is the tool's friendly display name, already
/// resolved by the caller (title, or a `polyc_proto::humanize_tool_name`
/// fallback).
#[must_use]
pub fn approval_decided_text(label: &str, choice: ApprovalChoice, decider: &str) -> String {
    match choice {
        ApprovalChoice::ApproveForSession => format!(
            "✅ Approved by {decider} — running \"{label}\"… (won't ask again this session)"
        ),
        ApprovalChoice::Approve => format!("✅ Approved by {decider} — running \"{label}\"…"),
        ApprovalChoice::Deny => format!("🚫 Denied by {decider} — \"{label}\" was not run."),
        ApprovalChoice::Abort => {
            format!("🛑 Aborted by {decider} — \"{label}\" was not run; the turn was stopped.")
        }
        ApprovalChoice::Defer => {
            format!("↩️ Sent back by {decider} — \"{label}\" is still waiting for a decision.")
        }
    }
}

/// The persisted outcome of an `ApprovalService.Respond` call.
///
/// On the THIN path the control plane is the signer: the caller submits an
/// UNSIGNED decision and the control plane appends a server-signed
/// `approval_response` event, returning the signature so the caller can show /
/// audit a verifiable outcome.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApprovalOutcome {
    /// `true` if a matching pending request existed and the response was
    /// persisted; `false` is the idempotent no-op (already answered / unknown).
    pub persisted: bool,
    /// Lowercase-hex ed25519 signature over the canonical response bytes.
    pub signature_hex: String,
    /// Lowercase-hex public key the signature verifies against.
    pub signed_by_hex: String,
}

/// One outstanding approval returned by [`ApprovalDialer::list_pending`].
///
/// Carries the fields an edge needs to re-render an Approve/Deny prompt for a
/// `request_id` it lost (e.g. the streamed `ApprovalPending` event never arrived).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingApproval {
    /// The id to answer via [`ApprovalDialer::respond`].
    pub request_id: String,
    /// The raw tool name the approval gates.
    pub tool_name: String,
    /// The tool-call arguments (JSON), for rendering the prompt.
    pub args_json: String,
    /// The override gate reason the durable `approval_request` recorded — empty
    /// for an ordinary gated call, non-empty for the lethal-trifecta /
    /// Rule-of-Two containment override. Lets a recovered card show the same
    /// explanation the live `ApprovalPending` event carried.
    pub reason: String,
}

/// Reusable handle for the control plane's `ApprovalService`.
///
/// The THIN human-in-the-loop path. Shares the `AgentService` endpoint — both
/// are served on one Connect port — so it is built from the same address.
#[derive(Clone)]
pub struct ApprovalDialer {
    client: Arc<ApprovalServiceClient<HttpClient>>,
}

impl ApprovalDialer {
    /// Build a dialer pointed at `addr` (expects `http://host:port`).
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI.
    pub fn new(addr: &str) -> Result<Self, DialError> {
        let uri = addr
            .parse::<http::Uri>()
            .map_err(|source| DialError::InvalidAddress {
                addr: addr.to_owned(),
                source,
            })?;
        let http = http_client_for(&uri)?;
        let client = ApprovalServiceClient::new(
            http,
            ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT),
        );
        Ok(Self {
            client: Arc::new(client),
        })
    }

    /// Submit a human's decision for a pending approval. The decision is
    /// UNSIGNED — the control plane signs and persists it (THIN path) and
    /// returns the signature. Idempotent: answering an already-decided or
    /// unknown `request_id` returns `persisted: false`.
    ///
    /// When `approved_for_session` is true (and `approved` is true), the
    /// control plane remembers the decision for the paused turn's caller and
    /// auto-approves that caller's later identical, idempotent tool calls for
    /// the rest of the session ("approve & don't ask again"). Ignored on a
    /// denial.
    ///
    /// When `abort` is true (only meaningful with `approved = false`), this is
    /// an abort: the call is declined AND the caller should stop the turn (not
    /// re-drive it). A plain denial (`approved = false`, `abort = false`)
    /// declines the call but lets the turn continue. See [`ApprovalChoice`],
    /// which maps a button decision to these flags.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn respond(
        &self,
        request_id: &str,
        choice: ApprovalChoice,
        reason: &str,
        conversation_id: &str,
        modified_args_json: &str,
        injected_context: &str,
    ) -> Result<ApprovalOutcome, DialError> {
        use polyc_proto::proto::polychrome::approval::v1::{
            Approve, Defer, Deny, approval_response_request::Decision,
        };
        // Map the choice into the structured `oneof` (#67). An approval carries
        // the approver's optional edit + injected context; a denial carries its
        // reason + abort hint; a defer ("send back") carries only its reason.
        let decision = if choice.is_defer() {
            Decision::Defer(Box::new(Defer {
                reason: reason.to_owned(),
                ..Default::default()
            }))
        } else if choice.approved() {
            Decision::Approve(Box::new(Approve {
                modified_args_json: modified_args_json.to_owned(),
                injected_context: injected_context.to_owned(),
                reason: reason.to_owned(),
                approved_for_session: choice.approved_for_session(),
                ..Default::default()
            }))
        } else {
            Decision::Deny(Box::new(Deny {
                reason: reason.to_owned(),
                abort: choice.is_abort(),
                ..Default::default()
            }))
        };
        let request = ApprovalResponseRequest {
            request_id: request_id.to_owned(),
            conversation_id: conversation_id.to_owned(),
            decision: Some(decision),
            ..Default::default()
        };
        let reply = self
            .client
            .respond_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(ApprovalOutcome {
            persisted: reply.persisted,
            signature_hex: reply.signature_hex,
            signed_by_hex: reply.signed_by_hex,
        })
    }

    /// List the conversation's outstanding approvals (a request with no later
    /// response). An edge calls this to recover `request_id`(s) it must prompt
    /// on after losing the streamed `ApprovalPending` event — turning a silent
    /// hang into a recoverable state. Read-only and idempotent.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn list_pending(
        &self,
        conversation_id: &str,
    ) -> Result<Vec<PendingApproval>, DialError> {
        let request = ListPendingRequest {
            conversation_id: conversation_id.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .list_pending_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(reply
            .pending
            .into_iter()
            .map(|p| PendingApproval {
                request_id: p.request_id,
                tool_name: p.tool_name,
                args_json: p.args_json,
                reason: p.reason,
            })
            .collect())
    }

    /// Remove outside content from a conversation (`#590`) so its web and
    /// outside access recover. Admin-gated server-side; the control plane
    /// verifies `actor` holds the admin role, appends the signed removal
    /// record, and reports which journal positions left the working context.
    ///
    /// `positions` names specific entries; `all_quarantined` removes every
    /// entry currently carrying outside content (`positions` is then
    /// ignored). `source_only` limits the removal to the named entries — the
    /// default also removes what the agent produced after them, which is the
    /// safer posture when the content may have steered it.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error, a
    /// permission refusal, or an invalid position.
    pub async fn excise_taint(
        &self,
        conversation_id: &str,
        positions: &[u64],
        all_quarantined: bool,
        source_only: bool,
        reason: &str,
        actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
    ) -> Result<ExcisionOutcome, DialError> {
        use polyc_proto::proto::polychrome::approval::v1::ExciseTaintRequest;
        let request = ExciseTaintRequest {
            conversation_id: conversation_id.to_owned(),
            positions: positions.to_vec(),
            all_quarantined,
            source_only,
            reason: reason.to_owned(),
            actor: buffa::MessageField::some(actor),
            ..Default::default()
        };
        let reply = self
            .client
            .excise_taint_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(ExcisionOutcome {
            persisted: reply.persisted,
            excised_positions: reply.excised_positions,
            signature_hex: reply.signature_hex,
            signed_by_hex: reply.signed_by_hex,
        })
    }
}

/// Result of an [`ApprovalDialer::excise_taint`] call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExcisionOutcome {
    /// Whether a removal record was durably appended (`false` = idempotent
    /// no-op: nothing was left to remove).
    pub persisted: bool,
    /// The journal positions that left the working context, after scope
    /// expansion.
    pub excised_positions: Vec<u64>,
    /// Hex signature over the removal record.
    pub signature_hex: String,
    /// Hex public key of the signer.
    pub signed_by_hex: String,
}

/// A freshly minted link-ceremony challenge: the code an edge must deliver
/// privately to the requesting user, plus its absolute expiry.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StartedLink {
    /// The single-use 6-digit code (a credential — never log or display it
    /// outside a private channel the user owns).
    pub code: String,
    /// Unix-ms after which the code is refused.
    pub expires_at_ms: u64,
    /// The persona the code is bound to.
    pub persona_id: String,
}

/// A freshly minted deep-link token: the opaque value an edge embeds in a
/// platform URL for a no-typing ceremony, plus its absolute expiry.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StartedDeepLink {
    /// The single-use, high-entropy token (a credential — only ever placed in
    /// a deep-link URL delivered to a private channel the user owns).
    pub token: String,
    /// Unix-ms after which the token is refused.
    pub expires_at_ms: u64,
    /// The persona the token is bound to.
    pub persona_id: String,
}

/// The outcome of completing a link ceremony, in edge-renderable terms.
///
/// `InvalidCode` and `Expired` are deliberately fused into one variant: the
/// proto contract is that edges present them identically so a guesser cannot
/// learn whether a code was ever valid. The distinction stays in the control
/// plane's logs.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LinkCeremony {
    /// The identity is now bound to the persona (directly or by merge).
    Linked {
        /// The surviving persona id.
        persona_id: String,
    },
    /// The identity already resolved to that persona; the code was consumed.
    AlreadyLinked {
        /// The persona id.
        persona_id: String,
    },
    /// No usable code (never minted, already consumed, or expired). Rendered
    /// identically to the user.
    InvalidOrExpired,
    /// The completing identity is locked out after repeated failures (or the
    /// deployment-wide backstop tripped). Rendered distinctly: "wait".
    Throttled,
    /// An unexpected/unspecified outcome — the edge should surface a generic
    /// failure rather than claim success.
    Failed,
}

impl LinkCeremony {
    /// The user-facing message for this outcome, shared by every edge.
    ///
    /// `InvalidCode` and `Expired` are fused into one variant upstream (so no
    /// edge can distinguish them — no oracle for a guesser); centralizing the
    /// rendering here keeps that security-relevant wording identical across
    /// surfaces. Channel-neutral phrasing; an edge that needs surface-specific
    /// text can still match the variant itself.
    #[must_use]
    pub const fn user_message(&self) -> &'static str {
        match self {
            Self::Linked { .. } => {
                "✅ Linked — this account now shares one Polychrome persona with your other channels."
            }
            Self::AlreadyLinked { .. } => {
                "✅ Already linked — this account was already on that persona."
            }
            Self::InvalidOrExpired => {
                "That link is invalid or expired. Start a fresh one from your other channel and try again."
            }
            Self::Throttled => "Too many attempts — wait a few minutes, then try again.",
            Self::Failed => "That link failed to complete. Start a fresh one and try again.",
        }
    }
}

/// The user-facing message for an admin-invite dial failure, shared by every
/// edge.
///
/// Centralizes the wording so the Slack and Telegram invite handlers can't
/// drift on copy for the identical state (the CLAUDE.md "route through a
/// shared helper" rule). Permission-denied maps to [`ADMIN_ONLY_INVITE`]; a
/// surface where echoing a refusal is unsafe (a group chat, where it would
/// let a non-admin make the bot speak / probe who is an admin) can suppress
/// the message itself rather than call this.
#[must_use]
pub const fn invite_error_message(err: &DialError) -> &'static str {
    match err.code() {
        Some(ErrorCode::PermissionDenied) => ADMIN_ONLY_INVITE,
        Some(ErrorCode::ResourceExhausted) => {
            "That's a lot of invites in a row — wait a couple of minutes, then try again."
        }
        _ => "I couldn't create the invite right now — try again in a moment.",
    }
}

/// The "only admins can invite" refusal copy, shared across edges.
pub const ADMIN_ONLY_INVITE: &str = "Only admins can send invites.";

/// Reusable handle for the control plane's `PersonaService`.
///
/// The verified identity-linking ceremonies. Shares the `AgentService`
/// endpoint — all three services are served on one internal Connect port —
/// so it is built from the same address.
#[derive(Clone)]
pub struct PersonaDialer {
    client: Arc<PersonaServiceClient<HttpClient>>,
}

impl PersonaDialer {
    /// Build a dialer pointed at `addr` (expects `http://host:port`).
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI.
    pub fn new(addr: &str) -> Result<Self, DialError> {
        let uri = addr
            .parse::<http::Uri>()
            .map_err(|source| DialError::InvalidAddress {
                addr: addr.to_owned(),
                source,
            })?;
        let http = http_client_for(&uri)?;
        let client = PersonaServiceClient::new(
            http,
            ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT),
        );
        Ok(Self {
            client: Arc::new(client),
        })
    }

    /// Mint a single-use link-ceremony code bound to `identity`'s persona
    /// (provisioning one on first contact). The returned [`StartedLink::code`]
    /// is a credential the caller must deliver only to a private channel the
    /// user owns.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn start_link(&self, identity: ExternalIdentity) -> Result<StartedLink, DialError> {
        let request = StartLinkRequest {
            identity: buffa::MessageField::some(identity),
            ..Default::default()
        };
        let reply = self
            .client
            .start_link_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(StartedLink {
            code: reply.code,
            expires_at_ms: reply.expires_at_ms,
            persona_id: reply.persona_id,
        })
    }

    /// Mint a high-entropy deep-link token bound to `identity`'s persona, for
    /// a no-typing ceremony (the caller embeds [`StartedDeepLink::token`] in a
    /// platform URL like `https://t.me/<bot>?start=<token>`). The token is a
    /// credential — deliver the URL only to a private channel the user owns.
    /// Completed via [`Self::complete_link`] from the target platform.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn start_deeplink(
        &self,
        identity: ExternalIdentity,
    ) -> Result<StartedDeepLink, DialError> {
        let request = StartDeepLinkRequest {
            identity: buffa::MessageField::some(identity),
            ..Default::default()
        };
        let reply = self
            .client
            .start_deep_link_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(StartedDeepLink {
            token: reply.token,
            expires_at_ms: reply.expires_at_ms,
            persona_id: reply.persona_id,
        })
    }

    /// Admin-gated: mint an invite code for a TARGET identity distinct from
    /// the acting admin. The code binds to the target — only the target can
    /// redeem it — and is a credential: deliver it only to a private channel
    /// the TARGET owns, never back through the admin's shared surfaces.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error; a
    /// non-admin actor surfaces as `permission_denied`.
    pub async fn admin_invite(
        &self,
        actor: ExternalIdentity,
        target: ExternalIdentity,
    ) -> Result<StartedLink, DialError> {
        let request = AdminInviteRequest {
            actor: buffa::MessageField::some(actor),
            target: buffa::MessageField::some(target),
            ..Default::default()
        };
        let reply = self
            .client
            .admin_invite_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(StartedLink {
            code: reply.code,
            expires_at_ms: reply.expires_at_ms,
            persona_id: reply.persona_id,
        })
    }

    /// Deterministic workspace-email auto-link: bind `identity` to the
    /// persona already holding a ceremony-verified email matching
    /// `asserted_email`. The email must come from the PLATFORM API (e.g.
    /// Slack `users.info`) — never from user-typed text. Returns whether a
    /// link happened so the edge can notify the user ("if this wasn't
    /// you…"); halting outcomes come back as `Ok(None)`.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn auto_link(
        &self,
        identity: ExternalIdentity,
        asserted_email: &str,
        basis: &str,
    ) -> Result<Option<String>, DialError> {
        let request = AutoLinkRequest {
            identity: buffa::MessageField::some(identity),
            asserted_email: asserted_email.to_owned(),
            basis: basis.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .auto_link_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(match reply.outcome.as_known() {
            Some(AutoLinkOutcome::Linked) => Some(reply.persona_id),
            _ => None,
        })
    }

    /// Consume `code` from the channel being claimed, binding `identity` to
    /// the minting persona (merging `identity`'s prior persona if it had one).
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn complete_link(
        &self,
        code: &str,
        identity: ExternalIdentity,
    ) -> Result<LinkCeremony, DialError> {
        let request = CompleteLinkRequest {
            code: code.to_owned(),
            identity: buffa::MessageField::some(identity),
            ..Default::default()
        };
        let reply = self
            .client
            .complete_link_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(match reply.outcome.as_known() {
            Some(LinkOutcome::Linked) => LinkCeremony::Linked {
                persona_id: reply.persona_id,
            },
            Some(LinkOutcome::AlreadyLinked) => LinkCeremony::AlreadyLinked {
                persona_id: reply.persona_id,
            },
            // Fused on purpose — see `LinkCeremony::InvalidOrExpired`.
            Some(LinkOutcome::InvalidCode | LinkOutcome::Expired) => LinkCeremony::InvalidOrExpired,
            Some(LinkOutcome::Throttled) => LinkCeremony::Throttled,
            Some(LinkOutcome::Unspecified) | None => LinkCeremony::Failed,
        })
    }

    /// Read-only: the profile the `identity` resolves to, as an edge-friendly
    /// [`PersonaView`]. Returns an all-empty view for an identity not yet in
    /// the directory; **never provisions** a persona.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn describe(&self, identity: ExternalIdentity) -> Result<PersonaView, DialError> {
        let request = DescribeRequest {
            identity: buffa::MessageField::some(identity),
            ..Default::default()
        };
        let reply = self
            .client
            .describe_with_options(request, traced_options())
            .await?
            .into_owned();
        // Unknown identity → empty profile → an all-default view (the
        // not-yet-claimed dashboard state).
        let Some(profile) = reply.profile.into_option() else {
            return Ok(PersonaView {
                persona_id: reply.persona_id,
                ..Default::default()
            });
        };
        Ok(PersonaView {
            persona_id: reply.persona_id,
            status: profile.status,
            identities: profile
                .identities
                .into_iter()
                .map(|id| LinkedIdentity {
                    provider: id.provider,
                    external_id: id.external_id,
                    display_name: id.display_name,
                })
                .collect(),
        })
    }
}

/// One external identity linked to a persona, as the dashboard renders it
/// (provider · display name · id).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LinkedIdentity {
    /// Edge namespace ("slack", "telegram", …).
    pub provider: String,
    /// The provider-native stable id.
    pub external_id: String,
    /// Display name as last observed (presentation only).
    pub display_name: String,
}

/// An edge-friendly read of a persona, for surfaces like the App Home
/// dashboard. Every field is empty when the queried identity is not in the
/// directory (a not-yet-claimed user).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PersonaView {
    /// The persona id; empty when the identity is unknown.
    pub persona_id: String,
    /// "provisional" | "linked" | "merged"; empty when unknown.
    pub status: String,
    /// Every identity currently linked to the persona.
    pub identities: Vec<LinkedIdentity>,
}

/// A turn-input message, re-exported so callers can build attributed
/// multi-party input for [`AgentDialer::run_turn_streaming_messages`] without
/// depending on `polyc-proto`.
pub use polyc_proto::proto::polychrome::agent::v1::Message as TurnMessage;
/// The attributed transcript line type the participation gate consumes.
/// Re-exported so callers build requests without importing `polyc-proto`.
pub use polyc_proto::proto::polychrome::agent::v1::ParticipantMessage as GateMessage;
/// A settled inbound payment receipt, re-exported so the public-edge layer can
/// thread it into [`AgentDialer::run_turn_streaming_messages_with`]
/// without depending on `polyc-proto` directly.
pub use polyc_proto::proto::polychrome::agent::v1::PaymentReceipt;

/// The external identity of a human observed at an edge (re-exported wire
/// type, shared with the persona store records): the per-edge `EdgeAdapter`
/// mapping produces these and the control plane resolves them to durable
/// personas.
pub use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;

/// The Connect error-code enum, re-exported so edges can branch on
/// [`DialError::code`] (e.g. render `permission_denied` differently from a
/// transient failure) without depending on `connectrpc` directly.
pub use connectrpc::ErrorCode;

/// Caller attribution for one turn.
///
/// Carries the identity whose message triggered the turn (`caller`) and any
/// other humans whose messages entered the turn's input (`participants`).
/// Edges without caller identity pass the default.
#[derive(Debug, Clone, Default)]
pub struct Attribution {
    /// The identity whose message triggered the turn.
    pub caller: Option<ExternalIdentity>,
    /// Other humans whose messages entered this turn's input.
    pub participants: Vec<ExternalIdentity>,
}

/// Build an attributed user-role turn message rendered as `"<speaker>: text"`,
/// so the model sees who said what in a multi-party thread.
#[must_use]
pub fn attributed_message(speaker: &str, text: &str) -> TurnMessage {
    text_message("user", &format!("{speaker}: {text}"))
}

/// Build a plain user-role turn message (no speaker attribution) — used for
/// 1:1 DMs where there is only one human in the conversation.
#[must_use]
pub fn user_message(text: &str) -> TurnMessage {
    text_message("user", text)
}

/// Build the documented **readable** namespaced conversation id,
/// `"{namespace}:{native_id}"`.
///
/// Every edge derives [`AgentRequest::conversation_id`] from its native unit
/// (a thread, a ticket, an issue, a session). The convention is a namespace
/// prefix that names the edge family, followed by the edge's own stable handle
/// for the conversation — e.g. `github:owner/repo#42`, `web:{session-uuid}`,
/// `mcp:{caller-id}`. The id is opaque to the orchestration core (it is the
/// event-log partition key, `conv-{id}`); only the *format* is a shared
/// convention so edges stay consistent and ids are greppable.
///
/// Edges whose native coordinate is unwieldy or sensitive — many chat
/// platforms — should instead use [`hashed_conversation_id`], which collapses
/// the coordinate into a fixed-length opaque `UUIDv5` under a pinned namespace.
#[must_use]
pub fn namespaced_id(namespace: &str, native_id: &str) -> String {
    format!("{namespace}:{native_id}")
}

/// Derive a stable, fixed-length conversation id by hashing `parts` into a
/// `UUIDv5` under a pinned `namespace`.
///
/// This is the "opaque hash" namespacing policy: a deterministic,
/// stateless-across-restarts id (no `native → id` table to keep) that is
/// globally unique enough to key the event-log partition. `parts` are joined
/// with `':'` before hashing, so a caller passing `["T1", "C1", "169…"]`
/// hashes exactly `"T1:C1:169…"`.
///
/// The pinned `namespace` UUID **must never change** for a given edge — every
/// existing conversation id for that edge depends on it. `polychrome-slack`
/// uses this policy (its `UUIDv5` over `(team, channel, thread_ts)`); new edges
/// pick their own frozen namespace.
#[must_use]
pub fn hashed_conversation_id(namespace: uuid::Uuid, parts: &[&str]) -> String {
    let joined = parts.join(":");
    uuid::Uuid::new_v5(&namespace, joined.as_bytes())
        .hyphenated()
        .to_string()
}

/// Collision-free variant of [`hashed_conversation_id`] for edges whose native
/// parts can themselves contain `':'` (e.g. an email `Message-ID`, a URL, a
/// path).
///
/// [`hashed_conversation_id`] joins parts with a bare `':'`, so `["a:b", "c"]`
/// and `["a", "b:c"]` both hash `"a:b:c"` and collide. This helper instead
/// **length-prefixes** each part (`"{len}:{part}"`), which is unambiguous
/// regardless of any `':'` inside a part — the length says exactly how many
/// bytes the part occupies, so distinct part boundaries always produce distinct
/// hash inputs. Use it for any edge whose coordinate fields are not guaranteed
/// `':'`-free.
///
/// Distinct from [`hashed_conversation_id`] precisely because the framing
/// differs; the two do not agree for the same `parts`.
#[must_use]
pub fn framed_conversation_id(namespace: uuid::Uuid, parts: &[&str]) -> String {
    let mut framed = String::new();
    for p in parts {
        framed.push_str(&p.len().to_string());
        framed.push(':');
        framed.push_str(p);
    }
    uuid::Uuid::new_v5(&namespace, framed.as_bytes())
        .hyphenated()
        .to_string()
}

/// Build the single [`AgentRequest`] that opens an `AgentService.connect`
/// stream. Shared by [`AgentDialer::run_turn`] and
/// [`AgentDialer::run_turn_streaming`] so the two paths can't drift.
fn build_request(
    conversation_id: &str,
    exec_id: &str,
    messages: Vec<Message>,
    payment_receipt: Option<PaymentReceipt>,
    attribution: Attribution,
) -> AgentRequest {
    AgentRequest {
        conversation_id: conversation_id.to_owned(),
        exec_id: exec_id.to_owned(),
        start: buffa::MessageField::some(AgentStart {
            agent_id: String::new(),
            agent_config: Vec::new(),
            messages,
            payment_receipt: payment_receipt
                .map_or_else(buffa::MessageField::none, buffa::MessageField::some),
            caller: attribution
                .caller
                .map_or_else(buffa::MessageField::none, buffa::MessageField::some),
            participants: attribution.participants,
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// Project a terminal [`AgentEnd`] into the [`TurnEvent`]s a live surface sees
/// at end-of-turn: one [`TurnEvent::ApprovalPending`] per paused tool call,
/// then a [`TurnEvent::HandoffStarted`] if the turn delegated to a sub-agent,
/// then the terminal [`TurnEvent::Done`].
///
/// Pure (no I/O) so the End-arm mapping can be unit-tested without a live
/// stream, mirroring [`message_to_event`] for the Outputs arm.
fn events_from_end(end: AgentEnd) -> Vec<TurnEvent> {
    let mut events: Vec<TurnEvent> = end
        .pending_approvals
        .into_iter()
        .map(|pa| TurnEvent::ApprovalPending {
            request_id: pa.request_id,
            tool_name: pa.tool_name,
            title: pa.title,
            args_json: pa.args_json,
            reason: pa.reason,
        })
        .collect();
    if let Some(h) = end.handoff.into_option() {
        events.push(TurnEvent::HandoffStarted {
            child_agent_id: h.child_agent_id,
            reason: h.reason,
        });
    }
    events.push(TurnEvent::Done);
    events
}

/// Project a wire [`ContextCompacted`] into the corresponding
/// [`TurnEvent::ContextCompacted`]. Pure (no I/O); an unknown/unspecified wire
/// reason maps to [`CompactionReason::Truncated`] (the quieter, no-preview
/// rendering) so a future wire reason never panics a live surface.
fn event_from_compacted(c: ContextCompacted) -> TurnEvent {
    let reason = if c.reason.to_i32() == WireCompactionReason::COMPACTION_REASON_SUMMARIZED as i32 {
        CompactionReason::Summarized
    } else {
        CompactionReason::Truncated
    };
    TurnEvent::ContextCompacted {
        reason,
        summarized_messages: c.summarized_messages,
        summary_preview: c.summary_preview,
    }
}

/// Map one output [`Message`] to an optional [`TurnEvent`] for the streaming
/// path.
///
/// Pure (no I/O) so the mapping can be unit-tested without a live stream:
///
/// - A tool-call content block, or any tool-role message, becomes a
///   [`TurnEvent::ToolStarted`] named for the called function (falling back to
///   the call id when the function name is absent).
/// - A non-empty text block on a model/assistant-role message becomes a
///   [`TurnEvent::TextDelta`].
/// - Tool-role text (a tool-result echo) and empty/non-textual blocks produce
///   no event.
fn message_to_event(msg: Message) -> Option<TurnEvent> {
    let content_block = msg.content.into_option()?;
    match content_block.r#type? {
        content::Type::ToolCall(tc) => {
            let name = match tc.r#type.as_ref() {
                Some(tool_call_content::Type::FunctionCall(fc)) if !fc.name.is_empty() => {
                    fc.name.clone()
                }
                _ => tc.id.clone(),
            };
            Some(TurnEvent::ToolStarted { name })
        }
        content::Type::Text(t) => {
            if is_assistant_role(&msg.role) && !t.text.is_empty() {
                Some(TurnEvent::TextDelta(t.text))
            } else {
                // Tool-role text is an intermediate result echo; empty text and
                // non-assistant roles carry nothing the user should see.
                None
            }
        }
        // Thoughts, tool results, and media variants have no streaming event in
        // this version.
        _ => None,
    }
}

/// Whether a message role denotes the assistant's own answer text.
///
/// The control plane uses `model` (provider-native) and `assistant`
/// interchangeably for the agent's replies; both count.
fn is_assistant_role(role: &str) -> bool {
    role == "model" || role == "assistant"
}

/// Render one [`content::Type`] variant as user-visible text.
///
/// Returns `None` for variants that have no useful textual representation
/// (image/audio/document/video/confirmation) so the `join("\n")` in
/// [`AgentDialer::run_turn`] doesn't emit stray blank lines.
///
/// Tool calls are rendered (rather than skipped) so a turn that ends on a
/// `tool_call` with no follow-up text — e.g. a HITL pause or approval-only
/// confirmation flow — gives the user something to see instead of an empty
/// body.
fn render_content(ty: Option<content::Type>) -> Option<String> {
    match ty? {
        content::Type::Text(t) => {
            if t.text.is_empty() {
                None
            } else {
                Some(t.text)
            }
        }
        content::Type::ToolCall(tc) => {
            let name = match tc.r#type.as_ref() {
                Some(tool_call_content::Type::FunctionCall(fc)) => fc.name.as_str(),
                None => "",
            };
            if name.is_empty() {
                Some(format!("[tool_call:{}]", tc.id))
            } else {
                Some(format!("[tool_call:{name} {}]", tc.id))
            }
        }
        content::Type::ToolResult(tr) => Some(format!("[tool_result:{}]", tr.call_id)),
        // Everything else renders to no reply text. Notably reasoning
        // (`Thought`): this helper feeds the buffered reply (and its no-text
        // scaffolding fallback), and surfacing reasoning would return raw
        // chain-of-thought as the answer on a tool-only / approval-pause turn.
        // Reasoning reaches the user via a separate path (the TUI builds a
        // collapsed thought line from the proto transcript), never here. Image /
        // audio / document / video / confirmation likewise have no v1 rendering.
        _ => None,
    }
}

// ---------------------------------------------------------------------------
// Operator mailbox — the system-initiated, operator-authorized approval path.
//
// The control plane holds no chat tokens; the edges do (docs/design/
// operator-mailbox.md). Delivery therefore splits: the control plane
// ORIGINATES + PERSISTS the ask and exposes a pending-notification projection
// (`NotificationService`); each edge DRAINS the notifications for ITS provider,
// DMs the operator with Approve/Deny controls, and `Ack`s. The decision comes
// home through `OperatorMailboxService.Decide`, where authorization lives
// (server-side, off the edge's already-verified inbound identity).
//
// Both services share the same internal Connect port as the other control-plane
// services, so they are dialed from the same `agent_addr` an edge already holds.
// ---------------------------------------------------------------------------

/// A control-plane action an operator is asked to authorize.
///
/// Lifted to an edge-renderable Rust enum so surfaces match on it without
/// touching the wire crate. Action-agnostic; the upgrade case is the first
/// (and only) variant.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OpsAction {
    /// Roll the cluster to a specific release.
    UpgradeTo {
        /// The target release version string, as rendered to the operator.
        version: String,
    },
    /// An action whose variant this client does not recognise (a newer wire
    /// shape). Rendered generically so an edge never silently drops a real ask.
    Unknown,
}

/// One undelivered `(item, target)` pair an edge must DM, in edge-friendly
/// terms.
///
/// Mirrors the wire [`PollPendingReply`](polyc_proto::proto::polychrome::ops::v1::PollPendingReply)
/// rows so an edge builds its prompt without importing `polyc-proto`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingNotice {
    /// Opaque idempotency id of the mailbox item — the only token an edge ever
    /// places in unsigned `callback_data`; state is looked up server-side.
    pub action_id: String,
    /// The DM coordinate for THIS provider: a Slack `U…` user id, or a Telegram
    /// `chat_id` as a decimal string.
    pub target: String,
    /// The action to render in the Approve/Deny prompt.
    pub action: OpsAction,
    /// Unix seconds at which the item expires (`0` when unset). An edge may
    /// surface it; a late decision is refused server-side regardless.
    pub expires_unix: u64,
    /// Lowercase-hex BLAKE3 of the canonical payload the operator is approving
    /// (WYSIWYS). The edge echoes this back in [`OperatorMailboxDialer::decide`].
    pub payload_hash: String,
    /// Whether this pair has already been delivered. A delivered pair is still
    /// surfaced so an edge can rehydrate its `action_id -> payload_hash` cache
    /// after a restart; the edge DMs only `delivered == false` pairs.
    pub delivered: bool,
}

/// Reusable handle for the control plane's `NotificationService` — the
/// pending-notification projection an edge drains to deliver operator DMs.
///
/// Shares the `AgentService` endpoint (one internal Connect port), so it is
/// built from the same address an edge already holds.
#[derive(Clone)]
pub struct NotificationDialer {
    client: Arc<NotificationServiceClient<HttpClient>>,
}

impl NotificationDialer {
    /// Build a dialer pointed at `addr` (expects `http://host:port`).
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::Tls`] if an `https` endpoint's TLS setup fails.
    pub fn new(addr: &str) -> Result<Self, DialError> {
        let uri = addr
            .parse::<http::Uri>()
            .map_err(|source| DialError::InvalidAddress {
                addr: addr.to_owned(),
                source,
            })?;
        let http = http_client_for(&uri)?;
        let client = NotificationServiceClient::new(
            http,
            ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT),
        );
        Ok(Self {
            client: Arc::new(client),
        })
    }

    /// Drain undelivered operator notifications for `provider` (`"slack"` |
    /// `"telegram"`). Read-only and idempotent — the durable mailbox is
    /// unchanged until [`Self::ack`].
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn poll_pending(&self, provider: &str) -> Result<Vec<PendingNotice>, DialError> {
        let request = PollPendingRequest {
            provider: provider.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .poll_pending_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(reply.pending.into_iter().map(pending_notice).collect())
    }

    /// Mark an `(action_id, target)` delivered after the DM is sent. Persists
    /// an `ops_action_delivered` marker server-side so the projection survives
    /// restart and never double-delivers. Idempotent: a redundant ack is
    /// harmless.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn ack(
        &self,
        provider: &str,
        action_id: &str,
        target: &str,
    ) -> Result<bool, DialError> {
        let request = AckRequest {
            provider: provider.to_owned(),
            action_id: action_id.to_owned(),
            target: target.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .ack_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(reply.persisted)
    }
}

/// Map a wire [`PendingNotification`](polyc_proto::proto::polychrome::ops::v1::PendingNotification)
/// to the edge-friendly [`PendingNotice`], lifting the action oneof to
/// [`OpsAction`].
fn pending_notice(
    p: polyc_proto::proto::polychrome::ops::v1::PendingNotification,
) -> PendingNotice {
    let action = p.action.into_option().and_then(|view| view.action).map_or(
        OpsAction::Unknown,
        |a| match a {
            ops_action_view::Action::UpgradeTo(u) => OpsAction::UpgradeTo { version: u.version },
        },
    );
    PendingNotice {
        action_id: p.action_id,
        target: p.target,
        action,
        expires_unix: p.expires_unix,
        payload_hash: p.payload_hash,
        delivered: p.delivered,
    }
}

/// The control plane's verdict on an operator decision, in edge-renderable
/// terms — the buffered analog of the wire `DecideReply.Outcome`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DecideOutcome {
    /// Approved, authorized, signed + recorded, handed to the executor.
    Applied,
    /// The operator denied it; the denial was recorded.
    Denied,
    /// Refused — not an operator / unknown action / already decided / expired /
    /// payload drift. `detail` carries the human-readable reason.
    Rejected,
    /// An unspecified/unknown outcome — the edge surfaces a generic failure
    /// rather than claiming success.
    Unknown,
}

/// An operator decision's result: the [`DecideOutcome`] plus the control
/// plane's human-readable `detail` (the reject reason when rejected).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DecideResult {
    /// The verdict.
    pub outcome: DecideOutcome,
    /// Human-readable detail to surface to the operator.
    pub detail: String,
}

/// Heading shown above an approval-mailbox prompt.
///
/// Shared across edges so the two surfaces never word the same ask differently,
/// and to keep internal jargon out of user-facing copy — an edge only wraps this
/// in its own markup.
pub const OPS_PROMPT_HEADING: &str = "Approval needed";

impl OpsAction {
    /// Plain description of the action for an approval prompt — the single
    /// wording both edges render, so an added action variant is described once
    /// and never diverges between surfaces.
    ///
    /// An `UpgradeTo` ask is worded through the shared
    /// [`update_copy`](polyc_runtime::update_copy::update_copy) helper — the same
    /// one the CLI `status` line and the dashboard banner use — so a person reads
    /// the identical "update ready" copy wherever it surfaces. Rolling the cluster
    /// to a new release replaces the running image, so the change is
    /// [`Compatibility::Warm`](polyc_runtime::compat::Compatibility::Warm): it
    /// restarts the service, and conversations already underway finish first.
    #[must_use]
    pub fn summary(&self) -> String {
        match self {
            Self::UpgradeTo { version } => {
                let copy = polyc_runtime::update_copy::update_copy(
                    &polyc_runtime::compat::Compatibility::Warm,
                    version,
                );
                format!("{} — {}", copy.headline, copy.detail)
            }
            // Jargon-free fallback for an action this client doesn't recognize.
            Self::Unknown => "Approve a pending action".to_owned(),
        }
    }

    /// The exact verb an edge puts on the approve affordance for this action, or
    /// `None` to fall back to a generic "Approve".
    ///
    /// An `UpgradeTo` ask that can be applied in place carries the shared
    /// [`APPLY_NOW`](polyc_runtime::update_copy::APPLY_NOW) verb, routed through
    /// the same helper as [`Self::summary`] so the button never words the update
    /// differently from its detail. A cold or incompatible change would carry no
    /// verb here; a cluster roll is always warm, so the upgrade ask carries the
    /// apply verb.
    #[must_use]
    pub fn approve_verb(&self) -> Option<&'static str> {
        match self {
            Self::UpgradeTo { version } => {
                polyc_runtime::update_copy::update_copy(
                    &polyc_runtime::compat::Compatibility::Warm,
                    version,
                )
                .action
            }
            Self::Unknown => None,
        }
    }
}

impl DecideOutcome {
    /// Metric label for this outcome (`applied` | `denied` | `rejected` |
    /// `unknown`) — shared so the two edges never emit divergent label sets for
    /// the same control-plane verdict.
    #[must_use]
    pub const fn metric_label(&self) -> &'static str {
        match self {
            Self::Applied => "applied",
            Self::Denied => "denied",
            Self::Rejected => "rejected",
            Self::Unknown => "unknown",
        }
    }
}

impl DecideResult {
    /// The decided-state line that REPLACES an approval prompt after a decision
    /// (the double-click guard drops the buttons). Surfaces the control plane's
    /// verdict and, on a rejection, its `detail`. Shared so both edges show the
    /// same wording for the same verdict.
    #[must_use]
    pub fn decided_line(&self, decider: &str) -> String {
        match self.outcome {
            DecideOutcome::Applied => format!("✅ Approved by {decider} — applying."),
            DecideOutcome::Denied => format!("🚫 Denied by {decider}."),
            DecideOutcome::Rejected => {
                let why = if self.detail.is_empty() {
                    "not authorized or no longer valid"
                } else {
                    self.detail.as_str()
                };
                format!("⛔ Rejected — {why}.")
            }
            DecideOutcome::Unknown => {
                "⚠️ Something went wrong recording that decision — try again.".to_owned()
            }
        }
    }
}

/// Reusable handle for the control plane's `OperatorMailboxService` — the
/// operator decision endpoint (authorize → evaluate → sign → record →
/// executor, all server-side).
///
/// Shares the `AgentService` endpoint (one internal Connect port), so it is
/// built from the same address an edge already holds.
#[derive(Clone)]
pub struct OperatorMailboxDialer {
    client: Arc<OperatorMailboxServiceClient<HttpClient>>,
}

impl OperatorMailboxDialer {
    /// Build a dialer pointed at `addr` (expects `http://host:port`).
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::Tls`] if an `https` endpoint's TLS setup fails.
    pub fn new(addr: &str) -> Result<Self, DialError> {
        let uri = addr
            .parse::<http::Uri>()
            .map_err(|source| DialError::InvalidAddress {
                addr: addr.to_owned(),
                source,
            })?;
        let http = http_client_for(&uri)?;
        let client = OperatorMailboxServiceClient::new(
            http,
            ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT),
        );
        Ok(Self {
            client: Arc::new(client),
        })
    }

    /// Submit an operator decision for `action_id`. Authorization is
    /// SERVER-SIDE: the edge forwards the `(provider, external_user_id)` it
    /// already authenticated (Slack HMAC / Telegram `secret_token`) and the
    /// control plane resolves it to an operator persona, refusing non-operators.
    /// `payload_hash` is the WYSIWYS hash the edge displayed; the control plane
    /// refuses an approval whose hash drifted from the live item.
    ///
    /// Idempotent: a duplicate Approve click on an already-decided item is
    /// [`DecideOutcome::Rejected`], never double-applied.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn decide(
        &self,
        action_id: &str,
        approved: bool,
        provider: &str,
        external_user_id: &str,
        reason: &str,
        payload_hash: &str,
    ) -> Result<DecideResult, DialError> {
        let request = DecideRequest {
            action_id: action_id.to_owned(),
            approved,
            provider: provider.to_owned(),
            external_user_id: external_user_id.to_owned(),
            reason: reason.to_owned(),
            payload_hash: payload_hash.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .decide_with_options(request, traced_options())
            .await?
            .into_owned();
        let outcome = match reply.outcome.as_known() {
            Some(decide_reply::Outcome::APPLIED) => DecideOutcome::Applied,
            Some(decide_reply::Outcome::DENIED) => DecideOutcome::Denied,
            Some(decide_reply::Outcome::REJECTED) => DecideOutcome::Rejected,
            Some(decide_reply::Outcome::OUTCOME_UNSPECIFIED) | None => DecideOutcome::Unknown,
        };
        Ok(DecideResult {
            outcome,
            detail: reply.detail,
        })
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]
    use super::*;
    use polyc_proto::proto::polychrome::agent::v1::{
        Content, FunctionCallContent, TextContent, ThoughtContent, ThoughtSummaryContent,
        ToolCallContent, ToolResultContent, thought_summary_content,
    };

    #[test]
    fn ops_copy_is_shared_and_jargon_free() {
        // The upgrade ask is worded through the shared update-copy helper: it
        // names the version, carries the exact "Apply now" verb (a cluster roll
        // is a warm, in-place restart), and reads identically to the CLI and
        // dashboard surfaces.
        let up = OpsAction::UpgradeTo {
            version: "1.2.3".to_owned(),
        };
        let summary = up.summary();
        assert!(
            summary.contains("1.2.3"),
            "summary names the version: {summary}"
        );
        assert!(
            summary.contains("restarts the service"),
            "warm upgrade summary is honest about the restart: {summary}"
        );
        assert_eq!(
            up.approve_verb(),
            Some(polyc_runtime::update_copy::APPLY_NOW),
            "an in-place upgrade carries the shared Apply now verb"
        );
        assert!(
            !summary.to_lowercase().contains("operator"),
            "no banned jargon in the upgrade ask: {summary}"
        );
        assert_eq!(OpsAction::Unknown.approve_verb(), None);
        let unknown = OpsAction::Unknown.summary();
        assert!(!unknown.to_lowercase().contains("control-plane"));
        assert!(!unknown.is_empty());
        // The shared heading carries no banned jargon.
        assert!(!OPS_PROMPT_HEADING.to_lowercase().contains("operator"));

        // Metric labels are stable and distinct.
        assert_eq!(DecideOutcome::Applied.metric_label(), "applied");
        assert_eq!(DecideOutcome::Denied.metric_label(), "denied");
        assert_eq!(DecideOutcome::Rejected.metric_label(), "rejected");
        assert_eq!(DecideOutcome::Unknown.metric_label(), "unknown");

        // The decided line surfaces each verdict, the decider, and reject detail,
        // and never says "please".
        let applied = DecideResult {
            outcome: DecideOutcome::Applied,
            detail: String::new(),
        }
        .decided_line("Chris");
        assert!(applied.contains("Approved") && applied.contains("Chris"));
        let rejected = DecideResult {
            outcome: DecideOutcome::Rejected,
            detail: "not an operator".to_owned(),
        }
        .decided_line("Chris");
        assert!(rejected.contains("Rejected") && rejected.contains("not an operator"));
        let unknown = DecideResult {
            outcome: DecideOutcome::Unknown,
            detail: String::new(),
        }
        .decided_line("Chris");
        assert!(!unknown.to_lowercase().contains("please"));
    }

    #[test]
    fn approval_choice_flags() {
        assert!(ApprovalChoice::Approve.approved());
        assert!(!ApprovalChoice::Approve.approved_for_session());
        assert!(!ApprovalChoice::Approve.is_abort());
        assert!(ApprovalChoice::ApproveForSession.approved());
        assert!(ApprovalChoice::ApproveForSession.approved_for_session());
        // Deny and Abort both decline; only Abort stops the turn.
        assert!(!ApprovalChoice::Deny.approved());
        assert!(!ApprovalChoice::Deny.is_abort());
        assert!(!ApprovalChoice::Abort.approved());
        assert!(ApprovalChoice::Abort.is_abort());
    }

    #[test]
    fn compacted_summarized_maps_with_preview() {
        let c = ContextCompacted {
            reason: buffa::EnumValue::from(
                WireCompactionReason::COMPACTION_REASON_SUMMARIZED as i32,
            ),
            summarized_messages: 12,
            summary_preview: "earlier work: fixed bug X".to_owned(),
            ..Default::default()
        };
        assert_eq!(
            event_from_compacted(c),
            TurnEvent::ContextCompacted {
                reason: CompactionReason::Summarized,
                summarized_messages: 12,
                summary_preview: "earlier work: fixed bug X".to_owned(),
            }
        );
    }

    #[test]
    fn compacted_truncated_maps_without_preview() {
        let c = ContextCompacted {
            reason: buffa::EnumValue::from(
                WireCompactionReason::COMPACTION_REASON_TRUNCATED as i32,
            ),
            ..Default::default()
        };
        assert_eq!(
            event_from_compacted(c),
            TurnEvent::ContextCompacted {
                reason: CompactionReason::Truncated,
                summarized_messages: 0,
                summary_preview: String::new(),
            }
        );
    }

    #[test]
    fn compacted_unknown_reason_falls_back_to_truncated() {
        // An unspecified/future wire reason must not panic a live surface; it
        // maps to the quieter, no-preview rendering.
        let c = ContextCompacted {
            reason: buffa::EnumValue::from(
                WireCompactionReason::COMPACTION_REASON_UNSPECIFIED as i32,
            ),
            ..Default::default()
        };
        assert!(matches!(
            event_from_compacted(c),
            TurnEvent::ContextCompacted {
                reason: CompactionReason::Truncated,
                ..
            }
        ));
    }

    #[test]
    fn dial_error_retryable_classification() {
        use connectrpc::ErrorCode;
        // Transient transport conditions → retryable (ask for redelivery).
        for code in [
            ErrorCode::Unavailable,
            ErrorCode::DeadlineExceeded,
            ErrorCode::ResourceExhausted,
            ErrorCode::Aborted,
        ] {
            let err = DialError::Connect(connectrpc::ConnectError::new(code, "x"));
            assert_eq!(err.code(), Some(code));
            assert!(err.is_retryable(), "{code:?} should be retryable");
        }
        // Terminal codes → NOT retryable (redelivery would loop forever).
        for code in [
            ErrorCode::InvalidArgument,
            ErrorCode::Unauthenticated,
            ErrorCode::NotFound,
            ErrorCode::PermissionDenied,
            ErrorCode::Internal,
        ] {
            let err = DialError::Connect(connectrpc::ConnectError::new(code, "x"));
            assert_eq!(err.code(), Some(code));
            assert!(!err.is_retryable(), "{code:?} should not be retryable");
        }
        // Local failures carry no Connect code and are never retryable.
        let bad_addr = DialError::InvalidAddress {
            addr: "http://a b".to_owned(),
            source: "http://a b".parse::<http::Uri>().unwrap_err(),
        };
        assert_eq!(bad_addr.code(), None);
        assert!(!bad_addr.is_retryable());
        let tls = DialError::Tls("no provider".to_owned());
        assert_eq!(tls.code(), None);
        assert!(!tls.is_retryable());
    }

    fn text(s: &str) -> Option<content::Type> {
        Some(content::Type::Text(Box::new(TextContent {
            text: s.to_owned(),
            ..Default::default()
        })))
    }

    fn tool_call(id: &str, name: &str) -> Option<content::Type> {
        Some(content::Type::ToolCall(Box::new(ToolCallContent {
            id: id.to_owned(),
            r#type: Some(tool_call_content::Type::FunctionCall(Box::new(
                FunctionCallContent {
                    name: name.to_owned(),
                    ..Default::default()
                },
            ))),
            ..Default::default()
        })))
    }

    fn tool_result(call_id: &str) -> Option<content::Type> {
        Some(content::Type::ToolResult(Box::new(ToolResultContent {
            call_id: call_id.to_owned(),
            ..Default::default()
        })))
    }

    fn thought(summary: &str) -> Option<content::Type> {
        Some(content::Type::Thought(Box::new(ThoughtContent {
            summary: vec![ThoughtSummaryContent {
                r#type: Some(thought_summary_content::Type::Text(Box::new(TextContent {
                    text: summary.to_owned(),
                    ..Default::default()
                }))),
                ..Default::default()
            }],
            ..Default::default()
        })))
    }

    #[test]
    fn renders_text_verbatim() {
        assert_eq!(render_content(text("hi")), Some("hi".to_owned()));
    }

    #[test]
    fn renders_tool_call_with_name_and_id() {
        assert_eq!(
            render_content(tool_call("call_42", "search")),
            Some("[tool_call:search call_42]".to_owned())
        );
    }

    #[test]
    fn renders_tool_call_without_function_name() {
        assert_eq!(
            render_content(Some(content::Type::ToolCall(Box::new(ToolCallContent {
                id: "call_bare".to_owned(),
                r#type: None,
                ..Default::default()
            })))),
            Some("[tool_call:call_bare]".to_owned())
        );
    }

    #[test]
    fn renders_tool_result_by_call_id() {
        assert_eq!(
            render_content(tool_result("call_42")),
            Some("[tool_result:call_42]".to_owned())
        );
    }

    #[test]
    fn reasoning_is_never_the_reply() {
        // Reasoning must NOT render into the buffered reply (raw chain-of-thought
        // as the answer). It is shown via the TUI's separate transcript path.
        assert_eq!(render_content(thought("considering options")), None);
    }

    #[test]
    fn thought_only_turn_yields_empty_reply() {
        // A turn that produced only reasoning (no answer text, no tool calls)
        // must come back empty, not with the chain-of-thought as the reply.
        assert_eq!(aggregate(vec![thought("secret reasoning")]), "");
    }

    #[test]
    fn empty_text_skipped() {
        assert_eq!(render_content(text("")), None);
    }

    #[test]
    fn unknown_variant_skipped() {
        // No content::Type set at all.
        assert_eq!(render_content(None), None);
    }

    // Helper used by aggregation tests: feed the same rendering loop
    // run_turn uses, but driven from a vector of fake content blocks rather
    // than a live connectrpc stream.
    fn aggregate(blocks: Vec<Option<content::Type>>) -> String {
        let mut parts: Vec<String> = Vec::new();
        for b in blocks {
            if let Some(s) = render_content(b) {
                parts.push(s);
            }
        }
        parts.join("\n")
    }

    #[test]
    fn aggregate_pure_text_turn() {
        assert_eq!(
            aggregate(vec![text("hello"), text("world")]),
            "hello\nworld"
        );
    }

    #[test]
    fn aggregate_tool_call_only_turn() {
        // A turn that ends in a tool call with no follow-up text used to
        // come back as "" and was silently dropped by the Slack handler.
        // Rendering a placeholder keeps the user informed.
        assert_eq!(
            aggregate(vec![tool_call("call_1", "lookup")]),
            "[tool_call:lookup call_1]"
        );
    }

    #[test]
    fn aggregate_mixed_text_and_tool_call() {
        assert_eq!(
            aggregate(vec![text("thinking..."), tool_call("call_1", "search")]),
            "thinking...\n[tool_call:search call_1]"
        );
    }

    // --- streaming mapping (message_to_event) ---

    fn message(role: &str, ty: Option<content::Type>) -> Message {
        Message {
            role: role.to_owned(),
            content: buffa::MessageField::some(Content {
                r#type: ty,
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    #[test]
    fn assistant_text_becomes_text_delta() {
        assert_eq!(
            message_to_event(message("assistant", text("hello"))),
            Some(TurnEvent::TextDelta("hello".to_owned()))
        );
    }

    #[test]
    fn model_role_also_counts_as_assistant() {
        assert_eq!(
            message_to_event(message("model", text("hi"))),
            Some(TurnEvent::TextDelta("hi".to_owned()))
        );
    }

    #[test]
    fn tool_role_text_is_skipped() {
        // A tool-result echo carried as text must not surface as answer text.
        assert_eq!(message_to_event(message("tool", text("result blob"))), None);
    }

    #[test]
    fn empty_assistant_text_is_skipped() {
        assert_eq!(message_to_event(message("assistant", text(""))), None);
    }

    #[test]
    fn tool_call_becomes_tool_started_with_name() {
        assert_eq!(
            message_to_event(message("model", tool_call("call_7", "search"))),
            Some(TurnEvent::ToolStarted {
                name: "search".to_owned()
            })
        );
    }

    #[test]
    fn tool_call_falls_back_to_call_id() {
        assert_eq!(
            message_to_event(message(
                "tool",
                Some(content::Type::ToolCall(Box::new(ToolCallContent {
                    id: "call_bare".to_owned(),
                    r#type: None,
                    ..Default::default()
                })))
            )),
            Some(TurnEvent::ToolStarted {
                name: "call_bare".to_owned()
            })
        );
    }

    #[test]
    fn tool_result_produces_no_event() {
        assert_eq!(
            message_to_event(message("tool", tool_result("call_7"))),
            None
        );
    }

    // Drives the same per-message mapping the streaming loop uses, over a
    // synthetic turn, then appends Done as the End arm would.
    fn map_turn(msgs: Vec<Message>) -> Vec<TurnEvent> {
        let mut events: Vec<TurnEvent> = msgs.into_iter().filter_map(message_to_event).collect();
        events.push(TurnEvent::Done);
        events
    }

    #[test]
    fn synthetic_turn_yields_expected_event_sequence() {
        // model Text delta, a tool-role ToolCall, another model Text, End.
        let turn = vec![
            message("model", text("Let me look that up.")),
            message("tool", tool_call("call_1", "search")),
            message("model", text("Found it.")),
        ];
        assert_eq!(
            map_turn(turn),
            vec![
                TurnEvent::TextDelta("Let me look that up.".to_owned()),
                TurnEvent::ToolStarted {
                    name: "search".to_owned()
                },
                TurnEvent::TextDelta("Found it.".to_owned()),
                TurnEvent::Done,
            ]
        );
    }

    // --- terminal-envelope projection (events_from_end) ---

    use polyc_proto::proto::polychrome::agent::v1::{
        Handoff as WireHandoff, PendingApproval as WirePendingApproval,
    };

    #[test]
    fn end_with_nothing_yields_only_done() {
        assert_eq!(events_from_end(AgentEnd::default()), vec![TurnEvent::Done]);
    }

    #[test]
    fn end_with_handoff_yields_handoff_then_done() {
        let end = AgentEnd {
            handoff: buffa::MessageField::some(WireHandoff {
                call_id: "call_1".to_owned(),
                child_agent_id: "researcher".to_owned(),
                reason: "needs deep dive".to_owned(),
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(
            events_from_end(end),
            vec![
                TurnEvent::HandoffStarted {
                    child_agent_id: "researcher".to_owned(),
                    reason: "needs deep dive".to_owned(),
                },
                TurnEvent::Done,
            ]
        );
    }

    #[test]
    fn end_orders_approvals_before_handoff_before_done() {
        let end = AgentEnd {
            pending_approvals: vec![WirePendingApproval {
                request_id: "r1".to_owned(),
                tool_name: "delete_file".to_owned(),
                args_json: "{}".to_owned(),
                title: "Delete a file".to_owned(),
                ..Default::default()
            }],
            handoff: buffa::MessageField::some(WireHandoff {
                child_agent_id: "child".to_owned(),
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(
            events_from_end(end),
            vec![
                TurnEvent::ApprovalPending {
                    request_id: "r1".to_owned(),
                    tool_name: "delete_file".to_owned(),
                    title: "Delete a file".to_owned(),
                    args_json: "{}".to_owned(),
                    reason: String::new(),
                },
                TurnEvent::HandoffStarted {
                    child_agent_id: "child".to_owned(),
                    reason: String::new(),
                },
                TurnEvent::Done,
            ]
        );
    }

    // --- namespaced conversation ids ---

    #[test]
    fn namespaced_id_is_prefix_colon_native() {
        assert_eq!(
            namespaced_id("github", "owner/repo#42"),
            "github:owner/repo#42"
        );
        assert_eq!(namespaced_id("web", "abc-123"), "web:abc-123");
    }

    #[test]
    fn hashed_conversation_id_is_deterministic_v5() {
        let ns = uuid::Uuid::from_u128(0x1234_5678_9abc_4def_8123_4567_89ab_cdef);
        let a = hashed_conversation_id(ns, &["T1", "C1", "169.000"]);
        let b = hashed_conversation_id(ns, &["T1", "C1", "169.000"]);
        assert_eq!(a, b);
        assert_ne!(a, hashed_conversation_id(ns, &["T1", "C2", "169.000"]));
        let parsed: uuid::Uuid = a.parse().unwrap();
        assert_eq!(parsed.get_version_num(), 5);
    }

    #[test]
    fn framed_conversation_id_resists_separator_collisions() {
        let ns = uuid::Uuid::from_u128(0x99);
        // The exact case that collides under the bare-join helper.
        assert_ne!(
            framed_conversation_id(ns, &["a:b", "c"]),
            framed_conversation_id(ns, &["a", "b:c"]),
        );
        // Sanity: the bare-join helper DOES collide here (documents why framed exists).
        assert_eq!(
            hashed_conversation_id(ns, &["a:b", "c"]),
            hashed_conversation_id(ns, &["a", "b:c"]),
        );
        // Deterministic + valid v5.
        let id = framed_conversation_id(ns, &["mail", "<abc@x>"]);
        assert_eq!(id, framed_conversation_id(ns, &["mail", "<abc@x>"]));
        assert_eq!(id.parse::<uuid::Uuid>().unwrap().get_version_num(), 5);
    }

    #[test]
    fn hashed_conversation_id_matches_slacks_inline_algorithm() {
        // Golden cross-check: the SDK helper must reproduce exactly what the
        // Slack edge used to compute inline (UUIDv5 of "team:channel:thread"
        // under a pinned namespace), so delegating in `polychrome-slack`
        // changes no existing conversation id.
        let ns = uuid::Uuid::from_u128(0xa1b2_c3d4_e5f6_4789_abcd_ef01_2345_6789);
        let parts = ["T01234ABCD", "C0B5V0JA401", "1700000000.000100"];
        let inline = uuid::Uuid::new_v5(&ns, parts.join(":").as_bytes())
            .hyphenated()
            .to_string();
        assert_eq!(hashed_conversation_id(ns, &parts), inline);
    }

    #[test]
    fn link_outcome_messages_honor_presentation_rules() {
        // INVALID and EXPIRED are fused upstream into one variant, so neither
        // edge can distinguish them — one shared message, no oracle.
        assert!(
            LinkCeremony::InvalidOrExpired
                .user_message()
                .contains("invalid or expired")
        );
        // THROTTLED reads distinctly (wait, don't retry).
        assert!(LinkCeremony::Throttled.user_message().contains("wait"));
        assert!(
            LinkCeremony::Linked {
                persona_id: "p".to_owned()
            }
            .user_message()
            .contains("Linked")
        );
    }
}