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 durable-receive dance: send
//! one [`AgentRequest`], obtain its receipt, then drain the attached
//! [`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).
//!
//! Buffered and incremental response projections build on the same receipt
//! and attachment client.

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

pub mod edge;
pub use edge::{
    ClaimedNamespace, ClaimedNamespaceError, EdgeAdapter, IngressDirective, IngressIdentity,
    IngressIdentityError, MAX_CLAIMED_NAMESPACE_BYTES, Priority, build_attribution,
};

pub mod edge_credentials;

/// Test-only access to this crate's envelope builder, behind the `test-util`
/// feature.
///
/// The control plane's ingest-gate tests must present envelopes an edge would
/// really send. Re-deriving the signing contract in a fixture would let the
/// two drift — the tests would keep passing against their own copy while real
/// edges broke — so they mint envelopes through the SAME
/// [`build_asserted_attribution`] production dials go through.
///
/// Mirrors `polyc-passkey`/`polyc-web-session`'s `test-util` shape. Never
/// compiled into a release binary: it is reached only via a dev-dependency.
#[cfg(feature = "test-util")]
pub mod test_util {
    use super::{
        AssertedAttribution, Attribution, ClaimedNamespace, EdgeCredentials, IngressIdentity,
        Message, ParticipantMessage, build_asserted_attribution, build_classify_attribution,
    };
    use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;

    /// Build and sign a real participation-gate envelope.
    ///
    /// The exact envelope [`crate::AgentDialer::should_respond`] puts on the
    /// wire, so a control-plane test cannot drift from what the edge sends.
    #[must_use]
    pub fn signed_classify_envelope(
        creds: &EdgeCredentials,
        conversation_id: &str,
        source_identity: &IngressIdentity,
        namespace: &ClaimedNamespace,
        transcript: &[ParticipantMessage],
        bot_name: &str,
        surface: &str,
    ) -> AssertedAttribution {
        build_classify_attribution(
            creds,
            conversation_id,
            source_identity,
            namespace,
            transcript,
            bot_name,
            surface,
        )
    }

    /// Build and sign a real [`AssertedAttribution`] for `messages`.
    ///
    /// Binds it to `conversation_id`/`exec_id` and asserts
    /// `caller`/`participants` — the exact envelope
    /// [`crate::AgentDialer::with_credentials`] puts on the wire for the same
    /// turn.
    #[must_use]
    #[allow(
        clippy::too_many_arguments,
        reason = "mirrors `build_asserted_attribution`'s parameter list on \
                  purpose: a fixture that could omit one would stop testing \
                  the envelope production dials really send"
    )]
    pub fn signed_envelope(
        creds: &EdgeCredentials,
        conversation_id: &str,
        exec_id: &str,
        source_identity: &IngressIdentity,
        namespace: &ClaimedNamespace,
        messages: &[Message],
        caller: Option<ExternalIdentity>,
        participants: Vec<ExternalIdentity>,
    ) -> AssertedAttribution {
        build_asserted_attribution(
            creds,
            conversation_id,
            exec_id,
            source_identity,
            namespace,
            messages,
            &Attribution {
                caller,
                participants,
            },
        )
    }
}
pub use edge_credentials::{
    CredentialError, EdgeCredentials, EdgeCredentialsError, edge_credentials_from_env_or_fail,
};

/// Re-exported so the thin edges (slack/telegram/discord/email/trigger/a2a)
/// can wrap their two secret config fields (the transport bearer, the
/// ed25519 signing key hex) without taking a direct `polyc-crypto`
/// dependency of their own — this crate already depends on it (signing an
/// [`edge_credentials::EdgeCredentials`]'s envelope), and every edge already
/// depends on this crate.
pub use polyc_crypto::sensitive::Sensitive;

/// Reads an optional wrapped secret, treating an empty string as unset.
///
/// Every thin edge (slack/telegram/trigger/...) reads its soft-required
/// secrets — a signing secret, a bot token — the same way: exposed through
/// [`Sensitive::expose`], then folded to `None` when the exposed value is
/// `""` so a blank env var or CLI placeholder reads as "not configured"
/// exactly like a genuinely absent field. This is the one sanctioned place to
/// perform that emptiness peek at an edge read site — call sites that only
/// need the raw (possibly empty) value keep calling `.expose()` directly.
#[must_use]
pub fn expose_nonempty(secret: Option<&Sensitive<String>>) -> Option<&str> {
    secret
        .map(Sensitive::expose)
        .map(String::as_str)
        .filter(|s| !s.is_empty())
}

/// Test-support assertion for edge crates' redaction tests: asserts `value`'s
/// `Debug` output carries the `Sensitive` redaction marker and contains none
/// of `must_not_contain`'s raw secret values.
///
/// Every edge (a2a/slack/telegram/discord/email/trigger) re-proves the same
/// `Sensitive` contract against its own `Config`/`Args` struct — this
/// collapses each of those near-identical bodies to a one-line call.
/// `#[doc(hidden)]` keeps it out of the crate's public rustdoc surface (it
/// exists for this workspace's tests, not downstream consumers of the
/// published API), while the doc comment above still satisfies
/// `missing_docs`.
///
/// # Panics
///
/// Panics (via `assert!`) if `value`'s `Debug` output doesn't contain
/// `Sensitive(<redacted>)`, or if it contains any of `must_not_contain`.
#[doc(hidden)]
pub fn assert_redacted(value: &impl core::fmt::Debug, must_not_contain: &[&str]) {
    let debug = format!("{value:?}");
    assert!(
        debug.contains("Sensitive(<redacted>)"),
        "expected a Sensitive redaction marker in Debug output, got {debug}"
    );
    for raw in must_not_contain {
        assert!(
            !debug.contains(raw),
            "Debug output leaked a raw secret value {raw:?}: {debug}"
        );
    }
}

use std::sync::Arc;

use chrono::{DateTime, Datelike as _, FixedOffset, Utc};
use connectrpc::client::{ClientConfig, HttpClient};
use futures::Stream;
use polyc_agent::text_message;
use polyc_proto::proto::polychrome::agent::v1::{
    AgentEnd, AgentRequest, AgentServiceClient, AgentStart,
    ApprovalPreview as WireAgentApprovalPreview,
    ApprovalPreviewFire as WireAgentApprovalPreviewFire, AssertedAttribution, AttachIngressRequest,
    ClassifyRequest, CompactionReason as WireCompactionReason, ContextCompacted,
    IngressDirective as WireIngressDirective, IngressReceipt as WireIngressReceipt,
    IngressSourceIdentity as WireIngressSourceIdentity, InterruptRequest, Message,
    ParticipantMessage, PendingApproval as WireAgentPendingApproval,
    PendingQuestion as WireAgentPendingQuestion, QuestionOption as WireAgentQuestionOption,
    TurnFailureKind as WireTurnFailureKind, Verdict, agent_response, content, tool_call_content,
};
use polyc_proto::proto::polychrome::approval::v1::{
    ApprovalPreview as WireListApprovalPreview, ApprovalPreviewFire as WireListApprovalPreviewFire,
    ApprovalResponseRequest, ApprovalServiceClient, ListApprovalRecoveryRequest,
    ListPendingRequest, PendingApprovalEntry, ReadPreview as WireListReadPreview,
    RecordedApprovalDecision as WireRecordedApprovalDecision, RecoverableApprovalEntry,
    RepromptRequired as WireRepromptRequired, SearchHitPreview as WireSearchHitPreview,
    SearchPreview as WireSearchPreview, read_preview as wire_list_read_preview,
    recoverable_approval_entry,
};
use polyc_proto::proto::polychrome::credential::v1::{
    AddCredentialKeyRequest, CredentialKeySummary as WireCredentialKeySummary,
    CredentialKeyVerifier as WireCredentialKeyVerifier, CredentialServiceClient,
    CredentialSummary as WireCredentialSummary, EnrollCredentialRequest,
    KeyState as WireCredentialKeyState, ListCredentialsRequest, RetireCredentialKeyRequest,
    RevokeCredentialRequest,
};
use polyc_proto::proto::polychrome::ops::v1::{
    AckRequest, DecideRequest, NotificationServiceClient, OperatorMailboxServiceClient,
    PollPendingRequest, SubscribeRequest, decide_reply, ops_action_view, upgrade_outcome,
};
use polyc_proto::proto::polychrome::persona::v1::{
    AdminInviteRequest, AttestVerifiedEmailOutcome, AttestVerifiedEmailRequest, AutoLinkOutcome,
    AutoLinkRequest, CompleteLinkRequest, DescribeRequest, LinkOutcome, PersonaServiceClient,
    RebuildUsageRollupsRequest, SetIncognitoRequest, StartDeepLinkRequest, StartLinkRequest,
};
use polyc_proto::proto::polychrome::question::v1::{
    Decline as WireDecline, ListPendingQuestionsRequest, PendingQuestionEntry,
    QuestionAnswerRequest, QuestionOptionEntry, QuestionServiceClient,
    SelectOption as WireSelectOption, question_answer_request::Answer as WireAnswer,
};
use polyc_proto::proto::polychrome::routine::v1::{FireRoutineRequest, RoutineServiceClient};
use polyc_proto::proto::polychrome::task::v1::{
    AgentTask as WireAgentTask, AgentTaskOwnership as WireAgentTaskOwnership,
    AgentTaskServiceClient, AgentTaskState as WireAgentTaskState, CancelAgentTaskRequest,
    ClaimAgentTaskRequest, CreateAgentTaskRequest, GetAgentTaskRequest, ListAgentTasksRequest,
    RenewAgentTaskClaimRequest, TransitionAgentTaskRequest,
};

/// 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),
    /// A bearer credential could not be encoded as an HTTP header value
    /// (e.g. contains a newline or non-ASCII byte). Produced by
    /// [`AgentDialer::with_credentials`], `PersonaDialer::new_admin`,
    /// `CredentialDialer::new_admin`, and `RoutineDialer::fire_routine`'s
    /// admin path. Fails the dial closed rather than silently sending the
    /// turn unauthenticated.
    #[error("bearer credential is not a valid header value: {0}")]
    InvalidBearer(String),
    /// A turn ingress was attempted with an unauthenticated dialer.
    ///
    /// Stable source identity rides only inside the signed edge envelope, so
    /// sending without credentials would necessarily drop the identity and is
    /// refused before transport I/O.
    #[error("turn ingress requires edge credentials so its source identity is signed")]
    MissingIngressCredentials,
    /// The server returned a success response that cannot prove the requested
    /// source event was durably received.
    #[error("invalid durable ingress receipt: {0}")]
    InvalidIngressReceipt(&'static str),
    /// The recovery service returned an entry without one of its required
    /// states.
    #[error("invalid approval recovery entry: {0}")]
    InvalidApprovalRecovery(&'static str),
}

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

    /// Whether this error is a `DeadlineExceeded` — the call's deadline
    /// elapsed rather than anything failing.
    ///
    /// A long-lived stream that is idle by design ([`NotificationDialer::subscribe`])
    /// ends this way on every cycle, so its caller reports the recycle as routine
    /// instead of as a failure. Callers of short unary RPCs should keep treating
    /// it as an error.
    ///
    /// The subscribe stream carries a deadline it cannot be exempted from
    /// where it is currently mounted, so reaching that deadline is the normal
    /// cycle, not a failure — reporting it as one buries the case where the
    /// notifier is genuinely broken (#1794).
    #[must_use]
    pub const fn is_deadline_exceeded(&self) -> bool {
        matches!(self.code(), Some(connectrpc::ErrorCode::DeadlineExceeded))
    }

    /// Whether a signed human-action capability is absent, expired, or no
    /// longer matches the durable occurrence it names.
    ///
    /// Approval edges use this to retire a stale card instead of inviting a
    /// deterministic retry. The server reports malformed/expired capability
    /// tokens as `Unauthenticated` and an occurrence that can no longer be
    /// answered as `FailedPrecondition`.
    #[must_use]
    pub const fn is_stale_capability(&self) -> bool {
        matches!(
            self.code(),
            Some(
                connectrpc::ErrorCode::Unauthenticated | connectrpc::ErrorCode::FailedPrecondition
            )
        )
    }
}

/// 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.
///
/// It must stay ABOVE the control plane's own stream cap, because the control
/// plane HONORS a client deadline shorter than that cap. A caller deadline at
/// or below the cap cuts the response stream at the same instant the turn's
/// own budget expires, so the edge reports a transport failure for a turn that
/// was about to end with a precise one. The control plane's
/// `grpc::tests::turn_deadline_chain_leaves_room_after_a_cold_start` asserts
/// that ordering against this constant, which is why it is public.
pub const AGENT_DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(4);

/// 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`].
///
/// No `pool_idle_timeout` is configured here (`#757` follow-up): this
/// `HttpClient` pools connections through hyper's legacy client, whose
/// `Client::builder` already defaults `pool_idle_timeout` to 90 seconds (see
/// `hyper_util::client::legacy::client::Config::default`) — comfortably under
/// any realistic gap between calls, so a connection genuinely idle in the pool
/// is already reaped rather than blackholed. connectrpc 0.8.1's
/// `HttpClientBuilder` doesn't expose a way to change this default anyway
/// (verified against its source — there is no `pool_idle_timeout` method).
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())
    }
}

/// Build the `Authorization: Bearer <bearer>` header value shared by every
/// dialer constructor that authenticates with an edge or admin bearer
/// (`AgentDialer::with_credentials`, each non-Agent dialer's `with_bearer` and
/// admin-gated `new_admin` — `PersonaDialer::new_admin`,
/// `CredentialDialer::new_admin` — via [`bearer_header`], and
/// `RoutineDialer::fire_routine`'s per-call admin bearer) — the ONE place a
/// bearer string turns into an HTTP header value.
///
/// Builds the `HeaderValue` explicitly and fails closed rather than using
/// connectrpc's `with_default_header`, which silently drops an invalid value
/// instead of erroring — a dropped header would send the turn unauthenticated
/// with no signal.
///
/// Marks the value [`set_sensitive`](http::HeaderValue::set_sensitive) so the
/// HTTP stack treats it as a secret: it's excluded from hyper/h2 debug
/// logging and, on HTTP/2, sent as an HPACK literal rather than added to the
/// dynamic table (where a compromised intermediary or a debug dump of the
/// table could otherwise recover it, and where CRIME/BREACH-style compression
/// side channels could otherwise probe it).
///
/// # Errors
/// Returns [`DialError::InvalidBearer`] if `bearer` can't be encoded as an
/// HTTP header value (e.g. contains a newline or non-ASCII byte).
fn bearer_header_value(bearer: &str) -> Result<http::HeaderValue, DialError> {
    let mut bearer_value = http::HeaderValue::from_str(&format!("Bearer {bearer}"))
        .map_err(|source| DialError::InvalidBearer(source.to_string()))?;
    bearer_value.set_sensitive(true);
    Ok(bearer_value)
}

/// Build a header map carrying the sensitive `Authorization: Bearer <bearer>`
/// value from [`bearer_header_value`], shared by every dialer constructor
/// that authenticates with an edge bearer (`AgentDialer::with_credentials`
/// and each non-Agent dialer's `with_bearer`).
///
/// This header map is folded into the generated Connect client's default
/// headers (see [`build_control_client`]) and held for the dialer's process
/// lifetime — a deliberate choice: the generated Connect client only accepts
/// a default `HeaderMap` at construction, not a per-request override, so
/// holding the sensitive, already-marked value for the dialer's lifetime is
/// the accepted shape rather than an oversight. Per-request construction
/// would require changing the generated-client plumbing, which is out of
/// scope here.
///
/// # Errors
/// Returns [`DialError::InvalidBearer`] if `bearer` can't be encoded as an
/// HTTP header value (e.g. contains a newline or non-ASCII byte).
fn bearer_header(bearer: &str) -> Result<http::HeaderMap, DialError> {
    let mut headers = http::HeaderMap::new();
    headers.insert(http::header::AUTHORIZATION, bearer_header_value(bearer)?);
    Ok(headers)
}

/// Build a control-plane dialer's inner client: parse `addr`, build the HTTP
/// transport, apply the shared [`CONTROL_DIAL_TIMEOUT`], and — when `bearer`
/// is `Some` — attach the `Authorization` header via [`bearer_header`].
/// `build_client` produces the generated Connect client from the resulting
/// `(HttpClient, ClientConfig)` pair (e.g. `PersonaServiceClient::new`).
///
/// Factors out the `new`/`with_bearer` body every non-`AgentDialer` dialer
/// (`ApprovalDialer`, `PersonaDialer`, `NotificationDialer`,
/// `LiveEnrollmentsDialer`, `OperatorMailboxDialer`) previously duplicated
/// byte-for-byte apart from its client type — mechanical dedup, no behavior
/// change. [`AgentDialer`] is NOT built through this helper: its
/// authenticated path additionally signs a per-turn envelope and uses a
/// different timeout ([`AGENT_DIAL_TIMEOUT`]), so it stays hand-written.
///
/// # Errors
///
/// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI,
/// [`DialError::Tls`] if an `https` endpoint's TLS setup fails, or
/// [`DialError::InvalidBearer`] if `bearer` (when `Some`) can't be encoded as
/// an HTTP header value.
fn build_control_client<C>(
    addr: &str,
    bearer: Option<&str>,
    build_client: impl FnOnce(HttpClient, ClientConfig) -> C,
) -> Result<Arc<C>, DialError> {
    let uri = addr
        .parse::<http::Uri>()
        .map_err(|source| DialError::InvalidAddress {
            addr: addr.to_owned(),
            source,
        })?;
    let http = http_client_for(&uri)?;
    let mut config = ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT);
    if let Some(bearer) = bearer {
        config = config.with_default_headers(bearer_header(bearer)?);
    }
    Ok(Arc::new(build_client(http, config)))
}

/// 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 {
        /// Turn that emitted this occurrence of `request_id`.
        turn_id: String,
        /// 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,
        /// Short-lived signed capability (`#787`), freshly minted for THIS
        /// card and scoped to `turn_id` + `request_id` + the conversation it
        /// belongs to.
        /// Opaque to the caller: carry it back unmodified to
        /// [`ApprovalDialer::respond`]. `ApprovalService.Respond` rejects a
        /// decision whose token is missing, expired, or bound to a different
        /// request or conversation.
        resolve_token: String,
        /// Computed-preview enrichment (`#1496`), set only for a
        /// `routine_create` call whose arguments parsed against the compiled
        /// spec shape — `None` for every other tool. Built by the control
        /// plane from the compiled spec, never from the model's narration.
        /// Render it with [`approval_preview_text`], never a hand-written
        /// per-surface rendering.
        preview: Option<ApprovalPreview>,
    },
    /// A question the model asked via `ask_question` is awaiting an answer
    /// (`#1660`). Emitted (one per pending question) from the terminal
    /// `AgentEnd` just before [`TurnEvent::Done`] — the question-pause
    /// SIBLING of [`TurnEvent::ApprovalPending`] above, not a reuse of it:
    /// `ask_question` is a decision the user makes, not a danger/permission
    /// gate, so it has its own event-log pair and its own signed-answer
    /// canonical (`polyc_crypto::question`). The caller submits a decision
    /// via `QuestionService.Respond` (the THIN path, mirroring
    /// `ApprovalService.Respond`) and re-drives the turn.
    QuestionPending {
        /// Turn that emitted this question occurrence. A provider re-mints a
        /// tool-call id across turns, so the turn is what separates two
        /// questions that share a `(call_id, index)` (`#2523`).
        turn_id: String,
        /// The `ask_question` call id this question came from. Several
        /// pending questions can share the same `call_id` (one call can ask
        /// up to three questions at once) — identity is
        /// `(turn_id, call_id, index)`.
        call_id: String,
        /// This question's position within its call's `questions` array
        /// (0-based).
        index: u32,
        /// The short label (fits a chat-surface button-row heading).
        header: String,
        /// The one-sentence question to ask.
        question: String,
        /// 2-4 mutually exclusive options to offer.
        options: Vec<QuestionOptionPrompt>,
        /// The raw `ask_question` call's full arguments JSON (every question
        /// in the call, not just this one) — the audit binding a signed
        /// answer must match against.
        args_json: String,
        /// Short-lived signed capability, freshly minted for THIS card and
        /// scoped to `turn_id` + `call_id` + `index` + the conversation it
        /// belongs to. Opaque to the caller: carry it back unmodified to
        /// [`QuestionDialer::respond`]. `QuestionService.Respond` rejects a
        /// decision whose token is missing, expired, or bound to a
        /// different occurrence or conversation.
        answer_token: String,
        /// Whether this still-unanswered question was already surfaced once
        /// (a later turn's boundary marker sits after its
        /// `question_request` in the event log while it stayed unanswered).
        /// `false`: render the full interactive card; `true`: render a
        /// compact reminder instead (`#1970`). Derived by the control plane
        /// at replay time — never edge-cached. Unlike
        /// [`TurnEvent::ApprovalPending`] (which never carries this bit —
        /// approvals only ride it through `PendingApproval`'s recovery
        /// path), THIS streamed event is itself the vehicle for
        /// re-surfacing a question on every redrive, so it copies the
        /// control plane's already-computed value straight through rather
        /// than always reporting `false`.
        already_surfaced: bool,
    },
    /// 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 runs in an
    /// independent conversation. The parent receives no child result. An edge
    /// can render "delegating…" instead of 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,
    },
    /// The control plane minted an admin invite this turn for the edge to
    /// deliver privately (agent-evaluable admin invite, `#698`). Surfaced from
    /// the terminal [`AgentEnd`] just before [`TurnEvent::Done`]. The edge opens
    /// the target's direct message and delivers the `code` there — and ONLY
    /// there. The code reached neither the agent nor a channel; the edge fails
    /// closed (telling the admin, delivering nothing) if it can't reach the
    /// target privately.
    InviteDelivery {
        /// The target's provider-native user id (from the mention markup). The
        /// edge opens THIS person's direct message.
        target_user_id: String,
        /// The single-use invite code — the one secret on this event. Deliver it
        /// only to the target's direct message; never log it or post it to a
        /// channel.
        code: String,
        /// The inviting admin's display name, for the target-facing copy. May be
        /// empty; the edge then uses a neutral phrasing.
        inviter_display: String,
    },
    /// A paid tool call this turn needed a USABLE linked wallet (`#519`;
    /// renewal branch `#2122`). Surfaced from the terminal [`AgentEnd`] just
    /// before [`TurnEvent::Done`]. Carries only the URL and the renewal bit —
    /// no copy — so every edge renders the identical "Link a
    /// wallet"/"Renew access" card text through the shared `polyc_proto`
    /// helpers instead of hand-writing its own wording.
    WalletLinkPrompt {
        /// The deployment's wallet-link URL, when known. `None` = point the
        /// reader at an admin instead of rendering a dangling button.
        link_url: Option<String>,
        /// True when the caller HAD a linked wallet whose delegation is no
        /// longer usable (expired, revoked, or otherwise broken) — false
        /// when they never linked one at all. Selects the "renew" card copy
        /// over the first-time "set one up" copy.
        renewal: bool,
        /// True when the caller directly asked to link (or replace) a
        /// wallet — the `wallet_link` tool called on its own, never the
        /// `paid_fetch`/`web_fetch` payment-interrupt path `renewal` covers.
        /// Selects the "here's a secure link" card copy over both the
        /// first-time and renewal payment-interrupt wording.
        requested: bool,
    },
    /// `wallet_update_limit` minted a fresh TIP-1011 in-place
    /// spending-limit-update ceremony this turn (issue #1041/#1159).
    /// Surfaced from the terminal [`AgentEnd`] just before
    /// [`TurnEvent::Done`]. Carries only the URL and the requested cap — no
    /// copy — so every edge renders the identical "Update spending cap"
    /// card text through the shared `polyc_proto` helpers instead of
    /// hand-writing its own wording.
    WalletUpdatePrompt {
        /// The one-time spending-limit-update ceremony URL. Always
        /// non-empty when this variant is emitted — the tool that produces
        /// this either mints a URL or refuses outright.
        update_url: String,
        /// The requested new daily spending cap, in the settlement
        /// currency's human units (e.g. `"20"`), echoed on the card.
        new_limit: String,
    },
    /// This turn's `unlink_self` (wallet target) call minted a TIP-1011
    /// hard-revoke ceremony for the just-unlinked delegated key (issue
    /// #1042/#1156). Surfaced from the terminal [`AgentEnd`] just before
    /// [`TurnEvent::Done`]. Carries only the URL — no copy — so every edge
    /// renders the identical "Revoke access" card text through the shared
    /// `polyc_proto` helpers instead of hand-writing its own wording.
    WalletRevokePrompt {
        /// The one-time hard-revoke ceremony URL. Always non-empty when
        /// this variant is emitted — only set for a delegated key that was
        /// provisioned onchain.
        revoke_url: String,
    },
    /// The turn failed durably instead of completing (`#756`). Surfaced from
    /// the terminal [`AgentEnd.failure`](AgentEnd) just before
    /// [`TurnEvent::Done`] — a structured, durable fact the control plane
    /// persisted, not merely the Connect RPC status a dial error would carry.
    /// `AgentEnd.failure` exists specifically so "external surfaces (Slack,
    /// the cockpit) can react" (see its wire doc comment); before this
    /// variant existed, no edge ever read it, and a durably-failed turn with
    /// no other content (no messages, no pending approvals, no handoff, no
    /// invites) surfaced as silence — every edge saw `Done` with nothing to
    /// show and posted nothing.
    TurnFailed {
        /// Provider-agnostic failure classification (mirrors the wire
        /// `TurnFailureKind` 1:1; `TURN_FAILURE_KIND_UNSPECIFIED` maps to
        /// [`TurnFailureKind::Other`] — a definite failure with an unknown
        /// reason is still a definite failure).
        kind: TurnFailureKind,
        /// Human-readable diagnostic text (the underlying provider/tool
        /// error). Log-only — an edge's user-facing wording comes from its
        /// own shared failure-notice helper keyed on
        /// [`TurnFailureKind::is_retryable`], not this string.
        message: String,
    },
    /// Terminal event: the turn has ended and no further events follow.
    Done,
}

/// Provider-agnostic classification of a durable turn failure (`#756`).
///
/// Local mirror of the wire `TurnFailureKind`, following the same
/// decouple-edges-from-the-wire-schema convention as [`CompactionReason`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TurnFailureKind {
    /// The provider rejected the call for exceeding its rate limit.
    RateLimit,
    /// The call exceeded its deadline.
    Timeout,
    /// The provider or a dependency it needs was unreachable.
    Unavailable,
    /// A credential/authentication failure.
    Auth,
    /// The request itself was invalid.
    BadRequest,
    /// Any other failure, including an unspecified wire kind.
    Other,
}

impl TurnFailureKind {
    /// Whether retrying the call could plausibly succeed, mirroring
    /// [`DialError::is_retryable`]'s transient-vs-terminal split: a provider
    /// hiccup (rate limit, timeout, unavailable) is worth retrying; a
    /// credential or malformed-request failure will fail identically again,
    /// and an unclassified `Other` failure is treated conservatively as
    /// non-retryable rather than implying a resend will help when it's
    /// unknown whether it would.
    #[must_use]
    pub const fn is_retryable(self) -> bool {
        matches!(self, Self::RateLimit | Self::Timeout | Self::Unavailable)
    }
}

/// Complete input for one durable ingress receive.
///
/// Construction requires a stable [`IngressIdentity`]. Optional turn details
/// are set through the consuming builder methods; there is no default that can
/// omit source identity.
#[derive(Debug, Clone)]
pub struct TurnIngress {
    conversation_id: String,
    exec_id: String,
    source_identity: IngressIdentity,
    namespace: ClaimedNamespace,
    messages: Vec<Message>,
    payment_receipt: Option<PaymentReceipt>,
    attribution: Attribution,
    ephemeral_history: bool,
    ingress_directive: IngressDirective,
    occurrence: String,
}

impl TurnIngress {
    /// Builds the required core of one source event.
    ///
    /// `namespace` is the tenancy namespace this conversation belongs to
    /// (#1691). It is positional rather than a builder option on purpose: a
    /// surface that mints conversations and names no namespace must fail to
    /// compile, because an unnamed namespace is how this field became
    /// decorative in the first place. Take it from
    /// [`EdgeAdapter::namespace`] so the two
    /// cannot drift.
    #[must_use]
    pub fn new(
        conversation_id: impl Into<String>,
        exec_id: impl Into<String>,
        source_identity: IngressIdentity,
        namespace: ClaimedNamespace,
        messages: Vec<Message>,
    ) -> Self {
        Self {
            conversation_id: conversation_id.into(),
            exec_id: exec_id.into(),
            source_identity,
            namespace,
            messages,
            payment_receipt: None,
            attribution: Attribution::default(),
            ephemeral_history: false,
            ingress_directive: IngressDirective::default(),
            occurrence: String::new(),
        }
    }

