meerkat_runtime/handles/

auth_lease.rs

//! Runtime impl of [`meerkat_core::handles::AuthLeaseHandle`].
//!
//! Per-binding [`AuthMachine`](crate::auth_machine) instances, keyed by
//! typed [`LeaseKey`](meerkat_core::handles::LeaseKey). Each trait
//! method looks up or creates the corresponding machine, then fires
//! the matching DSL input. `snapshot` projects the machine's phase
//! + expires_at back out.
//!
//! The original Phase 1.5-rev design absorbed auth state into
//! MeerkatMachine as Sets+Maps keyed by binding. Post-review that was
//! rejected: auth lifecycle is orthogonal to MeerkatMachine's
//! lifecycle, and carrying it there grew the TLC state space without
//! unifying any semantics (dogma §19). Splitting it out per-binding
//! keeps each machine small, aligns the fact "this binding's lease
//! state" with one canonical owner per dogma §1, and decouples auth
//! evolution from core-runtime evolution.

use std::collections::HashMap;
#[cfg(not(target_arch = "wasm32"))]
use std::sync::Weak;
use std::sync::{Arc, Mutex};

#[cfg(not(target_arch = "wasm32"))]
use meerkat_core::AuthBindingRef;
use meerkat_core::RefreshFailureObservation;
use meerkat_core::auth::TokenKey;
use meerkat_core::generated::auth_lease_durable_lifecycle_marker::AuthLeaseDurableRestorePublication;
use meerkat_core::handles::{
    AuthLeaseHandle, AuthLeasePhase, AuthLeaseRestoreSnapshot, AuthLeaseSnapshot,
    AuthLeaseTransition, DslTransitionError, LeaseKey,
};
use meerkat_core::time_compat::{SystemTime, UNIX_EPOCH};

use crate::auth_machine::dsl as auth_dsl;

fn current_time_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
        .unwrap_or(0)
}

/// Emit a structured audit record for every accepted auth-lease DSL
/// transition. REST/RPC surfaces (and any other `tracing::Subscriber`
/// consumer) can filter on `target = "meerkat::auth::audit"` to build
/// a persistent audit log without the shell inventing the fact set
/// (dogma §17 — surfaces observe, they do not own lifecycle truth).
///
/// Deferral §5: structured observability for auth events. We emit at
/// the single choke point where DSL transitions succeed, not from each
/// caller — that keeps the audit trail aligned with the machine's
/// actual acceptance of the transition and avoids duplication /
/// drift (dogma §1).
fn emit_audit(
    lease_key: &LeaseKey,
    action: &'static str,
    from_phase: AuthLeasePhase,
    to_phase: AuthLeasePhase,
) {
    tracing::info!(
        target: "meerkat::auth::audit",
        lease_key = %lease_key,
        realm = %lease_key.realm,
        binding = %lease_key.binding,
        profile = lease_key.profile.as_ref().map(meerkat_core::ProfileId::as_str),
        action = %action,
        from_phase = ?from_phase,
        to_phase = ?to_phase,
        "auth lease transition"
    );
}

/// Runtime-backed [`AuthLeaseHandle`] impl.
///
/// Holds a mutex-guarded registry of per-binding [`auth_dsl::AuthMachineAuthority`]
/// instances. Lookup-or-insert happens on first `acquire_lease`; release is
/// also allowed before acquire so token-clear surfaces remain idempotent after
/// process restart.
#[derive(Clone)]
pub struct RuntimeAuthLeaseHandle {
    machines: Arc<Mutex<AuthLeaseRegistry>>,
    #[cfg(not(target_arch = "wasm32"))]
    release_observers: Arc<Mutex<Vec<Weak<dyn AuthLeaseReleaseObserver>>>>,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone)]
pub(crate) struct ReleasedOAuthFlows {
    pub lease_key: LeaseKey,
    pub browser_flow_ids: Vec<String>,
    pub device_flow_ids: Vec<String>,
}

#[cfg(not(target_arch = "wasm32"))]
impl ReleasedOAuthFlows {
    fn empty(lease_key: LeaseKey) -> Self {
        Self {
            lease_key,
            browser_flow_ids: Vec::new(),
            device_flow_ids: Vec::new(),
        }
    }

    fn dedup(&mut self) {
        self.browser_flow_ids.sort();
        self.browser_flow_ids.dedup();
        self.device_flow_ids.sort();
        self.device_flow_ids.dedup();
    }
}

fn restore_phase_to_dsl(phase: AuthLeasePhase) -> auth_dsl::AuthLifecyclePhase {
    match phase {
        AuthLeasePhase::Valid => auth_dsl::AuthLifecyclePhase::Valid,
        AuthLeasePhase::Expiring => auth_dsl::AuthLifecyclePhase::Expiring,
        AuthLeasePhase::Expired => auth_dsl::AuthLifecyclePhase::Expired,
        AuthLeasePhase::Refreshing => auth_dsl::AuthLifecyclePhase::Refreshing,
        AuthLeasePhase::ReauthRequired => auth_dsl::AuthLifecyclePhase::ReauthRequired,
        AuthLeasePhase::Released => auth_dsl::AuthLifecyclePhase::Released,
    }
}

fn restore_input_from_lifecycle(
    lifecycle_phase: auth_dsl::AuthLifecyclePhase,
    expires_at: Option<u64>,
    last_refresh: Option<u64>,
    refresh_attempt: u64,
    credential_present: bool,
    credential_generation: u64,
    credential_published_at_millis: Option<u64>,
) -> auth_dsl::AuthMachineInput {
    auth_dsl::AuthMachineInput::RestoreAuthoritySnapshot {
        lifecycle_phase,
        expires_at,
        last_refresh,
        refresh_attempt,
        credential_present,
        credential_generation,
        credential_published_at_millis,
    }
}

#[allow(clippy::too_many_arguments)]
fn restore_credential_lifecycle_snapshot_input(
    lifecycle_phase: Option<auth_dsl::AuthLifecyclePhase>,
    expires_at: Option<u64>,
    last_refresh: Option<u64>,
    refresh_attempt: u64,
    credential_present: bool,
    credential_generation: u64,
    credential_published_at_millis: Option<u64>,
    restored_oauth_membership_observed: bool,
) -> auth_dsl::AuthMachineInput {
    auth_dsl::AuthMachineInput::RestoreCredentialLifecycleSnapshot {
        lifecycle_phase,
        expires_at,
        last_refresh,
        refresh_attempt,
        credential_present,
        credential_generation,
        credential_published_at_millis,
        restored_oauth_membership_observed,
    }
}

// Test-only restore-path fixture plumbing: the production restore arm was
// deleted by K7 (clear+release is one AuthMachine-authorized transition);
// these helpers remain solely to drive the generated restore guards in tests.
#[cfg(test)]
fn append_restore_oauth_inputs_from_state(
    inputs: &mut Vec<auth_dsl::AuthMachineInput>,
    poll_inputs: &mut Vec<auth_dsl::AuthMachineInput>,
    state: &auth_dsl::AuthMachineState,
) {
    for flow_id in &state.oauth_browser_flow_ids {
        inputs.push(auth_dsl::AuthMachineInput::RestoreOAuthBrowserFlow {
            flow_id: flow_id.clone(),
            provider: state.oauth_browser_flow_providers.get(flow_id).cloned(),
            redirect_uri: state.oauth_browser_flow_redirect_uris.get(flow_id).cloned(),
            expires_at_millis: state
                .oauth_browser_flow_expires_at_millis
                .get(flow_id)
                .copied(),
        });
    }
    for flow_id in &state.oauth_device_flow_ids {
        inputs.push(auth_dsl::AuthMachineInput::RestoreOAuthDeviceFlow {
            flow_id: flow_id.clone(),
            provider: state.oauth_device_flow_providers.get(flow_id).cloned(),
            expires_at_millis: state
                .oauth_device_flow_expires_at_millis
                .get(flow_id)
                .copied(),
        });
    }
    for poll_id in &state.oauth_device_poll_ids {
        poll_inputs.push(auth_dsl::AuthMachineInput::RestoreOAuthDevicePoll {
            flow_id: poll_id.clone(),
        });
    }
}

// Test-only restore-path fixture plumbing: the production restore arm was
// deleted by K7 (clear+release is one AuthMachine-authorized transition);
// these helpers remain solely to drive the generated restore guards in tests.
#[cfg(test)]
fn restore_oauth_inputs_from_states(
    states: &[&auth_dsl::AuthMachineState],
) -> Vec<auth_dsl::AuthMachineInput> {
    let mut inputs = Vec::new();
    let mut poll_inputs = Vec::new();
    for state in states {
        append_restore_oauth_inputs_from_state(&mut inputs, &mut poll_inputs, state);
    }
    inputs.extend(poll_inputs);
    inputs
}

// Test-only restore-path fixture plumbing: the production restore arm was
// deleted by K7 (clear+release is one AuthMachine-authorized transition);
// these helpers remain solely to drive the generated restore guards in tests.
#[cfg(test)]
fn restore_oauth_membership_observed(states: &[&auth_dsl::AuthMachineState]) -> bool {
    states
        .iter()
        .any(|state| state.oauth_outstanding_flow_count > 0)
}

fn map_auth_machine_error(
    err: auth_dsl::AuthMachineTransitionError,
    context: &'static str,
) -> DslTransitionError {
    let reason = err.to_string();
    match err {
        auth_dsl::AuthMachineTransitionError::GuardRejected { .. } => {
            DslTransitionError::guard_rejected(context, reason)
        }
        auth_dsl::AuthMachineTransitionError::NoMatchingTransition { .. } => {
            DslTransitionError::no_matching(context, reason)
        }
        auth_dsl::AuthMachineTransitionError::RecoveredStateInvariantRejected { .. } => {
            DslTransitionError::recovered_state_invariant_rejected(context, reason)
        }
    }
}

fn apply_restore_input(
    authority: &mut auth_dsl::AuthMachineAuthority,
    lease_key: &LeaseKey,
    input: auth_dsl::AuthMachineInput,
    context: &'static str,
) -> Result<(AuthLeasePhase, AuthLeaseTransition), DslTransitionError> {
    let transition = auth_dsl::AuthMachineMutator::apply(authority, input)
        .map_err(|err| map_auth_machine_error(err, context))?;
    let auth_transition = auth_lease_transition_from_generated_publication(
        lease_key,
        authority,
        &transition,
        context,
    )?;
    Ok((
        map_phase(authority.state().lifecycle_phase),
        auth_transition,
    ))
}

fn apply_restore_input_to_registry(
    registry: &mut AuthLeaseRegistry,
    lease_key: &LeaseKey,
    input: auth_dsl::AuthMachineInput,
    context: &'static str,
) -> Result<(AuthLeasePhase, AuthLeaseTransition), DslTransitionError> {
    let authority = registry
        .authorities
        .entry(lease_key.clone())
        .or_insert_with(auth_dsl::AuthMachineAuthority::new);
    apply_restore_input(authority, lease_key, input, context)
}

// Test-only restore-path fixture plumbing: the production restore arm was
// deleted by K7 (clear+release is one AuthMachine-authorized transition);
// these helpers remain solely to drive the generated restore guards in tests.
#[cfg(test)]
fn restore_authority_from_registry(
    registry: &AuthLeaseRegistry,
    lease_key: &LeaseKey,
    context: &'static str,
) -> Result<auth_dsl::AuthMachineAuthority, DslTransitionError> {
    match registry.authorities.get(lease_key) {
        Some(authority) => {
            auth_dsl::AuthMachineAuthority::recover_from_state(authority.state().clone())
                .map_err(|err| map_auth_machine_error(err, context))
        }
        None => Ok(auth_dsl::AuthMachineAuthority::new()),
    }
}

// Test-only restore-path fixture plumbing: the production restore arm was
// deleted by K7 (clear+release is one AuthMachine-authorized transition);
// these helpers remain solely to drive the generated restore guards in tests.
#[cfg(test)]
fn apply_restore_inputs_to_registry(
    registry: &mut AuthLeaseRegistry,
    lease_key: &LeaseKey,
    lifecycle_input: auth_dsl::AuthMachineInput,
    oauth_inputs: Vec<auth_dsl::AuthMachineInput>,
    context: &'static str,
) -> Result<(AuthLeasePhase, AuthLeaseTransition), DslTransitionError> {
    let mut authority = restore_authority_from_registry(registry, lease_key, context)?;
    let restored = apply_restore_input(&mut authority, lease_key, lifecycle_input, context)?;
    for input in oauth_inputs {
        apply_restore_input(&mut authority, lease_key, input, context)?;
    }
    registry.authorities.insert(lease_key.clone(), authority);
    Ok(restored)
}

