polyc_rpc_client/

lib.rs

//! 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, Message,
    ParticipantMessage, Verdict, agent_response, content, thought_summary_content,
    tool_call_content,
};
use polyc_proto::proto::polychrome::approval::v1::{
    ApprovalResponseRequest, ApprovalServiceClient,
};
use polyc_proto::proto::polychrome::persona::v1::{
    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);

/// 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::with_tls(std::sync::Arc::new(tls)))
    } else {
        Ok(HttpClient::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))))
}

/// 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 {
    /// 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,
    },
    /// 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(_)) => {}
                            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,
                None => {
                    // Empty envelope — defensive; the control plane shouldn't
                    // send these but the wire allows it. Ignore.
                }
            }
        }
        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 {
                    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)
    }
}

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

/// 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`.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn respond(
        &self,
        request_id: &str,
        approved: bool,
        reason: &str,
        conversation_id: &str,
    ) -> Result<ApprovalOutcome, DialError> {
        let request = ApprovalResponseRequest {
            request_id: request_id.to_owned(),
            approved,
            reason: reason.to_owned(),
            conversation_id: conversation_id.to_owned(),
            ..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,
        })
    }
}

/// 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. Please wait a few minutes and try again.",
            Self::Failed => "Sorry — something went wrong completing that link. Please try again.",
        }
    }
}

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

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

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

/// 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)),
        content::Type::Thought(t) => {
            // ThoughtContent has no top-level text; its `summary` repeated
            // field carries the user-visible reasoning when the provider
            // emits one. Collect any TextContent summaries; skip otherwise.
            let mut buf = String::new();
            for s in t.summary {
                if let Some(thought_summary_content::Type::Text(text)) = s.r#type
                    && !text.text.is_empty()
                {
                    if !buf.is_empty() {
                        buf.push(' ');
                    }
                    buf.push_str(&text.text);
                }
            }
            if buf.is_empty() {
                None
            } else {
                Some(format!("[thought] {buf}"))
            }
        }
        // image / audio / document / video / confirmation — no useful text
        // rendering for v1.
        _ => None,
    }
}

#[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,
    };

    #[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 renders_thought_summary() {
        assert_eq!(
            render_content(thought("considering options")),
            Some("[thought] considering options".to_owned())
        );
    }

    #[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(),
                },
                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")
        );
    }
}