    /// Adds a settled inbound payment receipt.
    #[must_use]
    pub fn with_payment_receipt(mut self, receipt: PaymentReceipt) -> Self {
        self.payment_receipt = Some(receipt);
        self
    }

    /// Adds caller and participant attribution asserted by the edge.
    #[must_use]
    pub fn with_attribution(mut self, attribution: Attribution) -> Self {
        self.attribution = attribution;
        self
    }

    /// Adds the edge-authored ingress policy.
    #[must_use]
    pub fn with_ingress_directive(mut self, directive: IngressDirective) -> Self {
        self.ingress_directive = directive;
        self
    }

    /// Declares that the turn starts without replaying prior transcript.
    #[must_use]
    pub const fn with_ephemeral_history(mut self) -> Self {
        self.ephemeral_history = true;
        self
    }

    /// Adds the stable scheduled occurrence that caused this ingress.
    #[must_use]
    pub fn with_occurrence(mut self, occurrence: impl Into<String>) -> Self {
        self.occurrence = occurrence.into();
        self
    }
}

/// Proof that State durably received one [`TurnIngress`].
///
/// Only [`AgentDialer::receive_ingress`] can construct this type. Opening a
/// transport stream or building a request cannot produce it.
#[derive(Debug, Clone)]
pub struct DurablyReceivedTurn {
    receipt: WireIngressReceipt,
    attach_attempt: String,
}

impl DurablyReceivedTurn {
    /// Returns State's stable dispatch handle for this source event.
    #[must_use]
    pub fn dispatch_id(&self) -> &str {
        &self.receipt.dispatch_id
    }

    /// Returns the conversation bound to the durable receipt.
    #[must_use]
    pub fn conversation_id(&self) -> &str {
        &self.receipt.conversation_id
    }

    /// Returns the opaque durable receipt bytes.
    #[must_use]
    pub fn receipt_bytes(&self) -> &[u8] {
        &self.receipt.receipt
    }
}

/// Reusable handle for dialing the polychrome control plane.
#[derive(Clone)]
pub struct AgentDialer {
    client: Arc<AgentServiceClient<HttpClient>>,
    /// Set only by [`Self::with_credentials`]. Turn ingress fails locally when
    /// this is `None`, because stable source identity must ride inside a signed
    /// assertion. Credential-free non-ingress calls remain available.
    credentials: Option<Arc<EdgeCredentials>>,
}

impl AgentDialer {
    /// Build a dialer pointed at `addr` (expects `http://host:port`).
    ///
    /// Unauthenticated: no bearer header rides its calls. This constructor is
    /// suitable only for non-ingress methods such as health/classification;
    /// [`Self::receive_ingress`] refuses it before network I/O.
    ///
    /// # 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),
            credentials: None,
        })
    }

    /// Build a dialer pointed at `addr`, authenticated with `creds`.
    ///
    /// Every call this dialer makes carries an `Authorization: Bearer
    /// <creds.bearer()>` header, and every turn it dials signs a fresh
    /// [`AssertedAttribution`] envelope (`creds.edge_id()`, this turn's
    /// `conversation_id`, a per-call nonce/timestamp, and the caller
    /// [`Attribution`] passed to the turn) onto `AgentStart.asserted_attribution`.
    ///
    /// # Errors
    ///
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::InvalidBearer`] if `creds.bearer()` can't be encoded as
    /// an HTTP header value.
    pub fn with_credentials(addr: &str, creds: EdgeCredentials) -> Result<Self, DialError> {
        let uri = addr
            .parse::<http::Uri>()
            .map_err(|source| DialError::InvalidAddress {
                addr: addr.to_owned(),
                source,
            })?;
        let http = http_client_for(&uri)?;
        let headers = bearer_header(creds.bearer())?;
        let config = ClientConfig::new(uri)
            .with_default_timeout(AGENT_DIAL_TIMEOUT)
            .with_default_headers(headers);
        let client = AgentServiceClient::new(http, config);
        Ok(Self {
            client: Arc::new(client),
            credentials: Some(Arc::new(creds)),
        })
    }

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

    /// Build an [`ApprovalDialer`] for the SAME control-plane endpoint,
    /// sharing this dialer's edge credentials — the bearer on every call, and
    /// the identity key that signs an approval's `AssertedApproval`
    /// (`#1553`). The authenticated sibling of [`Self::approval_dialer`], for
    /// an edge that has already built its `AgentDialer`.
    ///
    /// On the unauthenticated [`Self::new`] path there are no credentials to
    /// share, so this returns exactly what [`Self::approval_dialer`] does:
    /// no bearer, and no responder asserted on any decision.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::InvalidBearer`] if the credentials' bearer can't be
    /// encoded as an HTTP header value.
    pub fn approval_dialer_with_credentials(
        &self,
        addr: &str,
    ) -> Result<ApprovalDialer, DialError> {
        self.credentials.as_ref().map_or_else(
            || ApprovalDialer::new(addr),
            |creds| ApprovalDialer::with_credentials(addr, Arc::clone(creds)),
        )
    }

    /// Build a [`QuestionDialer`] for the SAME control-plane endpoint,
    /// carrying this dialer's edge bearer. The question sibling of
    /// [`Self::approval_dialer_with_credentials`].
    ///
    /// Shares the bearer and NOT the signing key. An approval decision is
    /// signed; a question answer is not, because answering authorizes
    /// nothing (see [`QuestionDialer::respond`]).
    ///
    /// The bearer is still required. `QuestionService` sits behind the
    /// control plane's `require_edge_bearer` layer like every other RPC on
    /// that listener, so an edge that builds an unauthenticated dialer sees
    /// every answer fail closed with 401 — the `#1660` incident, described on
    /// [`QuestionDialer::new`].
    ///
    /// On the unauthenticated [`Self::new`] path there is no bearer to share,
    /// so this returns exactly what [`QuestionDialer::new`] does.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::InvalidBearer`] if the credentials' bearer can't be
    /// encoded as an HTTP header value.
    pub fn question_dialer_with_credentials(
        &self,
        addr: &str,
    ) -> Result<QuestionDialer, DialError> {
        self.credentials.as_ref().map_or_else(
            || QuestionDialer::new(addr),
            |creds| QuestionDialer::with_bearer(addr, creds.bearer()),
        )
    }

    /// Durably receives one source event without waiting for its turn output.
    ///
    /// The returned [`DurablyReceivedTurn`] is acknowledgement authority: it
    /// exists only after the unary `ReceiveIngress` response carries a
    /// nonempty State receipt and a matching signed source identity. Response
    /// consumption starts separately through [`Self::attach_ingress`].
    ///
    /// # Errors
    ///
    /// Returns [`DialError::MissingIngressCredentials`] before network I/O for
    /// an unauthenticated dialer, [`DialError::Connect`] for transport errors,
    /// or [`DialError::InvalidIngressReceipt`] for an incomplete or mismatched
    /// success response.
    pub async fn receive_ingress(
        &self,
        ingress: TurnIngress,
    ) -> Result<DurablyReceivedTurn, DialError> {
        let expected_source = ingress.source_identity.to_wire();
        let expected_conversation = ingress.conversation_id.clone();
        let request = build_request(ingress, self.credentials.as_deref())?;
        let receipt = self
            .client
            .receive_ingress_with_options(request, traced_options())
            .await?
            .into_owned();
        validate_ingress_receipt(&receipt, &expected_source, &expected_conversation)?;
        Ok(DurablyReceivedTurn {
            receipt,
            // One proof object is one execution contender. Clones and exact
            // attachment retries preserve it; a crash/source redelivery that
            // receives a new proof mints a distinct contender identity.
            attach_attempt: uuid::Uuid::now_v7().to_string(),
        })
    }

    /// Attaches to a durably received turn and projects its response stream.
    ///
    /// The proof-carrying [`DurablyReceivedTurn`] is required; a dispatch id
    /// string or a successfully opened transport cannot call this method.
    /// Reattaching with the same proof never starts the turn twice. Until a
    /// durable output outbox exists, Control fails that duplicate attachment
    /// closed rather than pretending it can replay response bytes it did not
    /// persist; callers must treat a lost response stream as outcome-unknown.
    ///
    /// # Errors
    ///
    /// The outer result returns [`DialError::Connect`] if attachment fails.
    /// Each stream item carries response transport/decode failures inline.
    pub async fn attach_ingress(
        &self,
        received: &DurablyReceivedTurn,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>> + use<>, DialError> {
        self.run_turn_streaming_received(received).await
    }

    /// Attaches to a durably received turn and collects its buffered result.
    ///
    /// This is the response half for request/response edges: they await
    /// [`Self::receive_ingress`], acknowledge the source using that proof, and
    /// may then collect text and pending interactions without keeping source
    /// acknowledgement coupled to turn execution.
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for attachment, stream, or decode
    /// failures.
    pub async fn attach_ingress_buffered(
        &self,
        received: &DurablyReceivedTurn,
    ) -> Result<BufferedTurn, DialError> {
        self.run_turn_buffered_received(received).await
    }

    /// 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,
        source_identity: IngressIdentity,
        namespace: ClaimedNamespace,
        user_text: &str,
    ) -> Result<String, DialError> {
        self.run_turn_with(
            conversation_id,
            exec_id,
            source_identity,
            namespace,
            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,
        source_identity: IngressIdentity,
        namespace: ClaimedNamespace,
        user_text: &str,
        attribution: Attribution,
    ) -> Result<String, DialError> {
        Ok(self
            .run_turn_with_approvals(
                conversation_id,
                exec_id,
                source_identity,
                namespace,
                user_text,
                attribution,
                IngressDirective::default(),
            )
            .await?
            .reply)
    }

    /// Like [`run_turn_with`](Self::run_turn_with) but ALSO surfaces any
    /// [`PendingApprovalPrompt`]s the turn paused on, mirroring the streaming
    /// path's [`TurnEvent::ApprovalPending`] projection of the same terminal
    /// `AgentEnd.pending_approvals` field.
    ///
    /// Buffered edges (Discord/email/trigger/A2A) that don't drive the
    /// streaming API still need to render an approve/deny affordance instead
    /// of losing a gated call to the scaffolding-placeholder fallback — this
    /// is the buffered-API variant that lets them. Call
    /// [`ApprovalDialer::respond`] for a decision, then re-drive with this
    /// same method (empty `user_text`) to resume the turn.
    ///
    /// `ingress_directive` is the edge's own policy for this turn (`#68`) —
    /// a step-budget cap, an advisory priority, and/or a required approver.
    /// An edge with no such policy passes [`IngressDirective::default`]
    /// (empty; byte-for-byte unaffected).
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for any transport/stream/encoding
    /// error from the `AgentService` call.
    #[allow(
        clippy::too_many_arguments,
        reason = "each argument is an independent turn premise the signed \
                  envelope binds; grouping them would hide which fields the \
                  edge must supply, and a missing one must not compile"
    )]
    pub async fn run_turn_with_approvals(
        &self,
        conversation_id: &str,
        exec_id: &str,
        source_identity: IngressIdentity,
        namespace: ClaimedNamespace,
        user_text: &str,
        attribution: Attribution,
        ingress_directive: IngressDirective,
    ) -> Result<BufferedTurn, DialError> {
        let ingress = TurnIngress::new(
            conversation_id,
            exec_id,
            source_identity,
            namespace,
            vec![text_message("user", user_text)],
        )
        .with_attribution(attribution)
        .with_ingress_directive(ingress_directive);
        let received = self.receive_ingress(ingress).await?;
        self.attach_ingress_buffered(&received).await
    }

    /// Drive one **routine fire** turn: like
    /// [`run_turn_with_approvals`](Self::run_turn_with_approvals), but the
    /// conversation declares ephemeral history (#843), so the control plane
    /// starts the model on an empty transcript instead of replaying prior fires.
    /// Used by the trigger edge's fan-out — each periodic firing is independent,
    /// never re-feeding past digests / tool loops into context. The event log
    /// still records every fire for forensics, and a gated call still surfaces
    /// its [`PendingApprovalPrompt`]s so the pause is discoverable.
    ///
    /// The caller builds `ingress` with [`TurnIngress::with_ephemeral_history`],
    /// [`TurnIngress::with_occurrence`], and the routine's policy through
    /// [`TurnIngress::with_ingress_directive`]. Keeping those fields in one
    /// required-source-identity value prevents a routine variant from dropping
    /// its idempotency key as its options grow.
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for any transport/stream/encoding error.
    pub async fn run_routine_turn(&self, ingress: TurnIngress) -> Result<BufferedTurn, DialError> {
        let received = self.receive_ingress(ingress).await?;
        self.attach_ingress_buffered(&received).await
    }

    /// Attach to one durable receipt and fold the
    /// response envelopes into a [`BufferedTurn`] — the single transport body
    /// [`run_turn_with`](Self::run_turn_with) and
    /// [`run_turn_with_approvals`](Self::run_turn_with_approvals) share, so
    /// the two buffered call sites can't drift on what a terminal `AgentEnd`
    /// means.
    async fn run_turn_buffered_received(
        &self,
        received: &DurablyReceivedTurn,
    ) -> Result<BufferedTurn, DialError> {
        let mut stream = self
            .client
            .attach_ingress_with_options(
                AttachIngressRequest {
                    receipt: buffa::MessageField::some(received.receipt.clone()),
                    attach_attempt: received.attach_attempt.clone(),
                    ..Default::default()
                },
                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();
        // A paid call this turn needed a linked wallet (`#519`): absent by
        // default, `Present(url)` once `AgentEnd` carries the signal.
        let mut wallet_link_prompt = WalletLinkPrompt::None;
        // `wallet_update_limit` minted a fresh spending-limit-update
        // ceremony this turn (issue #1041/#1159): `None` by default,
        // `Some((update_url, new_limit))` once `AgentEnd` carries the
        // signal.
        let mut wallet_update_prompt: Option<(String, String)> = None;
        // This turn's `unlink_self` (wallet target) call minted a
        // hard-revoke ceremony (issue #1042/#1156): `None` by default,
        // `Some(revoke_url)` once `AgentEnd` carries the signal.
        let mut wallet_revoke_prompt: Option<String> = None;
        // Gated calls the turn paused on (`AgentEnd.pending_approvals`) — the
        // buffered analog of the streaming path's `TurnEvent::ApprovalPending`
        // (`events_from_end`). Empty unless the turn paused.
        let mut pending_approvals: Vec<PendingApprovalPrompt> = Vec::new();
        // `#1660`: the question-pause SIBLING of `pending_approvals` above —
        // `AgentEnd.pending_questions`, the buffered analog of
        // `TurnEvent::QuestionPending`.
        let mut pending_questions: Vec<PendingQuestionPrompt> = 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 {
                        aggregate_output_message(msg, &mut text_parts, &mut scaffolding);
                    }
                }
                // `End` carries the turn's answer already folded into
                // `text_parts`/`scaffolding` above; `wallet_link_prompt` and
                // `pending_approvals` mirror the streaming path's identical
                // projection (`events_from_end`) so the buffered and
                // streaming APIs cannot drift on either signal. A pre-turn
                // compaction notice carries no answer text; an empty
                // envelope is defensive (the wire allows it). Deliberately
                // NOT a `break` on `End`: returning there would close our
                // half of the stream while the transport may still have
                // frames (or only its trailer) in flight, which registers as
                // a client-cancelled call to anything that inspects gRPC
                // status on this path. Keep reading — mirrors the
                // drain-to-EOF shape in `harness_dialer::run_turn_streamed`
                // — until `stream.message()` itself returns `None` at
                // transport EOF.
                Some(agent_response::Type::End(end)) => {
                    if let Some(prompt) = end.wallet_link_prompt.into_option() {
                        let renewal = prompt.renewal;
                        let requested = prompt.requested;
                        let link_url =
                            (!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
                        wallet_link_prompt = WalletLinkPrompt::Present {
                            link_url,
                            renewal,
                            requested,
                        };
                    }
                    if let Some(prompt) = end.wallet_update_prompt.into_option() {
                        wallet_update_prompt = Some((prompt.update_url, prompt.new_limit));
                    }
                    if let Some(prompt) = end.wallet_revoke_prompt.into_option() {
                        wallet_revoke_prompt = Some(prompt.revoke_url);
                    }
                    pending_approvals = end
                        .pending_approvals
                        .into_iter()
                        .map(PendingApprovalPrompt::from)
                        .collect();
                    pending_questions = end
                        .pending_questions
                        .into_iter()
                        .map(PendingQuestionPrompt::from)
                        .collect();
                }
                Some(agent_response::Type::Compacted(_)) | None => {}
            }
        }
        Ok(BufferedTurn {
            reply: finalize_buffered_reply(
                &text_parts,
                &scaffolding,
                wallet_link_prompt,
                wallet_update_prompt
                    .as_ref()
                    .map(|(url, limit)| (url.as_str(), limit.as_str())),
                wallet_revoke_prompt.as_deref(),
            ),
            pending_approvals,
            pending_questions,
        })
    }

    /// 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,
        source_identity: IngressIdentity,
        namespace: ClaimedNamespace,
        user_text: &str,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
        self.run_turn_streaming_messages(
            conversation_id,
            exec_id,
            source_identity,
            namespace,
            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,
        source_identity: IngressIdentity,
        namespace: ClaimedNamespace,
        messages: Vec<Message>,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
        self.run_turn_streaming_ingress(TurnIngress::new(
            conversation_id,
            exec_id,
            source_identity,
            namespace,
            messages,
        ))
        .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),
    /// the caller [`Attribution`] (resolved to durable personas and recorded
    /// as `caller`/`participant` events), and the edge's own
    /// [`IngressDirective`] (`#68`) — a step-budget cap, an advisory
    /// priority, and/or a required approver. One method rather than a matrix
    /// of variants, so a paid *and* attributed *and* directed edge can't
    /// silently drop one of the three. An edge with no ingress policy passes
    /// [`IngressDirective::default`].
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for any transport/stream/encoding error.
    #[allow(clippy::too_many_arguments)] // compatibility surface gained required source identity
    pub async fn run_turn_streaming_messages_with(
        &self,
        conversation_id: &str,
        exec_id: &str,
        source_identity: IngressIdentity,
        namespace: ClaimedNamespace,
        messages: Vec<Message>,
        payment_receipt: Option<PaymentReceipt>,
        attribution: Attribution,
        ingress_directive: IngressDirective,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
        let mut ingress = TurnIngress::new(
            conversation_id,
            exec_id,
            source_identity,
            namespace,
            messages,
        )
        .with_attribution(attribution)
        .with_ingress_directive(ingress_directive);
        if let Some(receipt) = payment_receipt {
            ingress = ingress.with_payment_receipt(receipt);
        }
        self.run_turn_streaming_ingress(ingress).await
    }

    /// Runs a fully assembled durable ingress turn.
    ///
    /// Prefer this variant when optional turn fields are naturally composed
    /// through [`TurnIngress`]'s builders.
    ///
    /// # Errors
    ///
    /// Returns [`DialError::Connect`] for any transport/stream/encoding error.
    pub async fn run_turn_streaming_ingress(
        &self,
        ingress: TurnIngress,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
        let received = self.receive_ingress(ingress).await?;
        self.run_turn_streaming_received(&received).await
    }

    /// Attach to one durable receipt and project the
    /// response envelopes into [`TurnEvent`]s — the single transport body
    /// every streaming variant shares.
    async fn run_turn_streaming_received(
        &self,
        received: &DurablyReceivedTurn,
    ) -> Result<impl Stream<Item = Result<TurnEvent, DialError>> + use<>, DialError> {
        let mut stream = self
            .client
            .attach_ingress_with_options(
                AttachIngressRequest {
                    receipt: buffa::MessageField::some(received.receipt.clone()),
                    attach_attempt: received.attach_attempt.clone(),
                    ..Default::default()
                },
                traced_options(),
            )
            .await?;
        Ok(async_stream::try_stream! {
            // Set once an explicit `AgentEnd` has been seen and its events
            // (including the terminal `Done`) have been yielded, so the
            // fallback below doesn't yield a second `Done`.
            let mut ended = false;
            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;
                        }
                        ended = true;
                        // Deliberately NOT a `return`: returning here would
                        // drop our half of the stream while the transport may
                        // still have frames (or only its trailer) in flight,
                        // which registers as a client-cancelled call to
                        // anything that inspects gRPC status on this path.
                        // Keep reading — mirrors the drain-to-EOF shape in
                        // `harness_dialer::run_turn_streamed` — until
                        // `stream.message()` itself returns `None` at
                        // transport EOF.
                    }
                    // 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.
            if !ended {
                yield TurnEvent::Done;
            }
        })
    }

    /// Ask the control plane's participation gate whether to reply to the
    /// latest message of a (multi-party) thread. Returns `true` on `respond`
    /// and `false` on `ignore`. The gate runs a cheap classifier model
    /// server-side and never runs a turn.
    ///
    /// `surface` names the calling edge's surface (for example `"Slack"`);
    /// it is rendered into the classifier prompt so the gate reads the
    /// thread in its real setting. Empty keeps the prompt surface-neutral.
    ///
    /// `source_identity` names the source event this evaluation belongs to. It
    /// travels inside the signed envelope, which is what lets the control
    /// plane derive a model tenant: the gate spends against one, so it
    /// authenticates its caller exactly as a turn does.
    ///
    /// # Errors
    ///
    /// Returns [`DialError::MissingIngressCredentials`] when this client holds
    /// no edge credential to sign with, and [`DialError::Connect`] for any
    /// transport/encoding error.
    pub async fn should_respond(
        &self,
        conversation_id: &str,
        bot_name: &str,
        surface: &str,
        transcript: Vec<ParticipantMessage>,
        source_identity: &IngressIdentity,
        namespace: &ClaimedNamespace,
    ) -> Result<bool, DialError> {
        let credentials = self
            .credentials
            .as_ref()
            .ok_or(DialError::MissingIngressCredentials)?;
        let asserted_attribution = build_classify_attribution(
            credentials,
            conversation_id,
            source_identity,
            namespace,
            &transcript,
            bot_name,
            surface,
        );
        let request = ClassifyRequest {
            conversation_id: conversation_id.to_owned(),
            bot_name: bot_name.to_owned(),
            surface: surface.to_owned(),
            transcript,
            asserted_attribution: buffa::MessageField::some(asserted_attribution),
            ..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 one line of copy shown once an approved call's re-drive has actually
/// executed (`#743`).
///
/// The sibling of [`approval_decided_text`] one step later in the lifecycle:
/// that function's "running…" line covers the gap between the decision and
/// execution; this replaces it once the runtime knows the outcome, so the
/// card never sits on a stale "running…" after the tool has already
/// finished. `label` is the tool's friendly display name (title, or a
/// `humanize_tool_name` fallback), `decider` is who approved it, and
/// `success` distinguishes a clean run from one that errored — the runtime,
/// not the model, owns this line: it is built from the actual dispatch
/// outcome, never from anything the model said.
#[must_use]
pub fn approval_completed_text(label: &str, decider: &str, success: bool) -> String {
    if success {
        format!("✅ Approved by {decider} — \"{label}\" is done.")
    } else {
        format!("✅ Approved by {decider} — \"{label}\" ran but hit an error.")
    }
}

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

impl From<polyc_proto::proto::polychrome::approval::v1::ApprovalResponseReply> for ApprovalOutcome {
    fn from(reply: polyc_proto::proto::polychrome::approval::v1::ApprovalResponseReply) -> Self {
        Self {
            persisted: reply.persisted,
            signature_hex: reply.signature_hex,
            signed_by_hex: reply.signed_by_hex,
        }
    }
}

/// 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 {
    /// Turn that emitted this occurrence of `request_id`.
    pub turn_id: String,
    /// The id to answer via [`ApprovalDialer::respond`].
    pub request_id: String,
    /// The raw tool name the approval gates.
    pub tool_name: String,
    /// Curated title preserved from the durable approval request. Empty only
    /// for a historical request that predates the field.
    pub title: 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,
    /// Short-lived signed capability (`#787`), freshly minted for THIS
    /// listing and scoped to `turn_id` + `request_id` + the conversation it
    /// belongs to.
    /// Carry it back unmodified to [`ApprovalDialer::respond`].
    pub resolve_token: String,
    /// Computed-preview enrichment (`#1496`), read back from the durable
    /// `approval_request` event so a recovered card renders the exact
    /// preview the live one showed. `None` for every call but
    /// `routine_create`.
    pub preview: Option<ApprovalPreview>,
    /// Trusted-side preview for a read that crosses conversations.
    pub read_preview: Option<ReadPreview>,
    /// Whether this still-unanswered approval was already surfaced once (a
    /// later turn's boundary marker sits after its `approval_request` in the
    /// event log while it stayed unanswered). `false`: render the full
    /// interactive card; `true`: render a compact reminder instead (#1970).
    /// Derived by the control plane at listing time — never edge-cached.
    pub already_surfaced: bool,
}

/// Converts every durable pending-approval field to the edge-owned model.
impl TryFrom<PendingApprovalEntry> for PendingApproval {
    type Error = DialError;

    fn try_from(p: PendingApprovalEntry) -> Result<Self, Self::Error> {
        let PendingApprovalEntry {
            request_id,
            tool_name,
            args_json,
            reason,
            resolve_token,
            preview,
            read_preview,
            already_surfaced,
            turn_id,
            title,
            __buffa_unknown_fields: _,
        } = p;
        Ok(Self {
            turn_id,
            request_id,
            tool_name,
            title,
            args_json,
            reason,
            resolve_token,
            preview: preview.into_option().map(Into::into),
            read_preview: read_preview
                .into_option()
                .map(ReadPreview::try_from)
                .transpose()?,
            already_surfaced,
        })
    }
}

/// A decision durably recorded but not yet consumed by tool execution.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RecordedApprovalRecovery {
    /// Whether the saved decision approved the call.
    pub approved: bool,
}

/// The recovery state for an approval occurrence.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ApprovalRecoveryState {
    /// A verified decision is ready for a resume turn.
    DecisionRecorded(RecordedApprovalRecovery),
    /// The prior response was unverifiable, so the durable turn will re-prompt.
    RepromptRequired,
}

/// One durable approval occurrence that needs a resume turn.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RecoverableApproval {
    /// Turn that emitted this occurrence.
    pub turn_id: String,
    /// Request identifier within [`Self::turn_id`].
    pub request_id: String,
    /// Raw tool name.
    pub tool_name: String,
    /// Curated durable title, or empty for historic records.
    pub title: String,
    /// Full tool-call arguments JSON.
    pub args_json: String,
    /// Durable gating reason.
    pub reason: String,
    /// Routine preview, when the durable request carried one.
    pub preview: Option<ApprovalPreview>,
    /// Trusted-side preview for a read that crosses conversations.
    pub read_preview: Option<ReadPreview>,
    /// Exactly one recovery state.
    pub state: ApprovalRecoveryState,
}

impl TryFrom<RecoverableApprovalEntry> for RecoverableApproval {
    type Error = DialError;

    fn try_from(entry: RecoverableApprovalEntry) -> Result<Self, Self::Error> {
        let RecoverableApprovalEntry {
            turn_id,
            request_id,
            tool_name,
            title,
            args_json,
            reason,
            preview,
            read_preview,
            state,
            __buffa_unknown_fields: _,
        } = entry;
        let state = match state {
            Some(recoverable_approval_entry::State::DecisionRecorded(decision)) => {
                let WireRecordedApprovalDecision {
                    approved,
                    __buffa_unknown_fields: _,
                } = *decision;
                ApprovalRecoveryState::DecisionRecorded(RecordedApprovalRecovery { approved })
            }
            Some(recoverable_approval_entry::State::RepromptRequired(marker)) => {
                let WireRepromptRequired {
                    __buffa_unknown_fields: _,
                } = *marker;
                ApprovalRecoveryState::RepromptRequired
            }
            None => {
                return Err(DialError::InvalidApprovalRecovery(
                    "approval recovery entry omitted its state",
                ));
            }
        };
        Ok(Self {
            turn_id,
            request_id,
            tool_name,
            title,
            args_json,
            reason,
            preview: preview.into_option().map(Into::into),
            read_preview: read_preview
                .into_option()
                .map(ReadPreview::try_from)
                .transpose()?,
            state,
        })
    }
}