// Test-only restore-path fixture plumbing: the production restore arm was
// deleted by K7 (clear+release is one AuthMachine-authorized transition);
// these helpers remain solely to drive the generated restore guards in tests.
#[cfg(test)]
fn restore_state_to_registry(
    registry: &mut AuthLeaseRegistry,
    lease_key: &LeaseKey,
    state: &auth_dsl::AuthMachineState,
    context: &'static str,
) -> Result<(AuthLeasePhase, AuthLeaseTransition), DslTransitionError> {
    restore_state_with_oauth_sources_to_registry(registry, lease_key, state, &[state], context)
}

// Test-only restore-path fixture plumbing: the production restore arm was
// deleted by K7 (clear+release is one AuthMachine-authorized transition);
// these helpers remain solely to drive the generated restore guards in tests.
#[cfg(test)]
fn restore_state_with_oauth_sources_to_registry(
    registry: &mut AuthLeaseRegistry,
    lease_key: &LeaseKey,
    lifecycle_state: &auth_dsl::AuthMachineState,
    oauth_sources: &[&auth_dsl::AuthMachineState],
    context: &'static str,
) -> Result<(AuthLeasePhase, AuthLeaseTransition), DslTransitionError> {
    let oauth_inputs = restore_oauth_inputs_from_states(oauth_sources);
    apply_restore_inputs_to_registry(
        registry,
        lease_key,
        restore_credential_lifecycle_snapshot_input(
            Some(lifecycle_state.lifecycle_phase),
            lifecycle_state.expires_at,
            lifecycle_state.last_refresh,
            lifecycle_state.refresh_attempt,
            lifecycle_state.credential_present,
            lifecycle_state.credential_generation,
            lifecycle_state.credential_published_at_millis,
            restore_oauth_membership_observed(oauth_sources),
        ),
        oauth_inputs,
        context,
    )
}

fn auth_lease_transition_from_generated_publication(
    lease_key: &LeaseKey,
    authority: &auth_dsl::AuthMachineAuthority,
    transition: &auth_dsl::AuthMachineTransition,
    context: &'static str,
) -> Result<AuthLeaseTransition, DslTransitionError> {
    match maybe_auth_lease_transition_from_generated_publication(
        lease_key, authority, transition, context,
    )? {
        Some(transition) => Ok(transition),
        None => Err(DslTransitionError::no_matching(
            context,
            "AuthMachine transition emitted no lifecycle publication obligation",
        )),
    }
}

fn maybe_auth_lease_transition_from_generated_publication(
    lease_key: &LeaseKey,
    authority: &auth_dsl::AuthMachineAuthority,
    transition: &auth_dsl::AuthMachineTransition,
    context: &'static str,
) -> Result<Option<AuthLeaseTransition>, DslTransitionError> {
    let mut obligations =
        crate::protocol_auth_lease_lifecycle_publication::extract_obligations(transition);
    if obligations.is_empty() {
        return Ok(None);
    }
    if obligations.len() != 1 {
        return Err(DslTransitionError::no_matching(
            context,
            format!(
                "AuthMachine transition emitted {} lifecycle publication obligations",
                obligations.len()
            ),
        ));
    }
    let scope =
        crate::protocol_auth_lease_lifecycle_publication::AuthLeaseLifecyclePublicationScope::from_authority(
            lease_key.clone(),
            authority,
        );
    obligations
        .remove(0)
        .into_auth_lease_transition(scope)
        .map(Some)
        .map_err(|err| {
            DslTransitionError::no_matching(
                context,
                format!("AuthMachine lifecycle publication handoff failed: {err}"),
            )
        })
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) trait AuthLeaseReleaseObserver: Send + Sync {
    fn begin_auth_lease_release<'a>(
        &'a self,
        _lease_key: &LeaseKey,
    ) -> Result<Option<Box<dyn AuthLeaseReleasePermit + 'a>>, DslTransitionError> {
        Ok(None)
    }

    fn oauth_flows_for_release(
        &self,
        lease_key: &LeaseKey,
    ) -> Result<ReleasedOAuthFlows, DslTransitionError> {
        Ok(ReleasedOAuthFlows::empty(lease_key.clone()))
    }

    fn auth_lease_released(&self, released: &ReleasedOAuthFlows) -> Result<(), DslTransitionError>;
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) trait AuthLeaseReleasePermit {}

#[cfg(test)]
pub(crate) type ReleaseAfterAcceptHook = Arc<dyn Fn(&LeaseKey) + Send + Sync>;

#[cfg(test)]
static RELEASE_AFTER_ACCEPT_HOOK: std::sync::OnceLock<Mutex<Option<ReleaseAfterAcceptHook>>> =
    std::sync::OnceLock::new();

#[cfg(test)]
static RELEASE_AFTER_ACCEPT_HOOK_SERIAL: std::sync::OnceLock<Mutex<()>> =
    std::sync::OnceLock::new();

#[cfg(test)]
pub(crate) type ReleaseBeforeCommitHook = Arc<dyn Fn(&LeaseKey) + Send + Sync>;

#[cfg(test)]
static RELEASE_BEFORE_COMMIT_HOOK: std::sync::OnceLock<Mutex<Option<ReleaseBeforeCommitHook>>> =
    std::sync::OnceLock::new();

#[cfg(test)]
static RELEASE_BEFORE_COMMIT_HOOK_SERIAL: std::sync::OnceLock<Mutex<()>> =
    std::sync::OnceLock::new();

#[cfg(test)]
pub(crate) struct ReleaseAfterAcceptHookGuard {
    _serial: std::sync::MutexGuard<'static, ()>,
}

#[cfg(test)]
impl Drop for ReleaseAfterAcceptHookGuard {
    fn drop(&mut self) {
        set_release_after_accept_hook_for_test(None);
    }
}

#[cfg(test)]
pub(crate) struct ReleaseBeforeCommitHookGuard {
    _serial: std::sync::MutexGuard<'static, ()>,
}

#[cfg(test)]
impl Drop for ReleaseBeforeCommitHookGuard {
    fn drop(&mut self) {
        set_release_before_commit_hook_for_test(None);
    }
}

#[cfg(test)]
pub(crate) fn install_release_after_accept_hook_for_test(
    hook: ReleaseAfterAcceptHook,
) -> ReleaseAfterAcceptHookGuard {
    let serial = RELEASE_AFTER_ACCEPT_HOOK_SERIAL
        .get_or_init(|| Mutex::new(()))
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    set_release_after_accept_hook_for_test(Some(hook));
    ReleaseAfterAcceptHookGuard { _serial: serial }
}

#[cfg(test)]
pub(crate) fn install_release_before_commit_hook_for_test(
    hook: ReleaseBeforeCommitHook,
) -> ReleaseBeforeCommitHookGuard {
    let serial = RELEASE_BEFORE_COMMIT_HOOK_SERIAL
        .get_or_init(|| Mutex::new(()))
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    set_release_before_commit_hook_for_test(Some(hook));
    ReleaseBeforeCommitHookGuard { _serial: serial }
}

#[cfg(test)]
fn set_release_after_accept_hook_for_test(hook: Option<ReleaseAfterAcceptHook>) {
    *RELEASE_AFTER_ACCEPT_HOOK
        .get_or_init(|| Mutex::new(None))
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner) = hook;
}

#[cfg(test)]
fn set_release_before_commit_hook_for_test(hook: Option<ReleaseBeforeCommitHook>) {
    *RELEASE_BEFORE_COMMIT_HOOK
        .get_or_init(|| Mutex::new(None))
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner) = hook;
}

#[cfg(test)]
fn run_release_after_accept_hook(lease_key: &LeaseKey) {
    let hook = RELEASE_AFTER_ACCEPT_HOOK
        .get_or_init(|| Mutex::new(None))
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone();
    if let Some(hook) = hook {
        hook(lease_key);
    }
}

#[cfg(test)]
fn run_release_before_commit_hook(lease_key: &LeaseKey) {
    let hook = RELEASE_BEFORE_COMMIT_HOOK
        .get_or_init(|| Mutex::new(None))
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone();
    if let Some(hook) = hook {
        hook(lease_key);
    }
}

#[derive(Default)]
struct AuthLeaseRegistry {
    authorities: HashMap<LeaseKey, auth_dsl::AuthMachineAuthority>,
}

impl std::fmt::Debug for RuntimeAuthLeaseHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        f.debug_struct("RuntimeAuthLeaseHandle")
            .field("leases", &guard.authorities.keys().collect::<Vec<_>>())
            .finish()
    }
}

impl RuntimeAuthLeaseHandle {
    pub fn new() -> Self {
        Self {
            machines: Arc::new(Mutex::new(AuthLeaseRegistry::default())),
            #[cfg(not(target_arch = "wasm32"))]
            release_observers: Arc::new(Mutex::new(Vec::new())),
        }
    }

