supercode-harness 0.4.8

//! Transport-neutral observer and controller leases for live SDK runtimes.
//!
//! HTTP, ACP, local frontends, and compatibility adapters all project this
//! coordinator. Transports authenticate clients, but they do not get to
//! invent different ownership or expiry semantics.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use crate::{
    FrontendAttachment, FrontendOperationInvocation, FrontendOperationResult, FrontendResponse,
    FrontendRuntimeDescriptor, SdkError, SdkOperation, SdkRuntime,
};

/// Default controller duration. Successful mutations and explicit heartbeats
/// renew it; disconnect and detach release it immediately.
pub const DEFAULT_RUNTIME_LEASE_TTL_MS: u64 = 30_000;

/// Capability independently granted to an authenticated runtime client.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RuntimePermission {
    /// Discover, describe, attach, and watch canonical runtime events.
    Observe,
    /// Own the controller lease and mutate turn state.
    Interact,
    /// Answer approval or elicitation requests while controlling the runtime.
    Approve,
    /// Explicitly close the SDK-owned runtime.
    Terminate,
}

impl RuntimePermission {
    /// Stable permission spelling used by every binding and error envelope.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Observe => "observe",
            Self::Interact => "interact",
            Self::Approve => "approve",
            Self::Terminate => "terminate",
        }
    }
}

/// Permissions carried by one authenticated runtime client.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeAuthorization {
    permissions: BTreeSet<RuntimePermission>,
}

impl RuntimeAuthorization {
    /// Construct an authorization grant from its exact permission set.
    pub fn new(permissions: impl IntoIterator<Item = RuntimePermission>) -> Self {
        Self {
            permissions: permissions.into_iter().collect(),
        }
    }

    /// Full owner grant used by the process that started the runtime.
    pub fn owner() -> Self {
        Self::new([
            RuntimePermission::Observe,
            RuntimePermission::Interact,
            RuntimePermission::Approve,
            RuntimePermission::Terminate,
        ])
    }

    /// Interactive frontend grant. Frontends may observe, drive turns, and
    /// answer requests, but runtime termination remains an operator-only
    /// control-plane action.
    pub fn interactive() -> Self {
        Self::new([
            RuntimePermission::Observe,
            RuntimePermission::Interact,
            RuntimePermission::Approve,
        ])
    }

    /// Read-only observer grant.
    pub fn observer() -> Self {
        Self::new([RuntimePermission::Observe])
    }

    /// Whether this grant contains `permission`.
    pub fn allows(&self, permission: RuntimePermission) -> bool {
        self.permissions.contains(&permission)
    }

    /// Stable ordered permission inventory for descriptors and bindings.
    pub fn permissions(&self) -> impl Iterator<Item = RuntimePermission> + '_ {
        self.permissions.iter().copied()
    }

    /// Narrow this grant to permissions also present in `requested`.
    pub fn restrict_to(&self, requested: &Self) -> Self {
        Self::new(
            self.permissions
                .intersection(&requested.permissions)
                .copied(),
        )
    }

    /// Stable comma-separated representation used only to narrow an already
    /// authenticated transport credential.
    pub fn header_value(&self) -> String {
        self.permissions()
            .map(RuntimePermission::as_str)
            .collect::<Vec<_>>()
            .join(",")
    }

    /// Decode a narrowing request. Unknown or duplicate-independent names
    /// fail closed instead of being ignored.
    pub fn parse_header(value: &str) -> Result<Self, RuntimeLeaseError> {
        if value.is_empty() {
            return Ok(Self::default());
        }
        let mut permissions = Vec::new();
        for name in value.split(',') {
            let permission = match name {
                "observe" => RuntimePermission::Observe,
                "interact" => RuntimePermission::Interact,
                "approve" => RuntimePermission::Approve,
                "terminate" => RuntimePermission::Terminate,
                _ => return Err(RuntimeLeaseError::InvalidAuthorization),
            };
            permissions.push(permission);
        }
        Ok(Self::new(permissions))
    }
}