/// Trusted-side preview for a read that crosses conversations.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReadPreview {
    /// A canonical search query and its resolved conversation count.
    Search {
        /// Query text that will actually execute.
        canonical_query: String,
        /// Number of conversations in the resolved scope.
        scope_count: u32,
    },
    /// One bounded read of a search hit.
    Hit {
        /// Query that produced the hit.
        origin_query: String,
        /// Human-readable conversation label.
        conversation_label: String,
        /// Maximum returned text size in bytes.
        max_bytes: u32,
    },
}

impl TryFrom<WireListReadPreview> for ReadPreview {
    type Error = DialError;

    fn try_from(preview: WireListReadPreview) -> Result<Self, Self::Error> {
        let WireListReadPreview {
            detail,
            __buffa_unknown_fields: _,
        } = preview;
        match detail {
            Some(wire_list_read_preview::Detail::Search(search)) => {
                let WireSearchPreview {
                    canonical_query,
                    scope_count,
                    __buffa_unknown_fields: _,
                } = *search;
                Ok(Self::Search {
                    canonical_query,
                    scope_count,
                })
            }
            Some(wire_list_read_preview::Detail::Hit(hit)) => {
                let WireSearchHitPreview {
                    origin_query,
                    conversation_label,
                    max_bytes,
                    __buffa_unknown_fields: _,
                } = *hit;
                Ok(Self::Hit {
                    origin_query,
                    conversation_label,
                    max_bytes,
                })
            }
            None => Err(DialError::InvalidApprovalRecovery(
                "approval read preview omitted its detail",
            )),
        }
    }
}

/// One fire instant rendered for a human in [`ApprovalPreview`]: the same
/// real instant twice, RFC3339 in [`ApprovalPreview::zone_name`] and in UTC.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApprovalPreviewFire {
    /// RFC3339 instant in [`ApprovalPreview::zone_name`]'s local wall clock.
    pub local_time: String,
    /// The identical instant, RFC3339 in UTC. Carried for the audit path;
    /// [`approval_preview_text`] does not render it (`#1808`).
    pub utc_time: String,
    /// How this instant's zone reads to a human — `PDT (UTC-7)` — or EMPTY
    /// when [`ApprovalPreview::cadence`] already states it for every run.
    /// The control plane decides that placement (it holds both the schedule
    /// and the tz database); rendering it whenever it is non-empty is the
    /// whole of this side's job (`#1808`).
    pub zone_label: String,
}

/// The approval card's spec-computed preview (`#1496`).
///
/// The next fire times a routine will run at, plus the exact prompt text that
/// will execute — built by the control plane from the compiled spec, never
/// from the model's narration (INV-RL3). Carried on both
/// [`TurnEvent::ApprovalPending`] (the live card) and [`PendingApproval`]
/// (the `ListPending` recovery path) so neither surface can drop it. Render
/// with [`approval_preview_text`] — never a hand-written per-edge rendering,
/// so the four chat edges cannot word the same state differently.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ApprovalPreview {
    /// The exact prompt text the routine will run, copied verbatim from the
    /// compiled spec — never re-typed or summarized.
    pub prompt_text: String,
    /// Up to the next three fire instants, oldest first. Fewer than three
    /// (or none) when the schedule itself has fewer left.
    pub next_fires: Vec<ApprovalPreviewFire>,
    /// IANA name of the zone `next_fires[].local_time` is rendered in.
    pub zone_name: String,
    /// `true` when [`Self::zone_name`] is the schedule's OWN zone shown as a
    /// labeled fallback because the creating admin's zone could not be
    /// resolved — `false` when it is the admin's actual zone.
    pub zone_is_fallback: bool,
    /// The schedule in English — `every weekday at 9:00 AM PDT (UTC-7)` —
    /// EMPTY when it is not confidently describable, in which case the card
    /// omits its cadence line rather than assert a wrong sentence (`#1808`).
    pub cadence: String,
    /// Routine resource name for a delete preview; empty for create.
    pub routine_name: String,
}

/// Names every field so a field added on the agent-service wire fails to
/// compile here instead of silently defaulting (see CLAUDE.md's cross-type
/// conversion rule and #1241/#1238).
impl From<WireAgentApprovalPreviewFire> for ApprovalPreviewFire {
    fn from(f: WireAgentApprovalPreviewFire) -> Self {
        Self {
            local_time: f.local_time,
            utc_time: f.utc_time,
            zone_label: f.zone_label,
        }
    }
}

/// Names every field so a field added on the approval-service wire fails to
/// compile here instead of silently defaulting (see CLAUDE.md's cross-type
/// conversion rule and #1241/#1238).
impl From<WireListApprovalPreviewFire> for ApprovalPreviewFire {
    fn from(f: WireListApprovalPreviewFire) -> Self {
        Self {
            local_time: f.local_time,
            utc_time: f.utc_time,
            zone_label: f.zone_label,
        }
    }
}

impl From<WireAgentApprovalPreview> for ApprovalPreview {
    fn from(p: WireAgentApprovalPreview) -> Self {
        let WireAgentApprovalPreview {
            prompt_text,
            next_fires,
            zone_name,
            zone_is_fallback,
            cadence,
            routine_name,
            __buffa_unknown_fields: _,
        } = p;
        Self {
            prompt_text,
            next_fires: next_fires.into_iter().map(Into::into).collect(),
            zone_name,
            zone_is_fallback,
            cadence,
            routine_name,
        }
    }
}

impl From<WireListApprovalPreview> for ApprovalPreview {
    fn from(p: WireListApprovalPreview) -> Self {
        let WireListApprovalPreview {
            prompt_text,
            next_fires,
            zone_name,
            zone_is_fallback,
            cadence,
            routine_name,
            __buffa_unknown_fields: _,
        } = p;
        Self {
            prompt_text,
            next_fires: next_fires.into_iter().map(Into::into).collect(),
            zone_name,
            zone_is_fallback,
            cadence,
            routine_name,
        }
    }
}

/// Render an [`ApprovalPreview`] as the text block every chat edge appends to
/// an approval card (`#1496`, reformatted in `#1808`).
///
/// The ONE shared helper, so the four edges cannot word the same state
/// differently (this repo's house rule; see
/// `humanize_tool_name`/`tool_status_text` for the same pattern elsewhere).
/// Sentence case, no jargon, no destination line (the pivot dropped it — see
/// `docs/specifications/invariants-routine-lifecycle.md`).
///
/// The shape a reader gets, with the run table wrapped in a code fence:
///
/// ```text
/// This routine will run:
///
/// Post the daily standup summary.
///
/// Runs every weekday at 9:00 AM PDT (UTC-7)
///
/// Next 3 runs
///   Wed Jul 29    9:00 AM    tomorrow
///   Thu Jul 30    9:00 AM    in 2 days
///   Fri Jul 31    9:00 AM    in 3 days
/// ```
///
/// Three decisions worth knowing:
///
/// - **The cadence line states what the list cannot.** Three instants make a
///   reader infer "weekdays at 9" by diffing dates, and the recurrence is the
///   fact actually being approved. It is omitted entirely when the control
///   plane could not describe the schedule confidently — never guessed at.
/// - **The title counts the runs it actually has** ("Next 3 runs", "Next
///   run"), because a schedule can yield fewer than three.
/// - **The rows sit in a fenced block** so their columns align on a surface
///   that renders proportional text. The chat edges render fences natively;
///   the email edge strips the markers, which is markup, not wording.
///
/// `now` is caller-injected rather than read from the clock, matching this
/// workspace's other pure renderers. It anchors each run relatively
/// ("tomorrow", "in 2 days"), which matters on the `ListPending` recovery
/// path: that card is rebuilt from a preview frozen in the durable
/// `approval_request` event, so a stale "tomorrow" beside an absolute date
/// would mislead in a way a merely old date does not. A run already behind
/// `now` reads `passed`.
///
/// The zone-fallback disclosure is spelled out whenever
/// `preview.zone_is_fallback` is set, and keeps the IANA zone name: an
/// abbreviation cannot tell a reader WHICH zone got guessed, which is exactly
/// what they are checking there.
#[must_use]
pub fn approval_preview_text(preview: &ApprovalPreview, now: DateTime<Utc>) -> String {
    let mut lines = vec![
        "This routine will run:".to_owned(),
        String::new(),
        preview.prompt_text.clone(),
    ];
    if !preview.cadence.is_empty() {
        lines.push(String::new());
        lines.push(format!("Runs {}", preview.cadence));
    }
    lines.push(String::new());
    if preview.next_fires.is_empty() {
        lines.push(
            "This schedule has no upcoming runs, so approving it won't run anything.".to_owned(),
        );
    } else {
        lines.push(run_list_title(preview.next_fires.len()));
        lines.push(FENCE.to_owned());
        lines.extend(run_rows(&preview.next_fires, now));
        lines.push(FENCE.to_owned());
    }
    if preview.zone_is_fallback {
        lines.push(String::new());
        lines.push(format!(
            "We don't know your time zone yet, so these times are in the routine's own zone ({}).",
            preview.zone_name
        ));
    }
    lines.join("\n")
}

/// Render the shared approval preview without chat-markup fences.
///
/// Browser cards already place this text in a monospace block, so exposing the
/// literal Markdown fence would show implementation markup to the reader. The
/// wording and aligned rows remain shared with [`approval_preview_text`].
#[must_use]
pub fn approval_preview_plain_text(preview: &ApprovalPreview, now: DateTime<Utc>) -> String {
    approval_preview_text(preview, now)
        .lines()
        .filter(|line| *line != FENCE)
        .collect::<Vec<_>>()
        .join("\n")
}

/// The fence that wraps the run table. The chat edges that render fences
/// natively show it as a monospace block directly, and the chat edge that
/// speaks HTML converts it to `<pre>` on the way out — so one shared string
/// aligns on all of them.
const FENCE: &str = "```";

/// Title the run list by the number of runs it actually holds.
///
/// A schedule can yield fewer than three (a `once` schedule yields one; an
/// exhausted one yields none), so the count is derived, never assumed —
/// "Next 3 runs" beside two rows would misstate the schedule.
fn run_list_title(count: usize) -> String {
    if count == 1 {
        "Next run".to_owned()
    } else {
        format!("Next {count} runs")
    }
}

/// One run's three rendered cells: the day, the clock time (carrying the
/// zone label when this run owns it), and how far off it is.
struct RunRow {
    day: String,
    time: String,
    relative: String,
}

/// Lay the runs out as aligned `day / time / how-far-off` columns.
///
/// Column widths come from the widest cell present, so the table stays snug
/// whether or not the runs carry their own zone labels.
fn run_rows(fires: &[ApprovalPreviewFire], now: DateTime<Utc>) -> Vec<String> {
    let rows: Vec<RunRow> = fires.iter().map(|fire| run_cells(fire, now)).collect();
    let day_width = rows
        .iter()
        .map(|row| row.day.chars().count())
        .max()
        .unwrap_or(0);
    let time_width = rows
        .iter()
        .map(|row| row.time.chars().count())
        .max()
        .unwrap_or(0);
    rows.iter()
        .map(|row| {
            format!(
                "  {:day_width$}    {:time_width$}    {}",
                row.day, row.time, row.relative
            )
        })
        .collect()
}

/// One run's cells: the day, the clock time (carrying the zone label when
/// this run owns it), and how far off it is.
///
/// A `local_time` that does not parse falls back to the raw instant in the
/// day cell rather than dropping the run — a card that silently lists two of
/// three runs would understate the schedule.
fn run_cells(fire: &ApprovalPreviewFire, now: DateTime<Utc>) -> RunRow {
    let Ok(local) = DateTime::parse_from_rfc3339(&fire.local_time) else {
        return RunRow {
            day: fire.local_time.clone(),
            time: String::new(),
            relative: String::new(),
        };
    };
    // Compare in the run's own offset so "today" means the reader's calendar
    // day, not UTC's — those disagree for most of the world for part of each
    // day, and disagreeing here is exactly the mistake this column exists to
    // prevent.
    let now_local = now.with_timezone(local.offset());
    let day = if local.year() == now_local.year() {
        local.format("%a %b %e").to_string()
    } else {
        local.format("%a %b %e, %Y").to_string()
    };
    let mut time = local.format("%-I:%M %p").to_string();
    if !fire.zone_label.is_empty() {
        time.push(' ');
        time.push_str(&fire.zone_label);
    }
    RunRow {
        day,
        time,
        relative: relative_day(local, now_local),
    }
}

/// How far off a run is, in the reader's own calendar days.
fn relative_day(local: DateTime<FixedOffset>, now_local: DateTime<FixedOffset>) -> String {
    if local < now_local {
        return "passed".to_owned();
    }
    match (local.date_naive() - now_local.date_naive()).num_days() {
        0 => "today".to_owned(),
        1 => "tomorrow".to_owned(),
        days => format!("in {days} days"),
    }
}

/// 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>>,
    /// Set only by [`Self::with_credentials`]. `None` (the [`Self::new`] and
    /// [`Self::with_bearer`] paths) means [`Self::respond`] asserts no
    /// responder — there is no key to sign one with, and an unprovable
    /// responder is treated as an absent one (`#1553`).
    credentials: Option<Arc<EdgeCredentials>>,
}

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> {
        Ok(Self {
            client: build_control_client(addr, None, ApprovalServiceClient::new)?,
            credentials: None,
        })
    }

    /// Build a dialer pointed at `addr`, authenticated with `bearer`.
    ///
    /// Every call this dialer makes carries an `Authorization: Bearer
    /// <bearer>` header. `ApprovalService` doesn't send `AgentStart`, so no
    /// signed [`AssertedAttribution`] envelope rides these calls — bearer-only.
    /// A bearer alone cannot assert who resolved an approval; a dialer that
    /// also holds its edge's signing key uses [`Self::with_credentials`].
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value.
    pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            client: build_control_client(addr, Some(bearer), ApprovalServiceClient::new)?,
            credentials: None,
        })
    }

    /// Build a dialer pointed at `addr`, authenticated with `credentials`.
    ///
    /// Rides `credentials.bearer()` as the `Authorization` header, exactly
    /// like [`Self::with_bearer`], and additionally signs an
    /// `AssertedApproval` onto every [`Self::respond`] that names a responder
    /// (`#1553`) — one implementation every edge inherits, rather than one
    /// per surface.
    ///
    /// Takes an `Arc` because these are the SAME credentials the edge's
    /// [`AgentDialer`] holds: one edge, one identity key, whether it is
    /// dispatching a turn or asserting who approved one.
    /// [`AgentDialer::approval_dialer_with_credentials`] is the usual way to
    /// build this.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::InvalidBearer`] if the bearer can't be encoded as an HTTP
    /// header value.
    pub fn with_credentials(
        addr: &str,
        credentials: Arc<EdgeCredentials>,
    ) -> Result<Self, DialError> {
        let client =
            build_control_client(addr, Some(credentials.bearer()), ApprovalServiceClient::new)?;
        Ok(Self {
            client,
            credentials: Some(credentials),
        })
    }

    /// 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 (`turn_id`, `request_id`) occurrence 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.
    ///
    /// `resolve_token` is the short-lived signed capability (`#787`) carried
    /// unmodified off the [`TurnEvent::ApprovalPending`] event or
    /// [`PendingApproval`] entry this decision answers — required: the
    /// control plane rejects a `Respond` whose token is missing, expired, or
    /// bound to a different turn, request, or conversation.
    ///
    /// `responder` is the identity of the human answering (`#68`), as this
    /// edge's own provider verified it. A dialer built by
    /// [`Self::with_credentials`] signs it into the request as an
    /// `AssertedApproval` (`#1553`); the control plane enforces a turn's
    /// required approver (`IngressDirective.required_approver`) against the
    /// verified identity alone, so Approve of such a turn is refused unless
    /// the assertion verifies AND matches.
    ///
    /// Asserting nothing is a supported state, not an error: an edge that
    /// cannot attribute the responder passes `None`, a machine decision
    /// carries none, and a dialer without credentials has no key to sign one
    /// with. Deny and Defer are unaffected either way — refusing is safe
    /// regardless of who refuses — and so is Approve of a turn that names no
    /// required approver.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error
    /// (including the control plane's rejection of an invalid
    /// `resolve_token`, or a `permission_denied` when the verified responder
    /// doesn't match the turn's required approver).
    #[allow(clippy::too_many_arguments)] // each is a distinct field of the decision
    pub async fn respond(
        &self,
        turn_id: &str,
        request_id: &str,
        choice: ApprovalChoice,
        reason: &str,
        conversation_id: &str,
        modified_args_json: &str,
        injected_context: &str,
        resolve_token: &str,
        responder: Option<ExternalIdentity>,
    ) -> Result<ApprovalOutcome, DialError> {
        let request = self.respond_request(
            turn_id,
            request_id,
            choice,
            reason,
            conversation_id,
            modified_args_json,
            injected_context,
            resolve_token,
            responder,
        );
        let reply = self
            .client
            .respond_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(reply.into())
    }

    /// Build the [`ApprovalResponseRequest`] one [`Self::respond`] call sends,
    /// signing the responder into it when there is a responder to assert and a
    /// credential to assert it with (`#1553`).
    ///
    /// The assertion is attached LAST, and that ordering is load-bearing: its
    /// signature covers the whole request — the decision, an approve's
    /// modified arguments and injected context, the occurrence identity, and
    /// the `resolve_token` — so any field set after the attach would fall
    /// outside the signature it was supposed to be covered by, and the control
    /// plane would reject the assertion as unverifiable. Everything else about
    /// the request is therefore final before the last statement runs.
    ///
    /// A dialer built by [`Self::new`] or [`Self::with_bearer`] holds no
    /// credentials, and `responder` is `None` for an edge that cannot
    /// attribute the human (or for a machine decision). Either way the request
    /// carries no assertion, which the control plane treats exactly as it
    /// treats one that fails to verify — see
    /// `docs/reference/verified-approver.md`, decision 4.
    // One builder carrying the RPC's full field set, mirroring `build_request`
    // for a turn; each argument is a distinct field of the decision, and
    // routing them through a params struct would only add one indirection over
    // the same fields.
    #[allow(clippy::too_many_arguments)]
    fn respond_request(
        &self,
        turn_id: &str,
        request_id: &str,
        choice: ApprovalChoice,
        reason: &str,
        conversation_id: &str,
        modified_args_json: &str,
        injected_context: &str,
        resolve_token: &str,
        responder: Option<ExternalIdentity>,
    ) -> ApprovalResponseRequest {
        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(),
                __buffa_unknown_fields: buffa::UnknownFields::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(),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            }))
        } else {
            Decision::Deny(Box::new(Deny {
                reason: reason.to_owned(),
                abort: choice.is_abort(),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            }))
        };
        let mut request = ApprovalResponseRequest {
            turn_id: turn_id.to_owned(),
            request_id: request_id.to_owned(),
            conversation_id: conversation_id.to_owned(),
            decision: Some(decision),
            resolve_token: resolve_token.to_owned(),
            asserted_approval: buffa::MessageField::none(),
            ..Default::default()
        };
        // Last, over the finished request.
        if let (Some(responder), Some(creds)) = (responder, self.credentials.as_deref()) {
            creds.attach_approval_assertion(&mut request, responder);
        }
        request
    }

    /// 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 mut pending = Vec::new();
        let mut page_token = String::new();
        loop {
            let request = ListPendingRequest {
                conversation_id: conversation_id.to_owned(),
                page_token: page_token.clone(),
                ..Default::default()
            };
            let reply = self
                .client
                .list_pending_with_options(request, traced_options())
                .await?
                .into_owned();
            pending.extend(
                reply
                    .pending
                    .into_iter()
                    .map(PendingApproval::try_from)
                    .collect::<Result<Vec<_>, _>>()?,
            );
            if reply.next_page_token.is_empty() {
                break;
            }
            page_token = reply.next_page_token;
        }
        Ok(pending)
    }

    /// List saved approval decisions that need their paused turn resumed.
    ///
    /// Unlike [`Self::list_pending`], these entries intentionally carry no
    /// resolve capability: a decision has already been recorded, or the
    /// control plane must re-prompt it durably. A malformed repeated page token
    /// is refused rather than allowing a caller's recovery loop to spin.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for transport failures and
    /// [`DialError::InvalidApprovalRecovery`] when pagination repeats a token.
    pub async fn list_recovery(
        &self,
        conversation_id: &str,
    ) -> Result<Vec<RecoverableApproval>, DialError> {
        let mut recovery = Vec::new();
        let mut page_token = String::new();
        let mut seen_page_tokens = std::collections::HashSet::new();
        loop {
            let request = ListApprovalRecoveryRequest {
                conversation_id: conversation_id.to_owned(),
                page_size: 0,
                page_token: page_token.clone(),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            };
            let reply = self
                .client
                .list_recovery_with_options(request, traced_options())
                .await?
                .into_owned();
            recovery.extend(
                reply
                    .recovery
                    .into_iter()
                    .map(RecoverableApproval::try_from)
                    .collect::<Result<Vec<_>, _>>()?,
            );
            if reply.next_page_token.is_empty() {
                break;
            }
            if !seen_page_tokens.insert(reply.next_page_token.clone()) {
                return Err(DialError::InvalidApprovalRecovery(
                    "recovery listing repeated a continuation token",
                ));
            }
            page_token = reply.next_page_token;
        }
        Ok(recovery)
    }

    /// 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(reply.into())
    }

    /// Replay a recorded conversation against its committed event log (`#690`):
    /// re-execute its committed turns in process — no live model, no live tools,
    /// nothing sent or spent again — and return a per-turn verdict. Read-only:
    /// the control plane appends nothing and takes no writer lease, so it is safe
    /// to run on production data any number of times.
    ///
    /// Admin-gated (`#694`): the control plane verifies `actor` holds the admin
    /// role — the same gate `excise_taint` applies — before it reads or replays
    /// anything, so a non-admin (or absent) actor is refused.
    ///
    /// `from`/`to` bound the 0-based committed-turn ordinals (inclusive); `None`
    /// means the ends. A set `over` selects what-if mode (fork one recorded step
    /// and report where the conversation diverges); `None` is a plain verify
    /// (assert every turn reproduces its record).
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error, or a
    /// permission refusal when `actor` is not an admin.
    pub async fn replay_conversation(
        &self,
        conversation_id: &str,
        from: Option<usize>,
        to: Option<usize>,
        over: Option<ReplayOverrideSpec>,
        actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
    ) -> Result<ReplayReport, DialError> {
        use polyc_proto::proto::polychrome::approval::v1::{
            ReplayConversationRequest, ReplayOverride,
        };
        let override_msg = over.map(ReplayOverride::from);
        // An absent bound is the `-1` sentinel; the control plane maps a negative
        // ordinal to "the ends".
        let to_bound = |v: Option<usize>| v.and_then(|n| i64::try_from(n).ok()).unwrap_or(-1);
        let request = ReplayConversationRequest {
            conversation_id: conversation_id.to_owned(),
            from: to_bound(from),
            to: to_bound(to),
            r#override: override_msg
                .map_or_else(buffa::MessageField::none, buffa::MessageField::some),
            actor: buffa::MessageField::some(actor),
            ..Default::default()
        };
        let reply = self
            .client
            .replay_conversation_with_options(request, traced_options())
            .await?
            .into_owned();
        let turns = reply
            .turns
            .into_iter()
            .map(ReplayTurnVerdict::from)
            .collect();
        Ok(ReplayReport {
            turns,
            all_match: reply.all_match,
        })
    }

    /// Verify a conversation's tamper-evidence (`#799`): replay its event log
    /// and check every signed MMR root recorded along the way against a
    /// freshly rebuilt tree. Read-only — appends nothing.
    ///
    /// Admin-gated server-side, the same gate `excise_taint` uses.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error, or a
    /// permission refusal when `actor` is not an admin.
    pub async fn verify_conversation(
        &self,
        conversation_id: &str,
        actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
    ) -> Result<VerificationOutcome, DialError> {
        use polyc_proto::proto::polychrome::approval::v1::VerifyConversationRequest;
        let request = VerifyConversationRequest {
            conversation_id: conversation_id.to_owned(),
            actor: buffa::MessageField::some(actor),
            ..Default::default()
        };
        let reply = self
            .client
            .verify_conversation_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(reply.into())
    }

    /// Repair a conversation's event log (`#799`, the `conversation repair`
    /// quarantine verb): replay every position independently so one
    /// corrupted item cannot block the rest, then rewrite the partition
    /// retaining only what decoded. A partition with nothing corrupted is a
    /// no-op (empty quarantine list).
    ///
    /// Admin-gated server-side, the same gate `excise_taint` uses.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error, or a
    /// permission refusal when `actor` is not an admin.
    pub async fn repair_conversation(
        &self,
        conversation_id: &str,
        actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
        operation_id: &str,
    ) -> Result<Vec<u64>, DialError> {
        use polyc_proto::proto::polychrome::approval::v1::RepairConversationRequest;
        let request = RepairConversationRequest {
            conversation_id: conversation_id.to_owned(),
            actor: buffa::MessageField::some(actor),
            operation_id: operation_id.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .repair_conversation_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(reply.quarantined_positions)
    }
}

/// One counterfactual override for a what-if [`ApprovalDialer::replay_conversation`].
///
/// Mirrors the control plane's replay override so a `conversation replay
/// --what-if` command names an override without depending on `polyc-proto`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReplayOverrideSpec {
    /// Replace turn `turn`'s first recorded completion text.
    Completion {
        /// 0-based committed-turn ordinal to fork at.
        turn: u32,
        /// The counterfactual completion text.
        text: String,
    },
    /// Replace turn `turn`'s `index`-th recorded tool result.
    ToolResult {
        /// 0-based committed-turn ordinal to fork at.
        turn: u32,
        /// Which recorded tool result (in call order) to replace.
        index: u32,
        /// The counterfactual tool result JSON.
        result_json: String,
    },
}

/// Every field named explicitly — no `..Default::default()` spread on any of
/// the three wire messages this builds — so a field added to any of them
/// without updating this impl fails to compile instead of silently not
/// riding the wire.
impl From<ReplayOverrideSpec> for polyc_proto::proto::polychrome::approval::v1::ReplayOverride {
    fn from(spec: ReplayOverrideSpec) -> Self {
        use polyc_proto::proto::polychrome::approval::v1::{
            ReplayCompletionOverride, ReplayToolResultOverride, replay_override::Kind,
        };
        let kind = match spec {
            ReplayOverrideSpec::Completion { turn, text } => {
                Kind::Completion(Box::new(ReplayCompletionOverride {
                    turn,
                    text,
                    __buffa_unknown_fields: buffa::UnknownFields::default(),
                }))
            }
            ReplayOverrideSpec::ToolResult {
                turn,
                index,
                result_json,
            } => Kind::ToolResult(Box::new(ReplayToolResultOverride {
                turn,
                index,
                result_json,
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })),
        };
        Self {
            kind: Some(kind),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

/// How one replayed turn compared against its record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReplayTurnOutcome {
    /// Every compared field reproduced the record.
    Match,
    /// The turn diverged from the record (verify) or forked from it (what-if) at
    /// `field`, with a human-readable `detail`.
    Diverged {
        /// The first field that differs (e.g. `"messages"`, `"usage"`, `"stop"`).
        field: String,
        /// The divergence explanation.
        detail: String,
    },
    /// The turn's recorded inputs are incomplete (excised or malformed), so it
    /// cannot be faithfully reproduced — reported, never passed.
    Unreplayable {
        /// Why the turn could not be reproduced.
        reason: String,
    },
}

/// One turn's replay verdict, as [`ApprovalDialer::replay_conversation`] returns it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReplayTurnVerdict {
    /// The turn's 0-based committed ordinal.
    pub turn_index: u32,
    /// How the turn compared against its record.
    pub outcome: ReplayTurnOutcome,
}

/// Total over the wire `ReplayOutcome` enum — an unknown/unspecified value
/// maps to [`ReplayTurnOutcome::Unreplayable`] rather than being dropped, so
/// this is a `From`, not a `TryFrom`.
impl From<polyc_proto::proto::polychrome::approval::v1::ReplayTurnVerdict> for ReplayTurnVerdict {
    fn from(v: polyc_proto::proto::polychrome::approval::v1::ReplayTurnVerdict) -> Self {
        use polyc_proto::proto::polychrome::approval::v1::ReplayOutcome;
        let outcome = match v.outcome.as_known() {
            Some(ReplayOutcome::Match) => ReplayTurnOutcome::Match,
            Some(ReplayOutcome::Diverged) => ReplayTurnOutcome::Diverged {
                field: v.field,
                detail: v.detail,
            },
            Some(ReplayOutcome::Unreplayable) => {
                ReplayTurnOutcome::Unreplayable { reason: v.detail }
            }
            Some(ReplayOutcome::Unspecified) | None => ReplayTurnOutcome::Unreplayable {
                reason: "the control plane returned an unknown replay outcome".to_owned(),
            },
        };
        Self {
            turn_index: v.turn_index,
            outcome,
        }
    }
}

/// The whole-conversation replay report a `conversation replay` command renders.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReplayReport {
    /// Per-turn verdicts, in recorded order.
    pub turns: Vec<ReplayTurnVerdict>,
    /// True only when every replayed turn matched its record (the verify pass);
    /// false on any divergence or unreplayable turn.
    pub all_match: bool,
}

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

