polyc_rpc_client/

edge.rs

//! The edge/adapter contract.
//!
//! Every surface that puts the outside world in front of a polychrome
//! conversation — a chat app, a web UI, an inbound-mail handler, an MCP
//! server, a cron trigger — is an *edge*. The design
//! (`docs/reference/public-api-and-edges.md` §3) defines a single thin contract so
//! edges are interchangeable and vendor-agnostic: if an adapter satisfies it,
//! it works, regardless of which product or protocol it wraps.
//!
//! ## The five concerns
//!
//! The contract names five concerns. Two of them — **identity/namespacing** and
//! **ingress** — vary per edge and are pure mappings, so they are the methods of
//! the [`EdgeAdapter`] trait. The other three are satisfied by composing this
//! crate's transport, identically across edges, so they are documented here
//! rather than forced into awkward per-edge trait methods:
//!
//! 1. **Identity & namespacing** — [`EdgeAdapter::namespace`] +
//!    [`EdgeAdapter::conversation_id`], built on [`crate::namespaced_id`] /
//!    [`crate::hashed_conversation_id`].
//! 2. **Ingress** — [`EdgeAdapter::to_turn_input`] turns one native inbound unit
//!    into turn-input [`TurnMessage`]s. After authenticating the transport
//!    envelope, the handler separately constructs the required
//!    [`IngressIdentity`] from that envelope's source coordinate. It is not an
//!    `EdgeAdapter` method because `Inbound` deliberately contains mapped
//!    content, not provider delivery metadata. Any I/O the edge needs first
//!    (resolving display names, fetching a thread) happens before this pure
//!    mapping.
//! 3. **Egress / streaming** — drive [`crate::AgentDialer::run_turn_streaming_messages`]
//!    with the [`TurnMessage`]s and render the [`crate::TurnEvent`] stream where
//!    the transport allows incremental output.
//! 4. **Approval & handoff hooks** — react to [`crate::TurnEvent::ApprovalPending`]
//!    and [`crate::TurnEvent::HandoffStarted`] from that same stream, and answer
//!    approvals out-of-band via [`crate::ApprovalDialer`].
//! 5. **Auth & trust boundary** — authenticate the caller at the edge's own
//!    transport (a chat edge verifies an HMAC; a webhook checks a secret; the public
//!    HTTP surface will check a bearer token) and carry that identity inward.
//!    This stays edge-native because the mechanism differs fundamentally per
//!    transport; the contract only requires that it happens before ingress.
//!
//! `polychrome-slack` is the reference implementation.

use crate::{Attribution, ExternalIdentity, TurnMessage};
use polyc_proto::proto::polychrome::agent::v1::{
    IngressSourceIdentity as WireIngressSourceIdentity, ingress_source_identity,
};

/// Maximum bytes in a claimed tenancy namespace.
///
/// The stored grant side bounds an entry at 256 bytes, because an
/// administrator configures that set and it holds arbitrary tenancy names. A
/// claim is caller-supplied and drawn from a fixed vocabulary whose longest
/// member is eight bytes, so it takes the tighter bound. The value lands in a
/// partition name, a log field, and a query column; short and greppable is
/// worth more than the range it refuses.
pub const MAX_CLAIMED_NAMESPACE_BYTES: usize = 64;

/// Errors constructing a claimed tenancy namespace.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ClaimedNamespaceError {
    /// The claim was empty.
    #[error("a claimed conversation namespace must not be empty")]
    Empty,
    /// The claim was longer than [`MAX_CLAIMED_NAMESPACE_BYTES`].
    #[error(
        "a claimed conversation namespace is 1..={MAX_CLAIMED_NAMESPACE_BYTES} bytes, got {actual}"
    )]
    TooLong {
        /// The rejected length, in bytes.
        actual: usize,
    },
    /// The claim held a byte outside `[a-z0-9]`.
    #[error("a claimed conversation namespace holds lowercase ASCII letters and digits only")]
    InvalidCharacter,
}

/// The tenancy namespace a turn claims for its conversation (#1691).
///
/// The control plane refuses a claim outside the asserting credential's
/// `allowed_namespaces`, and binds the conversation to this value on its first
/// dispatch. This type has one fallible constructor and no empty case, which
/// is what makes "a conversation cannot hold a dispatched turn and no
/// namespace" true by construction rather than by convention (INV-N5).
///
/// This is deliberately narrower than what the state plane accepts in a stored
/// grant. A grant is administrator-controlled housekeeping; a claim is the
/// caller-adversarial boundary. The wildcard `"*"` is a legal grant entry and
/// is never a legal claim — it is a sentinel meaning "any namespace", and no
/// conversation belongs to it.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct ClaimedNamespace(String);