/// Opaque identity scoped to one authenticated frontend connection.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(transparent)]
pub struct RuntimeClientId(String);

impl RuntimeClientId {
    /// Validate a transport-provided client identity.
    pub fn parse(value: impl Into<String>) -> Result<Self, RuntimeLeaseError> {
        let value = value.into();
        if value.is_empty()
            || value.len() > 128
            || !value
                .bytes()
                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
        {
            return Err(RuntimeLeaseError::InvalidClientId);
        }
        Ok(Self(value))
    }

    /// Stable opaque spelling used on protocol boundaries.
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

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

/// Controller lease visible through runtime describe/list operations.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeControllerLease {
    /// Authenticated client that owns mutation rights.
    pub client_id: RuntimeClientId,
    /// Absolute epoch-millisecond deadline. At or after this value the lease
    /// is expired and another authorized client may claim it.
    pub expires_at_ms: u64,
}

/// One attached observer and its last authenticated activity.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeObserverLease {
    /// Authenticated observing client.
    pub client_id: RuntimeClientId,
    /// Last epoch-millisecond activity used for stale-client reconciliation.
    pub last_seen_ms: u64,
}

/// Serializable ownership snapshot included in registry descriptors.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeLeaseSnapshot {
    /// Current unexpired controller, if any.
    pub controller: Option<RuntimeControllerLease>,
    /// Attached observers in stable client-id order.
    pub observers: Vec<RuntimeObserverLease>,
    /// Configured lease duration.
    pub lease_ttl_ms: u64,
}

/// Stable coordination failure shared by every transport projection.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum RuntimeLeaseError {
    /// Client identity is absent or not a valid opaque identifier.
    #[error("invalid runtime client id")]
    InvalidClientId,
    /// A transport attempted to request an unknown permission spelling.
    #[error("invalid runtime authorization grant")]
    InvalidAuthorization,
    /// The authenticated client lacks the operation's required grant.
    #[error("runtime permission `{0:?}` is required")]
    Unauthorized(RuntimePermission),
    /// Another client still owns the controller lease.
    #[error("controller lease is held by `{holder}` until {expires_at_ms}")]
    ControllerHeld {
        /// Current controller.
        holder: RuntimeClientId,
        /// Current controller deadline.
        expires_at_ms: u64,
    },
    /// A mutation was attempted without owning the controller lease.
    #[error("controller lease required")]
    ControllerRequired,
    /// This client owned the lease, but its deadline has passed.
    #[error("controller lease expired")]
    LeaseExpired,
}

/// One-controller, many-observer coordination state for a live runtime.
#[derive(Debug)]
pub struct RuntimeLeaseCoordinator {
    lease_ttl_ms: u64,
    controller: Option<RuntimeControllerLease>,
    observers: BTreeMap<RuntimeClientId, RuntimeObserverLease>,
    expired_controller: Option<RuntimeClientId>,
}

impl RuntimeLeaseCoordinator {
    /// Create a coordinator with a non-zero controller lease duration.
    pub fn new(lease_ttl_ms: u64) -> Self {
        assert!(lease_ttl_ms > 0, "runtime lease TTL must be non-zero");
        Self {
            lease_ttl_ms,
            controller: None,
            observers: BTreeMap::new(),
            expired_controller: None,
        }
    }

    /// Register or refresh one authenticated observer.
    pub fn attach(
        &mut self,
        client_id: RuntimeClientId,
        authorization: &RuntimeAuthorization,
        now_ms: u64,
    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
        require(authorization, RuntimePermission::Observe)?;
        self.reconcile(now_ms);
        self.observers.insert(
            client_id.clone(),
            RuntimeObserverLease {
                client_id,
                last_seen_ms: now_ms,
            },
        );
        Ok(self.snapshot(now_ms))
    }