impl From<polyc_proto::proto::polychrome::approval::v1::ExciseTaintReply> for ExcisionOutcome {
    fn from(reply: polyc_proto::proto::polychrome::approval::v1::ExciseTaintReply) -> Self {
        Self {
            persisted: reply.persisted,
            excised_positions: reply.excised_positions,
            signature_hex: reply.signature_hex,
            signed_by_hex: reply.signed_by_hex,
        }
    }
}

/// Result of an [`ApprovalDialer::verify_conversation`] call (`#799`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerificationOutcome {
    /// Whether every signed root in the log matched what replay recomputed.
    pub verified: bool,
    /// Human-readable detail on the first violation found; empty when
    /// `verified` is true.
    pub violation: String,
    /// Number of events the (successful) replay read.
    pub event_count: u64,
}

impl From<polyc_proto::proto::polychrome::approval::v1::VerifyConversationReply>
    for VerificationOutcome
{
    fn from(reply: polyc_proto::proto::polychrome::approval::v1::VerifyConversationReply) -> Self {
        Self {
            verified: reply.verified,
            violation: reply.violation,
            event_count: reply.event_count,
        }
    }
}

/// What one [`PersonaDialer::rebuild_usage_rollups`] pass did.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct UsageRollupsRebuilt {
    /// Conversation partitions replayed.
    pub conversations_scanned: u32,
    /// Distinct personas that now hold a usage rollup.
    pub personas_rebuilt: u32,
}

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

impl From<polyc_proto::proto::polychrome::persona::v1::StartLinkReply> for StartedLink {
    fn from(reply: polyc_proto::proto::polychrome::persona::v1::StartLinkReply) -> Self {
        Self {
            code: reply.code,
            expires_at_ms: reply.expires_at_ms,
            persona_id: reply.persona_id,
        }
    }
}

impl From<polyc_proto::proto::polychrome::persona::v1::AdminInviteReply> for StartedLink {
    fn from(reply: polyc_proto::proto::polychrome::persona::v1::AdminInviteReply) -> Self {
        Self {
            code: reply.code,
            expires_at_ms: reply.expires_at_ms,
            persona_id: reply.persona_id,
        }
    }
}

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

impl From<polyc_proto::proto::polychrome::persona::v1::StartDeepLinkReply> for StartedDeepLink {
    fn from(reply: polyc_proto::proto::polychrome::persona::v1::StartDeepLinkReply) -> Self {
        Self {
            token: reply.token,
            expires_at_ms: reply.expires_at_ms,
            persona_id: reply.persona_id,
        }
    }
}

/// 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,
    /// The completing channel already belongs to its own established
    /// persona; absorbing it would be an unguarded merge, so the ceremony
    /// refused instead. The code was not consumed — the reader can retry by
    /// starting the link from that established account instead.
    EstablishedPersona,
}

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.",
            Self::EstablishedPersona => {
                "This account already has its own established persona. Start the link from this account instead, then complete it on the other channel."
            }
        }
    }
}

/// What [`PersonaDialer::attest_verified_email`] did. Not rendered to a real
/// person — the caller is a first-party backend, not an edge — so unlike
/// [`LinkCeremony`] this carries no `user_message`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AttestedEmail {
    /// The identity is now bound to the resolved persona (directly, or by
    /// absorbing its prior PROVISIONAL persona).
    Linked {
        /// The surviving persona id.
        persona_id: String,
    },
    /// The identity already resolved to that persona; now marked verified.
    AlreadyLinked {
        /// The persona id.
        persona_id: String,
    },
    /// The identity resolves to its own already-established persona;
    /// absorbing it would be an unguarded merge, so this refused instead.
    EstablishedPersona,
    /// The asserted email is on the deployment's blocked-identifier list.
    Blocked,
    /// Binding this email would exceed the deployment's per-provider
    /// identity cardinality cap on the target persona.
    CapHalted,
    /// An unexpected/unspecified outcome.
    Failed,
}

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

/// The opening line of the target-facing invite message, shared across
/// every edge so the greeting can't drift.
///
/// `#1628`, the CLAUDE.md "route through a shared helper" rule. Blank-safe:
/// with an inviter name it reads "Ada set you up with access", without one
/// it opens with a neutral "You're invited to Polychrome."
///
/// Each edge appends its OWN redemption instructions after this — the
/// mechanism genuinely differs per edge (Slack redeems via a `link <code>`
/// text command; Telegram via a `/start <code>` deep link), so that part
/// can't be shared without hardcoding one edge's UX into the others.
#[must_use]
pub fn invite_opener(inviter: &str) -> String {
    if inviter.trim().is_empty() {
        "🎟️ You're invited to Polychrome.".to_owned()
    } else {
        format!("🎟️ You're invited. {inviter} set you up with access.")
    }
}

/// The codeless acknowledgement the admin sees once an invite is on its way.
/// Never the code, never the redemption mechanism — just confirms who and
/// when it expires, identically on every edge.
///
/// `target_display` is the edge's own rendering of the target (Slack:
/// `<@U123>`; Telegram: `@username` or a first name) — this helper owns only
/// the shared sentence shape, not per-edge mention syntax.
#[must_use]
pub fn invite_sent_ack(target_display: &str) -> String {
    format!("📬 Invite sent to {target_display} by DM. It expires in 60 minutes.")
}

/// User-facing copy for a [`PersonaDialer::set_incognito`] failure (`#796`),
/// shared across edges so the toggle reads identically wherever it's
/// reachable from.
#[must_use]
pub const fn incognito_error_message(err: &DialError) -> &'static str {
    match err.code() {
        Some(ErrorCode::PermissionDenied) => {
            "I couldn't toggle incognito — that only works for a conversation you're part of."
        }
        _ => "I couldn't toggle incognito — try again.",
    }
}

/// A human's decision on a pending `ask_question` question (`#1660`).
///
/// The question-pause SIBLING of [`ApprovalChoice`], not a reuse of it:
/// `ask_question` offers a fixed, model-proposed option set (or an explicit
/// decline), never an approve/deny/session/abort/defer decision.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QuestionChoice {
    /// Pick the option at this index within the question's `options` array.
    SelectOption(u32),
    /// Explicitly decline to choose — "use your own judgment".
    Decline,
}

/// Outcome of a [`QuestionDialer::respond`] call — the question-pause
/// SIBLING of [`ApprovalOutcome`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QuestionOutcome {
    /// `true` if a matching pending question existed and the answer was
    /// persisted; `false` is the idempotent no-op (already answered /
    /// unknown).
    pub persisted: bool,
    /// Lowercase-hex ed25519 signature over the canonical answer bytes.
    /// Empty if `persisted` is false.
    pub signature_hex: String,
    /// Lowercase-hex public key the signature verifies against. Empty if
    /// `persisted` is false.
    pub signed_by_hex: String,
}

impl From<polyc_proto::proto::polychrome::question::v1::QuestionAnswerReply> for QuestionOutcome {
    fn from(reply: polyc_proto::proto::polychrome::question::v1::QuestionAnswerReply) -> Self {
        Self {
            persisted: reply.persisted,
            signature_hex: reply.signature_hex,
            signed_by_hex: reply.signed_by_hex,
        }
    }
}

/// Reusable handle for the control plane's `QuestionService` (`#1660`).
///
/// The THIN `ask_question` pause/resume path — the question-pause SIBLING of
/// [`ApprovalDialer`], not a reuse of it: its own event kinds, its own
/// signed-answer canonical (`polyc_crypto::question`), so an approval
/// signature can never verify as a question answer and vice versa. Shares
/// the `AgentService` endpoint — all these services are served on one
/// Connect port — so it is built from the same address.
#[derive(Clone)]
pub struct QuestionDialer {
    client: Arc<QuestionServiceClient<HttpClient>>,
}

impl QuestionDialer {
    /// Build a dialer pointed at `addr` (expects `http://host:port`),
    /// carrying NO `Authorization` header.
    ///
    /// `#1660` incident (2026-07-28): this used to be the only constructor,
    /// on the mistaken assumption that a `Respond` call's per-question
    /// `answer_token` was sufficient auth on its own. It authorizes THAT
    /// answer, but it is not what the control plane's `require_edge_bearer`
    /// layer checks — that layer guards every RPC on this listener
    /// regardless of the call's own payload, `QuestionService` included. An
    /// edge with real edge credentials MUST use [`Self::with_bearer`]
    /// instead, or every answer submission fails closed with 401. This
    /// constructor now exists only for the genuinely uncredentialed case
    /// (mirroring [`ApprovalDialer::new`]).
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI.
    pub fn new(addr: &str) -> Result<Self, DialError> {
        Ok(Self {
            client: build_control_client(addr, None, QuestionServiceClient::new)?,
        })
    }

    /// Build a dialer pointed at `addr`, authenticated with `bearer`.
    ///
    /// Every call this dialer makes carries an `Authorization: Bearer
    /// <bearer>` header — required by the control plane's
    /// `require_edge_bearer` layer, which guards `QuestionService` like
    /// every other RPC on this listener (see [`Self::new`]'s doc for why
    /// this constructor exists). `respond`'s per-question `answer_token` is
    /// a SEPARATE, per-call authorization on top of this — bearer proves
    /// "this is a registered edge", `answer_token` proves "this is a valid
    /// answer to this specific question".
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value.
    pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            client: build_control_client(addr, Some(bearer), QuestionServiceClient::new)?,
        })
    }

    /// Submit a human's answer for a pending question. The answer is
    /// UNSIGNED — the control plane resolves and signs it server-side (THIN
    /// path, mirrors [`ApprovalDialer::respond`]) and returns the signature.
    /// Idempotent: answering an already-decided or unknown occurrence returns
    /// `persisted: false`.
    ///
    /// `turn_id` is the occurrence this answer names, carried unmodified off
    /// the same card the `answer_token` came from (`#2523`). The control plane
    /// refuses a `Respond` that names no occurrence: a provider re-mints a
    /// tool-call id across turns, so `(call_id, index)` alone would let one
    /// person's decision land on a different question.
    ///
    /// `answer_token` is the short-lived signed capability carried unmodified
    /// off the [`TurnEvent::QuestionPending`] event or [`PendingQuestionPrompt`]
    /// entry this decision answers — required: the control plane rejects a
    /// `Respond` whose token is missing, expired, or bound to a different
    /// question or conversation.
    ///
    /// `responder` is the identity of the human answering — the durable
    /// `answered_by` fact. An edge that cannot supply caller identity passes
    /// `None`. Unsigned, unlike [`ApprovalDialer::respond`]'s: answering a
    /// question authorizes nothing, so no gate reads this.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error
    /// (including the control plane's rejection of an invalid
    /// `answer_token`).
    #[allow(clippy::too_many_arguments)] // each is a distinct field of the answer
    pub async fn respond(
        &self,
        turn_id: &str,
        call_id: &str,
        index: u32,
        choice: QuestionChoice,
        conversation_id: &str,
        answer_token: &str,
        responder: Option<ExternalIdentity>,
    ) -> Result<QuestionOutcome, DialError> {
        let answer = match choice {
            QuestionChoice::SelectOption(index) => {
                WireAnswer::SelectOption(Box::new(WireSelectOption {
                    index,
                    __buffa_unknown_fields: buffa::UnknownFields::default(),
                }))
            }
            QuestionChoice::Decline => WireAnswer::Decline(Box::new(WireDecline {
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            })),
        };
        let request = QuestionAnswerRequest {
            turn_id: turn_id.to_owned(),
            call_id: call_id.to_owned(),
            index,
            conversation_id: conversation_id.to_owned(),
            answer: Some(answer),
            answer_token: answer_token.to_owned(),
            responder: responder.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
            ..Default::default()
        };
        let reply = self
            .client
            .respond_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(reply.into())
    }

    /// List the conversation's outstanding questions (a `question_request`
    /// with no later `question_response`). An edge calls this to recover
    /// occurrences it must prompt on after losing the streamed
    /// [`TurnEvent::QuestionPending`] event — mirrors
    /// [`ApprovalDialer::list_pending`]. Read-only and idempotent.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error.
    pub async fn list_pending(
        &self,
        conversation_id: &str,
    ) -> Result<Vec<PendingQuestionPrompt>, DialError> {
        let mut pending = Vec::new();
        let mut page_token = String::new();
        loop {
            let request = ListPendingQuestionsRequest {
                conversation_id: conversation_id.to_owned(),
                page_token: page_token.clone(),
                ..Default::default()
            };
            let reply = self
                .client
                .list_pending_with_options(request, traced_options())
                .await?
                .into_owned();
            pending.extend(reply.pending.into_iter().map(PendingQuestionPrompt::from));
            if reply.next_page_token.is_empty() {
                break;
            }
            page_token = reply.next_page_token;
        }
        Ok(pending)
    }
}

/// 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> {
        Ok(Self {
            client: build_control_client(addr, None, PersonaServiceClient::new)?,
        })
    }

    /// Build a dialer pointed at `addr`, authenticated with `bearer`.
    ///
    /// Every call this dialer makes carries an `Authorization: Bearer
    /// <bearer>` header. `PersonaService` doesn't send `AgentStart`, so no
    /// signed [`AssertedAttribution`] envelope rides these calls — bearer-only.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value.
    pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            client: build_control_client(addr, Some(bearer), PersonaServiceClient::new)?,
        })
    }

    /// Build a dialer pointed at `addr`, presenting `bearer` as the `#803`
    /// admin service-credential (`Authorization: Bearer pc_<id>_<secret>`,
    /// a current record in State's credential authority carrying admin
    /// access) every `require_admin`-gated `PersonaService` RPC checks before
    /// it will even look at the body-supplied actor. [`Self::new`]'s dialer never presents this header
    /// — use `new_admin` (not `new`) for [`Self::rebuild_usage_rollups`] and
    /// any future RPC gated the same way.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI,
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value, or [`DialError::Tls`] if an `https` endpoint's TLS setup
    /// fails.
    pub fn new_admin(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            client: build_control_client(addr, Some(bearer), PersonaServiceClient::new)?,
        })
    }

    /// Admin-gated maintenance (`#457`): recompute every persona's usage
    /// rollup from a full eventlog replay and overwrite the maintained
    /// counters wholesale — the repair path for the crash window a
    /// best-effort post-commit rollup write leaves open. Build this dialer
    /// with [`Self::new_admin`], not [`Self::new`].
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error,
    /// including a `permission_denied` refusal (missing/invalid bearer, or a
    /// non-admin actor).
    pub async fn rebuild_usage_rollups(
        &self,
        actor: ExternalIdentity,
    ) -> Result<UsageRollupsRebuilt, DialError> {
        let request = RebuildUsageRollupsRequest {
            actor: buffa::MessageField::some(actor),
            ..Default::default()
        };
        let reply = self
            .client
            .rebuild_usage_rollups_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(UsageRollupsRebuilt {
            conversations_scanned: reply.conversations_scanned,
            personas_rebuilt: reply.personas_rebuilt,
        })
    }

    /// 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(reply.into())
    }

    /// 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(reply.into())
    }

    /// 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(reply.into())
    }

    /// Toggle incognito for a conversation (`#796`): suppresses memory
    /// extraction from here on when `on`, clears the suppression when not.
    /// The control plane authorizes `actor` — an admin may toggle any
    /// conversation, anyone else only one they are a caller or participant
    /// of — so a non-admin's own conversation is reachable through this same
    /// call; an unrelated conversation surfaces as `permission_denied`.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error,
    /// including a `permission_denied` refusal.
    pub async fn set_incognito(
        &self,
        actor: ExternalIdentity,
        conversation_id: &str,
        on: bool,
    ) -> Result<bool, DialError> {
        let request = SetIncognitoRequest {
            actor: buffa::MessageField::some(actor),
            conversation_id: conversation_id.to_owned(),
            on,
            ..Default::default()
        };
        let reply = self
            .client
            .set_incognito_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(reply.on)
    }

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

    /// First-party backend attestation (docs/reference/personas.md §2): assert
    /// that `identity`'s email is already verified by the calling backend's
    /// own signup flow — no code, no magic-link round trip. Requires a
    /// service credential carrying `Capabilities::attest_email`; the caller
    /// is authenticated as itself, never as `identity`. Resolves-or-
    /// provisions `identity`: a genuinely first-time identity (no prior
    /// agent contact) still gets a fresh persona in this one call — the
    /// caller may attest at raw account creation or lazily on/after first
    /// contact, whichever fits.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error,
    /// including `permission_denied` when the presented credential lacks
    /// `attest_email`.
    pub async fn attest_verified_email(
        &self,
        identity: ExternalIdentity,
        verified_email: &str,
    ) -> Result<AttestedEmail, DialError> {
        let request = AttestVerifiedEmailRequest {
            identity: buffa::MessageField::some(identity),
            verified_email: verified_email.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .attest_verified_email_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(match reply.outcome.as_known() {
            Some(AttestVerifiedEmailOutcome::Linked) => AttestedEmail::Linked {
                persona_id: reply.persona_id,
            },
            Some(AttestVerifiedEmailOutcome::AlreadyLinked) => AttestedEmail::AlreadyLinked {
                persona_id: reply.persona_id,
            },
            Some(AttestVerifiedEmailOutcome::EstablishedPersona) => {
                AttestedEmail::EstablishedPersona
            }
            Some(AttestVerifiedEmailOutcome::Blocked) => AttestedEmail::Blocked,
            Some(AttestVerifiedEmailOutcome::CapHalted) => AttestedEmail::CapHalted,
            Some(AttestVerifiedEmailOutcome::Unspecified) | None => AttestedEmail::Failed,
        })
    }

    /// 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::EstablishedPersona) => LinkCeremony::EstablishedPersona,
            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(LinkedIdentity::from)
                .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,
}

/// Deliberate subset projection: the dashboard doesn't render `scope`, the
/// one field the wire `ExternalIdentity` has beyond these three.
impl From<ExternalIdentity> for LinkedIdentity {
    fn from(id: ExternalIdentity) -> Self {
        Self {
            provider: id.provider,
            external_id: id.external_id,
            display_name: id.display_name,
        }
    }
}

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

/// The Connect transport error [`DialError::Connect`] wraps, re-exported for
/// the same reason [`ErrorCode`] is: an edge that maps a dial failure onto its
/// own surface's error taxonomy needs to name the type it is mapping from.
pub use connectrpc::ConnectError;

/// 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. `mail:{message-id}`, `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()
}

/// Every field named explicitly — no `..Default::default()` spread — so a
/// field added to `WireIngressDirective` without updating this impl fails to
/// compile instead of silently not riding the wire. `IngressDirective` is a
/// fail-closed policy envelope (`#68`): a dropped `required_approver` or
/// `budget_cap` here is a fail-open bug, not just data loss.
impl From<IngressDirective> for WireIngressDirective {
    fn from(directive: IngressDirective) -> Self {
        Self {
            budget_cap: directive.budget_cap.unwrap_or_default(),
            priority: directive.priority.map_or_else(
                || buffa::EnumValue::from(Priority::PRIORITY_UNSPECIFIED),
                buffa::EnumValue::from,
            ),
            required_approver: directive
                .required_approver
                .map_or_else(buffa::MessageField::none, buffa::MessageField::some),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

/// Convert the SDK-facing [`IngressDirective`] into the wire
/// `polychrome.agent.v1.IngressDirective`, `None` for an empty directive
/// (`#68`) — so an edge that sets no policy sends the field unset, matching
/// [`IngressDirective::is_empty`]'s "byte-for-byte unaffected" contract.
fn wire_ingress_directive(
    directive: IngressDirective,
) -> buffa::MessageField<WireIngressDirective> {
    if directive.is_empty() {
        return buffa::MessageField::none();
    }
    buffa::MessageField::some(directive.into())
}

/// Canonical bytes a turn's messages hash to for
/// [`AssertedAttribution::content_hash`].
///
/// Each [`Message`]'s buffa encoding, concatenated in order. The control
/// plane recomputes this same encoding over the messages it decodes off
/// `AgentStart.messages` (before its own smoke-test placeholder fallback,
/// which the signed envelope never covers) to check the two agree — see
/// `verify_ingest_envelope` in `crates/control-plane/src/grpc/mod.rs`.
/// Exported so both sides call one function instead of maintaining
/// independent encodings that could drift.
#[must_use]
pub fn encode_messages_for_content_hash(messages: &[Message]) -> Vec<u8> {
    use buffa::Message as _;
    let mut bytes = Vec::new();
    for m in messages {
        bytes.extend_from_slice(&m.encode_to_vec());
    }
    bytes
}

/// Canonical bytes the participation gate's envelope binds to.
///
/// Covers the judged transcript plus the two prompt inputs that change the
/// verdict: the bot's own name and the surface wording. Each field is
/// length-prefixed, so no boundary shift can move bytes between fields and
/// leave the hash unchanged.
///
/// This is deliberately NOT the turn encoder: a turn binds
/// `AgentStart.messages`. Two encodings mean a captured turn envelope cannot
/// authorize an evaluation, or the reverse.
#[must_use]
pub fn encode_transcript_for_content_hash(
    transcript: &[ParticipantMessage],
    bot_name: &str,
    surface: &str,
) -> Vec<u8> {
    use buffa::Message as _;
    let mut bytes = Vec::new();
    for field in [bot_name.as_bytes(), surface.as_bytes()] {
        bytes.extend_from_slice(&(field.len() as u64).to_be_bytes());
        bytes.extend_from_slice(field);
    }
    for message in transcript {
        let encoded = message.encode_to_vec();
        bytes.extend_from_slice(&(encoded.len() as u64).to_be_bytes());
        bytes.extend_from_slice(&encoded);
    }
    bytes
}

/// Builds the signed envelope for one participation evaluation.
///
/// It mirrors [`build_asserted_attribution`] with two deliberate differences:
/// `exec_id` is empty because no turn exists, and `content_hash` covers the
/// transcript rather than a turn's messages.
fn build_classify_attribution(
    creds: &EdgeCredentials,
    conversation_id: &str,
    source_identity: &IngressIdentity,
    namespace: &ClaimedNamespace,
    transcript: &[ParticipantMessage],
    bot_name: &str,
    surface: &str,
) -> AssertedAttribution {
    let content_hash = polyc_crypto::edge_identity::content_hash_hex(
        &encode_transcript_for_content_hash(transcript, bot_name, surface),
    );
    let mut envelope = AssertedAttribution {
        edge_id: creds.edge_id().to_owned(),
        conversation_id: conversation_id.to_owned(),
        nonce: uuid::Uuid::new_v4().to_string(),
        issued_unix_ms: issued_unix_ms(),
        caller: buffa::MessageField::none(),
        participants: Vec::new(),
        signature_hex: String::new(),
        exec_id: String::new(),
        content_hash,
        source_identity: buffa::MessageField::some(source_identity.to_wire()),
        // #1691: the gate claims the same tenancy namespace a turn does, and
        // the shared verifier refuses an envelope that names none.
        namespace: namespace.as_str().to_owned(),
        ..Default::default()
    };
    creds.sign_assertion(&mut envelope);
    envelope
}

/// Milliseconds since the Unix epoch, for [`AssertedAttribution::issued_unix_ms`].
/// A clock that reads before the epoch (never in practice) maps to `0` rather
/// than panicking — a turn-signing helper must not crash the dial.
fn issued_unix_ms() -> i64 {
    let millis = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| d.as_millis());
    i64::try_from(millis).unwrap_or(i64::MAX)
}

/// Build the signed [`AssertedAttribution`] envelope for one turn: `creds`'
/// edge id, this turn's `conversation_id` and `exec_id`, a fresh per-call
/// nonce/timestamp, the `caller`/`participants` this turn asserts — the
/// envelope is their SOLE carrier, since `AgentStart`'s loose fields 5/6 are
/// reserved and gone (see `AgentStart.asserted_attribution`'s doc) — and
/// stable `source_identity`, and `content_hash` binding the envelope to `messages` (this turn's
/// `AgentStart.messages`) via [`encode_messages_for_content_hash`] +
/// [`polyc_crypto::edge_identity::content_hash_hex`].
///
/// This is the one implementation of the envelope contract. Other crates'
/// tests reach it through the `test-util`-gated [`test_util`] module rather
/// than re-deriving it, so a change here cannot silently diverge from what
/// those tests assert the control plane accepts.
fn build_asserted_attribution(
    creds: &EdgeCredentials,
    conversation_id: &str,
    exec_id: &str,
    source_identity: &IngressIdentity,
    namespace: &ClaimedNamespace,
    messages: &[Message],
    attribution: &Attribution,
) -> AssertedAttribution {
    let content_hash =
        polyc_crypto::edge_identity::content_hash_hex(&encode_messages_for_content_hash(messages));
    let mut envelope = AssertedAttribution {
        edge_id: creds.edge_id().to_owned(),
        conversation_id: conversation_id.to_owned(),
        nonce: uuid::Uuid::new_v4().to_string(),
        issued_unix_ms: issued_unix_ms(),
        caller: attribution
            .caller
            .clone()
            .map_or_else(buffa::MessageField::none, buffa::MessageField::some),
        participants: attribution.participants.clone(),
        signature_hex: String::new(),
        exec_id: exec_id.to_owned(),
        content_hash,
        source_identity: buffa::MessageField::some(source_identity.to_wire()),
        namespace: namespace.as_str().to_owned(),
        ..Default::default()
    };
    creds.sign_assertion(&mut envelope);
    envelope
}

/// Builds the single signed [`AgentRequest`] sent to durable receive.
///
/// Shared by every buffered and streaming entry point so none can omit the
/// stable source identity. A credential-free dial fails closed instead of
/// placing that identity outside the signed assertion.
fn build_request(
    ingress: TurnIngress,
    credentials: Option<&EdgeCredentials>,
) -> Result<AgentRequest, DialError> {
    let credentials = credentials.ok_or(DialError::MissingIngressCredentials)?;
    let asserted_attribution = build_asserted_attribution(
        credentials,
        &ingress.conversation_id,
        &ingress.exec_id,
        &ingress.source_identity,
        &ingress.namespace,
        &ingress.messages,
        &ingress.attribution,
    );
    Ok(AgentRequest {
        conversation_id: ingress.conversation_id,
        exec_id: ingress.exec_id,
        start: buffa::MessageField::some(AgentStart {
            agent_id: String::new(),
            agent_config: Vec::new(),
            messages: ingress.messages,
            payment_receipt: ingress
                .payment_receipt
                .map_or_else(buffa::MessageField::none, buffa::MessageField::some),
            // #1514 review: the loose `caller`/`participants` fields were
            // removed from the wire (`reserved 5, 6` in `AgentStart`) — every
            // caller/participant now rides ONLY inside the signed
            // `asserted_attribution` envelope built below.
            // #843: a routine fire declares ephemeral history so each firing
            // starts the model on an empty transcript; every other edge leaves
            // this false (persistent).
            ephemeral_history: ingress.ephemeral_history,
            ingress_directive: wire_ingress_directive(ingress.ingress_directive),
            // #1103: the scheduled tick that produced this turn, when a routine's
            // cron binding fired it. Empty for every other caller — a chat turn
            // or a manual trigger send — so the field rides the wire unset and
            // never dedups.
            occurrence: ingress.occurrence,
            asserted_attribution: buffa::MessageField::some(asserted_attribution),
            ..Default::default()
        }),
        ..Default::default()
    })
}

fn validate_ingress_receipt(
    receipt: &WireIngressReceipt,
    expected_source: &WireIngressSourceIdentity,
    expected_conversation: &str,
) -> Result<(), DialError> {
    let Some(source) = receipt.source_identity.as_option() else {
        return Err(DialError::InvalidIngressReceipt(
            "source identity is absent",
        ));
    };
    if source != expected_source {
        return Err(DialError::InvalidIngressReceipt(
            "source identity does not match the signed request",
        ));
    }
    if receipt.receipt.is_empty() {
        return Err(DialError::InvalidIngressReceipt("State receipt is empty"));
    }
    if receipt.dispatch_id.is_empty() {
        return Err(DialError::InvalidIngressReceipt("dispatch id is empty"));
    }
    if receipt.conversation_id != expected_conversation {
        return Err(DialError::InvalidIngressReceipt(
            "conversation does not match the request",
        ));
    }
    Ok(())
}

/// 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 one [`TurnEvent::InviteDelivery`] per admin invite minted this turn,
/// then a [`TurnEvent::WalletLinkPrompt`] if a paid call needed a linked
/// wallet (or the caller directly requested one), then a
/// [`TurnEvent::WalletUpdatePrompt`]/[`TurnEvent::WalletRevokePrompt`] for
/// their respective mints, then a
/// [`TurnEvent::TurnFailed`] if the turn ended durably failed, 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(PendingApprovalPrompt::from)
        .map(TurnEvent::from)
        .collect();
    // `#1660`: the question-pause SIBLING of the `pending_approvals`
    // projection above — one `QuestionPending` per outstanding question.
    events.extend(
        end.pending_questions
            .into_iter()
            .map(PendingQuestionPrompt::from)
            .map(TurnEvent::from),
    );
    if let Some(h) = end.handoff.into_option() {
        events.push(TurnEvent::HandoffStarted {
            child_agent_id: h.child_agent_id,
            reason: h.reason,
        });
    }
    // Each minted invite becomes a delivery event for the edge to act on. The
    // code rides only here (control plane → edge); the edge routes it to the
    // target's DM and nowhere else.
    for d in end.invite_deliveries {
        events.push(TurnEvent::InviteDelivery {
            target_user_id: d.target_user_id,
            code: d.code,
            inviter_display: d.inviter_display,
        });
    }
    // A paid tool call this turn needed a linked wallet (`#519`). The wire
    // carries only the URL (empty = unknown); a blank/whitespace-only value
    // maps to `None` so the edge renders the no-button, admin-pointer card.
    if let Some(prompt) = end.wallet_link_prompt.into_option() {
        let renewal = prompt.renewal;
        let requested = prompt.requested;
        let link_url = (!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
        events.push(TurnEvent::WalletLinkPrompt {
            link_url,
            renewal,
            requested,
        });
    }
    // `wallet_update_limit` minted a fresh spending-limit-update ceremony
    // this turn (issue #1041/#1159). Always non-empty in practice — the
    // tool either mints a ceremony or refuses outright — mirrored here
    // unfiltered rather than re-adding the blank-collapses-to-`None` dance
    // the URL-optional prompts above need.
    if let Some(prompt) = end.wallet_update_prompt.into_option() {
        events.push(TurnEvent::WalletUpdatePrompt {
            update_url: prompt.update_url,
            new_limit: prompt.new_limit,
        });
    }
    // This turn's `unlink_self` (wallet target) call minted a hard-revoke
    // ceremony for the just-unlinked delegated key (issue #1042/#1156).
    if let Some(prompt) = end.wallet_revoke_prompt.into_option() {
        events.push(TurnEvent::WalletRevokePrompt {
            revoke_url: prompt.revoke_url,
        });
    }
    // A durably-failed turn (`#756`) ended without a `batch` — the wire's own
    // doc comment says this field exists so "external surfaces (Slack, the
    // cockpit) can react", but until this event existed no edge ever read it.
    if let Some(failure) = end.failure.into_option() {
        events.push(TurnEvent::TurnFailed {
            kind: turn_failure_kind_from_wire(failure.kind.to_i32()),
            message: failure.message,
        });
    }
    events.push(TurnEvent::Done);
    events
}

/// Map the wire `TurnFailureKind` onto the local, edge-facing
/// [`TurnFailureKind`]. An unrecognized or unspecified wire value maps to
/// [`TurnFailureKind::Other`] — a definite failure with an unclear reason is
/// still a definite failure, never silently dropped.
const fn turn_failure_kind_from_wire(kind: i32) -> TurnFailureKind {
    if kind == WireTurnFailureKind::TURN_FAILURE_KIND_RATE_LIMIT as i32 {
        TurnFailureKind::RateLimit
    } else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_TIMEOUT as i32 {
        TurnFailureKind::Timeout
    } else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_UNAVAILABLE as i32 {
        TurnFailureKind::Unavailable
    } else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_AUTH as i32 {
        TurnFailureKind::Auth
    } else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_BAD_REQUEST as i32 {
        TurnFailureKind::BadRequest
    } else {
        TurnFailureKind::Other
    }
}

/// 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.
/// - An `internal_only` message (`#743`: a paused turn's withheld model text,
///   or an approver/policy-injected note) produces no event — the proto
///   contract is that such a message is never emitted to a client, and the
///   streaming path is exactly that emission seam.
fn message_to_event(msg: Message) -> Option<TurnEvent> {
    if msg.internal_only {
        return None;
    }
    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,
    }
}

/// Classify one output [`Message`] for [`AgentDialer::run_turn_with`]'s
/// buffered aggregation, pushing into `text_parts` (the assistant's real
/// answer) or `scaffolding` (the tool-call/tool-result fallback rendering) —
/// pulled out of the loop body so the classification is unit-testable without
/// a live stream.
///
/// `#743`: an `internal_only` message — a paused turn's withheld model text,
/// or an approver/policy-injected note — is skipped entirely, mirroring
/// [`message_to_event`]'s identical rule on the streaming path so the two
/// APIs cannot drift on what a client is shown.
fn aggregate_output_message(
    msg: Message,
    text_parts: &mut Vec<String>,
    scaffolding: &mut Vec<String>,
) {
    if msg.internal_only {
        return;
    }
    // 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 {
        return;
    };
    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);
            }
        }
    }
}

