polyc-turn-runner 2026.9.0

//! Narrow, capability-bound broker admission for D7.
//!
//! # What this module's checks prove, and what they do not
//!
//! Read this before citing an admission check as a security property. Some
//! checks bind a decision made elsewhere; others are unreachable from any
//! production input because Control builds both sides of the comparison from
//! the same locals. Both are worth keeping — an unreachable check is a guard
//! against a future caller that does not share today's construction — but only
//! the first kind is evidence.
//!
//! **Load-bearing today** — a production input can fail these:
//!
//! - `registry.resolve` rejecting an unknown or wrong-family target;
//! - the encoded-request byte cap;
//! - the effective deadline, taken as the minimum of the capability, session,
//!   broker, and caller budgets;
//! - the one-use capability key, which stops a replayed capability starting a
//!   second call;
//! - the per-turn call cap and the pending-request cap.
//!
//! **Unreachable from production input today** — kept as guards, not evidence:
//!
//! - `BindingMismatch` and `UnadmittedLabel`. Control mints the capability and
//!   the request from the same locals, adjacent, with no await between them,
//!   so the fields cannot differ. They would fire only for a caller that
//!   builds the two from separate sources.
//! - `target.read_only` in the scope check. The only production
//!   [`crate::broker::TrustedRegistry`] for the connector family hardcodes `read_only: true`,
//!   because the real read-only decision already happened upstream in
//!   Control's catalog check — a tool whose catalog entry does not declare
//!   `read_only_hint` never reaches this module at all.
//!
//! **Carried but not read in production:** [`crate::broker::TrustedTarget::classification`].
//! Control computes it, and the value is asserted in tests, but no production
//! path branches on it. The untrusted marking that actually protects a reply
//! is unconditional and lives at the Harness boundary
//! (`polyc_harness::mcp_proxy`), so the `Quarantined` variant is currently
//! dead. It stays because the classification belongs to the trusted side and
//! the enforcement seam is the next thing to move here; a reader must not
//! mistake it for an active control.
//!
//! This target also used to carry a response-header cap and a decoded-body
//! cap, documented as "the peer transport applies this value". It did not:
//! the peer transport reads the `CONNECTOR_RESPONSE_*` constants directly, and
//! the two fields had no read site anywhere in the workspace. They are gone.
//! The peer bound itself is unchanged — it was never coming from here — and
//! the connector family still has no response-size bound, which is stated as a
//! residual rather than implied by a field.
//!
//! Execution names a logical family, target, method, and bounded resource. It
//! never supplies a URL, bearer credential, caller header, or peer address.
//! The trusted side resolves those values from its registry after this module
//! admits the request. This Component intentionally contains policy and
//! correlation only: Containers provide the registry and transport outward.

use std::collections::BTreeSet;

use thiserror::Error;

use crate::execution::ExecutionLabel;

/// One outbound family D7 must broker completely.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum BrokerFamily {
    /// A read-only connector resource request.
    Connector,
    /// A registered peer read request.
    Peer,
}

/// The trusted classification every broker return keeps after verification.
///
/// A valid signature can establish integrity but cannot turn external content
/// into trusted instruction text, so there is intentionally no `Trusted`
/// variant here.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReturnedContent {
    /// Ordinary external content, always untrusted at the boundary.
    Untrusted,
    /// Content subject to a stronger quarantine policy.
    Quarantined,
}

/// A correlation tuple the broker records without request or response bodies.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BrokerCorrelation {
    /// Tenant that owns the conversation.
    pub tenant: String,
    /// Conversation that owns the execution.
    pub conversation: String,
    /// D4 execution identity.
    pub execution: String,
    /// D4 attempt identity.
    pub attempt: String,
    /// D4 step identity.
    pub step: String,
    /// Provider tool-call identity, unique within the accepted step.
    pub tool_call: String,
    /// Capability audience the trusted registry granted.
    pub audience: String,
}

/// A capability minted by the trusted side for exactly one broker call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BrokerCapability {
    /// Durable correlation tuple.
    pub correlation: BrokerCorrelation,
    /// The only D4 label the session admits for this capability.
    pub admitted_label: ExecutionLabel,
    /// Logical outbound family.
    pub family: BrokerFamily,
    /// Trusted registry key, never a raw address.
    pub target: String,
    /// Exact method, rather than a broad method class.
    pub method: String,
    /// Exact resource name or route template chosen by the registry.
    pub resource: String,
    /// Largest encoded request body this capability may carry.
    pub max_request_bytes: usize,
    /// Absolute monotonic expiry supplied by the trusted session clock.
    pub expires_at_millis: u64,
}