    /// Refresh observation activity without changing controller ownership.
    pub fn heartbeat(
        &mut self,
        client_id: &RuntimeClientId,
        authorization: &RuntimeAuthorization,
        now_ms: u64,
    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
        require(authorization, RuntimePermission::Observe)?;
        self.reconcile(now_ms);
        self.observers
            .entry(client_id.clone())
            .and_modify(|observer| observer.last_seen_ms = now_ms)
            .or_insert_with(|| RuntimeObserverLease {
                client_id: client_id.clone(),
                last_seen_ms: now_ms,
            });
        Ok(self.snapshot(now_ms))
    }

    /// Validate a non-controller capability. Observation and lifecycle
    /// administration are authorized independently from turn ownership;
    /// interactive and approval mutations must use
    /// [`Self::authorize_controller`] instead.
    pub fn authorize(
        &mut self,
        authorization: &RuntimeAuthorization,
        permission: RuntimePermission,
        now_ms: u64,
    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
        require(authorization, permission)?;
        Ok(self.snapshot(now_ms))
    }

    /// Claim the controller when unowned, or renew it when already owned by
    /// this client. Another live controller is never displaced implicitly.
    pub fn claim_control(
        &mut self,
        client_id: RuntimeClientId,
        authorization: &RuntimeAuthorization,
        now_ms: u64,
    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
        require(authorization, RuntimePermission::Interact)?;
        self.attach(client_id.clone(), authorization, now_ms)?;
        if self.controller.is_none() && self.expired_controller.as_ref() == Some(&client_id) {
            return Err(RuntimeLeaseError::LeaseExpired);
        }
        match &self.controller {
            Some(lease) if lease.client_id != client_id => {
                return Err(RuntimeLeaseError::ControllerHeld {
                    holder: lease.client_id.clone(),
                    expires_at_ms: lease.expires_at_ms,
                });
            }
            _ => {}
        }
        self.controller = Some(RuntimeControllerLease {
            client_id,
            expires_at_ms: now_ms.saturating_add(self.lease_ttl_ms),
        });
        self.expired_controller = None;
        Ok(self.snapshot(now_ms))
    }

    /// Explicitly transfer control to another authorized interactive client.
    pub fn take_control(
        &mut self,
        client_id: RuntimeClientId,
        authorization: &RuntimeAuthorization,
        now_ms: u64,
    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
        require(authorization, RuntimePermission::Interact)?;
        self.attach(client_id.clone(), authorization, now_ms)?;
        self.controller = Some(RuntimeControllerLease {
            client_id,
            expires_at_ms: now_ms.saturating_add(self.lease_ttl_ms),
        });
        self.expired_controller = None;
        Ok(self.snapshot(now_ms))
    }

    /// Validate and renew mutation authority for `client_id`.
    pub fn authorize_controller(
        &mut self,
        client_id: &RuntimeClientId,
        authorization: &RuntimeAuthorization,
        permission: RuntimePermission,
        now_ms: u64,
    ) -> Result<RuntimeLeaseSnapshot, RuntimeLeaseError> {
        require(authorization, permission)?;
        let was_expired = self
            .controller
            .as_ref()
            .is_some_and(|lease| lease.client_id == *client_id && lease.expires_at_ms <= now_ms);
        self.reconcile(now_ms);
        let Some(controller) = &mut self.controller else {
            return Err(
                if was_expired || self.expired_controller.as_ref() == Some(client_id) {
                    RuntimeLeaseError::LeaseExpired
                } else {
                    RuntimeLeaseError::ControllerRequired
                },
            );
        };
        if controller.client_id != *client_id {
            return Err(RuntimeLeaseError::ControllerRequired);
        }
        controller.expires_at_ms = now_ms.saturating_add(self.lease_ttl_ms);
        self.expired_controller = None;
        if let Some(observer) = self.observers.get_mut(client_id) {
            observer.last_seen_ms = now_ms;
        }
        Ok(self.snapshot(now_ms))
    }

    /// Remove an observer and release its controller lease, if held.
    pub fn detach(&mut self, client_id: &RuntimeClientId, now_ms: u64) -> RuntimeLeaseSnapshot {
        self.reconcile(now_ms);
        self.observers.remove(client_id);
        if self
            .controller
            .as_ref()
            .is_some_and(|lease| lease.client_id == *client_id)
        {
            self.controller = None;
        }
        if self.expired_controller.as_ref() == Some(client_id) {
            self.expired_controller = None;
        }
        self.snapshot(now_ms)
    }