/// The buffered result of one turn: the aggregated reply text plus any
/// [`PendingApprovalPrompt`]s the turn paused on.
///
/// Returned by [`AgentDialer::run_turn_with_approvals`] — the buffered
/// sibling of the streaming path's `TurnEvent` sequence, folded into one
/// value since a buffered caller can't react mid-stream anyway.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct BufferedTurn {
    /// The turn's aggregated answer text (identical to what
    /// [`AgentDialer::run_turn_with`] returns).
    pub reply: String,
    /// Gated tool calls the turn paused on, one per entry — empty unless the
    /// turn paused. Submit a decision via [`ApprovalDialer::respond`] for
    /// each, then re-drive with an empty `user_text` to resume the turn.
    pub pending_approvals: Vec<PendingApprovalPrompt>,
    /// Questions from an `ask_question` call awaiting an answer (`#1660`),
    /// one per entry — empty unless the turn paused. Submit a decision via
    /// [`QuestionDialer::respond`] for each, then re-drive with an empty
    /// `user_text` to resume the turn. The question-pause SIBLING of
    /// `pending_approvals` above.
    pub pending_questions: Vec<PendingQuestionPrompt>,
}

/// One gated tool call a buffered turn paused on, surfaced from the terminal
/// `AgentEnd.pending_approvals` — the buffered-API mirror of
/// [`TurnEvent::ApprovalPending`]'s fields.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingApprovalPrompt {
    /// Turn that emitted this occurrence of `request_id`.
    pub turn_id: String,
    /// Tool-call id == the approval `request_id` to answer.
    pub request_id: String,
    /// The tool/function name awaiting approval. Raw machine identifier; the
    /// field of record for trust/audit.
    pub 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`.
    pub title: String,
    /// Arguments JSON for the call.
    pub 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.
    pub reason: String,
    /// The short-lived signed capability (`#787`) that must be replayed
    /// unmodified on the eventual [`ApprovalDialer::respond`] call — required,
    /// the control plane rejects a `Respond` whose token is missing, expired,
    /// or bound to a different request or conversation.
    pub resolve_token: String,
    /// Computed-preview enrichment (`#1496`); `None` for every call but
    /// `routine_create`. See [`ApprovalPreview`].
    pub preview: Option<ApprovalPreview>,
}

/// Every field named explicitly, shared by both the buffered
/// ([`AgentDialer::run_turn_with_approvals`]) and streaming
/// ([`TurnEvent::ApprovalPending`]) paths so the two can't drift on what a
/// pending approval carries — a field added to the wire type without
/// updating this impl fails to compile instead of one path silently not
/// carrying it.
impl From<WireAgentPendingApproval> for PendingApprovalPrompt {
    fn from(pa: WireAgentPendingApproval) -> Self {
        Self {
            turn_id: pa.turn_id,
            request_id: pa.request_id,
            tool_name: pa.tool_name,
            title: pa.title,
            args_json: pa.args_json,
            reason: pa.reason,
            resolve_token: pa.resolve_token,
            preview: pa.preview.into_option().map(Into::into),
        }
    }
}

/// One question a buffered turn paused on (`#1660`).
///
/// Surfaced from the terminal `AgentEnd.pending_questions` — the buffered-API
/// mirror of [`TurnEvent::QuestionPending`]'s fields. Every field named
/// explicitly, shared by both the buffered ([`BufferedTurn`]) and streaming
/// ([`TurnEvent::QuestionPending`]) paths so the two can't drift on what a
/// pending question carries.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingQuestionPrompt {
    /// Turn that emitted this question occurrence (`#2523`).
    pub turn_id: String,
    /// The `ask_question` call id this question came from.
    pub call_id: String,
    /// This question's position within its call's `questions` array
    /// (0-based).
    pub index: u32,
    /// The short label (fits a chat-surface button-row heading).
    pub header: String,
    /// The one-sentence question to ask.
    pub question: String,
    /// 2-4 mutually exclusive options to offer.
    pub options: Vec<QuestionOptionPrompt>,
    /// The raw `ask_question` call's full arguments JSON (every question in
    /// the call, not just this one).
    pub args_json: String,
    /// The short-lived signed capability that must be replayed unmodified on
    /// the eventual [`QuestionDialer::respond`] call — required, the control
    /// plane rejects a `Respond` whose token is missing, expired, or bound to
    /// a different occurrence or conversation.
    pub answer_token: String,
    /// Whether this still-unanswered question was already surfaced once (a
    /// later turn's boundary marker sits after its `question_request` in
    /// the event log while it stayed unanswered). `false`: render the full
    /// interactive card; `true`: render a compact reminder instead
    /// (`#1970`). Derived by the control plane at replay time — never
    /// edge-cached.
    pub already_surfaced: bool,
}

/// One option a pending question offers, mirroring the wire
/// `QuestionOption`/`QuestionOptionEntry` messages (`#1660`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QuestionOptionPrompt {
    /// A few words naming this option.
    pub label: String,
    /// The one-sentence consequence of picking this option.
    pub description: String,
    /// Whether this is the model's recommendation (at most one per
    /// question).
    pub recommended: bool,
}

impl From<WireAgentQuestionOption> for QuestionOptionPrompt {
    fn from(o: WireAgentQuestionOption) -> Self {
        Self {
            label: o.label,
            description: o.description,
            recommended: o.recommended,
        }
    }
}

impl From<QuestionOptionEntry> for QuestionOptionPrompt {
    fn from(o: QuestionOptionEntry) -> Self {
        Self {
            label: o.label,
            description: o.description,
            recommended: o.recommended,
        }
    }
}

impl From<WireAgentPendingQuestion> for PendingQuestionPrompt {
    fn from(q: WireAgentPendingQuestion) -> Self {
        Self {
            turn_id: q.turn_id,
            call_id: q.call_id,
            index: q.index,
            header: q.header,
            question: q.question,
            options: q.options.into_iter().map(Into::into).collect(),
            args_json: q.args_json,
            answer_token: q.answer_token,
            already_surfaced: q.already_surfaced,
        }
    }
}

/// A recovered [`PendingQuestionEntry`] (from [`QuestionDialer::list_pending`])
/// carries the same fields as [`PendingQuestionPrompt`] — one shared
/// conversion target so a `ListPending` recovery renders identically to the
/// live card.
impl From<PendingQuestionEntry> for PendingQuestionPrompt {
    fn from(q: PendingQuestionEntry) -> Self {
        Self {
            turn_id: q.turn_id,
            call_id: q.call_id,
            index: q.index,
            header: q.header,
            question: q.question,
            options: q.options.into_iter().map(Into::into).collect(),
            args_json: q.args_json,
            answer_token: q.answer_token,
            already_surfaced: q.already_surfaced,
        }
    }
}

/// [`TurnEvent::QuestionPending`] carries the exact same fields as
/// [`PendingQuestionPrompt`] — reuse that conversion rather than a second
/// hand-written field list that could drift from it.
impl From<PendingQuestionPrompt> for TurnEvent {
    fn from(p: PendingQuestionPrompt) -> Self {
        Self::QuestionPending {
            turn_id: p.turn_id,
            call_id: p.call_id,
            index: p.index,
            header: p.header,
            question: p.question,
            options: p.options,
            args_json: p.args_json,
            answer_token: p.answer_token,
            already_surfaced: p.already_surfaced,
        }
    }
}

/// [`TurnEvent::ApprovalPending`] carries the exact same fields as
/// [`PendingApprovalPrompt`] — reuse that conversion rather than a second
/// hand-written field list that could drift from it.
impl From<PendingApprovalPrompt> for TurnEvent {
    fn from(p: PendingApprovalPrompt) -> Self {
        Self::ApprovalPending {
            turn_id: p.turn_id,
            request_id: p.request_id,
            tool_name: p.tool_name,
            title: p.title,
            args_json: p.args_json,
            reason: p.reason,
            resolve_token: p.resolve_token,
            preview: p.preview,
        }
    }
}

/// Whether this turn's `AgentEnd` carried a wallet-link prompt (`#519`), and
/// if so, whether a URL was known. A dedicated enum instead of
/// `Option<Option<String>>` (`clippy::option_option`) for the same 3-way
/// distinction: not this turn / this turn with no known URL / this turn with
/// a known URL.
enum WalletLinkPrompt {
    /// This turn carried no wallet-link signal.
    None,
    /// This turn needed a USABLE linked wallet; `link_url` is `Some(url)`
    /// when the deployment or per-caller link URL was known, `None` when it
    /// wasn't (an admin pointer is rendered instead). `renewal` (`#2122`)
    /// distinguishes a caller who already linked and lapsed from one who
    /// never linked at all — see `polyc_proto::wallet_link_prompt`.
    Present {
        link_url: Option<String>,
        renewal: bool,
        /// True when the caller directly asked to link (or replace) a
        /// wallet — never the payment-interrupt path `renewal` covers. See
        /// `polyc_proto::wallet_link_requested_prompt`.
        requested: bool,
    },
}

/// Decide [`AgentDialer::run_turn_with`]'s final reply string from what the
/// buffered aggregation collected. Pulled out of the loop body so the
/// decision is unit-testable without a live stream, mirroring
/// [`aggregate_output_message`]'s identical rationale.
///
/// A wallet-link-needed turn (`#519` follow-up) REPLACES `text_parts`
/// outright with the deterministic shared-copy prompt — not an append. A
/// third review round found this path shares the same live-delta hazard as
/// Slack/Telegram (`crates/slack/src/handler.rs::interrupt_holdback_text`,
/// `crates/telegram/src/handler.rs`'s reply-override block): `run_turn_with`
/// and the streaming API dial the identical `AgentService.connect` RPC
/// (`build_request`), and the control plane forwards the harness's live
/// `TextDelta`s to EVERY client, buffered or not
/// (`crates/control-plane/src/grpc/mod.rs::should_emit_final_batch` skips
/// re-emitting the `internal_only`-filtered terminal batch whenever any delta
/// was forwarded — the normal case for a streaming provider). So `text_parts`
/// here can just as easily hold the model's own raw, unfiltered "you'll need
/// to link a wallet…" narration as `aggregate_output_message` never sees an
/// `internal_only` flag set on those live deltas (only the terminal batch
/// carries it, and that batch is what gets suppressed). Appending the
/// deterministic prompt on top of that would print both — the exact
/// duplicate-CTA bug this PR exists to fix, just inside one buffered message
/// instead of two Slack bubbles. Buffered edges (discord/email/
/// trigger/cli) only ever post this single final string, so a full replace
/// here is sufficient — there's no earlier partial render to leave stranded.
///
/// `wallet_update_prompt` (issue #1041/#1159) and `wallet_revoke_prompt`
/// (issue #1042/#1156) get the identical replace treatment, for the
/// identical reason — checked after `wallet_link_prompt`, in that order: an
/// arbitrary but documented tie-break for the rare case a turn somehow hits
/// more than one of these mutually-exclusive, distinct-tool signals at
/// once.
#[allow(clippy::too_many_arguments)] // each names a DISTINCT, mutually-exclusive prompt kind
fn finalize_buffered_reply(
    text_parts: &[String],
    scaffolding: &[String],
    wallet_link_prompt: WalletLinkPrompt,
    wallet_update_prompt: Option<(&str, &str)>,
    wallet_revoke_prompt: Option<&str>,
) -> String {
    if let WalletLinkPrompt::Present {
        link_url,
        renewal,
        requested,
    } = wallet_link_prompt
    {
        return if requested {
            // A direct `wallet_link` request always mints a URL before this
            // prompt is ever set — see `wallet_nav::link`'s doc comment —
            // so this always carries one.
            polyc_proto::wallet_link_requested_prompt(link_url.as_deref().unwrap_or_default())
        } else {
            polyc_proto::wallet_link_prompt(link_url.as_deref(), renewal)
        };
    }
    if let Some((url, new_limit)) = wallet_update_prompt {
        return polyc_proto::wallet_update_prompt(url, new_limit);
    }
    if let Some(url) = wallet_revoke_prompt {
        return polyc_proto::wallet_revoke_prompt(url);
    }
    if text_parts.is_empty() {
        scaffolding.join("\n")
    } else {
        text_parts.join("\n")
    }
}

// ---------------------------------------------------------------------------
// Operator mailbox — the system-initiated, operator-authorized approval path.
//
// The control plane holds no chat tokens; the edges do (docs/reference/
// 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
/// 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,
    },
    /// The closed-loop follow-up after an approved `UpgradeTo`: the executor
    /// launched the roll, then the control plane re-observed cluster state
    /// within a bounded window. A plain, no-decision DM reporting an outcome
    /// that already happened.
    UpgradeOutcome {
        /// The release version the roll targeted.
        version: String,
        /// What the closed-loop observation confirmed.
        outcome: UpgradeOutcomeKind,
    },
    /// 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,
}

/// What the closed-loop rollout observation confirmed after an approved
/// `UpgradeTo`, in edge-renderable terms — the buffered analog of the wire
/// `UpgradeOutcome.Kind`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UpgradeOutcomeKind {
    /// Every changed Deployment's rollout confirmed healthy at the target
    /// digests.
    Success,
    /// The observation window elapsed with at least one changed Deployment
    /// confirmed NOT healthy at the target digests.
    Failed,
    /// The observation window elapsed without a confirmed outcome either way
    /// — distinct from `Failed` so the copy never claims to know the roll
    /// broke when it might have simply succeeded unobserved.
    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> {
        Ok(Self {
            client: build_control_client(addr, None, NotificationServiceClient::new)?,
        })
    }

    /// Build a dialer pointed at `addr`, authenticated with `bearer`.
    ///
    /// Every call this dialer makes carries an `Authorization: Bearer
    /// <bearer>` header. `NotificationService` doesn't send `AgentStart`, so
    /// no signed [`AssertedAttribution`] envelope rides these calls —
    /// bearer-only.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value.
    pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            client: build_control_client(addr, Some(bearer), NotificationServiceClient::new)?,
        })
    }

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

    /// Server-streaming replacement for [`Self::poll_pending`] (#803): opens a
    /// long-lived `Subscribe` stream that pushes undelivered notifications for
    /// `provider` as they're durably recorded, instead of a fixed polling
    /// interval. The initial connect ALSO catches a freshly (re)started edge up
    /// on whatever is already undelivered — the same coverage a first
    /// `poll_pending` call would give, just pushed rather than pulled.
    ///
    /// # Errors
    /// The outer `Result` carries [`DialError::Connect`] if opening the stream
    /// fails. Each item is a `Result` so a later transport/decode error
    /// surfaces inline without tearing down the whole stream.
    pub async fn subscribe(
        &self,
        provider: &str,
    ) -> Result<impl Stream<Item = Result<PendingNotice, DialError>>, DialError> {
        let request = SubscribeRequest {
            provider: provider.to_owned(),
            ..Default::default()
        };
        let mut stream = self
            .client
            .subscribe_with_options(request, traced_options())
            .await?;
        Ok(async_stream::try_stream! {
            while let Some(view) = stream.message().await? {
                yield pending_notice(view.to_owned_message());
            }
        })
    }
}

/// 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 },
            ops_action_view::Action::UpgradeOutcome(u) => OpsAction::UpgradeOutcome {
                version: u.version,
                outcome: match u.kind.as_known() {
                    Some(upgrade_outcome::Kind::SUCCESS) => UpgradeOutcomeKind::Success,
                    Some(upgrade_outcome::Kind::FAILED) => UpgradeOutcomeKind::Failed,
                    Some(
                        upgrade_outcome::Kind::UNKNOWN | upgrade_outcome::Kind::KIND_UNSPECIFIED,
                    )
                    | None => UpgradeOutcomeKind::Unknown,
                },
            },
        },
    );
    PendingNotice {
        action_id: p.action_id,
        target: p.target,
        action,
        expires_unix: p.expires_unix,
        payload_hash: p.payload_hash,
        delivered: p.delivered,
    }
}

/// Reusable handle for the control plane's `RoutineService` — the
/// manual-fire dev-path affordance (`#1369`).
///
/// The routine-enrollment ceremony trigger this type used to also carry was
/// deleted in #1551: the prompt-routines pivot removed edge delivery and
/// made sharing duplication-first, so no subscription/enrollment
/// relationship returns.
///
/// Shares the `AgentService` endpoint (one internal Connect port), so it is
/// built from the same address an edge already holds.
#[derive(Clone)]
pub struct RoutineDialer {
    client: Arc<RoutineServiceClient<HttpClient>>,
}

impl RoutineDialer {
    /// 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> {
        Ok(Self {
            client: build_control_client(addr, None, RoutineServiceClient::new)?,
        })
    }

    /// Manually fire a routine right now (`#1369`) — the
    /// standard way to test any routine without waiting for its schedule.
    ///
    /// `occurrence` empty mints one from the current time
    /// (`{routine}-{epoch_minutes}`); non-empty is used verbatim, so a caller
    /// can pin (or replay) an exact occurrence. The control plane's admin
    /// gate (`#1489`) requires, in order: `admin_bearer` presented as
    /// `Authorization: Bearer <admin_bearer>` — the SAME admin-service
    /// credential scheme `POLYCHROME_ADMIN_SERVICE_KEYS` seeds — verified
    /// BEFORE the body is trusted at all, and only then `actor` resolved to
    /// an admin persona. Neither alone is enough — an asserted `actor` with
    /// no valid credential is refused just as a valid credential naming a
    /// non-admin `actor` is.
    ///
    /// Post-`#1595` this is a clean no-op beyond routine resolution and
    /// occurrence minting — the fixed-content fire path this call used to
    /// hand resolved content templates to is retired.
    ///
    /// `admin_bearer` is a per-call header, not one of this dialer's default
    /// headers (`RoutineDialer::new` dials without a bearer), so it goes
    /// through the crate's shared sensitive-`HeaderValue` construction site
    /// directly rather than the shared default-`HeaderMap` builder every
    /// other dialer's `with_bearer` uses — same construction, same
    /// `set_sensitive` marking, just built per call instead of once at dial
    /// time.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidBearer`] if `admin_bearer` can't be
    /// encoded as an HTTP header value, or [`DialError::Connect`] for any
    /// transport/encoding error, which carries the control plane's Connect
    /// error (unknown routine, a missing or invalid `admin_bearer`, a
    /// non-admin `actor`, or an unconfigured harness).
    pub async fn fire_routine(
        &self,
        routine: &str,
        occurrence: &str,
        actor: ExternalIdentity,
        admin_bearer: &str,
    ) -> Result<FiredRoutine, DialError> {
        let request = FireRoutineRequest {
            routine: routine.to_owned(),
            occurrence: occurrence.to_owned(),
            actor: buffa::MessageField::some(actor),
            ..Default::default()
        };
        let options = traced_options().with_header(
            http::header::AUTHORIZATION,
            bearer_header_value(admin_bearer)?,
        );
        let reply = self
            .client
            .fire_routine_with_options(request, options)
            .await?
            .into_owned();
        Ok(FiredRoutine {
            occurrence: reply.occurrence,
        })
    }
}

/// The result of [`RoutineDialer::fire_routine`]: the occurrence the firing used.
///
/// Post-`#1595` this is the whole reply — the fixed-content fire path that
/// used to report one outcome per content template is retired.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FiredRoutine {
    /// The occurrence identity this firing used (the caller's, or minted).
    pub occurrence: String,
}

/// A credential key's lifecycle state.
///
/// Buffered off the wire enum so downstream code (the CLI's `edge` verbs)
/// matches on a closed Rust enum instead of taking a `polyc-proto` dependency
/// of its own — this crate is the one place that talks connectrpc/buffa/proto
/// directly on this path, mirroring how [`CompactionReason`] already buffers
/// off its own wire enum.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CredentialKeyState {
    /// Verifies now — the credential's currently-presented key.
    Active,
    /// Still verifies, but is being phased out. The rotation runbook's step
    /// 4 watches this state's verification traffic go quiet before step 5
    /// (`CredentialDialer::retire_credential_key`) retires it.
    Retiring,
    /// Never verifies again.
    Revoked,
    /// A lifecycle value this client doesn't recognize yet. Reported
    /// distinctly rather than guessed at, since a mutation must never assert
    /// a state it can't read.
    Unknown,
}

impl From<buffa::EnumValue<WireCredentialKeyState>> for CredentialKeyState {
    fn from(state: buffa::EnumValue<WireCredentialKeyState>) -> Self {
        match state {
            buffa::EnumValue::Known(WireCredentialKeyState::KEY_STATE_ACTIVE) => Self::Active,
            buffa::EnumValue::Known(WireCredentialKeyState::KEY_STATE_RETIRING) => Self::Retiring,
            buffa::EnumValue::Known(WireCredentialKeyState::KEY_STATE_REVOKED) => Self::Revoked,
            buffa::EnumValue::Known(WireCredentialKeyState::KEY_STATE_UNSPECIFIED)
            | buffa::EnumValue::Unknown(_) => Self::Unknown,
        }
    }
}

/// One credential key's operator-visible summary — the buffered analog of
/// the wire `CredentialKeySummary`. Carries no salt, digest, or public key:
/// the control plane never sends one (INV-C3).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CredentialKeySummary {
    /// The key id — what `retire_credential_key` names, and what the audit
    /// trail and the retiring-verification metric record.
    pub kid: String,
    /// This key's lifecycle state.
    pub state: CredentialKeyState,
    /// When this key became active (unix ms).
    pub activated_at_ms: u64,
    /// Hard expiry (unix ms), if one was set.
    pub not_after_ms: Option<u64>,
    /// Unix ms this key FIRST verified a bearer successfully, if it ever
    /// has. `None` means it never has yet — the ordinary state for a
    /// freshly enrolled or freshly rotated-in key, not a fault (issue
    /// #1696). Recorded once and never updated again: a first-use
    /// timestamp, not a last-verified one.
    pub confirmed_at_ms: Option<u64>,
}

impl From<WireCredentialKeySummary> for CredentialKeySummary {
    fn from(wire: WireCredentialKeySummary) -> Self {
        Self {
            kid: wire.kid,
            state: wire.state.into(),
            activated_at_ms: wire.activated_at_ms,
            not_after_ms: wire.not_after_ms,
            confirmed_at_ms: wire.confirmed_at_ms,
        }
    }
}