/// The non-secret request Execution may present to the broker.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BrokerRequest {
    /// The capability's correlation tuple, repeated so any substitution fails.
    pub correlation: BrokerCorrelation,
    /// The D4 label on the envelope.
    pub label: ExecutionLabel,
    /// Requested family.
    pub family: BrokerFamily,
    /// Logical target key.
    pub target: String,
    /// Exact HTTP/RPC method selected by the capability.
    pub method: String,
    /// Exact bounded resource selected by the capability.
    pub resource: String,
    /// Encoded body size. The bytes are not retained or logged by this policy.
    pub encoded_request_bytes: usize,
    /// Caller-proposed deadline in the same monotonic clock domain.
    pub deadline_at_millis: u64,
}

/// Registry data the trusted broker resolves after capability admission.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TrustedTarget {
    /// Registry key that must equal [`BrokerCapability::target`].
    pub target: String,
    /// Family served by this target.
    pub family: BrokerFamily,
    /// Exact method the target admits.
    pub method: String,
    /// Exact resource it admits.
    pub resource: String,
    /// Whether D7 may execute it. D7 is a read-only surface.
    pub read_only: bool,
    /// Classification applied even after a successful signature check.
    pub classification: ReturnedContent,
}

/// The trusted registry boundary. Its implementation owns destination URL,
/// bearer credential, peer address, resolver pinning, and connection setup.
pub trait TrustedRegistry {
    /// Re-resolves a logical target from trusted data. `None` fails closed.
    fn resolve(&self, family: BrokerFamily, target: &str) -> Option<TrustedTarget>;
}

/// Independent limits that cap one broker session.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BrokerLimits {
    /// Maximum calls admitted in the session.
    pub max_calls: usize,
    /// Maximum requests waiting on a trusted transport at one time.
    pub max_pending: usize,
    /// Session deadline, applied even if a retry re-enters admission.
    pub session_deadline_at_millis: u64,
    /// Broker-local deadline, combined with all other budgets by minimum.
    pub broker_deadline_at_millis: u64,
}

/// A fully checked admission handed to the trusted transport.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BrokerAdmission {
    /// Registry result, never Execution-supplied destination data.
    pub target: TrustedTarget,
    /// Deadline that no retry is allowed to extend.
    pub deadline_at_millis: u64,
    /// Correlation data appropriate for body-free telemetry.
    pub correlation: BrokerCorrelation,
}

/// A fail-closed broker refusal.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
pub enum BrokerRefusal {
    /// Any tenant/conversation/execution/attempt/step/call/audience substitution.
    #[error("broker capability binding mismatch")]
    BindingMismatch,
    /// A label outside the D4 session. Equality admits a session frame only; it
    /// is intentionally not claimed to prove State freshness.
    #[error("Execution label is not admitted by this broker session")]
    UnadmittedLabel,
    /// A capability that reached its absolute expiry.
    #[error("broker capability expired")]
    Expired,
    /// An unknown logical target or a family mismatch in trusted registry data.
    #[error("trusted registry has no matching target")]
    UnknownTarget,
    /// Exact method/resource/read-only checks failed.
    #[error("broker request exceeds its exact target authority")]
    ScopeMismatch,
    /// Request body exceeded the encoded bound before any dial.
    #[error("broker request exceeds encoded body limit")]
    RequestTooLarge,
    /// The session's call count or pending map is full.
    #[error("broker session capacity exhausted")]
    CapacityExceeded,
    /// The end-to-end deadline has already elapsed.
    #[error("broker deadline exhausted")]
    DeadlineExceeded,
    /// One-use capability replay.
    #[error("broker capability was already used")]
    CapabilityReplayed,
}

/// Mutable admission state for one principal-and-conversation broker session.
///
/// Pending state is intentionally local to this tuple: callers never share
/// pending maps or cache keys across tenants or conversations.
#[derive(Debug)]
pub struct BrokerSession {
    limits: BrokerLimits,
    calls: usize,
    pending: usize,
    used: BTreeSet<String>,
}

impl BrokerSession {
    /// Opens an empty, bounded broker session.
    #[must_use]
    pub const fn new(limits: BrokerLimits) -> Self {
        Self {
            limits,
            calls: 0,
            pending: 0,
            used: BTreeSet::new(),
        }
    }