    /// Current state after deterministic expiry reconciliation.
    pub fn snapshot(&mut self, now_ms: u64) -> RuntimeLeaseSnapshot {
        self.reconcile(now_ms);
        RuntimeLeaseSnapshot {
            controller: self.controller.clone(),
            observers: self.observers.values().cloned().collect(),
            lease_ttl_ms: self.lease_ttl_ms,
        }
    }

    fn reconcile(&mut self, now_ms: u64) {
        if self
            .controller
            .as_ref()
            .is_some_and(|lease| lease.expires_at_ms <= now_ms)
        {
            self.expired_controller = self
                .controller
                .take()
                .map(|controller| controller.client_id);
        }
    }
}

/// Shared transport-neutral owner of a runtime and its client leases.
///
/// A transport authenticates a request, validates its opaque client id, and
/// asks this object for a [`CoordinatedRuntimeClient`]. HTTP, ACP, local
/// embedding, and compatibility adapters therefore execute identical
/// authorization, takeover, and expiry behavior.
pub struct CoordinatedRuntime {
    runtime: Arc<dyn SdkRuntime>,
    leases: Mutex<RuntimeLeaseCoordinator>,
}

impl CoordinatedRuntime {
    /// Wrap one SDK runtime with the default controller lease duration.
    pub fn new(runtime: Arc<dyn SdkRuntime>) -> Arc<Self> {
        Self::with_lease_ttl(runtime, DEFAULT_RUNTIME_LEASE_TTL_MS)
    }

    /// Wrap one SDK runtime with an explicit lease duration. Tests and
    /// embedders use this to exercise deterministic expiry without sleeping.
    pub fn with_lease_ttl(runtime: Arc<dyn SdkRuntime>, lease_ttl_ms: u64) -> Arc<Self> {
        Arc::new(Self {
            runtime,
            leases: Mutex::new(RuntimeLeaseCoordinator::new(lease_ttl_ms)),
        })
    }

    /// Bind an authenticated identity and exact permission set to this
    /// runtime. The returned client never owns the underlying runtime.
    pub fn client(
        self: &Arc<Self>,
        client_id: RuntimeClientId,
        authorization: RuntimeAuthorization,
    ) -> Arc<CoordinatedRuntimeClient> {
        Arc::new(CoordinatedRuntimeClient {
            coordinator: self.clone(),
            client_id,
            authorization,
        })
    }

    fn leases(&self) -> std::sync::MutexGuard<'_, RuntimeLeaseCoordinator> {
        self.leases
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }
}

/// Authenticated client projection of one coordinated SDK runtime.
pub struct CoordinatedRuntimeClient {
    coordinator: Arc<CoordinatedRuntime>,
    client_id: RuntimeClientId,
    authorization: RuntimeAuthorization,
}

impl CoordinatedRuntimeClient {
    /// Opaque authenticated identity used by lease snapshots and transports.
    pub fn client_id(&self) -> &RuntimeClientId {
        &self.client_id
    }

    /// Attach or refresh this observer and return current ownership state.
    pub fn observe(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
        self.coordinator
            .leases()
            .attach(self.client_id.clone(), &self.authorization, epoch_ms())
            .map_err(|error| lease_sdk_error(error, SdkOperation::Events))
    }

    /// Explicitly take the controller lease from another interactive client.
    pub fn take_control(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
        self.coordinator
            .leases()
            .take_control(self.client_id.clone(), &self.authorization, epoch_ms())
            .map_err(|error| lease_sdk_error(error, SdkOperation::Input))
    }

