polyc-rpc-client 2026.8.1

//! 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/design/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,
};

/// 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
    /// (`"slack"`, `"mail"`, `"web"`, `"mcp"`, …). Used to keep ids greppable
    /// and to route forensics by edge family.
    fn namespace(&self) -> &'static str;

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

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