impl ClaimedNamespace {
    /// Validates a claimed tenancy namespace.
    ///
    /// # Errors
    ///
    /// Returns [`ClaimedNamespaceError`] when the claim is empty, longer than
    /// [`MAX_CLAIMED_NAMESPACE_BYTES`], or holds a byte outside `[a-z0-9]`.
    /// The wildcard `"*"` fails the character rule.
    pub fn new(value: impl Into<String>) -> Result<Self, ClaimedNamespaceError> {
        let value = value.into();
        if value.is_empty() {
            return Err(ClaimedNamespaceError::Empty);
        }
        if value.len() > MAX_CLAIMED_NAMESPACE_BYTES {
            return Err(ClaimedNamespaceError::TooLong {
                actual: value.len(),
            });
        }
        if !value
            .bytes()
            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit())
        {
            return Err(ClaimedNamespaceError::InvalidCharacter);
        }
        Ok(Self(value))
    }

    /// Returns the claimed namespace.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for ClaimedNamespace {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

/// Errors constructing a stable ingress identity.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum IngressIdentityError {
    /// The source namespace was empty.
    #[error("ingress source namespace must not be empty")]
    EmptyNamespace,
    /// A source-reported event identifier was empty.
    #[error("reported ingress event id must not be empty")]
    EmptyReportedId,
}

/// Stable identity one source event keeps across every redelivery.
///
/// The namespace and event identifier come from the authenticated source
/// protocol. Neither an execution id nor an envelope nonce can construct this
/// type, because both change between delivery attempts.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct IngressIdentity {
    namespace: String,
    event: IngressEventId,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum IngressEventId {
    Reported(String),
    Derived([u8; 32]),
}

impl IngressIdentity {
    /// Builds an identity from an event identifier reported by the source.
    ///
    /// # Errors
    ///
    /// Returns [`IngressIdentityError::EmptyNamespace`] or
    /// [`IngressIdentityError::EmptyReportedId`] when either required part is
    /// empty.
    pub fn reported(
        namespace: impl Into<String>,
        event_id: impl Into<String>,
    ) -> Result<Self, IngressIdentityError> {
        let namespace = namespace.into();
        if namespace.is_empty() {
            return Err(IngressIdentityError::EmptyNamespace);
        }
        let event_id = event_id.into();
        if event_id.is_empty() {
            return Err(IngressIdentityError::EmptyReportedId);
        }
        Ok(Self {
            namespace,
            event: IngressEventId::Reported(event_id),
        })
    }

    /// Builds a reported identity from an ordered composite source key.
    ///
    /// Each component is length-framed, so values containing delimiters
    /// cannot collide with a different component split.
    ///
    /// # Errors
    ///
    /// Returns [`IngressIdentityError::EmptyNamespace`] when `namespace` is
    /// empty, or [`IngressIdentityError::EmptyReportedId`] when `components`
    /// is empty or contains an empty component.
    pub fn reported_components(
        namespace: impl Into<String>,
        components: &[&str],
    ) -> Result<Self, IngressIdentityError> {
        if components.is_empty() || components.iter().any(|part| part.is_empty()) {
            return Err(IngressIdentityError::EmptyReportedId);
        }
        let mut framed = String::new();
        for part in components {
            framed.push_str(&part.len().to_string());
            framed.push(':');
            framed.push_str(part);
            framed.push('/');
        }
        Self::reported(namespace, framed)
    }

    /// Builds an identity derived from the SHA-256 digest of authenticated
    /// source fields when that source genuinely reports no event identifier.
    ///
    /// # Errors
    ///
    /// Returns [`IngressIdentityError::EmptyNamespace`] when `namespace` is
    /// empty.
    pub fn derived(
        namespace: impl Into<String>,
        authenticated_fields_digest: [u8; 32],
    ) -> Result<Self, IngressIdentityError> {
        let namespace = namespace.into();
        if namespace.is_empty() {
            return Err(IngressIdentityError::EmptyNamespace);
        }
        Ok(Self {
            namespace,
            event: IngressEventId::Derived(authenticated_fields_digest),
        })
    }

    /// Returns the edge-native namespace in which the event id is unique.
    #[must_use]
    pub fn namespace(&self) -> &str {
        &self.namespace
    }