    /// Refresh this client's observer activity and controller lease when it
    /// currently owns control.
    pub fn heartbeat(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
        let now_ms = epoch_ms();
        let mut leases = self.coordinator.leases();
        let mut snapshot = leases
            .heartbeat(&self.client_id, &self.authorization, now_ms)
            .map_err(|error| lease_sdk_error(error, SdkOperation::Events))?;
        if snapshot
            .controller
            .as_ref()
            .is_some_and(|lease| lease.client_id == self.client_id)
        {
            snapshot = leases
                .authorize_controller(
                    &self.client_id,
                    &self.authorization,
                    RuntimePermission::Interact,
                    now_ms,
                )
                .map_err(|error| lease_sdk_error(error, SdkOperation::Input))?;
        }
        Ok(snapshot)
    }

    /// Release observation and controller ownership without closing the
    /// SDK-owned runtime.
    pub fn detach(&self) -> RuntimeLeaseSnapshot {
        self.coordinator
            .leases()
            .detach(&self.client_id, epoch_ms())
    }

    /// Current ownership state, visible only to observers.
    pub fn lease_snapshot(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
        self.coordinator
            .leases()
            .authorize(&self.authorization, RuntimePermission::Observe, epoch_ms())
            .map_err(|error| lease_sdk_error(error, SdkOperation::Events))
    }

    fn authorize_controller(
        &self,
        permission: RuntimePermission,
        operation: SdkOperation,
    ) -> Result<(), SdkError> {
        let now_ms = epoch_ms();
        let mut leases = self.coordinator.leases();
        leases
            .claim_control(self.client_id.clone(), &self.authorization, now_ms)
            .map_err(|error| lease_sdk_error(error, operation))?;
        if permission != RuntimePermission::Interact {
            leases
                .authorize_controller(&self.client_id, &self.authorization, permission, now_ms)
                .map_err(|error| lease_sdk_error(error, operation))?;
        }
        Ok(())
    }

    fn authorize_lifecycle(
        &self,
        permission: RuntimePermission,
        operation: SdkOperation,
    ) -> Result<(), SdkError> {
        self.coordinator
            .leases()
            .authorize(&self.authorization, permission, epoch_ms())
            .map(|_| ())
            .map_err(|error| lease_sdk_error(error, operation))
    }

    async fn descriptor(&self) -> Result<FrontendRuntimeDescriptor, SdkError> {
        let mut descriptor = self.coordinator.runtime.describe().await?;
        descriptor.actions.submit &= self.authorization.allows(RuntimePermission::Interact);
        descriptor.actions.interrupt &= self.authorization.allows(RuntimePermission::Interact);
        descriptor.actions.steer &= self.authorization.allows(RuntimePermission::Interact);
        descriptor.actions.respond &= self.authorization.allows(RuntimePermission::Approve)
            && self.authorization.allows(RuntimePermission::Interact);
        descriptor.actions.close &= self.authorization.allows(RuntimePermission::Terminate);
        descriptor.actions.detach &= self.authorization.allows(RuntimePermission::Observe);
        Ok(descriptor)
    }
}

#[async_trait]
impl SdkRuntime for CoordinatedRuntimeClient {
    async fn describe(&self) -> Result<FrontendRuntimeDescriptor, SdkError> {
        self.authorize_lifecycle(RuntimePermission::Observe, SdkOperation::Events)?;
        self.descriptor().await
    }

    async fn attach(&self, history_limit: usize) -> Result<FrontendAttachment, SdkError> {
        self.observe()?;
        let mut attachment = self.coordinator.runtime.attach(history_limit).await?;
        attachment.descriptor = self.descriptor().await?;
        Ok(attachment)
    }

    async fn send_input(self: Arc<Self>, prompt: String) -> Result<(), SdkError> {
        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
        self.coordinator.runtime.clone().send_input(prompt).await
    }

    async fn send_input_with_images(
        self: Arc<Self>,
        prompt: String,
        image_urls: Vec<String>,
    ) -> Result<(), SdkError> {
        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
        self.coordinator
            .runtime
            .clone()
            .send_input_with_images(prompt, image_urls)
            .await
    }

    async fn submit(&self, prompt: String) -> Result<String, SdkError> {
        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
        self.coordinator.runtime.submit(prompt).await
    }