    /// Validates and reserves one capability use before a trusted dial.
    ///
    /// `now_millis` is supplied by the trusted caller. The resulting deadline
    /// is the minimum of session, capability, broker, and caller budgets, so a
    /// retry can reuse it but cannot reset it.
    ///
    /// # Errors
    ///
    /// Returns a typed refusal when the capability, target, request bounds, or
    /// current broker budget is not admitted by trusted policy.
    pub fn admit(
        &mut self,
        capability: &BrokerCapability,
        request: &BrokerRequest,
        registry: &impl TrustedRegistry,
        now_millis: u64,
    ) -> Result<BrokerAdmission, BrokerRefusal> {
        if capability.correlation != request.correlation
            || capability.family != request.family
            || capability.target != request.target
            || capability.method != request.method
            || capability.resource != request.resource
        {
            return Err(BrokerRefusal::BindingMismatch);
        }
        if capability.admitted_label != request.label {
            return Err(BrokerRefusal::UnadmittedLabel);
        }
        if now_millis >= capability.expires_at_millis {
            return Err(BrokerRefusal::Expired);
        }
        if request.encoded_request_bytes > capability.max_request_bytes {
            return Err(BrokerRefusal::RequestTooLarge);
        }
        let deadline_at_millis = capability
            .expires_at_millis
            .min(self.limits.session_deadline_at_millis)
            .min(self.limits.broker_deadline_at_millis)
            .min(request.deadline_at_millis);
        if now_millis >= deadline_at_millis {
            return Err(BrokerRefusal::DeadlineExceeded);
        }
        let use_key = capability_use_key(capability);
        if self.used.contains(&use_key) {
            return Err(BrokerRefusal::CapabilityReplayed);
        }
        if self.calls >= self.limits.max_calls || self.pending >= self.limits.max_pending {
            return Err(BrokerRefusal::CapacityExceeded);
        }
        let target = registry
            .resolve(request.family, &request.target)
            .filter(|target| target.family == request.family)
            .ok_or(BrokerRefusal::UnknownTarget)?;
        if target.method != request.method
            || target.resource != request.resource
            || !target.read_only
        {
            return Err(BrokerRefusal::ScopeMismatch);
        }
        self.used.insert(use_key);
        self.calls += 1;
        self.pending += 1;
        Ok(BrokerAdmission {
            target,
            deadline_at_millis,
            correlation: capability.correlation.clone(),
        })
    }

    /// Releases one pending slot after cancellation, success, timeout, or an
    /// unknown result. The one-use reservation remains: a lost response is not
    /// a license to begin a second downstream call.
    pub const fn complete(&mut self) {
        self.pending = self.pending.saturating_sub(1);
    }

    /// Returns the pending count for bounded-state conformance tests.
    #[must_use]
    pub const fn pending(&self) -> usize {
        self.pending
    }
}