    pub(crate) fn to_wire(&self) -> WireIngressSourceIdentity {
        let event_id = match &self.event {
            IngressEventId::Reported(id) => {
                ingress_source_identity::EventId::ReportedId(id.clone())
            }
            IngressEventId::Derived(digest) => {
                ingress_source_identity::EventId::DerivedDigest(digest.to_vec())
            }
        };
        WireIngressSourceIdentity {
            namespace: self.namespace.clone(),
            event_id: Some(event_id),
            ..Default::default()
        }
    }
}

/// Advisory scheduling weight on an [`IngressDirective`]. Re-exported from the
/// wire type — no scheduler exists today (dispatch is immediate per
/// conversation-lease), so this value is only ever recorded (a signed
/// `ingress_directive` eventlog event, for forensics/operator display), never
/// used to reorder or defer a turn.
pub use polyc_proto::proto::polychrome::agent::v1::Priority;

/// Edge-authored policy for one turn (`#68`).
///
/// The edge's own decision, never something the model or a tool can set.
/// Built by [`EdgeAdapter::ingress_directive`]; threaded through
/// [`crate::AgentDialer`]'s dialer entry points into the wire `AgentStart`.
/// Every field absent (the [`Default`]) is the common case and costs nothing
/// on the wire — [`Self::is_empty`] says so without inspecting each field.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct IngressDirective {
    /// Upper bound on the turn's step budget. The control plane can only
    /// LOWER the budget it would otherwise resolve, never raise it. `None`
    /// means the edge expresses no cap.
    pub budget_cap: Option<u32>,
    /// Advisory scheduling hint; recorded in the eventlog, not scheduled (see
    /// [`Priority`]). `None` means the edge expressed no preference.
    pub priority: Option<Priority>,
    /// The identity that alone may APPROVE a gated call this turn (enforced
    /// fail-closed in `ApprovalService.Respond`); Deny/Defer stay open to any
    /// resolve-token holder. `None` means no approver restriction.
    pub required_approver: Option<ExternalIdentity>,
}

impl IngressDirective {
    /// True when every field is unset — the common case for an edge with no
    /// ingress policy, and the signal the control plane uses to skip
    /// appending an `ingress_directive` eventlog event for this turn.
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.budget_cap.is_none() && self.priority.is_none() && self.required_approver.is_none()
    }
}

/// The contract an edge implements to ride the public Connect API.
///
/// The trait is deliberately small: it captures only the per-edge *mapping*
/// decisions (how a native addressing unit becomes a namespaced conversation
/// id, and how a native inbound unit becomes turn input). Transport — opening
/// the turn stream, rendering deltas, answering approvals — is provided by
/// [`crate::AgentDialer`] / [`crate::ApprovalDialer`] and is the same for every
/// edge, so it is not part of the trait (see the module docs).
pub trait EdgeAdapter {
    /// This edge's native addressing unit — whatever it derives a conversation
    /// from. Slack: a `(team, channel, thread)` coordinate; an inbound-mail
    /// handler: a `(mailbox, thread)` pair; a web UI: a session id.
    type Native;

    /// One native inbound unit that [`Self::to_turn_input`] maps to turn input.
    /// Slack: one attributed thread line; a web UI: one request message. The
    /// edge resolves any I/O (display names, thread fetch) into this value
    /// *before* the pure mapping.
    type Inbound;

    /// The namespace prefix this edge stamps onto conversation ids, and the
    /// tenancy namespace it claims. Used to keep ids greppable, to route
    /// forensics by edge family, and to authorize the write (#1691).
    ///
    /// The live values are `slack`, `telegram`, `discord`, `email`, and `evt`.
    /// Note `email`, not `mail`: an operator granting the wrong spelling
    /// refuses every turn from that edge, and this doc comment is where a new
    /// edge author reads the convention.
    ///
    /// This value is also the edge's tenancy claim: see
    /// [`Self::claimed_namespace`]. Return the same string the operator grants
    /// in `allowed_namespaces`, or narrowing that grant refuses every turn
    /// this edge sends.
    fn namespace(&self) -> &'static str;