    async fn submit_with_images(
        &self,
        prompt: String,
        image_urls: Vec<String>,
    ) -> Result<String, SdkError> {
        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
        self.coordinator
            .runtime
            .submit_with_images(prompt, image_urls)
            .await
    }

    async fn interrupt(&self) -> Result<bool, SdkError> {
        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Interrupt)?;
        self.coordinator.runtime.interrupt().await
    }

    async fn steer(&self, prompt: String) -> Result<(), SdkError> {
        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Steer)?;
        self.coordinator.runtime.steer(prompt).await
    }

    async fn respond(&self, response: FrontendResponse) -> Result<(), SdkError> {
        self.authorize_controller(RuntimePermission::Approve, SdkOperation::Respond)?;
        self.coordinator.runtime.respond(response).await
    }

    async fn invoke(
        &self,
        operation: FrontendOperationInvocation,
    ) -> Result<FrontendOperationResult, SdkError> {
        self.authorize_controller(RuntimePermission::Interact, SdkOperation::Input)?;
        self.coordinator.runtime.invoke(operation).await
    }

    async fn lease_snapshot(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
        CoordinatedRuntimeClient::lease_snapshot(self)
    }

    async fn take_control(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
        CoordinatedRuntimeClient::take_control(self)
    }

    async fn heartbeat(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
        CoordinatedRuntimeClient::heartbeat(self)
    }

    async fn detach(&self) -> Result<RuntimeLeaseSnapshot, SdkError> {
        Ok(CoordinatedRuntimeClient::detach(self))
    }

    async fn close(&self) -> Result<(), SdkError> {
        self.authorize_lifecycle(RuntimePermission::Terminate, SdkOperation::Close)?;
        self.coordinator.runtime.close().await
    }
}

fn lease_sdk_error(error: RuntimeLeaseError, operation: SdkOperation) -> SdkError {
    match error {
        error @ (RuntimeLeaseError::InvalidClientId | RuntimeLeaseError::InvalidAuthorization) => {
            SdkError::InvalidArgument {
                operation,
                message: error.to_string(),
            }
        }
        RuntimeLeaseError::Unauthorized(permission) => SdkError::Unauthorized {
            permission: permission.as_str().into(),
        },
        RuntimeLeaseError::ControllerHeld {
            holder,
            expires_at_ms,
        } => SdkError::ControllerRequired {
            holder: Some(holder.to_string()),
            expires_at_ms: Some(expires_at_ms),
        },
        RuntimeLeaseError::ControllerRequired => SdkError::ControllerRequired {
            holder: None,
            expires_at_ms: None,
        },
        RuntimeLeaseError::LeaseExpired => SdkError::LeaseExpired,
    }
}

fn epoch_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
        .min(u64::MAX as u128) as u64
}