/// One stored credential's operator-visible summary — the buffered analog of
/// the wire `CredentialSummary`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CredentialRecordSummary {
    /// The caller identity string this record is keyed by.
    pub edge_id: String,
    /// Stable subject recorded for audit; survives key rotation.
    pub principal: String,
    /// Conversation-id namespace prefixes this record may write into.
    pub allowed_namespaces: Vec<String>,
    /// This record's key set, oldest entry first.
    pub keys: Vec<CredentialKeySummary>,
    /// Unix ms this credential was revoked, if it has been. Present means the
    /// record is a tombstone: it verifies nothing and never will again, and
    /// it is listed only so an operator can see when it was revoked. Any
    /// rendering of this record MUST say so — a revoked credential must never
    /// read as live.
    pub revoked_at_ms: Option<u64>,
    /// Whether this record clears the control plane's transport gate and may
    /// sign a turn's identity envelope.
    pub grants_edge: bool,
    /// Whether this record is accepted as the admin service credential.
    pub grants_admin: bool,
    /// Whether this record is accepted as the backend-email-attestation
    /// service credential (`PersonaService.AttestVerifiedEmail`).
    pub grants_attest_email: bool,
}

impl From<WireCredentialSummary> for CredentialRecordSummary {
    fn from(wire: WireCredentialSummary) -> Self {
        Self {
            edge_id: wire.edge_id,
            principal: wire.principal,
            allowed_namespaces: wire.allowed_namespaces,
            revoked_at_ms: wire.revoked_at_ms,
            grants_edge: wire.grants_edge,
            grants_admin: wire.grants_admin,
            grants_attest_email: wire.grants_attest_email,
            keys: wire
                .keys
                .into_iter()
                .map(CredentialKeySummary::from)
                .collect(),
        }
    }
}

/// The verifier half of one credential key, as an operator authors it.
///
/// The salted digest of a secret they wrote to their own secret manager, plus
/// the public key of the pair that secret's holder signs identity envelopes
/// with. Never a raw secret — [`CredentialDialer::enroll_credential`] and
/// [`CredentialDialer::add_credential_key`] don't accept one: the control
/// plane never mints or holds the secret itself.
#[derive(Debug, Clone)]
pub struct CredentialVerifier {
    /// Stable, operator-visible key id, unique within its record.
    pub kid: String,
    /// Per-key salt mixed into the digest (not secret).
    pub salt: String,
    /// `hex(sha256(salt || secret))`. The secret itself is never sent.
    pub secret_sha256: String,
    /// The ed25519 public key this key signs identity envelopes with,
    /// hex-encoded.
    pub signer_pk_hex: String,
    /// Hard expiry (unix ms), independent of lifecycle state. `None` means
    /// no expiry.
    pub not_after_ms: Option<u64>,
}

/// One idempotent credential-enrollment operation.
///
/// `operation_id` is caller-stable: reuse the same value, with the same
/// fields, after an ambiguous transport failure. State returns the original
/// outcome instead of applying the enrollment twice.
#[derive(Debug, Clone)]
pub struct CredentialEnrollment {
    /// Stable identity for this logical enrollment attempt.
    pub operation_id: String,
    /// Stable credential-record id.
    pub edge_id: String,
    /// Principal authenticated by the credential.
    pub principal: String,
    /// Namespace patterns this credential may enter.
    pub allowed_namespaces: Vec<String>,
    /// First active verifier for the record.
    pub key: CredentialVerifier,
    /// Whether this credential may enter through an edge transport.
    pub grants_edge: bool,
    /// Whether this credential may call admin-gated services.
    pub grants_admin: bool,
    /// Whether this credential is accepted as the backend-email-attestation
    /// service credential (`PersonaService.AttestVerifiedEmail`).
    pub grants_attest_email: bool,
}

impl From<CredentialVerifier> for WireCredentialKeyVerifier {
    fn from(verifier: CredentialVerifier) -> Self {
        Self {
            kid: verifier.kid,
            salt: verifier.salt,
            secret_sha256: verifier.secret_sha256,
            signer_pk_hex: verifier.signer_pk_hex,
            not_after_ms: verifier.not_after_ms,
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

/// What [`CredentialDialer::enroll_credential`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EnrolledCredential {
    /// The enrolled record's id.
    pub edge_id: String,
    /// The id of the key that is now active.
    pub kid: String,
}

/// One bounded page from credential lifecycle administration.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CredentialSummaryPage {
    /// Records in canonical credential-id order.
    pub records: Vec<CredentialRecordSummary>,
    /// Resume cursor when another record exists.
    pub next_after: Option<String>,
    /// Authority revision at which this page was read.
    pub snapshot_revision: u64,
}

/// What [`CredentialDialer::add_credential_key`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AddedCredentialKey {
    /// The record that was changed.
    pub edge_id: String,
    /// The id of the key that is now active.
    pub kid: String,
    /// The ids of the keys that dropped from active to retiring. They keep
    /// verifying until [`CredentialDialer::retire_credential_key`] names
    /// them, so nothing is disrupted (INV-C2).
    pub retiring_kids: Vec<String>,
}

/// What [`CredentialDialer::retire_credential_key`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RetiredCredentialKey {
    /// The record that was changed.
    pub edge_id: String,
    /// The key that is now revoked.
    pub kid: String,
}

/// What [`CredentialDialer::revoke_credential`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RevokedCredential {
    /// The record that was removed.
    pub edge_id: String,
}

/// Reusable handle for the control plane's `CredentialService` — the
/// admin-gated live enrollment, rotation, and revocation surface over State's
/// credential authority.
///
/// Shares the `AgentService` endpoint — all these services are served on one
/// internal Connect port — so it is built from the same address. Every RPC on
/// this service requires the `#803` admin service credential before it looks
/// at anything else in the request body, so unlike [`PersonaDialer`] this
/// dialer has no unauthenticated `new`: it is built ONLY through
/// [`Self::new_admin`].
#[derive(Clone)]
pub struct CredentialDialer {
    client: Arc<CredentialServiceClient<HttpClient>>,
}

impl CredentialDialer {
    /// Build a dialer pointed at `addr`, presenting `bearer` as the `#803`
    /// admin service credential (`Authorization: Bearer pc_<id>_<secret>`,
    /// a current record in State's credential authority carrying admin
    /// access) every `CredentialService` RPC checks before it will even look
    /// at the request body.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI,
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value, or [`DialError::Tls`] if an `https` endpoint's TLS setup
    /// fails.
    pub fn new_admin(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            client: build_control_client(addr, Some(bearer), CredentialServiceClient::new)?,
        })
    }

    /// One bounded page of stored records: id, principal, namespaces, and
    /// each key's id, lifecycle state, and expiry. Never any secret material
    /// — the control plane refuses to send a salt, digest, or public key
    /// (INV-C3).
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error,
    /// including `permission_denied` for a missing or invalid admin bearer.
    pub async fn list_credentials_page(
        &self,
        after: Option<&str>,
    ) -> Result<CredentialSummaryPage, DialError> {
        let reply = self
            .client
            .list_credentials_with_options(
                ListCredentialsRequest {
                    after: after.map(str::to_owned),
                    // CredentialService's public protocol pins the same
                    // count ceiling as State's typed page contract.
                    limit: 64,
                    __buffa_unknown_fields: buffa::UnknownFields::default(),
                },
                traced_options(),
            )
            .await?
            .into_owned();
        if reply.records.len() > 64
            || reply.next_after.as_ref().is_some_and(|next| {
                reply.records.last().map(|record| &record.edge_id) != Some(next)
                    || after.is_some_and(|prior| next.as_str() <= prior)
            })
        {
            return Err(connectrpc::ConnectError::internal(
                "credential authority returned an invalid page cursor",
            )
            .into());
        }
        Ok(CredentialSummaryPage {
            records: reply
                .records
                .into_iter()
                .map(CredentialRecordSummary::from)
                .collect(),
            next_after: reply.next_after,
            snapshot_revision: reply.snapshot_revision,
        })
    }

    /// Reads every bounded page in canonical order.
    ///
    /// The protocol remains paginated, so no response can exceed the
    /// credential-family wire bound. This convenience method is itself
    /// bounded by State's deployment-wide credential-count limit.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any page transport/encoding error.
    pub async fn list_credentials(&self) -> Result<Vec<CredentialRecordSummary>, DialError> {
        for _ in 0..3 {
            let mut records = Vec::new();
            let mut after = None;
            let mut revision = None;
            let coherent = loop {
                let page = self.list_credentials_page(after.as_deref()).await?;
                if revision.is_some_and(|expected| expected != page.snapshot_revision) {
                    break false;
                }
                revision.get_or_insert(page.snapshot_revision);
                records.extend(page.records);
                if records.len() > 1_024 {
                    return Err(connectrpc::ConnectError::internal(
                        "credential authority exceeded its directory bound",
                    )
                    .into());
                }
                let Some(next) = page.next_after else {
                    break true;
                };
                after = Some(next);
            };
            if coherent {
                return Ok(records);
            }
        }
        Err(connectrpc::ConnectError::aborted(
            "credential authority changed throughout the bounded listing retry",
        )
        .into())
    }

    /// Create a record with its first key from an operator-supplied
    /// verifier. Refuses (`already_exists`) if `edge_id` is already enrolled
    /// — add a key to rotate an existing record instead.
    ///
    /// `grants_edge` admits the record at the control plane's transport gate
    /// and lets its keys sign a turn's identity envelope; `grants_admin`
    /// makes it an admin service credential. A record needs at least one, and
    /// a credential that must reach a doubly-gated RPC — every
    /// `PersonaService` admin verb, `RoutineService.FireRoutine` — needs
    /// both, since one presented bearer clears two independent checks.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error,
    /// including `invalid_argument` when neither capability is granted.
    pub async fn enroll_credential(
        &self,
        enrollment: CredentialEnrollment,
    ) -> Result<EnrolledCredential, DialError> {
        let request = EnrollCredentialRequest {
            operation_id: enrollment.operation_id,
            edge_id: enrollment.edge_id,
            principal: enrollment.principal,
            allowed_namespaces: enrollment.allowed_namespaces,
            key: buffa::MessageField::some(enrollment.key.into()),
            grants_edge: enrollment.grants_edge,
            grants_admin: enrollment.grants_admin,
            grants_attest_email: enrollment.grants_attest_email,
            ..Default::default()
        };
        let reply = self
            .client
            .enroll_credential_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(EnrolledCredential {
            edge_id: reply.edge_id,
            kid: reply.kid,
        })
    }

    /// Append a new active key to `edge_id` — step 2 of a rotation. The
    /// key(s) currently active drop to retiring and keep verifying until
    /// [`Self::retire_credential_key`] names them (INV-C2): there is no
    /// instant in which a correctly-configured caller is refused.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error,
    /// including `not_found` for an unknown `edge_id`.
    pub async fn add_credential_key(
        &self,
        operation_id: &str,
        edge_id: &str,
        key: CredentialVerifier,
    ) -> Result<AddedCredentialKey, DialError> {
        let request = AddCredentialKeyRequest {
            operation_id: operation_id.to_owned(),
            edge_id: edge_id.to_owned(),
            key: buffa::MessageField::some(key.into()),
            ..Default::default()
        };
        let reply = self
            .client
            .add_credential_key_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(AddedCredentialKey {
            edge_id: reply.edge_id,
            kid: reply.kid,
            retiring_kids: reply.retiring_kids,
        })
    }

    /// Mark a retiring key revoked — step 5 of a rotation. It never
    /// verifies again.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error,
    /// including `failed_precondition` when `kid` isn't currently retiring.
    pub async fn retire_credential_key(
        &self,
        operation_id: &str,
        edge_id: &str,
        kid: &str,
    ) -> Result<RetiredCredentialKey, DialError> {
        let request = RetireCredentialKeyRequest {
            operation_id: operation_id.to_owned(),
            edge_id: edge_id.to_owned(),
            kid: kid.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .retire_credential_key_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(RetiredCredentialKey {
            edge_id: reply.edge_id,
            kid: reply.kid,
        })
    }

    /// Kill a whole credential immediately. Every one of its keys stops
    /// verifying and the record is removed from the store, so a restart
    /// cannot bring it back (INV-C1).
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport/encoding error,
    /// including `not_found` for an unknown `edge_id`.
    pub async fn revoke_credential(
        &self,
        operation_id: &str,
        edge_id: &str,
    ) -> Result<RevokedCredential, DialError> {
        let request = RevokeCredentialRequest {
            operation_id: operation_id.to_owned(),
            edge_id: edge_id.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .revoke_credential_with_options(request, traced_options())
            .await?
            .into_owned();
        Ok(RevokedCredential {
            edge_id: reply.edge_id,
        })
    }
}

/// 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 authorized to approve / 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)
            }
            // The closed-loop follow-up: reports what the approved roll
            // actually did, never claiming success it couldn't confirm.
            Self::UpgradeOutcome { version, outcome } => match outcome {
                UpgradeOutcomeKind::Success => {
                    format!("✅ The cluster is now running {version}.")
                }
                UpgradeOutcomeKind::Failed => format!(
                    "⚠️ The upgrade to {version} did not confirm healthy — check `polychrome status`."
                ),
                UpgradeOutcomeKind::Unknown => format!(
                    "❓ Could not confirm whether the upgrade to {version} finished — check `polychrome status`."
                ),
            },
            // 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::UpgradeOutcome { .. } | Self::Unknown => None,
        }
    }

    /// Whether this action carries an Approve/Deny decision at all.
    ///
    /// An [`OpsAction::UpgradeOutcome`] reports something that already happened,
    /// not an ask.
    /// Every other known action (and `Unknown`, defensively — a future wire
    /// shape this client doesn't recognize yet still gets rendered as a
    /// decidable ask rather than silently dropped) is decidable.
    #[must_use]
    pub const fn is_decision(&self) -> bool {
        !matches!(self, Self::UpgradeOutcome { .. })
    }
}

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> {
        Ok(Self {
            client: build_control_client(addr, None, OperatorMailboxServiceClient::new)?,
        })
    }

    /// Build a dialer pointed at `addr`, authenticated with `bearer`.
    ///
    /// Every call this dialer makes carries an `Authorization: Bearer
    /// <bearer>` header. `OperatorMailboxService` doesn't send `AgentStart`,
    /// so no signed [`AssertedAttribution`] envelope rides these calls —
    /// bearer-only.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI, or
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value.
    pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            client: build_control_client(addr, Some(bearer), OperatorMailboxServiceClient::new)?,
        })
    }

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

/// The lifecycle state of one durable agent task.
///
/// The wire enum, mapped once here so an edge branches on a Rust enum rather
/// than on generated proto constants. `InputRequired` and `AuthRequired`
/// pause a task without finishing it; the four terminal states are final.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AgentTaskState {
    /// Accepted, not yet started. The only state a new task is created in.
    Submitted,
    /// The turn is running.
    Working,
    /// Paused while a person decides.
    InputRequired,
    /// Paused while the caller authenticates.
    AuthRequired,
    /// Finished successfully.
    Completed,
    /// Finished unsuccessfully.
    Failed,
    /// Cancelled by the caller.
    Canceled,
    /// The agent declined the task.
    Rejected,
}

impl AgentTaskState {
    /// Reports whether the task has finished and can never move again.
    #[must_use]
    pub const fn is_terminal(self) -> bool {
        matches!(
            self,
            Self::Completed | Self::Failed | Self::Canceled | Self::Rejected
        )
    }
}

impl From<AgentTaskState> for WireAgentTaskState {
    fn from(value: AgentTaskState) -> Self {
        match value {
            AgentTaskState::Submitted => Self::AGENT_TASK_STATE_SUBMITTED,
            AgentTaskState::Working => Self::AGENT_TASK_STATE_WORKING,
            AgentTaskState::InputRequired => Self::AGENT_TASK_STATE_INPUT_REQUIRED,
            AgentTaskState::AuthRequired => Self::AGENT_TASK_STATE_AUTH_REQUIRED,
            AgentTaskState::Completed => Self::AGENT_TASK_STATE_COMPLETED,
            AgentTaskState::Failed => Self::AGENT_TASK_STATE_FAILED,
            AgentTaskState::Canceled => Self::AGENT_TASK_STATE_CANCELED,
            AgentTaskState::Rejected => Self::AGENT_TASK_STATE_REJECTED,
        }
    }
}

/// One durable task, as [`TaskDialer`] read it back.
///
/// Every content field is opaque bytes: the control plane and the state plane
/// behind it store and return these frames without interpreting them, so the
/// caller's own surface format never becomes a second copy of itself there.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AgentTaskRecord {
    /// Stable task identity.
    pub task_id: String,
    /// The conversation the task belongs to.
    pub context_id: String,
    /// Current lifecycle state.
    pub state: AgentTaskState,
    /// Opaque detail attached to the current status; empty when it carries none.
    pub status_detail: Vec<u8>,
    /// Opaque output frames.
    pub artifacts: Vec<Vec<u8>>,
    /// Opaque history frames, in order.
    pub history: Vec<Vec<u8>>,
    /// Opaque free-form metadata.
    pub metadata: Vec<u8>,
    /// When the task was created, in milliseconds since the Unix epoch.
    pub created_at_ms: u64,
    /// When the most recent accepted transition landed.
    pub updated_at_ms: u64,
}

impl TryFrom<WireAgentTask> for AgentTaskRecord {
    type Error = DialError;

    /// Names every field explicitly rather than spreading a default, so a
    /// field added to either side fails to compile instead of arriving zeroed
    /// (`#1241`/`#1238`).
    fn try_from(value: WireAgentTask) -> Result<Self, Self::Error> {
        let WireAgentTask {
            task_id,
            context_id,
            state,
            status_detail,
            artifacts,
            history,
            metadata,
            created_at_ms,
            updated_at_ms,
            __buffa_unknown_fields: _,
        } = value;
        let state = match state.as_known() {
            Some(WireAgentTaskState::AGENT_TASK_STATE_SUBMITTED) => AgentTaskState::Submitted,
            Some(WireAgentTaskState::AGENT_TASK_STATE_WORKING) => AgentTaskState::Working,
            Some(WireAgentTaskState::AGENT_TASK_STATE_INPUT_REQUIRED) => {
                AgentTaskState::InputRequired
            }
            Some(WireAgentTaskState::AGENT_TASK_STATE_AUTH_REQUIRED) => {
                AgentTaskState::AuthRequired
            }
            Some(WireAgentTaskState::AGENT_TASK_STATE_COMPLETED) => AgentTaskState::Completed,
            Some(WireAgentTaskState::AGENT_TASK_STATE_FAILED) => AgentTaskState::Failed,
            Some(WireAgentTaskState::AGENT_TASK_STATE_CANCELED) => AgentTaskState::Canceled,
            Some(WireAgentTaskState::AGENT_TASK_STATE_REJECTED) => AgentTaskState::Rejected,
            Some(WireAgentTaskState::AGENT_TASK_STATE_UNSPECIFIED) | None => {
                return Err(DialError::Connect(connectrpc::ConnectError::internal(
                    "the task record names no lifecycle state",
                )));
            }
        };
        Ok(Self {
            task_id,
            context_id,
            state,
            status_detail,
            artifacts,
            history,
            metadata,
            created_at_ms,
            updated_at_ms,
        })
    }
}

/// The successor a [`TaskDialer::transition_task`] call asks for.
///
/// Status detail, artifacts, and metadata replace what the record carries;
/// history is extended by [`Self::appended_history`] and never rewritten.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AgentTaskTransition {
    /// Opaque detail for the new status; empty clears it.
    pub status_detail: Vec<u8>,
    /// Opaque output frames the task carries after this transition.
    pub artifacts: Vec<Vec<u8>>,
    /// Opaque metadata the task carries after this transition.
    pub metadata: Vec<u8>,
    /// Opaque history frames this transition appends, in order.
    pub appended_history: Vec<Vec<u8>>,
}

/// Exact State-issued ownership for one task dispatch.
///
/// The dispatch, worker, and attempt identities bind the numeric fence to one
/// authenticated execution. A caller echoes the whole value on every later
/// update; a raw fence is never sufficient.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AgentTaskOwnership {
    /// Stable State dispatch identity of the source event.
    pub dispatch_id: String,
    /// Stable identity of the edge process handling this delivery.
    pub worker_instance: String,
    /// Stable identity of this one execution attempt.
    pub attempt_id: String,
    /// State-wide fencing token issued to the attempt.
    pub fence: u64,
    /// Server-derived digest of the history attached by `ClaimTask`.
    pub claim_digest: Vec<u8>,
}

impl From<&AgentTaskOwnership> for WireAgentTaskOwnership {
    fn from(value: &AgentTaskOwnership) -> Self {
        Self {
            dispatch_id: value.dispatch_id.clone(),
            worker_instance: value.worker_instance.clone(),
            attempt_id: value.attempt_id.clone(),
            fence: value.fence,
            claim_digest: value.claim_digest.clone(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        }
    }
}

impl From<WireAgentTaskOwnership> for AgentTaskOwnership {
    fn from(value: WireAgentTaskOwnership) -> Self {
        Self {
            dispatch_id: value.dispatch_id,
            worker_instance: value.worker_instance,
            attempt_id: value.attempt_id,
            fence: value.fence,
            claim_digest: value.claim_digest,
        }
    }
}

/// One page of a context's tasks, as [`TaskDialer::list_tasks`] read it back.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AgentTaskPage {
    /// The page's tasks, in identity order.
    pub tasks: Vec<AgentTaskRecord>,
    /// The identity to pass back as `after` for the next page; `None` when
    /// this page reached the end of the context's index.
    pub next_after: Option<String>,
}

/// Reusable handle for the control plane's `AgentTaskService`.
///
/// The durable task-record path a protocol edge drives its lifecycle over.
/// Shares the `AgentService` endpoint — both are served on one Connect port —
/// so it is built from the same address.
///
/// Bearer-only, exactly like [`ApprovalDialer::with_bearer`]: these calls send
/// no `AgentStart`, so no signed attribution envelope rides them.
#[derive(Clone)]
pub struct TaskDialer {
    client: Arc<AgentTaskServiceClient<HttpClient>>,
}

const fn is_ambiguous_task_error(error: &connectrpc::ConnectError) -> bool {
    matches!(
        error.code,
        connectrpc::ErrorCode::Unavailable
            | connectrpc::ErrorCode::DeadlineExceeded
            | connectrpc::ErrorCode::ResourceExhausted
            | connectrpc::ErrorCode::Aborted
    )
}