    /// The tenancy namespace this edge claims on every turn it dials (#1691).
    ///
    /// Derived from [`Self::namespace`] so the greppable id prefix and the
    /// authorization claim cannot drift apart. An edge does not override this.
    ///
    /// # Panics
    ///
    /// Panics when [`Self::namespace`] returns a value
    /// [`ClaimedNamespace`] refuses. Every namespace is a compile-time
    /// constant, so a panic here means a new edge shipped one no operator can
    /// ever grant. Failing loudly at startup beats refusing that edge's every
    /// turn.
    ///
    /// This crate cannot prove that for a given edge: the implementors live
    /// in downstream crates, so `every_real_namespace_is_a_valid_claim` below
    /// checks a hand-written list and cannot catch a NEW edge. Each edge
    /// crate carries its own `the_namespace_is_a_valid_claim` test against
    /// its own adapter, which is the only place that check is real.
    #[must_use]
    fn claimed_namespace(&self) -> ClaimedNamespace {
        ClaimedNamespace::new(self.namespace()).expect("an edge's namespace is a valid claim")
    }

    /// Derive the namespaced [`AgentRequest`](crate::AgentDialer) conversation
    /// id for a native unit. Implementations build it from
    /// [`crate::namespaced_id`] (readable) or [`crate::hashed_conversation_id`]
    /// (fixed-length opaque), keeping [`Self::namespace`] as the prefix /
    /// pinned-namespace policy.
    fn conversation_id(&self, native: &Self::Native) -> String;

    /// Map one inbound unit to zero-or-more turn-input messages. An empty
    /// result drops the unit (e.g. an empty or self-authored message). Pure:
    /// no I/O — the edge does any lookups before calling this, so the mapping
    /// is unit-testable in isolation.
    fn to_turn_input(&self, inbound: &Self::Inbound) -> Vec<TurnMessage>;

    /// The external identity of the human behind one inbound unit, when the
    /// edge knows it — the second pure *identity* mapping next to
    /// [`Self::conversation_id`] (docs/reference/personas.md §4). The tuple must
    /// use the provider's *stable* id with its disambiguating scope (a chat
    /// workspace id), never a mutable handle. `None` (the default) means this
    /// unit carries no identity; an edge that returns `None` for every unit
    /// (and sends no participants) is attribution-free.
    fn caller(&self, _inbound: &Self::Inbound) -> Option<ExternalIdentity> {
        None
    }

    /// This edge's ingress policy for one inbound unit (`#68`) — a step-budget
    /// cap, an advisory priority, and/or a required approver, attached to the
    /// event that starts (or resumes) the turn. Edge-authored, never derived
    /// from message content: an edge that has no such policy leaves the
    /// default, which is empty ([`IngressDirective::is_empty`]) and costs
    /// nothing downstream — every existing edge compiles unchanged.
    fn ingress_directive(&self, _inbound: &Self::Inbound) -> IngressDirective {
        IngressDirective::default()
    }
}