fn require(
    authorization: &RuntimeAuthorization,
    permission: RuntimePermission,
) -> Result<(), RuntimeLeaseError> {
    if authorization.allows(permission) {
        Ok(())
    } else {
        Err(RuntimeLeaseError::Unauthorized(permission))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn client(value: &str) -> RuntimeClientId {
        RuntimeClientId::parse(value).unwrap()
    }

    #[test]
    fn many_observers_share_one_explicit_controller() {
        let mut leases = RuntimeLeaseCoordinator::new(100);
        let owner = RuntimeAuthorization::owner();
        let observer = RuntimeAuthorization::observer();
        leases.attach(client("viewer-a"), &observer, 10).unwrap();
        leases.attach(client("viewer-b"), &observer, 11).unwrap();
        let snapshot = leases.claim_control(client("owner"), &owner, 12).unwrap();
        assert_eq!(snapshot.observers.len(), 3);
        assert_eq!(snapshot.controller.unwrap().client_id, client("owner"));

        let error = leases
            .claim_control(client("viewer-b"), &owner, 13)
            .unwrap_err();
        assert!(matches!(error, RuntimeLeaseError::ControllerHeld { .. }));
    }

    #[test]
    fn observer_cannot_claim_control_approve_or_terminate() {
        let mut leases = RuntimeLeaseCoordinator::new(100);
        let observer = RuntimeAuthorization::observer();
        leases.attach(client("viewer"), &observer, 1).unwrap();
        assert_eq!(
            leases
                .claim_control(client("viewer"), &observer, 2)
                .unwrap_err(),
            RuntimeLeaseError::Unauthorized(RuntimePermission::Interact)
        );
        assert_eq!(
            leases
                .authorize_controller(&client("viewer"), &observer, RuntimePermission::Approve, 2)
                .unwrap_err(),
            RuntimeLeaseError::Unauthorized(RuntimePermission::Approve)
        );
        assert_eq!(
            leases
                .authorize(&observer, RuntimePermission::Terminate, 2)
                .unwrap_err(),
            RuntimeLeaseError::Unauthorized(RuntimePermission::Terminate)
        );
    }

    #[test]
    fn expiry_is_deterministic_and_requires_a_new_claim() {
        let mut leases = RuntimeLeaseCoordinator::new(10);
        let owner = RuntimeAuthorization::owner();
        leases.claim_control(client("a"), &owner, 5).unwrap();
        assert_eq!(
            leases.claim_control(client("a"), &owner, 15).unwrap_err(),
            RuntimeLeaseError::LeaseExpired
        );
        assert_eq!(
            leases
                .authorize_controller(&client("a"), &owner, RuntimePermission::Interact, 15)
                .unwrap_err(),
            RuntimeLeaseError::LeaseExpired
        );
        let snapshot = leases.claim_control(client("b"), &owner, 15).unwrap();
        assert_eq!(snapshot.controller.unwrap().client_id, client("b"));
    }

    #[test]
    fn successful_mutation_renews_controller_and_observer_activity() {
        let mut leases = RuntimeLeaseCoordinator::new(10);
        let owner = RuntimeAuthorization::owner();
        leases.claim_control(client("a"), &owner, 5).unwrap();
        let snapshot = leases
            .authorize_controller(&client("a"), &owner, RuntimePermission::Approve, 9)
            .unwrap();
        assert_eq!(snapshot.controller.unwrap().expires_at_ms, 19);
        assert_eq!(snapshot.observers[0].last_seen_ms, 9);
    }

    #[test]
    fn takeover_and_detach_are_explicit_and_release_control() {
        let mut leases = RuntimeLeaseCoordinator::new(10);
        let owner = RuntimeAuthorization::owner();
        leases.claim_control(client("a"), &owner, 1).unwrap();
        let snapshot = leases.take_control(client("b"), &owner, 2).unwrap();
        assert_eq!(snapshot.controller.unwrap().client_id, client("b"));
        let snapshot = leases.detach(&client("b"), 3);
        assert!(snapshot.controller.is_none());
        assert_eq!(
            snapshot
                .observers
                .into_iter()
                .map(|observer| observer.client_id)
                .collect::<Vec<_>>(),
            vec![client("a")]
        );
    }

    #[test]
    fn client_ids_are_opaque_bounded_and_header_safe() {
        for invalid in ["", "space here", "slash/here", "💥"] {
            assert_eq!(
                RuntimeClientId::parse(invalid).unwrap_err(),
                RuntimeLeaseError::InvalidClientId
            );
        }
        assert_eq!(
            RuntimeClientId::parse("a".repeat(129)).unwrap_err(),
            RuntimeLeaseError::InvalidClientId
        );
        assert_eq!(
            RuntimeClientId::parse("client-1.v2_ok").unwrap().as_str(),
            "client-1.v2_ok"
        );
        let owner = RuntimeAuthorization::owner();
        let requested = RuntimeAuthorization::parse_header("observe,interact").unwrap();
        assert_eq!(
            owner.restrict_to(&requested).header_value(),
            "observe,interact"
        );
        assert_eq!(
            RuntimeAuthorization::parse_header("observe,admin").unwrap_err(),
            RuntimeLeaseError::InvalidAuthorization
        );
    }
}