/// Stable one-use identity, deliberately including every bound principal and
/// call component rather than an ambient process-global counter.
fn capability_use_key(capability: &BrokerCapability) -> String {
    let c = &capability.correlation;
    format!(
        "{}:{}:{}:{}:{}:{}:{}:{}",
        c.tenant,
        c.conversation,
        c.execution,
        c.attempt,
        c.step,
        c.tool_call,
        c.audience,
        capability.target,
    )
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use std::collections::BTreeMap;

    use polyc_state::{command::FencingToken, id::AttemptId};

    use super::*;
    use crate::execution::{ExecutionId, ExecutionIdentity};

    #[derive(Default)]
    struct Registry(BTreeMap<(BrokerFamily, String), TrustedTarget>);

    impl TrustedRegistry for Registry {
        fn resolve(&self, family: BrokerFamily, target: &str) -> Option<TrustedTarget> {
            self.0.get(&(family, target.to_owned())).cloned()
        }
    }

    fn label() -> ExecutionLabel {
        ExecutionLabel::new(
            ExecutionIdentity::new(
                ExecutionId::new("execution"),
                AttemptId::new("attempt"),
                FencingToken::new(4),
            )
            .expect("identity"),
            2,
        )
    }

    fn correlation() -> BrokerCorrelation {
        BrokerCorrelation {
            tenant: "tenant-a".to_owned(),
            conversation: "conversation-a".to_owned(),
            execution: "execution".to_owned(),
            attempt: "attempt".to_owned(),
            step: label().step().id().as_str().to_owned(),
            tool_call: "call-7".to_owned(),
            audience: "connector-read".to_owned(),
        }
    }

    fn capability() -> BrokerCapability {
        BrokerCapability {
            correlation: correlation(),
            admitted_label: label(),
            family: BrokerFamily::Connector,
            target: "calendar".to_owned(),
            method: "GET".to_owned(),
            resource: "/v1/events".to_owned(),
            max_request_bytes: 128,
            expires_at_millis: 1_000,
        }
    }

    fn request() -> BrokerRequest {
        BrokerRequest {
            correlation: correlation(),
            label: label(),
            family: BrokerFamily::Connector,
            target: "calendar".to_owned(),
            method: "GET".to_owned(),
            resource: "/v1/events".to_owned(),
            encoded_request_bytes: 8,
            deadline_at_millis: 900,
        }
    }

    fn registry() -> Registry {
        let mut registry = Registry::default();
        registry.0.insert(
            (BrokerFamily::Connector, "calendar".to_owned()),
            TrustedTarget {
                target: "calendar".to_owned(),
                family: BrokerFamily::Connector,
                method: "GET".to_owned(),
                resource: "/v1/events".to_owned(),
                read_only: true,
                classification: ReturnedContent::Untrusted,
            },
        );
        registry
    }

    fn session() -> BrokerSession {
        BrokerSession::new(BrokerLimits {
            max_calls: 2,
            max_pending: 1,
            session_deadline_at_millis: 800,
            broker_deadline_at_millis: 700,
        })
    }

    #[test]
    fn binds_every_identity_field_and_uses_the_smallest_deadline() {
        let mut session = session();
        let admission = session
            .admit(&capability(), &request(), &registry(), 100)
            .expect("the exact trusted request is admitted");
        assert_eq!(admission.deadline_at_millis, 700);
        assert_eq!(admission.target.classification, ReturnedContent::Untrusted);
    }

    #[test]
    fn changed_bound_fields_refuse_before_registry_resolution() {
        let mut changed = request();
        changed.correlation.tenant = "tenant-b".to_owned();
        assert_eq!(
            session().admit(&capability(), &changed, &registry(), 100),
            Err(BrokerRefusal::BindingMismatch)
        );
    }

    #[test]
    fn a_label_outside_the_admitted_session_is_not_freshness_authority() {
        let mut changed = request();
        changed.label = ExecutionLabel::new(
            ExecutionIdentity::new(
                ExecutionId::new("execution"),
                AttemptId::new("other-attempt"),
                FencingToken::new(5),
            )
            .expect("identity"),
            2,
        );
        assert_eq!(
            session().admit(&capability(), &changed, &registry(), 100),
            Err(BrokerRefusal::UnadmittedLabel)
        );
    }

    #[test]
    fn a_one_use_capability_cannot_start_a_second_call_after_completion() {
        let mut session = session();
        session
            .admit(&capability(), &request(), &registry(), 100)
            .expect("first call");
        session.complete();
        assert_eq!(
            session.admit(&capability(), &request(), &registry(), 100),
            Err(BrokerRefusal::CapabilityReplayed)
        );
    }

    #[test]
    fn unregistered_mutating_or_out_of_scope_targets_fail_closed() {
        let mut changed_request = request();
        changed_request.method = "POST".to_owned();
        assert_eq!(
            session().admit(&capability(), &changed_request, &registry(), 100),
            Err(BrokerRefusal::BindingMismatch)
        );
        let mut mutating = registry();
        mutating
            .0
            .get_mut(&(BrokerFamily::Connector, "calendar".to_owned()))
            .expect("target")
            .read_only = false;
        assert_eq!(
            session().admit(&capability(), &request(), &mutating, 100),
            Err(BrokerRefusal::ScopeMismatch)
        );
    }

    #[test]
    fn bounds_expiry_pending_and_deadline_are_enforced_before_a_dial() {
        let mut oversized = request();
        oversized.encoded_request_bytes = 129;
        assert_eq!(
            session().admit(&capability(), &oversized, &registry(), 100),
            Err(BrokerRefusal::RequestTooLarge)
        );
        assert_eq!(
            session().admit(&capability(), &request(), &registry(), 1_000),
            Err(BrokerRefusal::Expired)
        );
        assert_eq!(
            session().admit(&capability(), &request(), &registry(), 700),
            Err(BrokerRefusal::DeadlineExceeded)
        );
        let mut session = session();
        session
            .admit(&capability(), &request(), &registry(), 100)
            .expect("first pending call");
        let mut second = capability();
        second.correlation.tool_call = "call-8".to_owned();
        assert_eq!(
            session.admit(
                &second,
                &BrokerRequest {
                    correlation: second.correlation.clone(),
                    ..request()
                },
                &registry(),
                100
            ),
            Err(BrokerRefusal::CapacityExceeded)
        );
        assert_eq!(session.pending(), 1);
    }
}