impl TaskDialer {
    /// 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> {
        Ok(Self {
            client: build_control_client(addr, None, AgentTaskServiceClient::new)?,
        })
    }

    /// Build a dialer pointed at `addr`, authenticated with `bearer`.
    ///
    /// # Errors
    /// Returns [`DialError::InvalidAddress`] if `addr` isn't a valid URI,
    /// [`DialError::Tls`] if an `https` endpoint's TLS setup fails, or
    /// [`DialError::InvalidBearer`] if `bearer` can't be encoded as an HTTP
    /// header value.
    pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
        Ok(Self {
            client: build_control_client(addr, Some(bearer), AgentTaskServiceClient::new)?,
        })
    }

    /// Mint a task in [`AgentTaskState::Submitted`] and index it under
    /// `context_id`, carrying `history` as its opening frames.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport or encoding error,
    /// including the control plane's refusal of an identity that already
    /// exists (`already_exists`).
    pub async fn create_task(
        &self,
        task_id: &str,
        context_id: &str,
        history: Vec<Vec<u8>>,
    ) -> Result<AgentTaskRecord, DialError> {
        let request = CreateAgentTaskRequest {
            task_id: task_id.to_owned(),
            context_id: context_id.to_owned(),
            history,
            ..Default::default()
        };
        let reply = self
            .client
            .create_task_with_options(request, traced_options())
            .await?
            .into_owned();
        Self::require_task(reply.task.into_option())
    }

    /// Move a live task to `state`, carrying `successor`'s frames.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport or encoding error,
    /// including `not_found` for an unknown task and `failed_precondition`
    /// for a task that already finished.
    pub async fn transition_task(
        &self,
        task_id: &str,
        state: AgentTaskState,
        successor: AgentTaskTransition,
        ownership: &AgentTaskOwnership,
    ) -> Result<AgentTaskRecord, DialError> {
        let AgentTaskTransition {
            status_detail,
            artifacts,
            metadata,
            appended_history,
        } = successor;
        let request = TransitionAgentTaskRequest {
            task_id: task_id.to_owned(),
            state: WireAgentTaskState::from(state).into(),
            status_detail,
            artifacts,
            metadata,
            appended_history,
            ownership: buffa::MessageField::some(WireAgentTaskOwnership::from(ownership)),
            operation_id: uuid::Uuid::now_v7().to_string(),
            ..Default::default()
        };
        let first = self
            .client
            .transition_task_with_options(request.clone(), traced_options())
            .await;
        let reply = match first {
            Ok(reply) => reply,
            Err(error) if is_ambiguous_task_error(&error) => {
                self.client
                    .transition_task_with_options(request, traced_options())
                    .await?
            }
            Err(error) => return Err(DialError::Connect(error)),
        }
        .into_owned();
        Self::require_task(reply.task.into_option())
    }

    /// Claims one task for a durable source dispatch before its turn runs.
    ///
    /// `worker_instance` is stable for one edge process and `attempt_id` for
    /// one delivery attempt. Exact retries reuse both; a competing execution
    /// must use a different attempt.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for transport failures, malformed
    /// ownership replies, or a live competing claim.
    pub async fn claim_task(
        &self,
        task_id: &str,
        dispatch_id: &str,
        worker_instance: &str,
        attempt_id: &str,
        appended_history: Vec<Vec<u8>>,
    ) -> Result<(AgentTaskRecord, AgentTaskOwnership), DialError> {
        let request = ClaimAgentTaskRequest {
            task_id: task_id.to_owned(),
            dispatch_id: dispatch_id.to_owned(),
            worker_instance: worker_instance.to_owned(),
            attempt_id: attempt_id.to_owned(),
            appended_history,
            ..Default::default()
        };
        // A response can disappear after State commits the claim. Retry once
        // with the byte-for-byte same ownership identity; TaskService then
        // replays the receipt instead of consuming another grant. A retry
        // constructed by the outer edge would mint a new delivery attempt and
        // therefore cannot recover this ambiguous outcome.
        let first = self
            .client
            .claim_task_with_options(request.clone(), traced_options())
            .await;
        let reply = match first {
            Ok(reply) => reply,
            Err(error) if is_ambiguous_task_error(&error) => {
                self.client
                    .claim_task_with_options(request, traced_options())
                    .await?
            }
            Err(error) => return Err(DialError::Connect(error)),
        }
        .into_owned();
        let task = Self::require_task(reply.task.into_option())?;
        let ownership = reply.ownership.into_option().ok_or_else(|| {
            DialError::Connect(connectrpc::ConnectError::internal(
                "a successful task claim returned no ownership capability",
            ))
        })?;
        Ok((task, AgentTaskOwnership::from(ownership)))
    }

    /// Renews one exact task ownership capability without changing its fence.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for transport failures or a stale
    /// capability. Ambiguous transport outcomes replay the same ordinal.
    pub async fn renew_task_claim(
        &self,
        task_id: &str,
        ownership: &AgentTaskOwnership,
        ordinal: u64,
    ) -> Result<(), DialError> {
        let request = RenewAgentTaskClaimRequest {
            task_id: task_id.to_owned(),
            ownership: buffa::MessageField::some(WireAgentTaskOwnership::from(ownership)),
            ordinal,
            ..Default::default()
        };
        let first = self
            .client
            .renew_task_claim_with_options(request.clone(), traced_options())
            .await;
        match first {
            Ok(_) => Ok(()),
            Err(error) if is_ambiguous_task_error(&error) => self
                .client
                .renew_task_claim_with_options(request, traced_options())
                .await
                .map(|_| ())
                .map_err(DialError::Connect),
            Err(error) => Err(DialError::Connect(error)),
        }
    }

    /// Cancel a live task.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport or encoding error,
    /// including `not_found` for an unknown task and `failed_precondition`
    /// for a task that already finished — cancellation's refusal is part of
    /// its contract, never a quiet success.
    pub async fn cancel_task(&self, task_id: &str) -> Result<AgentTaskRecord, DialError> {
        let request = CancelAgentTaskRequest {
            task_id: task_id.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .cancel_task_with_options(request, traced_options())
            .await?
            .into_owned();
        Self::require_task(reply.task.into_option())
    }

    /// Read one task by identity. `None` means no task exists under it — a
    /// plain absence, not a failure.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport or encoding error.
    pub async fn get_task(&self, task_id: &str) -> Result<Option<AgentTaskRecord>, DialError> {
        let request = GetAgentTaskRequest {
            task_id: task_id.to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .get_task_with_options(request, traced_options())
            .await?
            .into_owned();
        reply
            .task
            .into_option()
            .map(AgentTaskRecord::try_from)
            .transpose()
    }

    /// Read one bounded page of `context_id`'s tasks, resuming after `after`
    /// when a previous page returned one.
    ///
    /// # Errors
    /// Returns [`DialError::Connect`] for any transport or encoding error,
    /// including `invalid_argument` for a page size above the server maximum.
    pub async fn list_tasks(
        &self,
        context_id: &str,
        page_size: u32,
        after: Option<&str>,
    ) -> Result<AgentTaskPage, DialError> {
        let request = ListAgentTasksRequest {
            context_id: context_id.to_owned(),
            page_size,
            after: after.unwrap_or_default().to_owned(),
            ..Default::default()
        };
        let reply = self
            .client
            .list_tasks_with_options(request, traced_options())
            .await?
            .into_owned();
        let tasks = reply
            .tasks
            .into_iter()
            .map(AgentTaskRecord::try_from)
            .collect::<Result<Vec<_>, _>>()?;
        Ok(AgentTaskPage {
            tasks,
            next_after: (!reply.next_after.is_empty()).then_some(reply.next_after),
        })
    }

    /// Unwrap the task every mutating reply carries, failing closed rather
    /// than inventing an empty record when one is somehow absent.
    fn require_task(task: Option<WireAgentTask>) -> Result<AgentTaskRecord, DialError> {
        task.ok_or_else(|| {
            DialError::Connect(connectrpc::ConnectError::internal(
                "the task operation returned no task record",
            ))
        })
        .and_then(AgentTaskRecord::try_from)
    }
}

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

    /// Invariant: absent, empty, and present wrapped secrets each read the
    /// way the edges depend on — no secret means `None`, an empty-string
    /// placeholder means `None` (unconfigured, not "configured as blank"),
    /// and a real value exposes back out unchanged.
    #[test]
    fn expose_nonempty_treats_empty_as_unset() {
        assert_eq!(expose_nonempty(None), None);

        let empty = Sensitive::new(String::new());
        assert_eq!(expose_nonempty(Some(&empty)), None);

        let value = Sensitive::new("real-secret".to_owned());
        assert_eq!(expose_nonempty(Some(&value)), Some("real-secret"));
    }

    #[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 authorized to approve".to_owned(),
        }
        .decided_line("Chris");
        assert!(rejected.contains("Rejected") && rejected.contains("not authorized to approve"));
        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());
    }

    /// `#743`: the completed-card line names the tool, the decider, and the
    /// outcome — the runtime's report of what actually happened, distinct
    /// from `approval_decided_text`'s earlier "running…" line.
    #[test]
    fn approval_completed_text_reports_the_runtime_outcome() {
        let done = approval_completed_text("Remove @vitor's admin role", "Chris", true);
        assert!(done.contains("Chris"), "names the decider: {done}");
        assert!(
            done.contains("Remove @vitor's admin role"),
            "names the tool label: {done}"
        );
        assert!(done.contains("done"), "a success reads as done: {done}");

        let failed = approval_completed_text("Remove @vitor's admin role", "Chris", false);
        assert_ne!(
            done, failed,
            "success and failure must not read identically"
        );
        assert!(
            failed.contains("error"),
            "a failed run must say so, not claim success: {failed}"
        );

        for copy in [&done, &failed] {
            let lower = copy.to_lowercase();
            for banned in ["please", "sorry", "unfortunately", "operator"] {
                assert!(
                    !lower.contains(banned),
                    "banned word {banned:?} in {copy:?}"
                );
            }
        }
    }

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

    #[test]
    fn deadline_exceeded_is_the_only_code_that_reads_as_a_routine_recycle() {
        let deadline = DialError::Connect(connectrpc::ConnectError::deadline_exceeded("idle"));
        assert!(deadline.is_deadline_exceeded());

        let unavailable = DialError::Connect(connectrpc::ConnectError::unavailable("gone"));
        assert!(
            !unavailable.is_deadline_exceeded(),
            "a genuinely dropped stream must still be reported as a failure"
        );

        let local = DialError::Tls("no provider".to_owned());
        assert!(
            !local.is_deadline_exceeded(),
            "a local setup failure has no Connect code and is never a routine recycle"
        );
    }

    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]"
        );
    }

    /// `#743`: the buffered aggregation path must skip `internal_only`
    /// messages entirely — neither the assistant-text nor the scaffolding
    /// leg — mirroring `message_to_event`'s streaming-path rule.
    #[test]
    fn buffered_aggregation_skips_internal_only_messages() {
        let mut withheld = message("model", text("pending your approval"));
        withheld.internal_only = true;
        let mut text_parts = Vec::new();
        let mut scaffolding = Vec::new();
        aggregate_output_message(withheld, &mut text_parts, &mut scaffolding);
        assert!(
            text_parts.is_empty(),
            "withheld text must not become the reply"
        );
        assert!(
            scaffolding.is_empty(),
            "withheld text must not fall back to scaffolding either"
        );
    }

    #[test]
    fn buffered_aggregation_keeps_visible_assistant_text() {
        // Regression: a normal (non-internal_only) assistant message must
        // still aggregate exactly as before.
        let mut text_parts = Vec::new();
        let mut scaffolding = Vec::new();
        aggregate_output_message(
            message("model", text("the answer")),
            &mut text_parts,
            &mut scaffolding,
        );
        assert_eq!(text_parts, vec!["the answer".to_owned()]);
        assert!(scaffolding.is_empty());
    }

    /// `#519` follow-up: a wallet-link-needed turn with NO preceding real
    /// content gets exactly the deterministic shared-copy prompt, byte-
    /// identical to what `polyc_proto::wallet_link_prompt` produces — not
    /// something a caller must special-case, since `finalize_buffered_reply`
    /// decides it centrally for every buffered edge (Discord, email,
    /// trigger).
    #[test]
    fn finalize_buffered_reply_wallet_link_prompt_wins_with_url() {
        let url = "https://polychrome.example/link/abc";
        let reply = finalize_buffered_reply(
            &[],
            &[],
            WalletLinkPrompt::Present {
                link_url: Some(url.to_owned()),
                renewal: false,
                requested: false,
            },
            None,
            None,
        );
        assert_eq!(reply, polyc_proto::wallet_link_prompt(Some(url), false));
        assert!(
            reply.contains(url),
            "the deterministic prompt carries the link: {reply}"
        );
    }

    #[test]
    fn finalize_buffered_reply_wallet_link_prompt_wins_without_url() {
        let reply = finalize_buffered_reply(
            &[],
            &[],
            WalletLinkPrompt::Present {
                link_url: None,
                renewal: false,
                requested: false,
            },
            None,
            None,
        );
        assert_eq!(reply, polyc_proto::wallet_link_prompt(None, false));
    }

    /// The originally reported bug this PR fixes: a directly-requested
    /// `wallet_link` call must get the requested-branch prompt, distinct
    /// from the payment-interrupt one, on a buffered edge.
    #[test]
    fn finalize_buffered_reply_wallet_link_prompt_requested_wins() {
        let url = "https://polychrome.example/link/abc";
        let reply = finalize_buffered_reply(
            &[],
            &[],
            WalletLinkPrompt::Present {
                link_url: Some(url.to_owned()),
                renewal: false,
                requested: true,
            },
            None,
            None,
        );
        assert_eq!(reply, polyc_proto::wallet_link_requested_prompt(url));
        assert!(
            reply.contains(url),
            "the requested-link prompt carries the link: {reply}"
        );
        assert_ne!(
            reply,
            polyc_proto::wallet_link_prompt(Some(url), false),
            "the requested branch must read differently from the payment-interrupt one"
        );
    }

    /// `#2122`: the renewal branch — `finalize_buffered_reply` must produce
    /// the renewal wording, not the first-time one, and it must never read
    /// as first-time setup.
    #[test]
    fn finalize_buffered_reply_wallet_link_prompt_renewal_wins() {
        let url = "https://polychrome.example/link/abc";
        let reply = finalize_buffered_reply(
            &[],
            &[],
            WalletLinkPrompt::Present {
                link_url: Some(url.to_owned()),
                renewal: true,
                requested: false,
            },
            None,
            None,
        );
        assert_eq!(reply, polyc_proto::wallet_link_prompt(Some(url), true));
        assert!(reply.to_lowercase().contains("expired"), "{reply}");
        assert!(
            !reply.to_lowercase().contains("no spending wallet set up"),
            "{reply}"
        );
    }

    /// Third-review finding: `text_parts` cannot be trusted as filtered,
    /// already-genuine content — `run_turn_with` dials the same
    /// `AgentService.connect` RPC as the streaming API, and the control plane
    /// forwards the harness's live `TextDelta`s to every client before the
    /// harness's post-hoc `internal_only` classification ever runs (and skips
    /// re-sending the filtered terminal batch once any delta was forwarded).
    /// So `text_parts` can hold the model's own raw, unfiltered wallet-link
    /// narration exactly like Slack/Telegram's live accumulator — a prior
    /// round's belief that appending was safe here was wrong. A
    /// wallet-link-needed turn must fully replace `text_parts`, matching
    /// Slack's `interrupt_holdback_text` and Telegram's reply-override.
    #[test]
    fn finalize_buffered_reply_wallet_link_prompt_replaces_preceding_content() {
        let url = "https://polychrome.example/link/abc";
        let reply = finalize_buffered_reply(
            &["It looks like you'll need to link a wallet first.".to_owned()],
            &[],
            WalletLinkPrompt::Present {
                link_url: Some(url.to_owned()),
                renewal: false,
                requested: false,
            },
            None,
            None,
        );
        assert_eq!(
            reply,
            polyc_proto::wallet_link_prompt(Some(url), false),
            "preceding text_parts must not survive — it may be the model's own \
             unfiltered wallet-link narration, not genuine unrelated content: {reply}"
        );
    }

    /// A turn with no wallet-link signal is unaffected — the same
    /// text-then-scaffolding fallback behavior as before this change.
    #[test]
    fn finalize_buffered_reply_without_wallet_link_prompt_is_unchanged() {
        assert_eq!(
            finalize_buffered_reply(
                &["the answer".to_owned()],
                &[],
                WalletLinkPrompt::None,
                None,
                None,
            ),
            "the answer"
        );
        assert_eq!(
            finalize_buffered_reply(
                &[],
                &["[tool_call:foo]".to_owned()],
                WalletLinkPrompt::None,
                None,
                None,
            ),
            "[tool_call:foo]"
        );
        assert_eq!(
            finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, None, None),
            ""
        );
    }

    /// Issue #1041/#1159 sibling of
    /// `finalize_buffered_reply_wallet_link_prompt_wins_with_url`: a
    /// `wallet_update_limit` mint with no preceding content gets exactly the
    /// deterministic shared-copy prompt, carrying both the link and the
    /// requested cap.
    #[test]
    fn finalize_buffered_reply_wallet_update_prompt_wins() {
        let url = "https://polychrome.example/update/xyz";
        let reply =
            finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, Some((url, "20")), None);
        assert_eq!(reply, polyc_proto::wallet_update_prompt(url, "20"));
        assert!(reply.contains(url), "{reply}");
        assert!(reply.contains("20"), "{reply}");
    }

    /// Issue #1042/#1156 sibling of the same: a `unlink_self` (wallet
    /// target) hard-revoke mint with no preceding content gets exactly the
    /// deterministic shared-copy prompt.
    #[test]
    fn finalize_buffered_reply_wallet_revoke_prompt_wins() {
        let url = "https://polychrome.example/revoke/xyz";
        let reply = finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, None, Some(url));
        assert_eq!(reply, polyc_proto::wallet_revoke_prompt(url));
        assert!(reply.contains(url), "{reply}");
    }

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

    /// `#743`: an `internal_only` message — a paused turn's withheld model
    /// text, or an approver/policy note — must never surface as a streamed
    /// event, even though it carries otherwise-eligible assistant text.
    #[test]
    fn internal_only_assistant_text_is_skipped() {
        let mut msg = message("assistant", text("pending your approval"));
        msg.internal_only = true;
        assert_eq!(message_to_event(msg), 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 {
                turn_id: "00000000-0000-0000-0000-000000000001".to_owned(),
                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 {
                    turn_id: "00000000-0000-0000-0000-000000000001".to_owned(),
                    request_id: "r1".to_owned(),
                    tool_name: "delete_file".to_owned(),
                    title: "Delete a file".to_owned(),
                    args_json: "{}".to_owned(),
                    reason: String::new(),
                    resolve_token: String::new(),
                    preview: None,
                },
                TurnEvent::HandoffStarted {
                    child_agent_id: "child".to_owned(),
                    reason: String::new(),
                },
                TurnEvent::Done,
            ]
        );
    }

    #[test]
    fn end_projects_invite_deliveries_after_handoff_before_done() {
        use polyc_proto::proto::polychrome::agent::v1::InviteDelivery as WireInviteDelivery;
        let end = AgentEnd {
            invite_deliveries: vec![WireInviteDelivery {
                target_user_id: "UVITOR".to_owned(),
                code: "482913".to_owned(),
                inviter_display: "Ada".to_owned(),
                ..Default::default()
            }],
            ..Default::default()
        };
        assert_eq!(
            events_from_end(end),
            vec![
                TurnEvent::InviteDelivery {
                    target_user_id: "UVITOR".to_owned(),
                    code: "482913".to_owned(),
                    inviter_display: "Ada".to_owned(),
                },
                TurnEvent::Done,
            ]
        );
    }

    #[test]
    fn end_projects_wallet_link_prompt_after_invite_deliveries_before_done() {
        use polyc_proto::proto::polychrome::agent::v1::WalletLinkPrompt as WireWalletLinkPrompt;
        let end = AgentEnd {
            wallet_link_prompt: buffa::MessageField::some(WireWalletLinkPrompt {
                link_url: "https://polychrome.example/link/abc".to_owned(),
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(
            events_from_end(end),
            vec![
                TurnEvent::WalletLinkPrompt {
                    link_url: Some("https://polychrome.example/link/abc".to_owned()),
                    renewal: false,
                    requested: false,
                },
                TurnEvent::Done,
            ]
        );
    }

    #[test]
    fn end_projects_wallet_link_prompt_without_url_as_none() {
        use polyc_proto::proto::polychrome::agent::v1::WalletLinkPrompt as WireWalletLinkPrompt;
        let end = AgentEnd {
            wallet_link_prompt: buffa::MessageField::some(WireWalletLinkPrompt {
                link_url: String::new(),
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(
            events_from_end(end),
            vec![
                TurnEvent::WalletLinkPrompt {
                    link_url: None,
                    renewal: false,
                    requested: false,
                },
                TurnEvent::Done,
            ]
        );
    }

    /// `#2122`: the renewal bit rides `events_from_end`'s projection too, not
    /// just the buffered path.
    #[test]
    fn end_projects_wallet_link_prompt_renewal_bit() {
        use polyc_proto::proto::polychrome::agent::v1::WalletLinkPrompt as WireWalletLinkPrompt;
        let end = AgentEnd {
            wallet_link_prompt: buffa::MessageField::some(WireWalletLinkPrompt {
                link_url: "https://polychrome.example/link/abc".to_owned(),
                renewal: true,
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(
            events_from_end(end),
            vec![
                TurnEvent::WalletLinkPrompt {
                    link_url: Some("https://polychrome.example/link/abc".to_owned()),
                    renewal: true,
                    requested: false,
                },
                TurnEvent::Done,
            ]
        );
    }

    /// The originally reported bug this PR fixes: a directly-requested
    /// `wallet_link` call must project with `requested: true`, distinct from
    /// the payment-interrupt signal, so an edge selects the requested-branch
    /// card copy.
    #[test]
    fn end_projects_wallet_link_prompt_requested_bit() {
        use polyc_proto::proto::polychrome::agent::v1::WalletLinkPrompt as WireWalletLinkPrompt;
        let end = AgentEnd {
            wallet_link_prompt: buffa::MessageField::some(WireWalletLinkPrompt {
                link_url: "https://polychrome.example/link/abc".to_owned(),
                requested: true,
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(
            events_from_end(end),
            vec![
                TurnEvent::WalletLinkPrompt {
                    link_url: Some("https://polychrome.example/link/abc".to_owned()),
                    renewal: false,
                    requested: true,
                },
                TurnEvent::Done,
            ]
        );
    }

    #[test]
    fn end_projects_wallet_update_prompt_before_done() {
        use polyc_proto::proto::polychrome::agent::v1::WalletUpdatePrompt as WireWalletUpdatePrompt;
        let end = AgentEnd {
            wallet_update_prompt: buffa::MessageField::some(WireWalletUpdatePrompt {
                update_url: "https://polychrome.example/update/xyz".to_owned(),
                new_limit: "20".to_owned(),
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(
            events_from_end(end),
            vec![
                TurnEvent::WalletUpdatePrompt {
                    update_url: "https://polychrome.example/update/xyz".to_owned(),
                    new_limit: "20".to_owned(),
                },
                TurnEvent::Done,
            ]
        );
    }

    #[test]
    fn end_projects_wallet_revoke_prompt_before_done() {
        use polyc_proto::proto::polychrome::agent::v1::WalletRevokePrompt as WireWalletRevokePrompt;
        let end = AgentEnd {
            wallet_revoke_prompt: buffa::MessageField::some(WireWalletRevokePrompt {
                revoke_url: "https://polychrome.example/revoke/xyz".to_owned(),
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(
            events_from_end(end),
            vec![
                TurnEvent::WalletRevokePrompt {
                    revoke_url: "https://polychrome.example/revoke/xyz".to_owned(),
                },
                TurnEvent::Done,
            ]
        );
    }

    // --- namespaced conversation ids ---

    #[test]
    fn namespaced_id_is_prefix_colon_native() {
        assert_eq!(
            namespaced_id("mail", "CAF=abc@mail.example"),
            "mail:CAF=abc@mail.example"
        );
        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", "C0000FAKEID", "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")
        );
    }

    fn sample_attribution() -> Attribution {
        Attribution {
            caller: Some(ExternalIdentity {
                provider: "slack".to_owned(),
                scope: "team-1".to_owned(),
                external_id: "U1".to_owned(),
                display_name: "Ada".to_owned(),
                ..Default::default()
            }),
            participants: vec![ExternalIdentity {
                provider: "slack".to_owned(),
                scope: "team-1".to_owned(),
                external_id: "U2".to_owned(),
                display_name: "Bea".to_owned(),
                ..Default::default()
            }],
        }
    }

    #[test]
    fn build_request_without_credentials_fails_before_transport() {
        let ingress = TurnIngress::new(
            "slack:team-1:general",
            "exec-1",
            IngressIdentity::reported("team-1", "message-1").expect("valid source identity"),
            ClaimedNamespace::new("slack").expect("a real namespace is a valid claim"),
            vec![text_message("user", "hi")],
        );
        assert!(
            matches!(
                build_request(ingress, None),
                Err(DialError::MissingIngressCredentials)
            ),
            "source identity cannot ride outside a signed envelope"
        );
    }

    #[test]
    fn build_request_with_credentials_signs_a_verifiable_envelope() {
        let key_bytes = [7u8; 32];
        let signer_pk = polyc_crypto::Signer::from_key_bytes(&key_bytes)
            .expect("valid test key material")
            .public_key_bytes();
        let creds = EdgeCredentials::from_parts(
            "slack".to_owned(),
            "pc_slack_test-secret".to_owned(),
            &polyc_crypto::hex::lower(&key_bytes),
        )
        .expect("valid test credentials");
        let attribution = sample_attribution();

        let source_identity =
            IngressIdentity::reported("team-1", "message-1").expect("valid source identity");
        let ingress = TurnIngress::new(
            "slack:team-1:general",
            "exec-1",
            source_identity.clone(),
            ClaimedNamespace::new("slack").expect("a real namespace is a valid claim"),
            vec![text_message("user", "hi")],
        )
        .with_attribution(attribution.clone());
        let request = build_request(ingress, Some(&creds)).expect("credentialed request builds");

        let start = request.start.into_option().expect("start set");
        let envelope = start
            .asserted_attribution
            .into_option()
            .expect("a credentialed dial signs an asserted_attribution envelope");

        assert_eq!(envelope.edge_id, "slack");
        assert_eq!(envelope.conversation_id, "slack:team-1:general");
        assert!(!envelope.nonce.is_empty(), "nonce must be set per turn");
        assert!(envelope.issued_unix_ms > 0);
        assert!(
            polyc_crypto::edge_identity::verify_edge_assertion(&signer_pk, &envelope),
            "the signed envelope must verify under the credential's own public key"
        );
        let mut tampered_source = envelope.clone();
        tampered_source.source_identity = buffa::MessageField::some(
            IngressIdentity::reported("team-1", "message-2")
                .expect("valid source identity")
                .to_wire(),
        );
        assert!(
            !polyc_crypto::edge_identity::verify_edge_assertion(&signer_pk, &tampered_source),
            "source identity must be covered by the edge signature"
        );
        assert_eq!(
            envelope.caller.into_option().map(|c| c.external_id),
            attribution.caller.map(|c| c.external_id),
            "the envelope's caller must match the turn's attribution"
        );
        assert_eq!(
            envelope
                .participants
                .into_iter()
                .map(|p| p.external_id)
                .collect::<Vec<_>>(),
            attribution
                .participants
                .into_iter()
                .map(|p| p.external_id)
                .collect::<Vec<_>>(),
            "the envelope's participants must match the turn's attribution"
        );
        assert_eq!(
            envelope.exec_id, "exec-1",
            "the envelope must bind this turn's exec_id"
        );
        assert_eq!(
            envelope.content_hash,
            polyc_crypto::edge_identity::content_hash_hex(&encode_messages_for_content_hash(&[
                text_message("user", "hi")
            ])),
            "the envelope's content_hash must match the turn's messages"
        );
        assert_eq!(
            envelope.source_identity.into_option(),
            Some(source_identity.to_wire()),
            "the stable source identity must ride inside the signed envelope"
        );
    }

    #[test]
    fn source_identity_and_content_digest_do_not_follow_exec_id() {
        let creds = EdgeCredentials::from_parts(
            "edge".to_owned(),
            "bearer".to_owned(),
            &polyc_crypto::hex::lower(&[8; 32]),
        )
        .expect("valid credentials");
        let source =
            IngressIdentity::reported("source", "event-42").expect("valid source identity");
        let messages = vec![text_message("user", "same authenticated content")];

        let envelope = |exec_id: &str| {
            build_request(
                TurnIngress::new(
                    "conv",
                    exec_id,
                    source.clone(),
                    ClaimedNamespace::new("slack").expect("a real namespace is a valid claim"),
                    messages.clone(),
                ),
                Some(&creds),
            )
            .expect("request builds")
            .start
            .into_option()
            .expect("start set")
            .asserted_attribution
            .into_option()
            .expect("assertion set")
        };
        let first = envelope("attempt-1");
        let retry = envelope("attempt-2");

        assert_ne!(first.exec_id, retry.exec_id);
        assert_eq!(first.source_identity, retry.source_identity);
        assert_eq!(first.content_hash, retry.content_hash);
    }

    #[test]
    fn transport_success_without_complete_receipt_is_not_acknowledgement_authority() {
        let source = IngressIdentity::reported("source", "event-42")
            .expect("valid source identity")
            .to_wire();
        let valid = WireIngressReceipt {
            source_identity: buffa::MessageField::some(source.clone()),
            receipt: b"receipt".to_vec(),
            dispatch_id: "dispatch".to_owned(),
            conversation_id: "conv".to_owned(),
            ..Default::default()
        };
        assert!(validate_ingress_receipt(&valid, &source, "conv").is_ok());

        let malformed = [
            WireIngressReceipt {
                source_identity: buffa::MessageField::none(),
                ..valid.clone()
            },
            WireIngressReceipt {
                receipt: Vec::new(),
                ..valid.clone()
            },
            WireIngressReceipt {
                dispatch_id: String::new(),
                ..valid.clone()
            },
            WireIngressReceipt {
                conversation_id: "other".to_owned(),
                ..valid
            },
        ];
        for receipt in malformed {
            assert!(matches!(
                validate_ingress_receipt(&receipt, &source, "conv"),
                Err(DialError::InvalidIngressReceipt(_))
            ));
        }
    }

    // `#1553`: the responder an edge names on a decision is signed here, once,
    // so every edge inherits it. The tests below pin what the assertion covers,
    // the two states that legitimately carry none, and the wiring that gets the
    // edge's key from its credentials to the request.

    /// A dial address nothing connects to — every dialer below is built for
    /// the request it produces, never for a call.
    const TEST_ADDR: &str = "http://127.0.0.1:0";

    /// Test credentials plus the public key their assertions verify under.
    fn approval_credentials() -> (EdgeCredentials, Vec<u8>) {
        let key_bytes = [11u8; 32];
        let public_key = polyc_crypto::Signer::from_key_bytes(&key_bytes)
            .expect("valid test key material")
            .public_key_bytes();
        let creds = EdgeCredentials::from_parts(
            "slack".to_owned(),
            "pc_slack_test-secret".to_owned(),
            &polyc_crypto::hex::lower(&key_bytes),
        )
        .expect("valid test credentials");
        (creds, public_key)
    }

    fn sample_responder() -> ExternalIdentity {
        ExternalIdentity {
            provider: "slack".to_owned(),
            scope: "team-1".to_owned(),
            external_id: "U1".to_owned(),
            display_name: "Ada".to_owned(),
            ..Default::default()
        }
    }

    /// The request an approving edge that named a responder sends — the shape
    /// every assertion test below starts from.
    fn approve_naming_a_responder(dialer: &ApprovalDialer) -> ApprovalResponseRequest {
        dialer.respond_request(
            "00000000-0000-0000-0000-000000000001",
            "req-1",
            ApprovalChoice::Approve,
            "looks right",
            "slack:team-1:general",
            r#"{"path":"/tmp/a"}"#,
            "context",
            "resolve-token-1",
            Some(sample_responder()),
        )
    }

    #[test]
    fn a_credentialed_dialer_signs_a_verifiable_assertion() {
        let (creds, public_key) = approval_credentials();
        let dialer = ApprovalDialer::with_credentials(TEST_ADDR, Arc::new(creds))
            .expect("a credentialed approval dialer builds");
        let request = approve_naming_a_responder(&dialer);

        let assertion = request
            .asserted_approval
            .as_option()
            .expect("a credentialed edge naming a responder asserts one");
        assert_eq!(assertion.edge_id, "slack");
        assert_eq!(
            assertion
                .responder
                .as_option()
                .map(|r| r.external_id.as_str()),
            Some("U1"),
            "the assertion must carry the responder the edge named"
        );
        assert!(
            polyc_crypto::approval_assertion::verify_approval_assertion(
                [public_key.as_slice()],
                &request
            ),
            "the assertion must verify under the edge's own identity key"
        );
    }

    #[test]
    fn a_decision_naming_no_responder_asserts_none() {
        // Decision 4: a machine decision, or an edge that cannot attribute the
        // human, asserts nothing — a supported state, not an error.
        let (creds, _) = approval_credentials();
        let dialer = ApprovalDialer::with_credentials(TEST_ADDR, Arc::new(creds))
            .expect("a credentialed approval dialer builds");
        let request = dialer.respond_request(
            "00000000-0000-0000-0000-000000000001",
            "req-1",
            ApprovalChoice::Deny,
            "no",
            "slack:team-1:general",
            "",
            "",
            "resolve-token-1",
            None,
        );
        assert!(
            request.asserted_approval.as_option().is_none(),
            "no responder means no assertion, not an assertion of nobody"
        );
    }

    #[test]
    fn an_uncredentialed_dialer_asserts_none() {
        // The `new`/`with_bearer` paths: no key to sign with, so the responder
        // the edge named goes unasserted rather than riding unsigned.
        for (label, dialer) in [
            (
                "new",
                ApprovalDialer::new(TEST_ADDR).expect("an unauthenticated dialer builds"),
            ),
            (
                "with_bearer",
                ApprovalDialer::with_bearer(TEST_ADDR, "pc_slack_test-secret")
                    .expect("a bearer-only dialer builds"),
            ),
        ] {
            let request = approve_naming_a_responder(&dialer);
            assert!(
                request.asserted_approval.as_option().is_none(),
                "{label}: an uncredentialed dialer must not put an unsigned responder on the wire"
            );
        }
    }

    #[test]
    fn the_assertion_covers_every_field_of_the_final_request() {
        // The attach-last ordering, pinned from the outside: the signature is
        // over the FINISHED request, so mutating ANY signed field after the
        // builder returns must break verification. A builder that attached the
        // assertion before some field was final would ship a request that fails
        // this same check on the control plane.
        let (creds, public_key) = approval_credentials();
        let dialer = ApprovalDialer::with_credentials(TEST_ADDR, Arc::new(creds))
            .expect("a credentialed approval dialer builds");
        let keys = || [public_key.as_slice()];

        /// One "change a signed field after the fact" case.
        type Mutation = (&'static str, fn(&mut ApprovalResponseRequest));

        let mutations: Vec<Mutation> = vec![
            ("request_id", |r| r.request_id = "req-2".to_owned()),
            ("conversation_id", |r| {
                r.conversation_id = "slack:team-1:secrets".to_owned();
            }),
            ("resolve_token", |r| {
                r.resolve_token = "resolve-token-2".to_owned();
            }),
            ("decision", |r| {
                use polyc_proto::proto::polychrome::approval::v1::{
                    Deny, approval_response_request::Decision,
                };
                r.decision = Some(Decision::Deny(Box::new(Deny {
                    reason: "looks right".to_owned(),
                    abort: false,
                    __buffa_unknown_fields: buffa::UnknownFields::default(),
                })));
            }),
            ("modified_args_json", |r| {
                use polyc_proto::proto::polychrome::approval::v1::approval_response_request::Decision;
                let Some(Decision::Approve(approve)) = r.decision.as_mut() else {
                    panic!("the builder produced an Approve");
                };
                approve.modified_args_json = r#"{"path":"/etc/shadow"}"#.to_owned();
            }),
            ("responder", |r| {
                r.asserted_approval
                    .as_option_mut()
                    .expect("the request carries an assertion")
                    .responder
                    .as_option_mut()
                    .expect("the assertion carries a responder")
                    .external_id = "U0EVE".to_owned();
            }),
            ("edge_id", |r| {
                r.asserted_approval
                    .as_option_mut()
                    .expect("the request carries an assertion")
                    .edge_id = "messaging".to_owned();
            }),
        ];

        for (field, mutate) in mutations {
            let mut request = approve_naming_a_responder(&dialer);
            assert!(
                polyc_crypto::approval_assertion::verify_approval_assertion(keys(), &request),
                "{field}: the unmutated request must verify, or this case proves nothing"
            );
            mutate(&mut request);
            assert!(
                !polyc_crypto::approval_assertion::verify_approval_assertion(keys(), &request),
                "{field}: changing it after signing must invalidate the assertion"
            );
        }
    }

    #[test]
    fn an_approval_dialer_inherits_the_agent_dialers_edge_identity() {
        // The wiring an edge actually uses: one credential, one key, shared by
        // the turn-dispatch dialer and the approval dialer built from it.
        let (creds, public_key) = approval_credentials();
        let agent = AgentDialer::with_credentials(TEST_ADDR, creds)
            .expect("a credentialed agent dialer builds");
        let approval = agent
            .approval_dialer_with_credentials(TEST_ADDR)
            .expect("the approval sibling builds");
        assert!(
            polyc_crypto::approval_assertion::verify_approval_assertion(
                [public_key.as_slice()],
                &approve_naming_a_responder(&approval)
            ),
            "the approval dialer must sign with the SAME key the agent dialer dispatches with"
        );

        // And the unauthenticated path has nothing to inherit.
        let plain = AgentDialer::new(TEST_ADDR)
            .expect("an unauthenticated agent dialer builds")
            .approval_dialer_with_credentials(TEST_ADDR)
            .expect("the approval sibling builds");
        assert!(
            approve_naming_a_responder(&plain)
                .asserted_approval
                .as_option()
                .is_none(),
            "an unauthenticated agent dialer has no key to lend, so its approvals assert nobody"
        );
    }

    #[test]
    fn with_credentials_rejects_a_bearer_that_is_not_a_valid_header_value() {
        let creds = EdgeCredentials::from_parts(
            "slack".to_owned(),
            "pc_slack_bad\nbearer".to_owned(),
            &polyc_crypto::hex::lower(&[9u8; 32]),
        )
        .expect("valid test credentials");
        // `AgentDialer` isn't `Debug` (it wraps a connect client), so match
        // directly instead of `.expect_err`, which requires `T: Debug`.
        let err = match AgentDialer::with_credentials("http://127.0.0.1:0", creds) {
            Ok(_) => panic!("a newline in the bearer must fail the dial closed"),
            Err(err) => err,
        };
        assert!(matches!(err, DialError::InvalidBearer(_)));
    }

    // [`bearer_header`] is the ONE place a bearer becomes an HTTP header, and
    // every bearer-carrying constructor reaches it through
    // [`build_control_client`]. The two tests below pin that logic directly;
    // the table-driven one after them proves each dialer actually routes
    // through it rather than re-implementing the header itself.

    #[test]
    fn bearer_header_encodes_a_valid_bearer() {
        let headers = bearer_header("pc_slack_good").expect("a valid bearer encodes");
        let value = headers
            .get(http::header::AUTHORIZATION)
            .expect("authorization header is set");
        assert_eq!(value, "Bearer pc_slack_good");
        assert!(
            value.is_sensitive(),
            "the bearer header must be marked sensitive so the HTTP stack \
             never logs it or HPACK-indexes it"
        );
    }

    #[test]
    fn bearer_header_rejects_a_bearer_that_is_not_a_valid_header_value() {
        let err =
            bearer_header("pc_slack_bad\nbearer").expect_err("a newline must fail the dial closed");
        assert!(matches!(err, DialError::InvalidBearer(_)));
    }

    #[test]
    fn bearer_header_value_is_sensitive() {
        let value = bearer_header_value("pc_admin_good").expect("a valid bearer encodes");
        assert_eq!(value, "Bearer pc_admin_good");
        assert!(
            value.is_sensitive(),
            "RoutineDialer::fire_routine's per-call admin bearer must be \
             marked sensitive, same as every other dialer's bearer header"
        );
    }

    /// Wiring guard for every dialer whose auth is a bearer header (none of
    /// them sends `AgentStart`, so no turn envelope rides their calls): a
    /// valid bearer constructs, and one that isn't a valid header value fails
    /// closed with
    /// [`DialError::InvalidBearer`]. Table-driven so adding a dialer is one
    /// line here rather than another pair of copy-pasted tests. Each closure
    /// erases the dialer's own (non-`Debug`) type so both outcomes assert
    /// uniformly.
    #[test]
    fn every_bearer_dialer_routes_through_bearer_header() {
        const ADDR: &str = "http://127.0.0.1:0";
        const GOOD: &str = "pc_slack_good";
        const BAD: &str = "pc_slack_bad\nbearer";

        type Build = fn(&str, &str) -> Result<(), DialError>;
        let dialers: &[(&str, Build)] = &[
            ("ApprovalDialer", |a, b| {
                ApprovalDialer::with_bearer(a, b).map(|_| ())
            }),
            // The credentialed approval path rides the SAME header (its
            // credentials' bearer), so it belongs to this guard too.
            ("ApprovalDialer::with_credentials", |a, b| {
                let creds = EdgeCredentials::from_parts(
                    "slack".to_owned(),
                    b.to_owned(),
                    &polyc_crypto::hex::lower(&[7u8; 32]),
                )
                .expect("valid test credentials");
                ApprovalDialer::with_credentials(a, Arc::new(creds)).map(|_| ())
            }),
            ("PersonaDialer", |a, b| {
                PersonaDialer::with_bearer(a, b).map(|_| ())
            }),
            // Admin-gated `new_admin` constructors are a separate code path
            // from `with_bearer` — they must independently route through
            // `bearer_header` rather than `with_default_header`, which
            // silently drops an invalid value instead of failing the dial
            // closed. See `#1811`.
            ("PersonaDialer::new_admin", |a, b| {
                PersonaDialer::new_admin(a, b).map(|_| ())
            }),
            ("CredentialDialer::new_admin", |a, b| {
                CredentialDialer::new_admin(a, b).map(|_| ())
            }),
            ("NotificationDialer", |a, b| {
                NotificationDialer::with_bearer(a, b).map(|_| ())
            }),
            ("OperatorMailboxDialer", |a, b| {
                OperatorMailboxDialer::with_bearer(a, b).map(|_| ())
            }),
            // `#1660` incident (2026-07-28): QuestionDialer shipped with
            // ONLY `new()` (no bearer) on the assumption that the
            // per-question `answer_token` was sufficient auth. The control
            // plane's `require_edge_bearer` layer guards EVERY RPC on this
            // listener regardless of per-call authorization, so every real
            // answer submission 401'd in production — this dialer was never
            // in this guard, so nothing caught the gap before a live user
            // did.
            ("QuestionDialer", |a, b| {
                QuestionDialer::with_bearer(a, b).map(|_| ())
            }),
            // `RoutineDialer::fire_routine` isn't a dialer constructor —
            // `RoutineDialer::new` dials without a bearer — but its per-call
            // `admin_bearer` header routes through the SAME
            // `bearer_header_value` construction site every other row here
            // reaches through `bearer_header`, so it belongs in this guard
            // too. `_a` is unused: there's no address to dial.
            ("RoutineDialer::fire_routine", |_a, b| {
                bearer_header_value(b).map(|_| ())
            }),
        ];

        for (label, build) in dialers {
            build(ADDR, GOOD)
                .unwrap_or_else(|err| panic!("{label}: a valid bearer must build a dialer: {err}"));
            match build(ADDR, BAD) {
                Ok(()) => panic!("{label}: a newline in the bearer must fail the dial closed"),
                Err(err) => assert!(
                    matches!(err, DialError::InvalidBearer(_)),
                    "{label}: expected InvalidBearer, got {err}"
                ),
            }
        }
    }

    /// Wiring guard for `RoutineDialer::fire_routine` itself (as opposed to
    /// the `bearer_header_value` unit call the table above uses as a stand-in
    /// for it): a malformed `admin_bearer` must fail closed with
    /// [`DialError::InvalidBearer`] before any dial is attempted, so
    /// `127.0.0.1:0` (nothing listening) never gets reached.
    #[tokio::test]
    async fn fire_routine_rejects_a_malformed_admin_bearer_before_dialing() {
        let dialer = RoutineDialer::new("http://127.0.0.1:0").expect("valid address");

        let err = dialer
            .fire_routine("standup", "", sample_responder(), "pc_admin_bad\nbearer")
            .await
            .expect_err("a newline in admin_bearer must fail the dial closed");

        assert!(
            matches!(err, DialError::InvalidBearer(_)),
            "expected InvalidBearer, got {err}"
        );
    }

    #[test]
    fn recovery_conversion_preserves_every_display_projection() {
        use polyc_proto::proto::polychrome::approval::v1::{
            RecordedApprovalDecision, SearchPreview,
        };

        let wire = RecoverableApprovalEntry {
            turn_id: "turn-1".to_owned(),
            request_id: "call-1".to_owned(),
            tool_name: "conversation_search".to_owned(),
            title: "Search past conversations".to_owned(),
            args_json: r#"{"query":"release"}"#.to_owned(),
            reason: "Review the search scope.".to_owned(),
            preview: buffa::MessageField::none(),
            read_preview: buffa::MessageField::some(WireListReadPreview {
                detail: Some(wire_list_read_preview::Detail::Search(Box::new(
                    SearchPreview {
                        canonical_query: "release decision".to_owned(),
                        scope_count: 7,
                        __buffa_unknown_fields: buffa::UnknownFields::default(),
                    },
                ))),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            }),
            state: Some(recoverable_approval_entry::State::DecisionRecorded(
                Box::new(RecordedApprovalDecision {
                    approved: true,
                    __buffa_unknown_fields: buffa::UnknownFields::default(),
                }),
            )),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };

        let recovered = RecoverableApproval::try_from(wire).expect("valid recovery projection");
        assert_eq!(recovered.turn_id, "turn-1");
        assert_eq!(recovered.request_id, "call-1");
        assert_eq!(recovered.tool_name, "conversation_search");
        assert_eq!(recovered.title, "Search past conversations");
        assert_eq!(recovered.args_json, r#"{"query":"release"}"#);
        assert_eq!(recovered.reason, "Review the search scope.");
        assert_eq!(
            recovered.read_preview,
            Some(ReadPreview::Search {
                canonical_query: "release decision".to_owned(),
                scope_count: 7,
            })
        );
        assert_eq!(
            recovered.state,
            ApprovalRecoveryState::DecisionRecorded(RecordedApprovalRecovery { approved: true })
        );
    }

    /// Build a preview whose runs all sit on the cadence line's zone.
    fn standup_preview() -> ApprovalPreview {
        ApprovalPreview {
            routine_name: "standup".to_owned(),
            prompt_text: "Post the daily standup summary.".to_owned(),
            next_fires: vec![
                fire("2026-07-29T09:00:00-07:00", "2026-07-29T16:00:00Z", ""),
                fire("2026-07-30T09:00:00-07:00", "2026-07-30T16:00:00Z", ""),
                fire("2026-07-31T09:00:00-07:00", "2026-07-31T16:00:00Z", ""),
            ],
            zone_name: "America/Los_Angeles".to_owned(),
            zone_is_fallback: false,
            cadence: "every weekday at 9:00 AM PDT (UTC-7)".to_owned(),
        }
    }

    fn fire(local: &str, utc: &str, zone_label: &str) -> ApprovalPreviewFire {
        ApprovalPreviewFire {
            local_time: local.to_owned(),
            utc_time: utc.to_owned(),
            zone_label: zone_label.to_owned(),
        }
    }

    fn at(rfc3339: &str) -> DateTime<Utc> {
        rfc3339.parse().expect("valid RFC3339")
    }

    /// `#1496`/`#1808`: the shared preview-render helper — the ONE place
    /// every chat edge's approval-card text comes from, so they can't word
    /// the same preview differently. Asserted whole, because the layout IS
    /// the deliverable here: a per-line spot check would pass on a table
    /// whose columns had silently stopped lining up.
    #[test]
    fn approval_preview_text_renders_cadence_then_an_aligned_run_table() {
        let text = approval_preview_text(&standup_preview(), at("2026-07-28T17:00:00Z"));
        assert_eq!(
            text,
            "This routine will run:\n\
             \n\
             Post the daily standup summary.\n\
             \n\
             Runs every weekday at 9:00 AM PDT (UTC-7)\n\
             \n\
             Next 3 runs\n\
             ```\n\
             \u{20} Wed Jul 29    9:00 AM    tomorrow\n\
             \u{20} Thu Jul 30    9:00 AM    in 2 days\n\
             \u{20} Fri Jul 31    9:00 AM    in 3 days\n\
             ```"
        );
    }

    #[test]
    fn approval_preview_plain_text_keeps_copy_and_removes_markup() {
        let preview = standup_preview();
        let now = at("2026-07-28T17:00:00Z");
        let chat = approval_preview_text(&preview, now);
        let plain = approval_preview_plain_text(&preview, now);
        assert!(chat.contains("```"));
        assert!(!plain.contains("```"));
        assert!(plain.contains("This routine will run:"));
        assert!(plain.contains("Wed Jul 29    9:00 AM    tomorrow"));
    }

    /// The card states the schedule, never the raw instants it was computed
    /// from: no RFC3339, and no UTC column duplicating the local time.
    #[test]
    fn approval_preview_text_shows_no_machine_timestamps() {
        let text = approval_preview_text(&standup_preview(), at("2026-07-28T17:00:00Z"));
        assert!(!text.contains("2026-07-29T09:00:00-07:00"), "{text}");
        assert!(!text.contains("2026-07-29T16:00:00Z"), "{text}");
        assert!(
            !text.contains('Z'),
            "no UTC instants belong on the card: {text}"
        );
    }

    /// "fire" is internal vocabulary — a person reading this card sees
    /// "runs" everywhere, per this repo's user-facing-copy rule.
    #[test]
    fn approval_preview_text_never_says_fire() {
        let mut empty = standup_preview();
        empty.next_fires = Vec::new();
        for preview in [standup_preview(), empty] {
            let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
            assert!(
                !text.to_lowercase().contains("fire"),
                "user-facing copy must say runs, not fires: {text}"
            );
        }
    }

    /// The title counts the runs it actually has — a schedule with fewer
    /// than three left must not claim three, and one run reads as singular.
    #[test]
    fn approval_preview_text_titles_the_list_by_its_real_count() {
        let mut preview = standup_preview();
        preview.next_fires.truncate(2);
        let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
        assert!(text.contains("Next 2 runs"), "{text}");

        preview.next_fires.truncate(1);
        let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
        assert!(text.contains("Next run"), "{text}");
        assert!(!text.contains("Next 1 runs"), "{text}");
    }

    /// When the control plane could not describe the schedule, the card says
    /// nothing about cadence rather than guessing — the run list still
    /// carries the truth.
    #[test]
    fn approval_preview_text_omits_an_absent_cadence_line() {
        let mut preview = standup_preview();
        preview.cadence = String::new();
        let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
        assert!(!text.contains("Runs "), "{text}");
        assert!(text.contains("Next 3 runs"), "{text}");
    }

    /// A run carrying its own zone label (a daylight-saving straddle, or a
    /// cadence with no single clock time) prints it in the time column.
    #[test]
    fn approval_preview_text_puts_a_runs_own_zone_label_beside_its_time() {
        let preview = ApprovalPreview {
            routine_name: "standup".to_owned(),
            prompt_text: "prompt".to_owned(),
            next_fires: vec![
                fire(
                    "2026-10-31T09:00:00-07:00",
                    "2026-10-31T16:00:00Z",
                    "PDT (UTC-7)",
                ),
                fire(
                    "2026-11-01T09:00:00-08:00",
                    "2026-11-01T17:00:00Z",
                    "PST (UTC-8)",
                ),
            ],
            zone_name: "America/Los_Angeles".to_owned(),
            zone_is_fallback: false,
            cadence: "every day at 9:00 AM".to_owned(),
        };
        let text = approval_preview_text(&preview, at("2026-10-30T17:00:00Z"));
        assert!(text.contains("9:00 AM PDT (UTC-7)    tomorrow"), "{text}");
        assert!(text.contains("9:00 AM PST (UTC-8)    in 2 days"), "{text}");
    }

    /// The relative column is anchored at RENDER time, so a preview frozen
    /// in the durable event and recovered days later cannot still claim
    /// "tomorrow". A run already behind `now` reads `passed`.
    #[test]
    fn approval_preview_text_anchors_runs_against_render_time() {
        let preview = standup_preview();
        let text = approval_preview_text(&preview, at("2026-07-30T17:00:00Z"));
        assert!(text.contains("Wed Jul 29    9:00 AM    passed"), "{text}");
        assert!(text.contains("Thu Jul 30    9:00 AM    passed"), "{text}");
        assert!(text.contains("Fri Jul 31    9:00 AM    tomorrow"), "{text}");
    }

    /// "today" means the reader's calendar day, not UTC's — the two disagree
    /// for part of every day across most of the world.
    #[test]
    fn approval_preview_text_reckons_days_in_the_runs_own_offset() {
        let preview = standup_preview();
        // 16:00Z is 09:00 in UTC-7, so the first run is later TODAY locally
        // even though UTC has not yet reached it.
        let text = approval_preview_text(&preview, at("2026-07-29T15:00:00Z"));
        assert!(text.contains("Wed Jul 29    9:00 AM    today"), "{text}");
    }

    /// A run in another calendar year carries the year, so a schedule that
    /// crosses New Year cannot read as if it ran this year.
    #[test]
    fn approval_preview_text_adds_the_year_when_it_differs() {
        let preview = ApprovalPreview {
            routine_name: "standup".to_owned(),
            prompt_text: "prompt".to_owned(),
            next_fires: vec![fire(
                "2027-01-01T09:00:00-08:00",
                "2027-01-01T17:00:00Z",
                "",
            )],
            zone_name: "America/Los_Angeles".to_owned(),
            zone_is_fallback: false,
            cadence: "once".to_owned(),
        };
        let text = approval_preview_text(&preview, at("2026-12-30T17:00:00Z"));
        assert!(text.contains("Fri Jan  1, 2027"), "{text}");
    }

    /// A `once` schedule renders `Runs once`, a `Next run` title (singular —
    /// there is only ever one), and the run's own zone label beside its time,
    /// since "once" names no clock time for the label to ride on.
    #[test]
    fn approval_preview_text_renders_a_once_schedule_end_to_end() {
        let preview = ApprovalPreview {
            routine_name: "standup".to_owned(),
            prompt_text: "Post the launch announcement.".to_owned(),
            next_fires: vec![fire(
                "2026-07-29T09:00:00-07:00",
                "2026-07-29T16:00:00Z",
                "PDT (UTC-7)",
            )],
            zone_name: "America/Los_Angeles".to_owned(),
            zone_is_fallback: false,
            cadence: "once".to_owned(),
        };
        let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
        assert!(text.contains("Runs once"), "{text}");
        assert!(text.contains("Next run"), "{text}");
        assert!(!text.contains("Next 1 run"), "{text}");
        assert!(text.contains("9:00 AM PDT (UTC-7)    tomorrow"), "{text}");
    }

    /// The zone fallback is LABELED, never a silent guess (acceptance
    /// criterion 3 / INV-RL3's "best-effort" assumption), and keeps the IANA
    /// name — an abbreviation cannot tell a reader WHICH zone got guessed.
    #[test]
    fn approval_preview_text_labels_the_zone_fallback() {
        let mut preview = standup_preview();
        preview.zone_is_fallback = true;
        let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
        assert!(
            text.contains("don't know your time zone"),
            "a fallback zone must be disclosed, not silently guessed: {text}"
        );
        assert!(text.contains("America/Los_Angeles"), "{text}");
    }

    /// An empty schedule states the CONSEQUENCE, not just the fact — the
    /// reader is deciding whether to approve it.
    #[test]
    fn approval_preview_text_handles_no_upcoming_runs() {
        let mut preview = standup_preview();
        preview.next_fires = Vec::new();
        let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
        assert!(
            text.contains("no upcoming runs, so approving it won't run anything"),
            "{text}"
        );
        assert!(
            !text.contains("Next"),
            "an empty schedule has no run list: {text}"
        );
    }

    /// An unparseable instant keeps its run on the card rather than dropping
    /// it — a card silently listing two of three runs understates the
    /// schedule being approved.
    #[test]
    fn approval_preview_text_keeps_a_run_whose_instant_cannot_be_parsed() {
        let mut preview = standup_preview();
        preview.next_fires[1] = fire("not-an-instant", "", "");
        let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
        assert!(text.contains("not-an-instant"), "{text}");
        assert!(text.contains("Next 3 runs"), "{text}");
    }

    /// The `ListPending` recovery wire type and the live-card wire type
    /// convert to the SAME [`ApprovalPreview`] — a recovered card cannot
    /// render a different preview than the live one showed.
    #[test]
    fn agent_and_list_wire_previews_convert_to_the_same_shape() {
        let agent_wire = WireAgentApprovalPreview {
            prompt_text: "prompt".to_owned(),
            next_fires: vec![WireAgentApprovalPreviewFire {
                local_time: "2026-07-21T09:00:00-04:00".to_owned(),
                utc_time: "2026-07-21T13:00:00Z".to_owned(),
                ..Default::default()
            }],
            zone_name: "America/New_York".to_owned(),
            zone_is_fallback: true,
            ..Default::default()
        };
        let list_wire = WireListApprovalPreview {
            prompt_text: "prompt".to_owned(),
            next_fires: vec![WireListApprovalPreviewFire {
                local_time: "2026-07-21T09:00:00-04:00".to_owned(),
                utc_time: "2026-07-21T13:00:00Z".to_owned(),
                ..Default::default()
            }],
            zone_name: "America/New_York".to_owned(),
            zone_is_fallback: true,
            ..Default::default()
        };
        assert_eq!(
            ApprovalPreview::from(agent_wire),
            ApprovalPreview::from(list_wire)
        );
    }

    /// A recovered `PendingApprovalEntry` (no live `title` field) still
    /// carries the preview through to the shared [`PendingApproval`] shape.
    #[test]
    fn pending_approval_entry_conversion_carries_the_preview() {
        let entry = PendingApprovalEntry {
            turn_id: "018f47f0-5f70-7cc5-98df-123456789abc".to_owned(),
            request_id: "r1".to_owned(),
            tool_name: "routine_create".to_owned(),
            args_json: "{}".to_owned(),
            preview: buffa::MessageField::some(WireListApprovalPreview {
                prompt_text: "prompt".to_owned(),
                zone_name: "UTC".to_owned(),
                ..Default::default()
            }),
            ..Default::default()
        };
        let pending = PendingApproval::try_from(entry).expect("valid pending approval projection");
        assert_eq!(pending.turn_id, "018f47f0-5f70-7cc5-98df-123456789abc");
        assert_eq!(
            pending.preview.as_ref().map(|p| p.prompt_text.as_str()),
            Some("prompt")
        );
    }

    #[test]
    fn pending_approval_rejects_a_present_read_preview_without_a_detail() {
        let entry = PendingApprovalEntry {
            read_preview: buffa::MessageField::some(WireListReadPreview {
                detail: None,
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            }),
            ..Default::default()
        };

        assert!(matches!(
            PendingApproval::try_from(entry),
            Err(DialError::InvalidApprovalRecovery(
                "approval read preview omitted its detail"
            ))
        ));
    }

    /// `#1660`: a wire `PendingQuestion` round-trips into the client-facing
    /// [`PendingQuestionPrompt`] with every field carried — the question-pause
    /// SIBLING of `pending_approval_entry_conversion_carries_the_preview`
    /// above.
    #[test]
    fn pending_question_conversion_carries_every_field() {
        let wire = WireAgentPendingQuestion {
            call_id: "call-1".to_owned(),
            index: 0,
            header: "Deploy target".to_owned(),
            question: "Which environment?".to_owned(),
            options: vec![
                WireAgentQuestionOption {
                    label: "Staging".to_owned(),
                    description: "Deploys to staging only.".to_owned(),
                    recommended: false,
                    ..Default::default()
                },
                WireAgentQuestionOption {
                    label: "Production".to_owned(),
                    description: "Deploys straight to production.".to_owned(),
                    recommended: true,
                    ..Default::default()
                },
            ],
            args_json: r#"{"questions":[]}"#.to_owned(),
            answer_token: "token-abc".to_owned(),
            already_surfaced: true,
            ..Default::default()
        };
        let prompt = PendingQuestionPrompt::from(wire);
        assert_eq!(prompt.call_id, "call-1");
        assert_eq!(prompt.index, 0);
        assert_eq!(prompt.header, "Deploy target");
        assert_eq!(prompt.question, "Which environment?");
        assert_eq!(prompt.options.len(), 2);
        assert_eq!(prompt.options[1].label, "Production");
        assert!(prompt.options[1].recommended);
        assert_eq!(prompt.args_json, r#"{"questions":[]}"#);
        assert_eq!(prompt.answer_token, "token-abc");
        assert!(prompt.already_surfaced);
    }

    /// `#1660`: the terminal `AgentEnd.pending_questions` projects into ONE
    /// [`TurnEvent::QuestionPending`] per entry — the question-pause SIBLING
    /// of `AgentEnd.pending_approvals`'s own `TurnEvent::ApprovalPending`
    /// projection in `events_from_end`.
    #[test]
    fn events_from_end_projects_one_question_pending_per_entry() {
        let end = AgentEnd {
            pending_questions: vec![WireAgentPendingQuestion {
                call_id: "call-1".to_owned(),
                index: 0,
                header: "h".to_owned(),
                question: "q?".to_owned(),
                options: vec![WireAgentQuestionOption {
                    label: "A".to_owned(),
                    ..Default::default()
                }],
                args_json: "{}".to_owned(),
                answer_token: "tok".to_owned(),
                already_surfaced: true,
                ..Default::default()
            }],
            ..Default::default()
        };
        let events = events_from_end(end);
        let question_events: Vec<&TurnEvent> = events
            .iter()
            .filter(|e| matches!(e, TurnEvent::QuestionPending { .. }))
            .collect();
        assert_eq!(question_events.len(), 1);
        let TurnEvent::QuestionPending {
            call_id,
            answer_token,
            already_surfaced,
            ..
        } = question_events[0]
        else {
            unreachable!("filtered above")
        };
        assert_eq!(call_id, "call-1");
        assert_eq!(answer_token, "tok");
        assert!(
            already_surfaced,
            "the streamed event must copy the control plane's already_surfaced bit through, \
             not hardcode false (unlike the approval side's streamed path)"
        );
        assert!(matches!(events.last(), Some(TurnEvent::Done)));
    }
}