    /// Alias for [`Self::new`]; kept for parity with other runtime
    /// handles that distinguish session-owned vs. ephemeral
    /// constructors. AuthMachine instances are always ephemeral from
    /// the registry's perspective — the registry itself is owned by
    /// the session bindings.
    pub fn ephemeral() -> Self {
        Self::new()
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn add_release_observer(&self, observer: Weak<dyn AuthLeaseReleaseObserver>) {
        self.release_observers
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .push(observer);
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn live_release_observers(&self) -> Vec<Arc<dyn AuthLeaseReleaseObserver>> {
        {
            let mut guard = self
                .release_observers
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            let mut observers = Vec::new();
            guard.retain(|observer| match observer.upgrade() {
                Some(observer) => {
                    observers.push(observer);
                    true
                }
                None => false,
            });
            observers
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn collect_release_observer_flows(
        &self,
        observers: &[Arc<dyn AuthLeaseReleaseObserver>],
        lease_key: &LeaseKey,
    ) -> Result<ReleasedOAuthFlows, DslTransitionError> {
        let mut released = ReleasedOAuthFlows::empty(lease_key.clone());
        for observer in observers {
            let mut observed = observer.oauth_flows_for_release(lease_key)?;
            released
                .browser_flow_ids
                .append(&mut observed.browser_flow_ids);
            released
                .device_flow_ids
                .append(&mut observed.device_flow_ids);
        }
        released.dedup();
        Ok(released)
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn notify_release_observers(
        &self,
        observers: &[Arc<dyn AuthLeaseReleaseObserver>],
        released: &ReleasedOAuthFlows,
    ) -> Result<(), DslTransitionError> {
        for observer in observers {
            observer.auth_lease_released(released)?;
        }
        Ok(())
    }

    fn apply(
        &self,
        lease_key: &LeaseKey,
        input: auth_dsl::AuthMachineInput,
        context: &'static str,
        create_if_missing: bool,
    ) -> Result<AuthLeaseTransition, DslTransitionError> {
        let action = Self::audit_action_for(&input);
        let mut guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if !create_if_missing && !guard.authorities.contains_key(lease_key) {
            return Err(DslTransitionError::no_matching(
                context,
                format!("no auth lease registered for lease_key `{lease_key}`"),
            ));
        }
        let (from_phase, to_phase, auth_transition) = {
            let entry = if create_if_missing {
                guard
                    .authorities
                    .entry(lease_key.clone())
                    .or_insert_with(auth_dsl::AuthMachineAuthority::new)
            } else {
                match guard.authorities.get_mut(lease_key) {
                    Some(m) => m,
                    None => {
                        return Err(DslTransitionError::no_matching(
                            context,
                            format!("no auth lease registered for lease_key `{lease_key}`"),
                        ));
                    }
                }
            };
            let from_phase = map_phase(entry.state().lifecycle_phase);
            let transition = auth_dsl::AuthMachineMutator::apply(entry, input)
                .map_err(|err| map_auth_machine_error(err, context))?;
            let auth_transition = auth_lease_transition_from_generated_publication(
                lease_key,
                entry,
                &transition,
                context,
            )?;
            let to_phase = map_phase(entry.state().lifecycle_phase);
            (from_phase, to_phase, auth_transition)
        };
        emit_audit(lease_key, action, from_phase, to_phase);
        Ok(auth_transition)
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn oauth_global_outstanding_flow_count(registry: &AuthLeaseRegistry) -> u64 {
        registry
            .authorities
            .values()
            .map(|authority| authority.state().oauth_outstanding_flow_count)
            .fold(0u64, u64::saturating_add)
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn attach_oauth_global_observation(
        input: auth_dsl::AuthMachineInput,
        observed_global_outstanding_flows: u64,
    ) -> auth_dsl::AuthMachineInput {
        match input {
            auth_dsl::AuthMachineInput::AdmitOAuthBrowserFlow {
                flow_id,
                provider,
                redirect_uri,
                expires_at_millis,
                max_outstanding_flows,
                ..
            } => auth_dsl::AuthMachineInput::AdmitOAuthBrowserFlow {
                flow_id,
                provider,
                redirect_uri,
                expires_at_millis,
                max_outstanding_flows,
                observed_global_outstanding_flows,
            },
            auth_dsl::AuthMachineInput::AdmitOAuthDeviceFlow {
                flow_id,
                provider,
                expires_at_millis,
                max_outstanding_flows,
                ..
            } => auth_dsl::AuthMachineInput::AdmitOAuthDeviceFlow {
                flow_id,
                provider,
                expires_at_millis,
                max_outstanding_flows,
                observed_global_outstanding_flows,
            },
            other => other,
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn apply_oauth_input(
        &self,
        target: &AuthBindingRef,
        input: auth_dsl::AuthMachineInput,
        context: &'static str,
        create_if_missing: bool,
    ) -> Result<(), DslTransitionError> {
        let lease_key = LeaseKey::from_auth_binding(target);
        let mut guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let input = Self::attach_oauth_global_observation(
            input,
            Self::oauth_global_outstanding_flow_count(&guard),
        );
        let action = Self::audit_action_for(&input);
        if create_if_missing && !guard.authorities.contains_key(&lease_key) {
            let mut authority = auth_dsl::AuthMachineAuthority::new();
            let transition = auth_dsl::AuthMachineMutator::apply(
                &mut authority,
                auth_dsl::AuthMachineInput::MarkReauthRequired,
            )
            .map_err(|err| map_auth_machine_error(err, context))?;
            auth_lease_transition_from_generated_publication(
                &lease_key,
                &authority,
                &transition,
                context,
            )?;
            guard.authorities.insert(lease_key.clone(), authority);
        }
        let (from_phase, to_phase) = {
            let entry = match guard.authorities.get_mut(&lease_key) {
                Some(m) => m,
                None => {
                    return Err(DslTransitionError::no_matching(
                        context,
                        format!("no auth machine registered for lease_key `{lease_key}`"),
                    ));
                }
            };
            let from_phase = map_phase(entry.state().lifecycle_phase);
            let transition = auth_dsl::AuthMachineMutator::apply(entry, input)
                .map_err(|err| map_auth_machine_error(err, context))?;
            maybe_auth_lease_transition_from_generated_publication(
                &lease_key,
                entry,
                &transition,
                context,
            )?;
            let to_phase = map_phase(entry.state().lifecycle_phase);
            (from_phase, to_phase)
        };
        emit_audit(&lease_key, action, from_phase, to_phase);
        Ok(())
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn confirm_oauth_durable_admission(
        &self,
        target: &AuthBindingRef,
        observed_global_outstanding_flows: u64,
        max_outstanding_flows: u64,
        context: &'static str,
    ) -> Result<(), DslTransitionError> {
        let lease_key = LeaseKey::from_auth_binding(target);
        let input = auth_dsl::AuthMachineInput::ConfirmOAuthDurableAdmission {
            observed_global_outstanding_flows,
            max_outstanding_flows,
        };
        let mut guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let (from_phase, to_phase) = {
            let entry = match guard.authorities.get_mut(&lease_key) {
                Some(m) => m,
                None => {
                    return Err(DslTransitionError::no_matching(
                        context,
                        format!("no auth machine registered for lease_key `{lease_key}`"),
                    ));
                }
            };
            let from_phase = map_phase(entry.state().lifecycle_phase);
            let transition = auth_dsl::AuthMachineMutator::apply(entry, input)
                .map_err(|err| map_auth_machine_error(err, context))?;
            maybe_auth_lease_transition_from_generated_publication(
                &lease_key,
                entry,
                &transition,
                context,
            )?;
            let to_phase = map_phase(entry.state().lifecycle_phase);
            (from_phase, to_phase)
        };
        emit_audit(
            &lease_key,
            "confirm_oauth_durable_admission",
            from_phase,
            to_phase,
        );
        Ok(())
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn has_oauth_browser_flow(&self, target: &AuthBindingRef, flow_id: &str) -> bool {
        let lease_key = LeaseKey::from_auth_binding(target);
        self.machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .authorities
            .get(&lease_key)
            .is_some_and(|authority| authority.state().oauth_browser_flow_ids.contains(flow_id))
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn has_oauth_device_flow(&self, target: &AuthBindingRef, flow_id: &str) -> bool {
        let lease_key = LeaseKey::from_auth_binding(target);
        self.machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .authorities
            .get(&lease_key)
            .is_some_and(|authority| authority.state().oauth_device_flow_ids.contains(flow_id))
    }

    #[cfg(test)]
    pub(crate) fn has_oauth_browser_flow_for_test(
        &self,
        target: &AuthBindingRef,
        flow_id: &str,
    ) -> bool {
        self.has_oauth_browser_flow(target, flow_id)
    }

    #[cfg(test)]
    pub(crate) fn has_oauth_device_flow_for_test(
        &self,
        target: &AuthBindingRef,
        flow_id: &str,
    ) -> bool {
        self.has_oauth_device_flow(target, flow_id)
    }

    fn audit_action_for(input: &auth_dsl::AuthMachineInput) -> &'static str {
        match input {
            auth_dsl::AuthMachineInput::Acquire { .. } => "acquire_lease",
            auth_dsl::AuthMachineInput::MarkExpiring => "mark_expiring",
            auth_dsl::AuthMachineInput::ObserveCredentialFreshness { .. } => {
                "observe_credential_freshness"
            }
            auth_dsl::AuthMachineInput::BeginRefresh => "begin_refresh",
            auth_dsl::AuthMachineInput::CompleteRefresh { .. } => "complete_refresh",
            auth_dsl::AuthMachineInput::RefreshFailed { .. } => "refresh_failed",
            auth_dsl::AuthMachineInput::MarkReauthRequired => "mark_reauth_required",
            auth_dsl::AuthMachineInput::ClearCredentialLifecycle => "clear_credential_lifecycle",
            auth_dsl::AuthMachineInput::ReleaseCredentialLifecycle => {
                "release_credential_lifecycle"
            }
            auth_dsl::AuthMachineInput::BeginRelease => "begin_release_lease",
            auth_dsl::AuthMachineInput::Release => "release_lease",
            auth_dsl::AuthMachineInput::RestoreAuthoritySnapshot { .. } => {
                "restore_authority_snapshot"
            }
            auth_dsl::AuthMachineInput::RestoreCredentialLifecycleSnapshot { .. } => {
                "restore_credential_lifecycle_snapshot"
            }
            auth_dsl::AuthMachineInput::RestoreOAuthBrowserFlow { .. } => {
                "restore_oauth_browser_flow"
            }
            auth_dsl::AuthMachineInput::RestoreOAuthDeviceFlow { .. } => {
                "restore_oauth_device_flow"
            }
            auth_dsl::AuthMachineInput::RestoreOAuthDevicePoll { .. } => {
                "restore_oauth_device_poll"
            }
            auth_dsl::AuthMachineInput::AdmitOAuthBrowserFlow { .. } => "admit_oauth_browser_flow",
            auth_dsl::AuthMachineInput::VerifyOAuthBrowserFlow { .. } => {
                "verify_oauth_browser_flow"
            }
            auth_dsl::AuthMachineInput::ConsumeOAuthBrowserFlow { .. } => {
                "consume_oauth_browser_flow"
            }
            auth_dsl::AuthMachineInput::ExpireOAuthBrowserFlow { .. } => {
                "expire_oauth_browser_flow"
            }
            auth_dsl::AuthMachineInput::AdmitOAuthDeviceFlow { .. } => "admit_oauth_device_flow",
            auth_dsl::AuthMachineInput::ConfirmOAuthDurableAdmission { .. } => {
                "confirm_oauth_durable_admission"
            }
            auth_dsl::AuthMachineInput::VerifyOAuthDeviceFlow { .. } => "verify_oauth_device_flow",
            auth_dsl::AuthMachineInput::BeginOAuthDevicePoll { .. } => "begin_oauth_device_poll",
            auth_dsl::AuthMachineInput::FinishOAuthDevicePoll { .. } => "finish_oauth_device_poll",
            auth_dsl::AuthMachineInput::ConsumeOAuthDeviceFlow { .. } => {
                "consume_oauth_device_flow"
            }
            auth_dsl::AuthMachineInput::ExpireOAuthDeviceFlow { .. } => "expire_oauth_device_flow",
            auth_dsl::AuthMachineInput::ResolveCredentialUseAdmission { .. } => {
                "resolve_credential_use_admission"
            }
            auth_dsl::AuthMachineInput::ResolveOAuthLoginCredentialDisposition { .. } => {
                "resolve_oauth_login_credential_disposition"
            }
        }
    }
}

/// Exhaustive 1-to-1 projection of the AuthMachine DSL's typed lifecycle
/// phase into the cross-crate [`AuthLeasePhase`] contract. Compiler enforces
/// completeness.
fn map_phase(phase: auth_dsl::AuthLifecyclePhase) -> AuthLeasePhase {
    match phase {
        auth_dsl::AuthLifecyclePhase::Valid => AuthLeasePhase::Valid,
        auth_dsl::AuthLifecyclePhase::Expiring => AuthLeasePhase::Expiring,
        auth_dsl::AuthLifecyclePhase::Expired => AuthLeasePhase::Expired,
        auth_dsl::AuthLifecyclePhase::Refreshing => AuthLeasePhase::Refreshing,
        auth_dsl::AuthLifecyclePhase::ReauthRequired => AuthLeasePhase::ReauthRequired,
        auth_dsl::AuthLifecyclePhase::Released => AuthLeasePhase::Released,
    }
}

fn credential_use_intent_to_dsl(
    intent: meerkat_core::handles::CredentialUseIntent,
) -> auth_dsl::CredentialUseIntent {
    match intent {
        meerkat_core::handles::CredentialUseIntent::UseCredential => {
            auth_dsl::CredentialUseIntent::UseCredential
        }
        meerkat_core::handles::CredentialUseIntent::HoldAuthority => {
            auth_dsl::CredentialUseIntent::HoldAuthority
        }
        meerkat_core::handles::CredentialUseIntent::BeginRefresh => {
            auth_dsl::CredentialUseIntent::BeginRefresh
        }
    }
}

fn credential_use_disposition_from_dsl(
    disposition: auth_dsl::CredentialUseDisposition,
) -> meerkat_core::handles::CredentialUseDisposition {
    match disposition {
        auth_dsl::CredentialUseDisposition::Authorized => {
            meerkat_core::handles::CredentialUseDisposition::Authorized
        }
        auth_dsl::CredentialUseDisposition::RefreshRequired => {
            meerkat_core::handles::CredentialUseDisposition::RefreshRequired
        }
        auth_dsl::CredentialUseDisposition::RefreshDisallowed => {
            meerkat_core::handles::CredentialUseDisposition::RefreshDisallowed
        }
        auth_dsl::CredentialUseDisposition::ReauthRequired => {
            meerkat_core::handles::CredentialUseDisposition::ReauthRequired
        }
        auth_dsl::CredentialUseDisposition::LeaseAbsent => {
            meerkat_core::handles::CredentialUseDisposition::LeaseAbsent
        }
        auth_dsl::CredentialUseDisposition::AlreadyRefreshing => {
            meerkat_core::handles::CredentialUseDisposition::AlreadyRefreshing
        }
    }
}

fn restore_phase(phase: AuthLeasePhase) -> auth_dsl::AuthLifecyclePhase {
    match phase {
        AuthLeasePhase::Valid => auth_dsl::AuthLifecyclePhase::Valid,
        AuthLeasePhase::Expiring => auth_dsl::AuthLifecyclePhase::Expiring,
        AuthLeasePhase::Expired => auth_dsl::AuthLifecyclePhase::Expired,
        AuthLeasePhase::Refreshing => auth_dsl::AuthLifecyclePhase::Refreshing,
        AuthLeasePhase::ReauthRequired => auth_dsl::AuthLifecyclePhase::ReauthRequired,
        AuthLeasePhase::Released => auth_dsl::AuthLifecyclePhase::Released,
    }
}

impl Default for RuntimeAuthLeaseHandle {
    fn default() -> Self {
        Self::new()
    }
}

impl AuthLeaseHandle for RuntimeAuthLeaseHandle {
    fn acquire_lease(
        &self,
        lease_key: &LeaseKey,
        expires_at: u64,
    ) -> Result<AuthLeaseTransition, DslTransitionError> {
        let expires_at_ts = if expires_at == u64::MAX {
            None
        } else {
            Some(expires_at)
        };
        self.apply(
            lease_key,
            auth_dsl::AuthMachineInput::Acquire {
                expires_at_ts,
                credential_published_at_millis: current_time_millis(),
            },
            "AuthLeaseHandle::acquire_lease",
            true,
        )
    }

    fn mark_expiring(&self, lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
        self.apply(
            lease_key,
            auth_dsl::AuthMachineInput::MarkExpiring,
            "AuthLeaseHandle::mark_expiring",
            false,
        )
        .map(|_| ())
    }

    fn observe_credential_freshness(
        &self,
        lease_key: &LeaseKey,
        now: u64,
        refresh_window_secs: u64,
    ) -> Result<(), DslTransitionError> {
        #[allow(clippy::unwrap_used)]
        if !self
            .machines
            .lock()
            .unwrap()
            .authorities
            .contains_key(lease_key)
        {
            return Ok(());
        }
        self.apply(
            lease_key,
            auth_dsl::AuthMachineInput::ObserveCredentialFreshness {
                now_ts: now,
                refresh_window_secs,
            },
            "AuthLeaseHandle::observe_credential_freshness",
            false,
        )
        .map(|_| ())
    }

    fn begin_refresh(&self, lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
        self.apply(
            lease_key,
            auth_dsl::AuthMachineInput::BeginRefresh,
            "AuthLeaseHandle::begin_refresh",
            false,
        )
        .map(|_| ())
    }

    fn complete_refresh(
        &self,
        lease_key: &LeaseKey,
        new_expires_at: u64,
        now: u64,
    ) -> Result<AuthLeaseTransition, DslTransitionError> {
        let new_expires_at = if new_expires_at == u64::MAX {
            None
        } else {
            Some(new_expires_at)
        };
        self.apply(
            lease_key,
            auth_dsl::AuthMachineInput::CompleteRefresh {
                new_expires_at,
                now_ts: now,
                credential_published_at_millis: current_time_millis(),
            },
            "AuthLeaseHandle::complete_refresh",
            false,
        )
    }

    fn refresh_failed(
        &self,
        lease_key: &LeaseKey,
        observation: RefreshFailureObservation,
    ) -> Result<(), DslTransitionError> {
        let input = auth_dsl::AuthMachineInput::RefreshFailed {
            http_status: observation.http_status,
            oauth_error_code: observation.oauth_error_code,
            local_credential_unusable: observation.local_credential_unusable,
        };
        self.apply(lease_key, input, "AuthLeaseHandle::refresh_failed", false)
            .map(|_| ())
    }

    fn mark_reauth_required(&self, lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
        self.apply(
            lease_key,
            auth_dsl::AuthMachineInput::MarkReauthRequired,
            "AuthLeaseHandle::mark_reauth_required",
            false,
        )
        .map(|_| ())
    }

    fn release_lease(&self, lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
        let context = "AuthLeaseHandle::release_lease";
        // Two-phase machine-owned release drain (D1).
        //
        // Phase 1 — BeginRelease: fire the machine-owned drain input to record
        // release intent and obtain the typed CancelOAuthFlowsForRelease
        // obligation carrying the in-flight membership. Only fires if a local
        // machine exists; the stale-authority case (no local machine) has an
        // empty obligation and relies entirely on the observer's durable-store
        // scan (phase 2) for cleanup.
        //
        // Phase 2 — observer collect: gather registry-payload-only ids from the
        // durable store. Merged with the machine-emitted obligation so the
        // observer's auth_lease_released prune covers both the machine-owned
        // flows and any flows admitted by a stale authority that the local
        // machine never saw.
        //
        // Fail-closed ordering: observer notify (phase 3) and drain discharge
        // (phase 4) are both fallible and both precede the Release commit
        // (phase 5). An error in either aborts the release — the machine stays
        // in release_draining (or un-created in the stale case) and the caller
        // retries. Retried release re-fires BeginRelease which re-emits the
        // remaining membership (idempotent). No machine entry is synthesized in
        // the registry for the stale-authority path until Release commits.
        #[cfg(not(target_arch = "wasm32"))]
        let release_observers = self.live_release_observers();
        #[cfg(not(target_arch = "wasm32"))]
        let release_permits = release_observers
            .iter()
            .map(|observer| observer.begin_auth_lease_release(lease_key))
            .collect::<Result<Vec<_>, _>>()?;
        #[cfg(not(target_arch = "wasm32"))]
        {
            // Phase 1: BeginRelease under the machines lock.
            // Extract the machine-owned drain obligation (empty for stale authority).
            let machine_drain_obligation = {
                let mut guard = self
                    .machines
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                if let Some(entry) = guard.authorities.get_mut(lease_key) {
                    let transition = auth_dsl::AuthMachineMutator::apply(
                        entry,
                        auth_dsl::AuthMachineInput::BeginRelease,
                    )
                    .map_err(|err| map_auth_machine_error(err, context))?;
                    // BeginRelease emits no EmitLifecycleEvent; use the
                    // maybe_-variant (strict variant errors on missing
                    // publication which BeginRelease legitimately omits).
                    maybe_auth_lease_transition_from_generated_publication(
                        lease_key,
                        entry,
                        &transition,
                        context,
                    )?;
                    crate::protocol_auth_release_oauth_flow_drain::extract_obligations(&transition)
                        .into_iter()
                        .next()
                } else {
                    None
                }
            };

            // Phase 2: Observer-only durable-store scan (stale-authority
            // flows the local machine never admitted). Merged with the
            // machine-emitted obligation so observers see the full prune set.
            let mut released =
                self.collect_release_observer_flows(&release_observers, lease_key)?;
            if let Some(ref obligation) = machine_drain_obligation {
                released
                    .browser_flow_ids
                    .extend(obligation.browser_flow_ids.iter().cloned());
                released
                    .device_flow_ids
                    .extend(obligation.device_flow_ids.iter().cloned());
            }
            released.dedup();

            // Phase 3: Notify observers — prune payloads from registry and
            // durable store. Fallible: abort on error, machine stays draining.
            self.notify_release_observers(&release_observers, &released)?;

            // Phase 4: Discharge machine-owned drain obligations.
            // Fire typed terminal Expire* feedback per drained flow id.
            // Each is total (Released no-op + Absent no-op exist); Err is
            // a genuine fault (machine absent/poisoned), propagated.
            if let Some(obligation) = machine_drain_obligation {
                let mut guard = self
                    .machines
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                if let Some(entry) = guard.authorities.get_mut(lease_key) {
                    for flow_id in obligation.browser_flow_ids {
                        let transition = auth_dsl::AuthMachineMutator::apply(
                            entry,
                            auth_dsl::AuthMachineInput::ExpireOAuthBrowserFlow { flow_id },
                        )
                        .map_err(|err| map_auth_machine_error(err, context))?;
                        maybe_auth_lease_transition_from_generated_publication(
                            lease_key,
                            entry,
                            &transition,
                            context,
                        )?;
                    }
                    for flow_id in obligation.device_flow_ids {
                        let transition = auth_dsl::AuthMachineMutator::apply(
                            entry,
                            auth_dsl::AuthMachineInput::ExpireOAuthDeviceFlow { flow_id },
                        )
                        .map_err(|err| map_auth_machine_error(err, context))?;
                        maybe_auth_lease_transition_from_generated_publication(
                            lease_key,
                            entry,
                            &transition,
                            context,
                        )?;
                    }
                }
            }
        }
        #[cfg(test)]
        run_release_before_commit_hook(lease_key);
        // Phase 5: Commit — Release transition through generated machine
        // authority. The oauth_release_drained guard passes (count == 0 after
        // the drain discharge above). For the stale-authority path (no local
        // machine), or_insert_with creates a fresh machine whose initial
        // count is already 0. Nothing fallible follows the Release commit.
        let (from_phase, to_phase) = {
            let mut guard = self
                .machines
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            let entry = guard
                .authorities
                .entry(lease_key.clone())
                .or_insert_with(auth_dsl::AuthMachineAuthority::new);
            let from_phase = map_phase(entry.state().lifecycle_phase);
            let transition =
                auth_dsl::AuthMachineMutator::apply(entry, auth_dsl::AuthMachineInput::Release)
                    .map_err(|err| map_auth_machine_error(err, context))?;
            auth_lease_transition_from_generated_publication(
                lease_key,
                entry,
                &transition,
                context,
            )?;
            let to_phase = map_phase(entry.state().lifecycle_phase);
            (from_phase, to_phase)
        };
        #[cfg(not(target_arch = "wasm32"))]
        drop(release_permits);
        #[cfg(not(target_arch = "wasm32"))]
        drop(release_observers);
        emit_audit(lease_key, "release_lease", from_phase, to_phase);
        #[cfg(test)]
        run_release_after_accept_hook(lease_key);
        Ok(())
    }

    fn release_credential_lifecycle(&self, lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
        let context = "AuthLeaseHandle::release_credential_lifecycle";
        let mut guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let (input, from_phase, to_phase) = {
            let entry = guard
                .authorities
                .entry(lease_key.clone())
                .or_insert_with(auth_dsl::AuthMachineAuthority::new);
            let input = auth_dsl::AuthMachineInput::ReleaseCredentialLifecycle;
            let from_phase = map_phase(entry.state().lifecycle_phase);
            let transition = auth_dsl::AuthMachineMutator::apply(entry, input.clone())
                .map_err(|err| map_auth_machine_error(err, context))?;
            auth_lease_transition_from_generated_publication(
                lease_key,
                entry,
                &transition,
                context,
            )?;
            let to_phase = map_phase(entry.state().lifecycle_phase);
            (input, from_phase, to_phase)
        };
        emit_audit(
            lease_key,
            Self::audit_action_for(&input),
            from_phase,
            to_phase,
        );
        Ok(())
    }

    fn restore_auth_lifecycle_snapshot(
        &self,
        captured: &AuthLeaseRestoreSnapshot,
    ) -> Result<Option<AuthLeaseTransition>, DslTransitionError> {
        if captured.captured_by_type_id() != std::any::TypeId::of::<RuntimeAuthLeaseHandle>()
            || captured.captured_by_instance_id() != self.auth_lifecycle_restore_instance_id()
        {
            return Err(DslTransitionError::no_matching(
                "AuthLeaseHandle::restore_auth_lifecycle_snapshot",
                "auth lifecycle restore snapshot was not captured from this RuntimeAuthLeaseHandle",
            ));
        }
        let lease_key = captured.lease_key();
        let snapshot = captured.snapshot();
        let mut guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let from_phase = guard
            .authorities
            .get(lease_key)
            .map(|authority| map_phase(authority.state().lifecycle_phase))
            .unwrap_or(AuthLeasePhase::Released);
        let (to_phase, auth_transition) = apply_restore_input_to_registry(
            &mut guard,
            lease_key,
            restore_credential_lifecycle_snapshot_input(
                snapshot.phase.map(restore_phase),
                snapshot.expires_at,
                None,
                0,
                snapshot.credential_present,
                snapshot.generation,
                snapshot.credential_published_at_millis,
                false,
            ),
            "AuthLeaseHandle::restore_auth_lifecycle_snapshot",
        )?;
        emit_audit(
            lease_key,
            "restore_auth_lifecycle_snapshot",
            from_phase,
            to_phase,
        );
        if snapshot.credential_present
            && snapshot.phase.is_some()
            && snapshot.phase != Some(AuthLeasePhase::Released)
        {
            Ok(Some(auth_transition))
        } else {
            Ok(None)
        }
    }

    fn auth_lifecycle_restore_instance_id(&self) -> usize {
        Arc::as_ptr(&self.machines) as usize
    }

    fn restore_published_credential_lifecycle(
        &self,
        lease_key: &LeaseKey,
        publication: &AuthLeaseDurableRestorePublication,
    ) -> Result<AuthLeaseTransition, DslTransitionError> {
        let context = "AuthLeaseHandle::restore_published_credential_lifecycle";
        let lease_token_key = TokenKey::new_with_profile(
            lease_key.realm.clone(),
            lease_key.binding.clone(),
            lease_key.profile.clone(),
        );
        if publication.token_key() != &lease_token_key {
            return Err(DslTransitionError::no_matching(
                context,
                "durable auth lifecycle marker identity does not match restore lease key",
            ));
        }
        let mut guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let from_phase = guard
            .authorities
            .get(lease_key)
            .map(|authority| map_phase(authority.state().lifecycle_phase))
            .unwrap_or(AuthLeasePhase::Released);
        let (to_phase, auth_transition) = apply_restore_input_to_registry(
            &mut guard,
            lease_key,
            restore_input_from_lifecycle(
                restore_phase_to_dsl(publication.phase()),
                (publication.expires_at() != u64::MAX).then_some(publication.expires_at()),
                None,
                0,
                publication.phase() != AuthLeasePhase::Released,
                publication.generation(),
                Some(publication.credential_published_at_millis()),
            ),
            context,
        )?;
        emit_audit(
            lease_key,
            "restore_published_credential_lifecycle",
            from_phase,
            to_phase,
        );
        Ok(auth_transition)
    }

    fn resolve_credential_use_admission(
        &self,
        lease_key: &LeaseKey,
        intent: meerkat_core::handles::CredentialUseIntent,
    ) -> Result<meerkat_core::handles::CredentialUseDisposition, DslTransitionError> {
        const CONTEXT: &str = "AuthLeaseHandle::resolve_credential_use_admission";
        let guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // No registered lease for this binding mirrors the resolver's prior
        // credential-filtered `None` phase: the lease is absent.
        let Some(authority) = guard.authorities.get(lease_key) else {
            return Ok(meerkat_core::handles::CredentialUseDisposition::LeaseAbsent);
        };
        // Read-only classification: recover a transient authority over the live
        // state so the registry is never mutated and no lifecycle audit fires.
        let mut transient =
            auth_dsl::AuthMachineAuthority::recover_from_state(authority.state().clone())
                .map_err(|err| map_auth_machine_error(err, CONTEXT))?;
        let transition = auth_dsl::AuthMachineMutator::apply(
            &mut transient,
            auth_dsl::AuthMachineInput::ResolveCredentialUseAdmission {
                intent: credential_use_intent_to_dsl(intent),
            },
        )
        .map_err(|err| map_auth_machine_error(err, CONTEXT))?;

        let mut resolved = None;
        for effect in transition.effects() {
            if let auth_dsl::AuthMachineEffect::CredentialUseAdmissionResolved { disposition } =
                effect
                && resolved.replace(*disposition).is_some()
            {
                return Err(DslTransitionError::no_matching(
                    CONTEXT,
                    format!(
                        "AuthMachine emitted multiple credential-use dispositions for `{lease_key}`"
                    ),
                ));
            }
        }

        resolved
            .map(credential_use_disposition_from_dsl)
            .ok_or_else(|| {
                DslTransitionError::no_matching(
                    CONTEXT,
                    format!("AuthMachine emitted no credential-use disposition for `{lease_key}`"),
                )
            })
    }

    fn resolve_oauth_login_credential_disposition(
        &self,
        lease_key: &LeaseKey,
        facts: meerkat_core::handles::OAuthLoginCredentialFacts,
    ) -> Result<meerkat_core::handles::CredentialUseDisposition, DslTransitionError> {
        const CONTEXT: &str = "AuthLeaseHandle::resolve_oauth_login_credential_disposition";
        let guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // No registered lease for this binding mirrors the resolver's prior
        // credential-filtered `None` phase: the lease is absent.
        let Some(authority) = guard.authorities.get(lease_key) else {
            return Ok(meerkat_core::handles::CredentialUseDisposition::LeaseAbsent);
        };
        // Read-only classification: recover a transient authority over the live
        // state so the registry is never mutated and no lifecycle audit fires.
        let mut transient =
            auth_dsl::AuthMachineAuthority::recover_from_state(authority.state().clone())
                .map_err(|err| map_auth_machine_error(err, CONTEXT))?;
        let transition = auth_dsl::AuthMachineMutator::apply(
            &mut transient,
            auth_dsl::AuthMachineInput::ResolveOAuthLoginCredentialDisposition {
                credential_present: facts.credential_present,
                force_refresh: facts.force_refresh,
                refresh_allowed: facts.refresh_allowed,
            },
        )
        .map_err(|err| map_auth_machine_error(err, CONTEXT))?;

        let mut resolved = None;
        for effect in transition.effects() {
            if let auth_dsl::AuthMachineEffect::CredentialUseAdmissionResolved { disposition } =
                effect
                && resolved.replace(*disposition).is_some()
            {
                return Err(DslTransitionError::no_matching(
                    CONTEXT,
                    format!(
                        "AuthMachine emitted multiple OAuth-login credential dispositions for `{lease_key}`"
                    ),
                ));
            }
        }

        resolved
            .map(credential_use_disposition_from_dsl)
            .ok_or_else(|| {
                DslTransitionError::no_matching(
                    CONTEXT,
                    format!(
                        "AuthMachine emitted no OAuth-login credential disposition for `{lease_key}`"
                    ),
                )
            })
    }

    fn snapshot(&self, lease_key: &LeaseKey) -> AuthLeaseSnapshot {
        let guard = self
            .machines
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        match guard.authorities.get(lease_key) {
            Some(machine) => {
                let state = machine.state();
                let phase = (state.lifecycle_phase != auth_dsl::AuthLifecyclePhase::Released)
                    .then(|| map_phase(state.lifecycle_phase));
                AuthLeaseSnapshot {
                    phase,
                    expires_at: state.expires_at,
                    credential_present: state.credential_present,
                    generation: state.credential_generation,
                    credential_published_at_millis: state
                        .credential_present
                        .then_some(state.credential_published_at_millis)
                        .flatten(),
                }
            }
            None => AuthLeaseSnapshot {
                phase: None,
                expires_at: None,
                credential_present: false,
                generation: 0,
                credential_published_at_millis: None,
            },
        }
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;
    use meerkat_core::connection::{BindingId, BindingOrigin, RealmId};

    fn lease(realm: &str, binding: &str) -> LeaseKey {
        LeaseKey::new(
            RealmId::parse(realm).expect("valid realm"),
            BindingId::parse(binding).expect("valid binding"),
            None,
        )
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn auth_binding(realm: &str, binding: &str) -> AuthBindingRef {
        AuthBindingRef {
            realm: RealmId::parse(realm).expect("valid realm"),
            binding: BindingId::parse(binding).expect("valid binding"),
            profile: None,
            origin: BindingOrigin::Configured,
        }
    }

    fn empty_auth_state() -> auth_dsl::AuthMachineState {
        auth_dsl::AuthMachineAuthority::new().state().clone()
    }

    #[test]
    fn acquire_and_snapshot_roundtrip() {
        let h = RuntimeAuthLeaseHandle::new();
        let key = lease("dev", "default_openai");
        h.acquire_lease(&key, 1_800_000_000).unwrap();
        let snap = h.snapshot(&key);
        assert_eq!(snap.phase, Some(AuthLeasePhase::Valid));
        assert_eq!(snap.expires_at, Some(1_800_000_000));
    }

    #[test]
    fn lifecycle_transitions() {
        let h = RuntimeAuthLeaseHandle::new();
        let k = lease("dev", "default_anthropic");

        h.acquire_lease(&k, 1_800_000_000).unwrap();
        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Valid));

        h.mark_expiring(&k).unwrap();
        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Expiring));

        h.begin_refresh(&k).unwrap();
        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Refreshing));

        h.complete_refresh(&k, 1_800_000_900, 1_800_000_000)
            .unwrap();
        let snap = h.snapshot(&k);
        assert_eq!(snap.phase, Some(AuthLeasePhase::Valid));
        assert_eq!(snap.expires_at, Some(1_800_000_900));
    }

    #[test]
    fn credential_use_admission_is_decided_by_authmachine() {
        use meerkat_core::handles::CredentialUseDisposition as Disp;
        use meerkat_core::handles::CredentialUseIntent as Intent;

        let h = RuntimeAuthLeaseHandle::new();
        let k = lease("dev", "creduse_openai");

        // No registered lease -> LeaseAbsent for every intent.
        for intent in [
            Intent::UseCredential,
            Intent::HoldAuthority,
            Intent::BeginRefresh,
        ] {
            assert_eq!(
                h.resolve_credential_use_admission(&k, intent).unwrap(),
                Disp::LeaseAbsent,
                "unregistered lease must classify LeaseAbsent for {intent:?}"
            );
        }

        // Valid + credential present.
        h.acquire_lease(&k, 1_800_000_000).unwrap();
        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Valid));
        assert_eq!(
            h.resolve_credential_use_admission(&k, Intent::UseCredential)
                .unwrap(),
            Disp::Authorized
        );
        assert_eq!(
            h.resolve_credential_use_admission(&k, Intent::HoldAuthority)
                .unwrap(),
            Disp::Authorized
        );
        assert_eq!(
            h.resolve_credential_use_admission(&k, Intent::BeginRefresh)
                .unwrap(),
            Disp::RefreshRequired
        );

        // Expiring: UseCredential -> RefreshRequired, HoldAuthority -> Authorized,
        // BeginRefresh -> RefreshRequired.
        h.mark_expiring(&k).unwrap();
        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Expiring));
        assert_eq!(
            h.resolve_credential_use_admission(&k, Intent::UseCredential)
                .unwrap(),
            Disp::RefreshRequired
        );
        assert_eq!(
            h.resolve_credential_use_admission(&k, Intent::HoldAuthority)
                .unwrap(),
            Disp::Authorized
        );
        assert_eq!(
            h.resolve_credential_use_admission(&k, Intent::BeginRefresh)
                .unwrap(),
            Disp::RefreshRequired
        );

        // Refreshing: UseCredential -> RefreshRequired, HoldAuthority -> Authorized,
        // BeginRefresh -> AlreadyRefreshing.
        h.begin_refresh(&k).unwrap();
        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Refreshing));
        assert_eq!(
            h.resolve_credential_use_admission(&k, Intent::UseCredential)
                .unwrap(),
            Disp::RefreshRequired
        );
        assert_eq!(
            h.resolve_credential_use_admission(&k, Intent::HoldAuthority)
                .unwrap(),
            Disp::Authorized
        );
        assert_eq!(
            h.resolve_credential_use_admission(&k, Intent::BeginRefresh)
                .unwrap(),
            Disp::AlreadyRefreshing
        );

        // Expired: every intent -> RefreshRequired.
        let kx = lease("dev", "creduse_expired");
        h.acquire_lease(&kx, 1_800_000_000).unwrap();
        h.observe_credential_freshness(&kx, 1_800_000_001, 60)
            .unwrap();
        assert_eq!(h.snapshot(&kx).phase, Some(AuthLeasePhase::Expired));
        for intent in [
            Intent::UseCredential,
            Intent::HoldAuthority,
            Intent::BeginRefresh,
        ] {
            assert_eq!(
                h.resolve_credential_use_admission(&kx, intent).unwrap(),
                Disp::RefreshRequired,
                "expired lease must classify RefreshRequired for {intent:?}"
            );
        }

        // ReauthRequired: every intent -> ReauthRequired.
        let kr = lease("dev", "creduse_reauth");
        h.acquire_lease(&kr, 1_800_000_000).unwrap();
        h.mark_reauth_required(&kr).unwrap();
        assert_eq!(h.snapshot(&kr).phase, Some(AuthLeasePhase::ReauthRequired));
        for intent in [
            Intent::UseCredential,
            Intent::HoldAuthority,
            Intent::BeginRefresh,
        ] {
            assert_eq!(
                h.resolve_credential_use_admission(&kr, intent).unwrap(),
                Disp::ReauthRequired,
                "reauth-required lease must classify ReauthRequired for {intent:?}"
            );
        }

        // Released: every intent -> LeaseAbsent.
        let kl = lease("dev", "creduse_released");
        h.acquire_lease(&kl, 1_800_000_000).unwrap();
        h.release_lease(&kl).unwrap();
        for intent in [
            Intent::UseCredential,
            Intent::HoldAuthority,
            Intent::BeginRefresh,
        ] {
            assert_eq!(
                h.resolve_credential_use_admission(&kl, intent).unwrap(),
                Disp::LeaseAbsent,
                "released lease must classify LeaseAbsent for {intent:?}"
            );
        }

        // Classification is read-only: the live phase is unchanged after probing.
        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Refreshing));
    }

    // P0 Dogma Invariant 1, FOLD 2: the OAuth-login cached-vs-refresh
    // disposition is now owned by the AuthMachine
    // `ResolveOAuthLoginCredentialDisposition` classifier. This test pins EXACT
    // behavioral parity with the prior provider-runtime shell composite
    // (`lifecycle == Authorized && primary_secret.is_some() && !force_refresh`
    // use-cached; else `refresh_allowed ? begin-refresh : Err(RefreshRequired)`)
    // for every combination of phase, persisted-secret presence, force_refresh,
    // and refresh_allowed — including the four required cases (cached-use;
    // force_refresh -> refresh; lifecycle RefreshRequired -> refresh;
    // refresh-not-allowed -> Err).
    #[test]
    fn oauth_login_credential_disposition_matches_legacy_composite() {
        use meerkat_core::handles::CredentialUseDisposition as Disp;
        use meerkat_core::handles::OAuthLoginCredentialFacts;

        // Drive a lease into each lifecycle phase and return its handle/key. The
        // machine's own `credential_present` is true in all of these (acquired
        // leases carry a published credential), exactly as in production where
        // the provider composite is only reached for credential-bearing leases.
        fn lease_in_phase(
            phase: AuthLeasePhase,
            binding: &str,
        ) -> (RuntimeAuthLeaseHandle, LeaseKey) {
            let h = RuntimeAuthLeaseHandle::new();
            let k = lease("dev", binding);
            h.acquire_lease(&k, 1_800_000_000).unwrap();
            match phase {
                AuthLeasePhase::Valid => {}
                AuthLeasePhase::Expiring => h.mark_expiring(&k).unwrap(),
                AuthLeasePhase::Expired => {
                    h.observe_credential_freshness(&k, 1_800_000_001, 60)
                        .unwrap();
                }
                AuthLeasePhase::Refreshing => h.begin_refresh(&k).unwrap(),
                AuthLeasePhase::ReauthRequired => h.mark_reauth_required(&k).unwrap(),
                AuthLeasePhase::Released => h.release_lease(&k).unwrap(),
            }
            (h, k)
        }

        let phases = [
            AuthLeasePhase::Valid,
            AuthLeasePhase::Expiring,
            AuthLeasePhase::Expired,
            AuthLeasePhase::Refreshing,
            AuthLeasePhase::ReauthRequired,
            AuthLeasePhase::Released,
        ];
        let mut counter = 0u32;
        for phase in phases {
            for credential_present in [false, true] {
                for force_refresh in [false, true] {
                    for refresh_allowed in [false, true] {
                        counter += 1;
                        let (h, k) = lease_in_phase(phase, &format!("oauthdisp_{counter}"));
                        let got = h
                            .resolve_oauth_login_credential_disposition(
                                &k,
                                OAuthLoginCredentialFacts {
                                    credential_present,
                                    force_refresh,
                                    refresh_allowed,
                                },
                            )
                            .unwrap();

                        // Reference composite: use-cached iff lifecycle==Authorized
                        // (phase Valid with the machine's published credential) AND
                        // a persisted secret is present AND not force-refreshing;
                        // else refresh_allowed ? begin-refresh : refresh-disallowed.
                        let use_cached =
                            phase == AuthLeasePhase::Valid && credential_present && !force_refresh;
                        let expected = if use_cached {
                            Disp::Authorized
                        } else if refresh_allowed {
                            Disp::RefreshRequired
                        } else {
                            Disp::RefreshDisallowed
                        };
                        assert_eq!(
                            got, expected,
                            "phase={phase:?} cred_present={credential_present} \
                             force_refresh={force_refresh} refresh_allowed={refresh_allowed}"
                        );
                    }
                }
            }
        }

        // Unregistered lease -> LeaseAbsent regardless of observations.
        let h = RuntimeAuthLeaseHandle::new();
        let absent = lease("dev", "oauthdisp_absent");
        assert_eq!(
            h.resolve_oauth_login_credential_disposition(
                &absent,
                OAuthLoginCredentialFacts {
                    credential_present: true,
                    force_refresh: false,
                    refresh_allowed: true,
                },
            )
            .unwrap(),
            Disp::LeaseAbsent
        );
    }

    #[test]
    fn observe_credential_freshness_marks_expired_through_authmachine() {
        let h = RuntimeAuthLeaseHandle::new();
        let k = lease("dev", "expired_openai");

        h.acquire_lease(&k, 1_800_000_000).unwrap();
        h.observe_credential_freshness(&k, 1_800_000_001, 60)
            .unwrap();

        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Expired));
        h.begin_refresh(&k).unwrap();
        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Refreshing));
    }

    #[test]
    fn refresh_failed_permanent_routes_to_reauth() {
        let h = RuntimeAuthLeaseHandle::new();
        let k = lease("dev", "default_google");

        h.acquire_lease(&k, 1_800_000_000).unwrap();
        h.begin_refresh(&k).unwrap();
        h.refresh_failed(&k, RefreshFailureObservation::local_credential_unusable())
            .unwrap();

        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::ReauthRequired));
    }

    #[test]
    fn refresh_failed_transient_routes_to_expiring() {
        let h = RuntimeAuthLeaseHandle::new();
        let k = lease("dev", "foo");

        h.acquire_lease(&k, 1_800_000_000).unwrap();
        h.begin_refresh(&k).unwrap();
        h.refresh_failed(&k, RefreshFailureObservation::transient())
            .unwrap();

        assert_eq!(h.snapshot(&k).phase, Some(AuthLeasePhase::Expiring));
    }

    #[test]
    fn transient_refresh_failure_preserves_credential_marker_generation() {
        let h = RuntimeAuthLeaseHandle::new();
        let k = lease("dev", "retryable_refresh");

        h.acquire_lease(&k, 1_800_000_000).unwrap();
        let before = h.snapshot(&k);
        h.begin_refresh(&k).unwrap();
        h.refresh_failed(&k, RefreshFailureObservation::transient())
            .unwrap();
        let after = h.snapshot(&k);

        assert_eq!(after.phase, Some(AuthLeasePhase::Expiring));
        assert_eq!(after.expires_at, before.expires_at);
        assert_eq!(after.generation, before.generation);
        assert_eq!(
            after.credential_published_at_millis,
            before.credential_published_at_millis
        );
    }

    #[test]
    fn snapshot_for_unknown_binding_is_none() {
        let h = RuntimeAuthLeaseHandle::new();
        let snap = h.snapshot(&lease("dev", "never_registered"));
        assert!(snap.phase.is_none());
        assert!(snap.expires_at.is_none());
    }

    #[test]
    fn mark_expiring_before_acquire_errors() {
        let h = RuntimeAuthLeaseHandle::new();
        let err = h.mark_expiring(&lease("dev", "ghost")).unwrap_err();
        assert_eq!(err.context, "AuthLeaseHandle::mark_expiring");
    }

    #[test]
    fn release_before_acquire_is_idempotent() {
        let h = RuntimeAuthLeaseHandle::new();
        let key = lease("dev", "ghost");

        h.release_lease(&key).unwrap();

        let snap = h.snapshot(&key);
        assert!(snap.phase.is_none());
        assert!(snap.expires_at.is_none());
    }

    #[test]
    fn release_does_not_remove_concurrent_reacquire() {
        let h = RuntimeAuthLeaseHandle::new();
        let key = lease("dev", "shared");
        h.acquire_lease(&key, 1_800_000_000).unwrap();

        let acquire_handle = h.clone();
        let acquire_key = key.clone();
        let hook_key = key.clone();
        let acquired_generation = Arc::new(Mutex::new(None));
        let hook_generation = Arc::clone(&acquired_generation);
        let _hook_guard =
            install_release_after_accept_hook_for_test(Arc::new(move |released_key| {
                if released_key != &hook_key {
                    return;
                }
                let generation = acquire_handle
                    .acquire_lease(&acquire_key, 1_800_000_000)
                    .unwrap()
                    .generation();
                *hook_generation.lock().unwrap() = Some(generation);
            }));

        h.release_lease(&key).unwrap();

        let acquired_generation = acquired_generation
            .lock()
            .unwrap()
            .expect("release hook should reacquire the lease");

        let snap = h.snapshot(&key);
        assert_eq!(
            snap.phase,
            Some(AuthLeasePhase::Valid),
            "accepted reacquire generation {acquired_generation} must remain visible after release completes; snapshot was {snap:?}"
        );
        assert_eq!(snap.expires_at, Some(1_800_000_000));
        assert_eq!(snap.generation, acquired_generation);
    }

    #[cfg(not(target_arch = "wasm32"))]
    struct FailingReleaseObserver;

    #[cfg(not(target_arch = "wasm32"))]
    impl AuthLeaseReleaseObserver for FailingReleaseObserver {
        fn auth_lease_released(
            &self,
            _released: &ReleasedOAuthFlows,
        ) -> Result<(), DslTransitionError> {
            Err(DslTransitionError::no_matching(
                "test_release_observer",
                "injected release observer failure",
            ))
        }
    }

    /// An observer fault aborts the release BEFORE the authoritative Release
    /// transition: the typed fault propagates and the original live lease is
    /// untouched — no transition happened, so there is nothing to restore.
    /// (Regression pin for the deleted post-transition restore arm, which
    /// could resurrect a live lease over a credential the token-clear flow
    /// was destroying.)
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn release_observer_failure_aborts_release_fail_closed() {
        let h = RuntimeAuthLeaseHandle::new();
        let key = lease("dev", "staged_failure");
        let acquired = h.acquire_lease(&key, 1_800_000_000).unwrap();

        let observer: Arc<dyn AuthLeaseReleaseObserver> = Arc::new(FailingReleaseObserver);
        h.add_release_observer(Arc::downgrade(&observer));

        // The after-accept hook must NOT fire: the release aborts before the
        // Release transition is ever applied.
        let hook_key = key.clone();
        let hook_fired = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let hook_fired_flag = Arc::clone(&hook_fired);
        let _hook_guard =
            install_release_after_accept_hook_for_test(Arc::new(move |released_key| {
                if released_key == &hook_key {
                    hook_fired_flag.store(true, std::sync::atomic::Ordering::Release);
                }
            }));

        let err = h
            .release_lease(&key)
            .expect_err("observer fault must surface typed");
        assert!(
            err.to_string()
                .contains("injected release observer failure"),
            "typed fault must carry the observer failure, got: {err}"
        );
        assert!(
            !hook_fired.load(std::sync::atomic::Ordering::Acquire),
            "Release transition must not be applied when an observer faulted"
        );

        // Fail-closed: the original lease is fully intact and retryable.
        let snap = h.snapshot(&key);
        assert_eq!(snap.phase, Some(AuthLeasePhase::Valid));
        assert_eq!(snap.expires_at, Some(1_800_000_000));
        assert!(snap.credential_present);
        assert_eq!(snap.generation, acquired.generation());
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn oauth_global_capacity_rejection_comes_from_generated_authority() {
        let h = RuntimeAuthLeaseHandle::new();
        let first = auth_binding("dev", "oauth_a");
        let second = auth_binding("dev", "oauth_b");

        h.apply_oauth_input(
            &first,
            auth_dsl::AuthMachineInput::AdmitOAuthBrowserFlow {
                flow_id: "first".to_string(),
                provider: "provider".to_string(),
                redirect_uri: "http://localhost/first".to_string(),
                expires_at_millis: 1_900_000_000,
                max_outstanding_flows: 1,
                observed_global_outstanding_flows: u64::MAX,
            },
            "test_admit_oauth_browser_flow",
            true,
        )
        .unwrap();

        let err = h
            .apply_oauth_input(
                &second,
                auth_dsl::AuthMachineInput::AdmitOAuthBrowserFlow {
                    flow_id: "second".to_string(),
                    provider: "provider".to_string(),
                    redirect_uri: "http://localhost/second".to_string(),
                    expires_at_millis: 1_900_000_000,
                    max_outstanding_flows: 1,
                    observed_global_outstanding_flows: 0,
                },
                "test_admit_oauth_browser_flow",
                true,
            )
            .unwrap_err();

        assert!(
            err.is_guard_rejected(),
            "expected generated guard rejection: {err:?}"
        );
        assert_eq!(err.context, "test_admit_oauth_browser_flow");
        assert!(
            err.reason.contains("AdmitOAuthBrowserFlow"),
            "guard rejection should come from the generated AuthMachine transition: {err:?}"
        );
        assert!(h.has_oauth_browser_flow_for_test(&first, "first"));
        assert!(!h.has_oauth_browser_flow_for_test(&second, "second"));
    }

    #[test]
    fn restore_oauth_missing_payload_rejected_by_generated_authority() {
        let mut registry = AuthLeaseRegistry::default();
        let key = lease("dev", "restore_missing_provider");
        let mut state = empty_auth_state();
        state.lifecycle_phase = auth_dsl::AuthLifecyclePhase::ReauthRequired;
        state
            .oauth_browser_flow_ids
            .insert("browser-flow".to_string());
        state.oauth_browser_flow_redirect_uris.insert(
            "browser-flow".to_string(),
            "http://localhost/callback".to_string(),
        );
        state
            .oauth_browser_flow_expires_at_millis
            .insert("browser-flow".to_string(), 1_900_000_000);
        state.oauth_outstanding_flow_count = 1;

        let err = restore_state_to_registry(
            &mut registry,
            &key,
            &state,
            "test_restore_oauth_missing_payload",
        )
        .unwrap_err();

        assert!(
            err.is_guard_rejected(),
            "missing restored OAuth payload must be rejected by generated guards: {err:?}"
        );
        assert!(
            err.reason.contains("RestoreOAuthBrowserFlow"),
            "rejection should name the generated restore input: {err:?}"
        );
        assert!(!registry.authorities.contains_key(&key));
    }

    #[test]
    fn restore_oauth_orphan_device_poll_rejected_by_generated_authority() {
        let mut registry = AuthLeaseRegistry::default();
        let key = lease("dev", "restore_orphan_poll");
        let mut state = empty_auth_state();
        state.lifecycle_phase = auth_dsl::AuthLifecyclePhase::ReauthRequired;
        state
            .oauth_device_poll_ids
            .insert("orphan-device".to_string());

        let err = restore_state_to_registry(
            &mut registry,
            &key,
            &state,
            "test_restore_oauth_orphan_device_poll",
        )
        .unwrap_err();

        assert!(
            err.is_guard_rejected(),
            "orphan restored OAuth poll must be rejected by generated guards: {err:?}"
        );
        assert!(
            err.reason.contains("RestoreOAuthDevicePoll"),
            "rejection should name the generated restore input: {err:?}"
        );
        assert!(!registry.authorities.contains_key(&key));
    }

    #[test]
    fn restore_released_oauth_membership_reauths_through_generated_authority() {
        let mut registry = AuthLeaseRegistry::default();
        let key = lease("dev", "restore_released_oauth");
        let mut state = empty_auth_state();
        state.lifecycle_phase = auth_dsl::AuthLifecyclePhase::Released;
        state
            .oauth_browser_flow_ids
            .insert("browser-flow".to_string());
        state
            .oauth_browser_flow_providers
            .insert("browser-flow".to_string(), "provider".to_string());
        state.oauth_browser_flow_redirect_uris.insert(
            "browser-flow".to_string(),
            "http://localhost/callback".to_string(),
        );
        state
            .oauth_browser_flow_expires_at_millis
            .insert("browser-flow".to_string(), 1_900_000_000);
        state.oauth_outstanding_flow_count = 1;

        let (phase, transition) = restore_state_to_registry(
            &mut registry,
            &key,
            &state,
            "test_restore_released_oauth_membership",
        )
        .unwrap();

        assert_eq!(phase, AuthLeasePhase::ReauthRequired);
        assert_eq!(transition.phase(), AuthLeasePhase::ReauthRequired);
        let restored = registry.authorities.get(&key).unwrap().state();
        assert_eq!(
            restored.lifecycle_phase,
            auth_dsl::AuthLifecyclePhase::ReauthRequired
        );
        assert!(restored.oauth_browser_flow_ids.contains("browser-flow"));
        assert_eq!(restored.oauth_outstanding_flow_count, 1);
    }

    #[test]
    fn repeated_acquire_updates_existing_lease() {
        let h = RuntimeAuthLeaseHandle::new();
        let key = lease("dev", "default");

        h.acquire_lease(&key, 1_800_000_000).unwrap();
        h.acquire_lease(&key, 1_900_000_000).unwrap();

        let snap = h.snapshot(&key);
        assert_eq!(snap.phase, Some(AuthLeasePhase::Valid));
        assert_eq!(snap.expires_at, Some(1_900_000_000));
    }

    #[test]
    fn restore_snapshot_preserves_publication_marker_without_lowering_generation() {
        let h = RuntimeAuthLeaseHandle::new();
        let key = lease("dev", "shared");

        h.acquire_lease(&key, 1_800_000_000).unwrap();
        let before = h.snapshot(&key);
        let before_restore = h.capture_auth_lifecycle_restore_snapshot(&key);
        assert_eq!(before.phase, Some(AuthLeasePhase::Valid));
        assert!(before.credential_present);
        assert!(before.credential_published_at_millis.is_some());

        h.acquire_lease(&key, 1_900_000_000).unwrap();
        let advanced = h.snapshot(&key);
        assert!(advanced.generation > before.generation);

        h.restore_auth_lifecycle_snapshot(&before_restore).unwrap();

        let restored = h.snapshot(&key);
        assert_eq!(restored.phase, before.phase);
        assert_eq!(restored.expires_at, before.expires_at);
        assert_eq!(restored.credential_present, before.credential_present);
        assert_eq!(
            restored.credential_published_at_millis,
            before.credential_published_at_millis
        );
        assert_eq!(restored.generation, advanced.generation);
    }

    #[test]
    fn restore_empty_zero_generation_snapshot_releases_through_generated_authority() {
        let h = RuntimeAuthLeaseHandle::new();
        let key = lease("dev", "shared");
        let empty = h.capture_auth_lifecycle_restore_snapshot(&key);
        h.acquire_lease(&key, 1_800_000_000).unwrap();
        let acquired_generation = h.snapshot(&key).generation;
        assert!(acquired_generation > empty.snapshot().generation);

        h.restore_auth_lifecycle_snapshot(&empty).unwrap();

        let restored = h.snapshot(&key);
        assert_eq!(restored.phase, None);
        assert_eq!(restored.expires_at, None);
        assert!(!restored.credential_present);
        assert_eq!(restored.generation, acquired_generation);
        assert_eq!(restored.credential_published_at_millis, None);
    }

    #[test]
    fn restore_snapshot_rejects_capture_from_different_runtime_handle() {
        let first = RuntimeAuthLeaseHandle::new();
        let second = RuntimeAuthLeaseHandle::new();
        let key = lease("dev", "shared");
        first.acquire_lease(&key, 1_800_000_000).unwrap();
        let captured = first.capture_auth_lifecycle_restore_snapshot(&key);

        let err = second
            .restore_auth_lifecycle_snapshot(&captured)
            .unwrap_err();

        assert_eq!(
            err.context,
            "AuthLeaseHandle::restore_auth_lifecycle_snapshot"
        );
        assert!(
            err.reason.contains("this RuntimeAuthLeaseHandle"),
            "restore must reject snapshots captured from a different runtime handle: {err:?}"
        );
        assert_eq!(second.snapshot(&key).phase, None);
    }

    #[tokio::test]
    async fn restore_published_credential_lifecycle_uses_generated_authority() {
        struct SingleTokenStore {
            key: meerkat_core::auth::TokenKey,
            tokens: meerkat_core::auth::PersistedTokens,
        }

        #[async_trait::async_trait]
        impl meerkat_core::auth::TokenStore for SingleTokenStore {
            async fn load(
                &self,
                key: &meerkat_core::auth::TokenKey,
            ) -> Result<
                Option<meerkat_core::auth::PersistedTokens>,
                meerkat_core::auth::TokenStoreError,
            > {
                Ok((key == &self.key).then(|| self.tokens.clone()))
            }

            async fn save(
                &self,
                _key: &meerkat_core::auth::TokenKey,
                _tokens: &meerkat_core::auth::PersistedTokens,
            ) -> Result<(), meerkat_core::auth::TokenStoreError> {
                Ok(())
            }

            async fn clear(
                &self,
                _key: &meerkat_core::auth::TokenKey,
            ) -> Result<(), meerkat_core::auth::TokenStoreError> {
                Ok(())
            }

            async fn list(
                &self,
            ) -> Result<Vec<meerkat_core::auth::TokenKey>, meerkat_core::auth::TokenStoreError>
            {
                Ok(vec![self.key.clone()])
            }

            fn backend_name(&self) -> &'static str {
                "single-token-test"
            }
        }

        let source = Arc::new(RuntimeAuthLeaseHandle::new());
        let restored = Arc::new(RuntimeAuthLeaseHandle::new());
        let generated_restored =
            crate::protocol_auth_lease_lifecycle_publication::generated_auth_lease_handle(
                Arc::clone(&restored),
            )
            .expect("test AuthLeaseHandle must be generated-authority certified");
        let key = lease("dev", "shared");
        let transition = source.acquire_lease(&key, 1_800_000_000).unwrap();
        let published_at = transition
            .credential_published_at_millis()
            .expect("acquire transition carries publication time");
        let key_for_tokens = meerkat_core::auth::TokenKey::new_with_profile(
            key.realm.clone(),
            key.binding.clone(),
            key.profile.clone(),
        );
        let tokens = meerkat_core::auth::PersistedTokens {
            auth_mode: meerkat_core::auth::PersistedAuthMode::ChatgptOauth,
            primary_secret: Some("access-token".into()),
            refresh_token: Some("refresh-token".into()),
            id_token: None,
            expires_at: Some(
                chrono::DateTime::from_timestamp(transition.expires_at() as i64, 0)
                    .expect("fixture expiry is representable"),
            ),
            last_refresh: None,
            scopes: Vec::new(),
            account_id: None,
            metadata: serde_json::Value::Null,
        };
        let marked = meerkat_core::mark_tokens_lifecycle_published_for_transition(
            &key_for_tokens,
            &tokens,
            &transition,
        )
        .expect("generated transition marks durable publication");

        let auth_binding = AuthBindingRef {
            realm: key.realm.clone(),
            binding: key.binding.clone(),
            profile: key.profile.clone(),
            origin: BindingOrigin::Configured,
        };
        let store = SingleTokenStore {
            key: key_for_tokens,
            tokens: marked,
        };
        meerkat_core::rehydrate_marked_tokens_for_status(
            &store,
            &generated_restored,
            &auth_binding,
            meerkat_core::auth::PersistedAuthMode::ChatgptOauth,
            chrono::Utc::now(),
        )
        .await
        .expect("generated marker restores through AuthMachine")
        .expect("marker is present");

        let snapshot = restored.snapshot(&key);
        assert_eq!(snapshot.phase, Some(AuthLeasePhase::Valid));
        assert_eq!(snapshot.expires_at, Some(transition.expires_at()));
        assert_eq!(snapshot.generation, transition.generation());
        assert_eq!(snapshot.credential_published_at_millis, Some(published_at));
        assert!(snapshot.credential_present);
    }

    #[test]
    fn per_binding_isolation() {
        let h = RuntimeAuthLeaseHandle::new();
        let openai = lease("dev", "openai");
        let anthropic = lease("dev", "anthropic");
        h.acquire_lease(&openai, 1_800_000_000).unwrap();
        h.acquire_lease(&anthropic, 1_900_000_000).unwrap();
        h.mark_expiring(&openai).unwrap();

        assert_eq!(h.snapshot(&openai).phase, Some(AuthLeasePhase::Expiring));
        assert_eq!(h.snapshot(&anthropic).phase, Some(AuthLeasePhase::Valid));
        assert_eq!(h.snapshot(&anthropic).expires_at, Some(1_900_000_000));
    }
    /// K7 interleaving pin: the token-clear flow stages the lease release
    /// BEFORE the durable clear commits, and the release itself notifies
    /// observers before its authoritative transition. A release-observer
    /// fault therefore aborts the whole clear fail-closed: the typed fault
    /// surfaces, the durable clear never ran, and the lease is untouched —
    /// no interleaving can produce a live lease over a cleared credential,
    /// and no post-clear lease operation (the old restore arm) exists.
    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn clear_flow_observer_failure_fails_closed_and_never_clears_durable() {
        use std::sync::Mutex as StdMutex;

        struct RecordingStore {
            tokens: StdMutex<Option<meerkat_core::auth::PersistedTokens>>,
            key: meerkat_core::auth::TokenKey,
            clear_called: std::sync::atomic::AtomicBool,
        }

        #[async_trait::async_trait]
        impl meerkat_core::auth::TokenStore for RecordingStore {
            async fn load(
                &self,
                key: &meerkat_core::auth::TokenKey,
            ) -> Result<
                Option<meerkat_core::auth::PersistedTokens>,
                meerkat_core::auth::TokenStoreError,
            > {
                if key == &self.key {
                    Ok(self.tokens.lock().unwrap().clone())
                } else {
                    Ok(None)
                }
            }

            async fn save(
                &self,
                _key: &meerkat_core::auth::TokenKey,
                tokens: &meerkat_core::auth::PersistedTokens,
            ) -> Result<(), meerkat_core::auth::TokenStoreError> {
                *self.tokens.lock().unwrap() = Some(tokens.clone());
                Ok(())
            }

            async fn clear(
                &self,
                _key: &meerkat_core::auth::TokenKey,
            ) -> Result<(), meerkat_core::auth::TokenStoreError> {
                self.clear_called
                    .store(true, std::sync::atomic::Ordering::Release);
                *self.tokens.lock().unwrap() = None;
                Ok(())
            }

            async fn list(
                &self,
            ) -> Result<Vec<meerkat_core::auth::TokenKey>, meerkat_core::auth::TokenStoreError>
            {
                Ok(vec![self.key.clone()])
            }

            fn backend_name(&self) -> &'static str {
                "recording-clear-test"
            }
        }

        let handle = Arc::new(RuntimeAuthLeaseHandle::new());
        let generated =
            crate::protocol_auth_lease_lifecycle_publication::generated_auth_lease_handle(
                Arc::clone(&handle),
            )
            .expect("test AuthLeaseHandle must be generated-authority certified");
        let binding = auth_binding("dev", "clear_stage");
        let key = meerkat_core::auth::TokenKey::from_auth_binding(&binding);

        let tokens = meerkat_core::auth::PersistedTokens {
            auth_mode: meerkat_core::auth::PersistedAuthMode::ChatgptOauth,
            primary_secret: Some("access-token".into()),
            refresh_token: Some("refresh-token".into()),
            id_token: None,
            expires_at: None,
            last_refresh: None,
            scopes: Vec::new(),
            account_id: None,
            metadata: serde_json::Value::Null,
        };
        meerkat_core::publish_token_lifecycle_acquired(&generated, &binding, &tokens)
            .expect("acquire lease");
        let store = RecordingStore {
            tokens: StdMutex::new(Some(tokens)),
            key: key.clone(),
            clear_called: std::sync::atomic::AtomicBool::new(false),
        };

        let observer: Arc<dyn AuthLeaseReleaseObserver> = Arc::new(FailingReleaseObserver);
        handle.add_release_observer(Arc::downgrade(&observer));

        let err =
            meerkat_core::clear_tokens_and_publish_lifecycle_released(&store, &generated, &binding)
                .await
                .expect_err("staged release observer fault must fail the clear typed");
        assert!(
            matches!(
                err,
                meerkat_core::TokenLifecycleClearError::AuthMachineRelease(_)
            ),
            "expected typed AuthMachineRelease fault, got: {err:?}"
        );

        // The durable clear must never run after a staging fault: fail closed.
        assert!(
            !store
                .clear_called
                .load(std::sync::atomic::Ordering::Acquire),
            "durable clear must not run when release staging faulted"
        );
        assert!(store.tokens.lock().unwrap().is_some());

        // Fail-closed consistency: durable truth still holds the credential
        // and the lease projection still matches it (live) — the clear is
        // retryable; nothing diverged. In particular no interleaving exists
        // where a live lease outlives a CLEARED credential: the durable clear
        // only commits after the release fully succeeded.
        let lease_key = meerkat_core::handles::LeaseKey::from_auth_binding(&binding);
        let snap = handle.snapshot(&lease_key);
        assert_eq!(snap.phase, Some(AuthLeasePhase::Valid));
        assert!(
            snap.credential_present,
            "aborted clear must leave the live lease aligned with the retained durable record, got {snap:?}"
        );
    }

    // --- 0.7.2 disciplined shell inputs (lane L3 auth-release) ---
    //
    // Machine-level pins for the two-phase release drain (D1) and the
    // post-teardown totality of observation/cleanup-shaped OAuth inputs
    // (D2a). These drive the generated authority directly.

    fn apply_machine(
        authority: &mut auth_dsl::AuthMachineAuthority,
        input: auth_dsl::AuthMachineInput,
    ) -> Result<auth_dsl::AuthMachineTransition, auth_dsl::AuthMachineTransitionError> {
        auth_dsl::AuthMachineMutator::apply(authority, input)
    }

    fn admit_browser_flow_input(flow_id: &str) -> auth_dsl::AuthMachineInput {
        auth_dsl::AuthMachineInput::AdmitOAuthBrowserFlow {
            flow_id: flow_id.to_string(),
            provider: "openai-chatgpt".to_string(),
            redirect_uri: "http://127.0.0.1/callback".to_string(),
            expires_at_millis: u64::MAX,
            max_outstanding_flows: 16,
            observed_global_outstanding_flows: 0,
        }
    }

    fn admit_device_flow_input(flow_id: &str) -> auth_dsl::AuthMachineInput {
        auth_dsl::AuthMachineInput::AdmitOAuthDeviceFlow {
            flow_id: flow_id.to_string(),
            provider: "openai-chatgpt".to_string(),
            expires_at_millis: u64::MAX,
            max_outstanding_flows: 16,
            observed_global_outstanding_flows: 0,
        }
    }

    /// D1: release is two-phase. With flows in flight, Release alone is
    /// rejected (drain obligations open); BeginRelease records the draining
    /// intent and emits the typed cancellation obligation carrying exactly
    /// the in-flight membership; the Expire* feedback inputs close the
    /// obligations; only then does Release commit.
    #[test]
    fn begin_release_emits_machine_owned_drain_obligation_and_guards_release() {
        let mut machine = auth_dsl::AuthMachineAuthority::new();
        apply_machine(
            &mut machine,
            auth_dsl::AuthMachineInput::Acquire {
                expires_at_ts: Some(2_000_000_000),
                credential_published_at_millis: 1,
            },
        )
        .expect("acquire");
        apply_machine(&mut machine, admit_browser_flow_input("b-1")).expect("admit browser flow");
        apply_machine(&mut machine, admit_device_flow_input("d-1")).expect("admit device flow");

        // Release with in-flight flows is guard-rejected: the drain
        // obligations are still open.
        let err = apply_machine(&mut machine, auth_dsl::AuthMachineInput::Release)
            .expect_err("Release must not commit while flows are in flight");
        assert!(
            matches!(
                err,
                auth_dsl::AuthMachineTransitionError::GuardRejected { .. }
            ),
            "expected guard rejection, got: {err:?}"
        );

        // BeginRelease enters the draining sub-state and emits the typed
        // cancellation obligation with the exact in-flight membership.
        let transition = apply_machine(&mut machine, auth_dsl::AuthMachineInput::BeginRelease)
            .expect("begin release");
        assert!(machine.state().release_draining);
        let cancel = transition
            .effects()
            .iter()
            .find_map(|effect| match effect {
                auth_dsl::AuthMachineEffect::CancelOAuthFlowsForRelease {
                    browser_flow_ids,
                    device_flow_ids,
                } => Some((browser_flow_ids.clone(), device_flow_ids.clone())),
                _ => None,
            })
            .expect("BeginRelease with in-flight flows must emit CancelOAuthFlowsForRelease");
        assert_eq!(
            cancel.0.iter().collect::<Vec<_>>(),
            vec!["b-1"],
            "cancellation obligation must carry the in-flight browser flow"
        );
        assert_eq!(
            cancel.1.iter().collect::<Vec<_>>(),
            vec!["d-1"],
            "cancellation obligation must carry the in-flight device flow"
        );

        // New admissions are refused while the drain is in flight, so the
        // obligation set cannot grow behind the shell's back.
        let err = apply_machine(&mut machine, admit_browser_flow_input("b-2"))
            .expect_err("admission during release drain must be refused");
        assert!(
            matches!(
                err,
                auth_dsl::AuthMachineTransitionError::GuardRejected { .. }
            ),
            "expected guard rejection, got: {err:?}"
        );

        // Typed terminal cancellation closes each obligation.
        apply_machine(
            &mut machine,
            auth_dsl::AuthMachineInput::ExpireOAuthBrowserFlow {
                flow_id: "b-1".to_string(),
            },
        )
        .expect("terminal browser cancellation");
        apply_machine(
            &mut machine,
            auth_dsl::AuthMachineInput::ExpireOAuthDeviceFlow {
                flow_id: "d-1".to_string(),
            },
        )
        .expect("terminal device cancellation");
        assert_eq!(machine.state().oauth_outstanding_flow_count, 0);

        // All drain obligations closed: Release commits.
        apply_machine(&mut machine, auth_dsl::AuthMachineInput::Release).expect("drained release");
        assert!(matches!(
            machine.state().lifecycle_phase,
            auth_dsl::AuthLifecyclePhase::Released
        ));
        assert!(!machine.state().release_draining);
        assert_eq!(machine.state().oauth_outstanding_flow_count, 0);
        assert!(machine.state().oauth_browser_flow_ids.is_empty());
        assert!(machine.state().oauth_device_flow_ids.is_empty());
    }

    /// D1: with no flows in flight, BeginRelease records the drain intent
    /// without emitting a cancellation obligation, and Release commits
    /// immediately.
    #[test]
    fn begin_release_without_flows_emits_no_drain_obligation() {
        let mut machine = auth_dsl::AuthMachineAuthority::new();
        apply_machine(
            &mut machine,
            auth_dsl::AuthMachineInput::Acquire {
                expires_at_ts: Some(2_000_000_000),
                credential_published_at_millis: 1,
            },
        )
        .expect("acquire");

        let transition = apply_machine(&mut machine, auth_dsl::AuthMachineInput::BeginRelease)
            .expect("begin release");
        assert!(machine.state().release_draining);
        assert!(
            transition.effects().iter().all(|effect| !matches!(
                effect,
                auth_dsl::AuthMachineEffect::CancelOAuthFlowsForRelease { .. }
            )),
            "no cancellation obligation without in-flight flows"
        );

        apply_machine(&mut machine, auth_dsl::AuthMachineInput::Release).expect("release");
        assert!(matches!(
            machine.state().lifecycle_phase,
            auth_dsl::AuthLifecyclePhase::Released
        ));
        assert!(!machine.state().release_draining);
    }

    /// D2a: late poll/prune/compensation arrivals after teardown — Expire*,
    /// durable-admission Confirm, poll Finish, and a release retry — are
    /// accepted as no-ops in Released with zero state change and zero
    /// emitted effects, never a guard rejection (worklist entries 24-30).
    #[test]
    fn post_release_oauth_observation_inputs_are_total_noops() {
        let mut machine = auth_dsl::AuthMachineAuthority::new();
        apply_machine(&mut machine, auth_dsl::AuthMachineInput::Release).expect("release");
        assert!(matches!(
            machine.state().lifecycle_phase,
            auth_dsl::AuthLifecyclePhase::Released
        ));
        let before = machine.state().clone();

        for input in [
            auth_dsl::AuthMachineInput::ExpireOAuthBrowserFlow {
                flow_id: "ghost-browser".to_string(),
            },
            auth_dsl::AuthMachineInput::ExpireOAuthDeviceFlow {
                flow_id: "ghost-device".to_string(),
            },
            auth_dsl::AuthMachineInput::FinishOAuthDevicePoll {
                flow_id: "ghost-poll".to_string(),
            },
            auth_dsl::AuthMachineInput::ConfirmOAuthDurableAdmission {
                observed_global_outstanding_flows: 0,
                max_outstanding_flows: 16,
            },
            auth_dsl::AuthMachineInput::BeginRelease,
        ] {
            let description = format!("{input:?}");
            let transition = apply_machine(&mut machine, input).unwrap_or_else(|err| {
                panic!("`{description}` must be a total no-op in Released, got: {err:?}")
            });
            assert!(
                transition.effects().is_empty(),
                "`{description}` must not emit effects in Released"
            );
            assert_eq!(
                machine.state(),
                &before,
                "`{description}` must not mutate Released state"
            );
        }
    }

    /// D2a: stale expiry/finish observations for flows that are no longer
    /// members are total no-ops in live phases too — a consume or release
    /// drain legitimately removes the flow between a producer's membership
    /// observation and its input delivery.
    #[test]
    fn stale_oauth_cleanup_observations_are_total_noops_in_live_phases() {
        let mut machine = auth_dsl::AuthMachineAuthority::new();
        apply_machine(
            &mut machine,
            auth_dsl::AuthMachineInput::Acquire {
                expires_at_ts: Some(2_000_000_000),
                credential_published_at_millis: 1,
            },
        )
        .expect("acquire");
        apply_machine(&mut machine, admit_browser_flow_input("b-1")).expect("admit browser flow");
        let before = machine.state().clone();

        for input in [
            auth_dsl::AuthMachineInput::ExpireOAuthBrowserFlow {
                flow_id: "ghost-browser".to_string(),
            },
            auth_dsl::AuthMachineInput::ExpireOAuthDeviceFlow {
                flow_id: "ghost-device".to_string(),
            },
            auth_dsl::AuthMachineInput::FinishOAuthDevicePoll {
                flow_id: "ghost-poll".to_string(),
            },
        ] {
            let description = format!("{input:?}");
            apply_machine(&mut machine, input).unwrap_or_else(|err| {
                panic!("stale `{description}` must be a total no-op, got: {err:?}")
            });
            assert_eq!(
                machine.state(),
                &before,
                "stale `{description}` must not mutate live-phase state"
            );
        }

        // The genuine member is still expirable afterwards.
        apply_machine(
            &mut machine,
            auth_dsl::AuthMachineInput::ExpireOAuthBrowserFlow {
                flow_id: "b-1".to_string(),
            },
        )
        .expect("present flow still expires");
        assert_eq!(machine.state().oauth_outstanding_flow_count, 0);
    }
}