/// Assemble a turn's [`Attribution`] from an edge's inbound units.
///
/// The triggering unit's identity becomes the caller; every *other* distinct
/// identity among `observed` becomes a participant. Pass as `observed` only
/// the units whose content actually enters the turn's input — attribution
/// must not record speakers the turn never saw. Deduplication is by
/// `(provider, scope, external_id)` — display names don't identify.
///
/// SDK-composed (not a trait method) so every edge shares one dedupe policy;
/// the per-edge part is exactly [`EdgeAdapter::caller`].
pub fn build_attribution<E: EdgeAdapter>(
    edge: &E,
    trigger: Option<&E::Inbound>,
    observed: &[E::Inbound],
) -> Attribution {
    // Identity equality is the (provider, scope, external_id) tuple —
    // display names don't identify.
    fn same(a: &ExternalIdentity, b: &ExternalIdentity) -> bool {
        a.provider == b.provider && a.scope == b.scope && a.external_id == b.external_id
    }

    let caller = trigger.and_then(|t| edge.caller(t));

    let mut participants: Vec<ExternalIdentity> = Vec::new();
    for unit in observed {
        let Some(identity) = edge.caller(unit) else {
            continue;
        };
        let duplicate = caller.as_ref().is_some_and(|c| same(c, &identity))
            || participants.iter().any(|p| same(p, &identity));
        if !duplicate {
            participants.push(identity);
        }
    }

    Attribution {
        caller,
        participants,
    }
}

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

    /// Every namespace a real surface claims is accepted. The list is the
    /// vocabulary in ADR 0011's minter table; a rule that refused one of these
    /// would refuse that surface's every turn.
    #[test]
    fn every_real_namespace_is_a_valid_claim() {
        for namespace in [
            "slack", "telegram", "discord", "email", "evt", "a2a", "web", "app", "mcp", "routine",
            "eval", "cli",
        ] {
            assert_eq!(
                ClaimedNamespace::new(namespace)
                    .expect("a real namespace is a valid claim")
                    .as_str(),
                namespace,
            );
        }
    }

    /// The wildcard is a grant sentinel, never a claim. A conversation cannot
    /// belong to "any namespace", and every credential deployed today holds
    /// the wildcard — so an unconstrained claim would let any of them bind a
    /// conversation to it permanently, with no adoption path.
    #[test]
    fn the_wildcard_is_not_a_valid_claim() {
        assert_eq!(
            ClaimedNamespace::new("*"),
            Err(ClaimedNamespaceError::InvalidCharacter),
        );
    }

    #[test]
    fn an_empty_claim_is_refused() {
        assert_eq!(ClaimedNamespace::new(""), Err(ClaimedNamespaceError::Empty));
    }

    /// A colon would split under any namespaced-id reader, and the stored
    /// grant side refuses one for the same reason.
    #[test]
    fn a_claim_holding_a_colon_is_refused() {
        assert_eq!(
            ClaimedNamespace::new("slack:team"),
            Err(ClaimedNamespaceError::InvalidCharacter),
        );
    }

    #[test]
    fn a_claim_is_lowercase_ascii_only() {
        for rejected in ["Slack", "web-chat", "web_chat", "café", "web.chat", " web"] {
            assert_eq!(
                ClaimedNamespace::new(rejected),
                Err(ClaimedNamespaceError::InvalidCharacter),
                "{rejected} must not be a valid claim",
            );
        }
    }

    #[test]
    fn a_claim_is_bounded() {
        let longest = "a".repeat(MAX_CLAIMED_NAMESPACE_BYTES);
        assert!(ClaimedNamespace::new(longest).is_ok());
        let over = "a".repeat(MAX_CLAIMED_NAMESPACE_BYTES + 1);
        assert_eq!(
            ClaimedNamespace::new(over),
            Err(ClaimedNamespaceError::TooLong {
                actual: MAX_CLAIMED_NAMESPACE_BYTES + 1,
            }),
        );
    }

    #[test]
    fn source_identity_is_stable_when_execution_identity_changes() {
        let identity = crate::IngressIdentity::reported("workspace-1", "message-42")
            .expect("reported source identity is valid");

        let first = identity.to_wire();
        let second = identity.to_wire();
        assert_eq!(first, second);
        assert_eq!(identity.namespace(), "workspace-1");
    }

    #[test]
    fn composite_reported_identity_is_injective() {
        let left =
            IngressIdentity::reported_components("source", &["a/b", "c"]).expect("valid identity");
        let right =
            IngressIdentity::reported_components("source", &["a", "b/c"]).expect("valid identity");
        assert_ne!(left, right);
    }

    // A minimal reference edge proving the contract is implementable and pure.
    struct ExampleEdge {
        namespace_uuid: uuid::Uuid,
    }

    struct Thread {
        team: String,
        channel: String,
        thread_ts: String,
    }

    struct Line {
        speaker: String,
        text: String,
    }

    impl EdgeAdapter for ExampleEdge {
        type Native = Thread;
        type Inbound = Line;

        fn namespace(&self) -> &'static str {
            "example"
        }

        fn conversation_id(&self, native: &Thread) -> String {
            hashed_conversation_id(
                self.namespace_uuid,
                &[&native.team, &native.channel, &native.thread_ts],
            )
        }

        fn to_turn_input(&self, inbound: &Line) -> Vec<TurnMessage> {
            if inbound.text.trim().is_empty() {
                return Vec::new();
            }
            vec![attributed_message(&inbound.speaker, &inbound.text)]
        }
    }

    #[test]
    fn conversation_id_is_stable_per_native_unit() {
        let edge = ExampleEdge {
            namespace_uuid: uuid::Uuid::from_u128(0x42),
        };
        let t = Thread {
            team: "T1".to_owned(),
            channel: "C1".to_owned(),
            thread_ts: "169.0".to_owned(),
        };
        assert_eq!(edge.conversation_id(&t), edge.conversation_id(&t));
        assert_eq!(edge.namespace(), "example");
    }

    #[test]
    fn ingress_drops_empty_and_attributes_speakers() {
        let edge = ExampleEdge {
            namespace_uuid: uuid::Uuid::from_u128(0x42),
        };
        assert!(
            edge.to_turn_input(&Line {
                speaker: "Alice".to_owned(),
                text: "   ".to_owned(),
            })
            .is_empty()
        );
        assert_eq!(
            edge.to_turn_input(&Line {
                speaker: "Alice".to_owned(),
                text: "hi".to_owned(),
            }),
            vec![attributed_message("Alice", "hi")]
        );
    }
}