meerkat-mobkit 0.8.21

//! Unified runtime — combines mob lifecycle, module management, and operational subsystems.

use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Duration;

use futures::stream::{BoxStream, SelectAll, StreamExt};
use meerkat_core::comms::EventStream;
use meerkat_core::event::{AgentEvent, agent_event_type};
use meerkat_mob::{
    AgentIdentity, AgentRuntimeId, AttributedEvent, FenceToken, MobError, MobHandle,
    MobMemberStatus, MobState, ProfileName, SpawnMemberSpec,
};
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::task::JoinHandle;

pub(crate) use self::console_events::ConsoleEventStore;
use self::mob_events::MobEventsStore;
use crate::console_aggregator::{ConsoleLogStore, InMemoryConsoleLogStore};
use crate::mob_handle_runtime::{MobBootstrapSpec, MobRuntime, MobRuntimeError};
use crate::runtime::{
    InMemoryMetadataStore, MetadataScope, MobkitRuntimeHandle, PersistentMetadataStore,
    RuntimeMetadataTable, RuntimeOptions, start_mobkit_runtime_with_options,
};
use crate::types::{
    AgentDiscoverySpec, EventEnvelope, MobKitConfig, MobStructuralEventEnvelope, UnifiedEvent,
};

pub mod builder;
pub(crate) mod console_events;
pub mod cross_mob;
pub mod edge_reconcile;
pub mod edge_types;
pub mod event_log;
pub mod http;
pub(crate) mod implicit_delegate_retirement;
pub mod lifecycle;
pub mod mob_events;
pub mod mob_ops;
pub mod module_ops;
pub mod types;

pub use crate::identity_first::IdentityBootstrapMode;
pub use builder::UnifiedRuntimeBuilder;
pub use edge_types::{
    DesiredPeerEdge, DesiredPeerEdgeError, Discovery, EdgeDiscovery, EdgeReconcileFailure,
    PreSpawnContext, PreSpawnHook,
};
pub use event_log::{
    EventLogConfig, EventLogError, EventLogStore, EventQuery, NullEventLogStore, PersistedEvent,
};
pub use http::DEFAULT_REFERENCE_APP_MAX_CONCURRENT_REQUESTS;
pub use mob_ops::MemberTurnAdmission;
pub use types::{
    CompactionPreservedHistoryFit, ErrorEvent, IdentityAuthorityReleaseOutcome, MobStopOutcome,
    RediscoverReport, ShutdownDrainReport, UnifiedRuntimeBootstrapError,
    UnifiedRuntimeBuilderError, UnifiedRuntimeBuilderField, UnifiedRuntimeError,
    UnifiedRuntimeReconcileEdgesReport, UnifiedRuntimeReconcileError,
    UnifiedRuntimeReconcileReport, UnifiedRuntimeReconcileRoutingReport, UnifiedRuntimeRunReport,
    UnifiedRuntimeShutdownReport,
};

/// Called after members are spawned. Receives the list of spawned member IDs.
pub type PostSpawnHook =
    Arc<dyn Fn(Vec<String>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;

/// Called after reconcile completes. Receives the reconcile report.
pub type PostReconcileHook = Arc<
    dyn Fn(UnifiedRuntimeReconcileReport) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync,
>;

/// Called when a runtime operation fails. Fire-and-forget — the hook's
/// result is not checked and a failing hook cannot break the runtime.
pub type ErrorHook =
    Arc<dyn Fn(ErrorEvent) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;

/// Late-bound shared error hook slot (the same pattern as the identity
/// authority and `gateway_peer_keys`): gateways install the hook via
/// [`UnifiedRuntime::set_error_hook`] AFTER construction, while runtime-owned
/// background tasks that must fire it (the actor-loop probe) start AT
/// construction. Tasks hold the slot and read the hook at fire time.
type SharedErrorHook = Arc<std::sync::RwLock<Option<ErrorHook>>>;

fn current_error_hook(slot: &SharedErrorHook) -> Option<ErrorHook> {
    slot.read()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone()
}

/// The default sink every [`ErrorEvent`] passes through, whether or not a
/// host registered an [`ErrorHook`].
///
/// A paging channel that discards when nobody wired it is indistinguishable
/// from a healthy fleet. A host with zero `on_error` call sites used to get
/// silence for every operational failure mobkit detected — 136 compaction
/// rejections in one observed fleet, found only by reading a console table
/// by hand. Logging here costs a wired host nothing (the hook still fires,
/// exactly once) and gives an unwired one somewhere an operator or a
/// log-shipper can see the event.
///
/// Emitted with the typed event (`Debug`, so the variant name and its fields
/// both land) rather than only the `Display` rendering, and synchronously on
/// the caller's thread so the record is not stranded on a detached task.
pub(crate) fn log_error_event(event: &ErrorEvent, hook_registered: bool) {
    // `ActorLoopRecovered` is the one variant that reports a failure ENDING.
    // Logging a recovery at ERROR would make the level a lie and would put a
    // second scary line in the log for every stall that resolved fine.
    if matches!(event, ErrorEvent::ActorLoopRecovered { .. }) {
        tracing::info!(
            error_event = ?event,
            hook_registered,
            "mobkit runtime error event resolved: {event}"
        );
    } else {
        tracing::error!(
            error_event = ?event,
            hook_registered,
            "mobkit runtime error event: {event}"
        );
    }
    if !hook_registered {
        warn_error_hook_absent_once();
    }
}

/// State the missing-hook condition plainly: nobody is listening, and here is
/// where to fix it. Emitted at build time by `UnifiedRuntimeBuilder::build`
/// for hosts that never called `on_error`, and once per process from the fire
/// path for hosts that bootstrap directly and install the hook afterwards.
pub(crate) fn emit_error_hook_absent_notice() {
    tracing::warn!(
        "no error hook is registered, so runtime error events reach logs only; \
         register one with UnifiedRuntimeBuilder::on_error (or \
         UnifiedRuntime::set_error_hook) to route them to paging"
    );
}

/// The fire-path guard for the notice: the condition is per-process, and
/// repeating it on every event would bury the events themselves.
fn warn_error_hook_absent_once() {
    static NOTICED: std::sync::Once = std::sync::Once::new();
    NOTICED.call_once(emit_error_hook_absent_notice);
}

/// Fire an error event on the hook currently installed in `slot`, if any.
/// Truly fire-and-forget — spawns a detached task so slow hooks (HTTP to
/// Slack, PagerDuty) never block the caller.
///
/// The event reaches [`log_error_event`] either way: an unregistered hook
/// must not be the difference between an operator seeing a failure and not.
fn fire_error_hook(slot: &SharedErrorHook, event: ErrorEvent) {
    let hook = current_error_hook(slot);
    log_error_event(&event, hook.is_some());
    if let Some(hook) = hook {
        tokio::spawn(async move {
            let () = hook(event).await;
        });
    }
}

const ROSTER_ROUTE_PREFIX: &str = "mob.member.";
const ROSTER_ROUTE_CHANNEL: &str = "notification";
const ROSTER_ROUTE_SINK: &str = "mob_member";
const ROSTER_ROUTE_TARGET_MODULE: &str = "delivery";

const DEFAULT_DRAIN_TIMEOUT: Duration = Duration::from_secs(30);

/// Map an [`AgentDiscoverySpec`] to a [`SpawnMemberSpec`] for spawning.
///
/// `additional_instructions` maps directly to `SpawnMemberSpec.additional_instructions`,
/// which flows through Meerkat's build pipeline to `AgentBuildConfig.additional_instructions`.
pub fn discovery_spec_to_spawn_spec(spec: &AgentDiscoverySpec) -> SpawnMemberSpec {
    let resume_session_id = spec
        .resume_session_id
        .as_deref()
        .and_then(|s| meerkat_core::types::SessionId::parse(s).ok());
    let additional_instructions = if spec.additional_instructions.is_empty() {
        None
    } else {
        Some(spec.additional_instructions.clone())
    };
    let mut spawn = SpawnMemberSpec::new(
        meerkat_mob::ProfileName::from(spec.profile.as_str()),
        // The spec stays in the public alias space: the hook-aware
        // `UnifiedRuntime::spawn`/`spawn_many` own the encode to the
        // comms-safe roster id (meerkat 0.7 MemberCommsName), and the encode
        // is deliberately not idempotent (`mk--` is a reserved marker), so
        // encoding here too would double-encode `:`-bearing identities.
        meerkat_mob::ids::AgentIdentity::from(spec.meerkat_id.as_str()),
    );
    if let Some(context) = spec.context.clone() {
        spawn = spawn.with_context(context);
    }
    if let Some(labels) = spec.labels.clone() {
        spawn = spawn.with_labels(labels);
    }
    if let Some(sid) = resume_session_id {
        spawn = spawn.with_resume_bridge_session_id(sid);
    }
    if let Some(instructions) = additional_instructions {
        spawn = spawn.with_additional_instructions(instructions);
    }
    spawn
}

pub struct UnifiedRuntime {
    // Immutable after construction — &self access
    mob_runtime: MobRuntime,
    post_spawn_hook: Option<PostSpawnHook>,
    post_reconcile_hook: Option<PostReconcileHook>,
    error_hook: SharedErrorHook,
    drain_timeout: Duration,
    discovery: Option<Box<dyn Discovery>>,
    edge_discovery: Option<Arc<dyn EdgeDiscovery>>,

    // Fine-grained interior mutability
    module_runtime: Arc<tokio::sync::Mutex<MobkitRuntimeHandle>>,
    managed_dynamic_edges: Arc<tokio::sync::RwLock<BTreeSet<(String, String)>>>,
    shutting_down: AtomicBool,
    mob_event_ingress: tokio::sync::Mutex<Option<MobEventIngress>>,
    bootstrap_edges_report: tokio::sync::RwLock<Option<UnifiedRuntimeReconcileEdgesReport>>,
    event_log: Option<event_log::EventLogHandle>,
    console_log_store: Arc<dyn ConsoleLogStore>,
    console_events: ConsoleEventStore,
    mob_events: MobEventsStore,
    mob_events_subscriber_task: tokio::sync::Mutex<Option<JoinHandle<()>>>,
    /// Actor-loop liveness probe: a periodic O(1) round trip through the
    /// serialized mob command loop that pages (`ErrorEvent::ActorLoopStalled`)
    /// when the loop stops draining. Observation only — aborted first during
    /// shutdown so the intentional actor stop cannot fire a false page.
    actor_loop_probe_task: tokio::sync::Mutex<Option<JoinHandle<()>>>,
    implicit_delegate_retirement_task: tokio::sync::Mutex<Option<JoinHandle<()>>>,
    /// Late-bound identity authority observed by the already-running idle
    /// retirement sweeper. Gateways attach identity-first after base runtime
    /// bootstrap, so capturing `identity_runtime()` when the task starts would
    /// permanently capture `None`.
    implicit_delegate_identity_runtime:
        Arc<std::sync::RwLock<Option<Arc<crate::identity_first::IdentityRuntime>>>>,
    identity_lease_renewal_task:
        tokio::sync::Mutex<Option<crate::identity_first::runtime::TrackedLeaseRenewalTask>>,
    identity_continuity_repair_task:
        tokio::sync::Mutex<Option<crate::identity_first::runtime::TrackedContinuityRepairTask>>,
    /// Cleanups for supervisors displaced by `start_identity_first_supervisors`.
    ///
    /// Replacement previously did `tokio::spawn(previous.cancel_and_join())` and
    /// dropped the handle, and a `JoinHandle` detaches on drop, so the cleanup
    /// could outlive `shutdown()` while still holding the authority it was
    /// releasing. Owning them here makes them joinable at shutdown.
    retired_supervisor_cleanups:
        tokio::sync::Mutex<tokio::task::JoinSet<types::RetiredSupervisorKind>>,
    agent_memory_observer_task:
        tokio::sync::Mutex<Option<crate::memory::taint::TaintObserverGuard>>,
    agent_memory_steward_task: tokio::sync::Mutex<Option<JoinHandle<()>>>,

    // Cross-mob communication
    contact_directory: Option<crate::contact_directory::ContactDirectory>,
    peer_mob_handles: tokio::sync::RwLock<BTreeMap<String, cross_mob::PeerMobAuthority>>,
    /// Serve task of the cross-mob control listener, when one was started
    /// via [`UnifiedRuntime::start_control_listener`]. Aborted on shutdown
    /// like the other runtime-owned background tasks.
    cross_mob_control_task: tokio::sync::Mutex<Option<JoinHandle<()>>>,
    /// The dialable address the control listener actually bound
    /// (`tcp://ip:port` with the real port for `host:0` binds, or
    /// `uds:///path`). This is the address remote peers are told to put in
    /// their contact directories, and the address stamped into outbound
    /// remote wire requests as this gateway's control endpoint.
    cross_mob_control_advertised: std::sync::RwLock<Option<String>>,
    /// Long-lived Ed25519 signing identity for cross-process peering.
    /// `None` is the default for inproc-only deployments and tests;
    /// production gateways set this via
    /// [`UnifiedRuntime::set_gateway_peer_keys`] during bootstrap so the
    /// `mobkit/peer_pubkey` RPC can advertise it and the cross-mob control
    /// listener can sign its responses. A shared late-bound slot (the same
    /// pattern as the identity authority): the control listener may start
    /// before the host installs keys, and its serve task re-reads this
    /// slot per request.
    gateway_peer_keys: crate::runtime::cross_mob_control::ControlSignerSlot,
    /// Late-bound read-only host projection paired with `gateway_peer_keys`.
    /// The listener re-reads both slots, so listener-first and keys-first
    /// bootstrap orders converge on the same authenticated host identity.
    remote_host_facts: crate::runtime::cross_mob_control::HostFactsProviderSlot,
    /// Controller-owned durable endpoint-identity pins plus transient
    /// authenticated reachability. Installed from the same durable state root
    /// as `gateway_peer_keys`; absent for explicitly ephemeral gateways.
    remote_host_lifecycle:
        std::sync::RwLock<Option<Arc<crate::runtime::remote_host::RemoteHostLifecycle>>>,
    /// A corrupt/unreadable pairing file is retained as a typed refusal. The
    /// runtime may continue serving local work, but remote placement cannot
    /// silently start from an empty pin set.
    remote_host_lifecycle_error:
        std::sync::RwLock<Option<crate::runtime::remote_host::HostPairingError>>,
    /// Sole reconnect task. Probes are observations only and are aborted and
    /// joined before the control listener and mob authority shut down.
    remote_host_reconnect_task: tokio::sync::Mutex<Option<JoinHandle<()>>>,

    // Identity-first session bridge
    session_bridge: Option<Arc<dyn crate::identity_first::bridge::SessionBridge>>,
    identity_first_context: Option<Arc<crate::identity_first::IdentityFirstRuntimeContext>>,

    // Optional ABAC enforcement shared by the console/SSE surfaces.
    access_controller: Option<crate::access::AccessController>,

    // Optional product-level topology authority. The controller always
    // exists so query can remain available, but its policy defaults to
    // disabled and mutation methods are then absent/denied.
    topology_controller: crate::topology_control::TopologyController,

    // Optional panel-capable store handle backing the console Memory
    // panel's read-only RPCs (§9.3). Any provider advertising
    // `MemoryPanelStore` serves it (M4 de-weld). Interior-mutable so
    // gateways can wire it after the runtime is shared (`Arc`), wherever
    // the store is constructed.
    memory_panel_store:
        std::sync::RwLock<Option<Arc<dyn crate::memory::capabilities::MemoryPanelStore>>>,
    /// Rebuildable detached-job health/status projection supplied by the
    /// host that owns the canonical Meerkat job service.
    job_health_projection: Arc<std::sync::RwLock<Option<serde_json::Value>>>,
    // Realm-scoped WorkGraph service backing the `mobkit/workgraph/*` RPC
    // group and the console experience section. Seeded from the bootstrap
    // spec and deliberately FIXED from then on: the admission guards
    // (cross-process sidecar + agent tool-plane slots) freeze at
    // `MobRuntime::bootstrap`, so a service wired in later would run
    // guard-degraded. The spec (`MobBootstrapSpec::with_workgraph_service`
    // plus admission slot/sidecar) is the only blessed wiring.
    workgraph_service: Option<meerkat::WorkGraphService>,
    /// Runtime-owned lossy wake fan-out. It exists exactly when the
    /// authoritative WorkGraph service exists and is never an authority of
    /// its own.
    workgraph_fact_hub: Option<crate::workgraph_events::WorkGraphFactHub>,
    /// Sole WorkGraph cursor tail feeding `workgraph_fact_hub`. It waits on a
    /// subscriber notification without reading the store while idle.
    workgraph_fact_tail_task: tokio::sync::Mutex<WorkGraphFactTailTask>,
    /// Identity-first console gateways: the mutable desired-identity roster
    /// that `mobkit/ensure_member` extends at runtime (ask K0). Set by the
    /// host beside `attach_identity_first_context`.
    console_identity_roster:
        std::sync::RwLock<Option<Arc<crate::identity_first::MutableRosterProvider>>>,
    /// §16 Q1 provisional operator keying: the console-principal resolver,
    /// shared between the memory coordinator (reads) and the console send
    /// path (notes interactions). `&self`-settable like the panel store.
    console_operator_resolver: std::sync::RwLock<
        Option<Arc<crate::memory::coordinator::ConsolePrincipalOperatorResolver>>,
    >,

    // Mobkit-side label sidecar for mob- and run-scoped metadata
    metadata_table: Arc<RuntimeMetadataTable>,

    // Persistent metadata adapter (currently used for the structural-events
    // subscription cursor). Falls back to `InMemoryMetadataStore` when not
    // explicitly configured — see `UnifiedRuntimeBuilder::persistent_metadata`.
    persistent_metadata: Arc<dyn PersistentMetadataStore>,
}

enum MobEventIngress {
    Forwarder(MobEventForwarder),
}

struct MobEventForwarder {
    event_rx: Receiver<ForwardedMemberEvent>,
    task: JoinHandle<()>,
    identity_stream_health_task: JoinHandle<()>,
}

/// A forwarded member event: the wire-facing unified envelope plus a
/// drain-side alert extracted at ingest, while the member `AgentEvent` was
/// still typed. The alert never reaches the wire — the drain fires it on
/// the error hook (which gateways install after construction, so the
/// forwarder task cannot capture it) and forwards only the envelope.
struct ForwardedMemberEvent {
    envelope: EventEnvelope<UnifiedEvent>,
    alert: Option<ErrorEvent>,
}

struct WorkGraphFactTailTask(Option<JoinHandle<()>>);

impl WorkGraphFactTailTask {
    fn take(&mut self) -> Option<JoinHandle<()>> {
        self.0.take()
    }
}

impl Drop for WorkGraphFactTailTask {
    fn drop(&mut self) {
        // A JoinHandle detaches on drop. Failed construction paths do not get
        // an async shutdown boundary, so abort here; graceful shutdown takes
        // the handle first and also joins it.
        if let Some(task) = self.0.take() {
            task.abort();
        }
    }
}

impl UnifiedRuntime {
    pub fn builder() -> UnifiedRuntimeBuilder {
        UnifiedRuntimeBuilder::default()
    }

    #[allow(
        unknown_lints,
        clippy::unused_async_trait_impl,
        reason = "preserve the async construction seam used by runtime bootstrap"
    )]
    pub(crate) async fn from_parts(
        mob_runtime: MobRuntime,
        module_runtime: MobkitRuntimeHandle,
        persistent_metadata: Arc<dyn PersistentMetadataStore>,
    ) -> Self {
        // Construct the metadata table first so the structural-events store
        // can be wired with it — every projected envelope picks up the
        // matching mob/run labels at projection time.
        let metadata_table = Arc::new(RuntimeMetadataTable::new());
        let mob_events_store = MobEventsStore::new().with_metadata_table(metadata_table.clone());
        let identity_runtime_authority = Arc::new(std::sync::RwLock::new(None));
        let mob_event_ingress = Some(Self::create_event_ingress(
            mob_runtime.handle(),
            mob_runtime.agent_mob_mcp_state(),
            mob_events_store.clone(),
            Arc::clone(&identity_runtime_authority),
        ));
        let mob_events_task = Self::spawn_mob_events_subscriber(
            mob_runtime.handle(),
            mob_events_store.clone(),
            persistent_metadata.clone(),
        );
        // The probe starts at construction while the error hook is installed
        // later (`set_error_hook`), so it holds the late-bound slot and reads
        // the hook at fire time.
        let error_hook: SharedErrorHook = Arc::new(std::sync::RwLock::new(None));
        let actor_loop_probe_task =
            Self::spawn_actor_loop_probe(mob_runtime.handle(), Arc::clone(&error_hook));
        let console_events = ConsoleEventStore::new();
        // Agent-tool spawns (mob_spawn_member/delegate) project their members
        // into this runtime's console event store so spawned workers are
        // visible in the console without embedder-side workarounds.
        mob_runtime.install_console_spawn_sink(crate::console_spawn::ConsoleSpawnSink::new(
            console_events.clone(),
        ));
        let workgraph_service = mob_runtime.workgraph_service();
        let (workgraph_fact_hub, workgraph_fact_tail_task) =
            if let Some(service) = workgraph_service.clone() {
                let hub = crate::workgraph_events::WorkGraphFactHub::new();
                let task = crate::workgraph_events::spawn_workgraph_fact_tail(
                    service,
                    hub.clone(),
                    crate::workgraph_events::WorkGraphFactTailOptions::default(),
                );
                (Some(hub), Some(task))
            } else {
                (None, None)
            };
        let definition_edge_discovery =
            edge_reconcile::DefinitionWiringEdgeDiscovery::from_definition(
                mob_runtime.handle().definition(),
            )
            .map(|policy| Arc::new(policy) as Arc<dyn EdgeDiscovery>);
        Self {
            mob_runtime,
            post_spawn_hook: None,
            post_reconcile_hook: None,
            error_hook,
            drain_timeout: DEFAULT_DRAIN_TIMEOUT,
            discovery: None,
            // Default the edge policy to the definition's declared wiring
            // (auto_wire_orchestrator / role_wiring): upstream applies those
            // rules only at spawn time and only from the non-orchestrator
            // side, so bring-up order and restarts leave declared crews
            // unwired (HomeCore, 2026-07-09). With the default installed,
            // `reconcile_edges` converges the roster onto the declaration;
            // embedder-supplied policies (builder) override it.
            edge_discovery: definition_edge_discovery,
            module_runtime: Arc::new(tokio::sync::Mutex::new(module_runtime)),
            managed_dynamic_edges: Arc::new(tokio::sync::RwLock::new(BTreeSet::new())),
            shutting_down: AtomicBool::new(false),
            mob_event_ingress: tokio::sync::Mutex::new(mob_event_ingress),
            bootstrap_edges_report: tokio::sync::RwLock::new(None),
            event_log: None,
            console_log_store: Arc::new(InMemoryConsoleLogStore::new()),
            console_events,
            mob_events: mob_events_store,
            mob_events_subscriber_task: tokio::sync::Mutex::new(mob_events_task),
            actor_loop_probe_task: tokio::sync::Mutex::new(actor_loop_probe_task),
            implicit_delegate_retirement_task: tokio::sync::Mutex::new(None),
            implicit_delegate_identity_runtime: identity_runtime_authority,
            identity_lease_renewal_task: tokio::sync::Mutex::new(None),
            identity_continuity_repair_task: tokio::sync::Mutex::new(None),
            retired_supervisor_cleanups: tokio::sync::Mutex::new(tokio::task::JoinSet::new()),
            agent_memory_observer_task: tokio::sync::Mutex::new(None),
            agent_memory_steward_task: tokio::sync::Mutex::new(None),
            contact_directory: None,
            peer_mob_handles: tokio::sync::RwLock::new(BTreeMap::new()),
            cross_mob_control_task: tokio::sync::Mutex::new(None),
            cross_mob_control_advertised: std::sync::RwLock::new(None),
            gateway_peer_keys: crate::runtime::cross_mob_control::unsigned_control_signer(),
            remote_host_facts: crate::runtime::cross_mob_control::empty_host_facts_provider(),
            remote_host_lifecycle: std::sync::RwLock::new(None),
            remote_host_lifecycle_error: std::sync::RwLock::new(None),
            remote_host_reconnect_task: tokio::sync::Mutex::new(None),
            session_bridge: None,
            identity_first_context: None,
            access_controller: None,
            topology_controller: crate::topology_control::TopologyController::default(),
            memory_panel_store: std::sync::RwLock::new(None),
            job_health_projection: Arc::new(std::sync::RwLock::new(None)),
            workgraph_service,
            workgraph_fact_hub,
            workgraph_fact_tail_task: tokio::sync::Mutex::new(WorkGraphFactTailTask(
                workgraph_fact_tail_task,
            )),
            console_identity_roster: std::sync::RwLock::new(None),
            console_operator_resolver: std::sync::RwLock::new(None),
            metadata_table,
            persistent_metadata,
        }
    }

    /// Spawn a background task that opens a streaming subscription to
    /// the meerkat mob event ledger and projects each [`MobEvent`] into
    /// the runtime's [`MobEventsStore`]. The task resumes from the
    /// last-projected cursor recorded in `persistent_metadata`, so the
    /// SDK-side cursor is durable across mobkit restarts on
    /// SQLite-backed deployments.
    ///
    /// Returns `None` when there is no current tokio runtime (e.g. unit
    /// tests outside an async context); in that case the store is still
    /// usable via direct projection.
    fn spawn_mob_events_subscriber(
        handle: MobHandle,
        store: MobEventsStore,
        persistent_metadata: Arc<dyn PersistentMetadataStore>,
    ) -> Option<JoinHandle<()>> {
        let runtime_handle = tokio::runtime::Handle::try_current().ok()?;
        Some(runtime_handle.spawn(run_mob_events_subscription(
            handle,
            store,
            persistent_metadata,
        )))
    }

    /// Spawn the actor-loop liveness probe (see [`run_actor_loop_probe`]).
    ///
    /// The probe round trip is `MobHandle::status()` → `MobCommand::QueryPhase`:
    /// the cheapest command the handle exposes — the actor's handler is a pure
    /// in-memory phase read (`reply_tx.send(Ok(self.state()))`), read-only and
    /// O(1) — while still riding the same serialized command loop whose stall
    /// it exists to detect.
    ///
    /// Returns `None` when there is no current tokio runtime (e.g. unit
    /// tests outside an async context), like the events subscriber above.
    fn spawn_actor_loop_probe(
        handle: MobHandle,
        error_hook: SharedErrorHook,
    ) -> Option<JoinHandle<()>> {
        let runtime_handle = tokio::runtime::Handle::try_current().ok()?;
        Some(runtime_handle.spawn(run_actor_loop_probe(
            move || {
                let handle = handle.clone();
                async move { handle.status().await }
            },
            error_hook,
            actor_loop_probe_interval(),
            actor_loop_probe_budget(),
        )))
    }

    pub async fn bootstrap(
        mob_spec: MobBootstrapSpec,
        module_config: MobKitConfig,
        timeout: Duration,
    ) -> Result<Self, UnifiedRuntimeBootstrapError> {
        Box::pin(Self::bootstrap_with_options(
            mob_spec,
            module_config,
            Vec::new(),
            timeout,
            RuntimeOptions::default(),
            Arc::new(InMemoryMetadataStore::new()),
        ))
        .await
    }

    pub async fn bootstrap_with_options(
        mob_spec: MobBootstrapSpec,
        module_config: MobKitConfig,
        module_agent_events: Vec<EventEnvelope<UnifiedEvent>>,
        timeout: Duration,
        options: RuntimeOptions,
        persistent_metadata: Arc<dyn PersistentMetadataStore>,
    ) -> Result<Self, UnifiedRuntimeBootstrapError> {
        Self::bootstrap_with_options_and_topology(
            mob_spec,
            module_config,
            module_agent_events,
            timeout,
            options,
            persistent_metadata,
            crate::topology_control::TopologyBootstrapConfig::default(),
        )
        .await
    }

    /// Bootstrap the legacy runtime with the ordinary defaults plus an
    /// explicit optional topology-control configuration.
    ///
    /// This is the concise opt-in for embedders that do not otherwise need
    /// custom module events, runtime options, or metadata storage.
    pub async fn bootstrap_with_topology(
        mob_spec: MobBootstrapSpec,
        module_config: MobKitConfig,
        timeout: Duration,
        topology: crate::topology_control::TopologyBootstrapConfig,
    ) -> Result<Self, UnifiedRuntimeBootstrapError> {
        Self::bootstrap_with_options_and_topology(
            mob_spec,
            module_config,
            Vec::new(),
            timeout,
            RuntimeOptions::default(),
            Arc::new(InMemoryMetadataStore::new()),
            topology,
        )
        .await
    }

    /// Legacy bootstrap with an explicit optional topology-control seam.
    ///
    /// The default remains query-only with mutation disabled. Supplying an
    /// editable policy does not bypass console authentication or ABAC; every
    /// RPC mutation is still authorized against both endpoint resources.
    /// Supplying `state_path` makes desired additions, suppression tombstones,
    /// revisions, idempotency records, and recovery journals durable.
    #[allow(clippy::too_many_arguments)]
    pub async fn bootstrap_with_options_and_topology(
        mob_spec: MobBootstrapSpec,
        module_config: MobKitConfig,
        module_agent_events: Vec<EventEnvelope<UnifiedEvent>>,
        timeout: Duration,
        options: RuntimeOptions,
        persistent_metadata: Arc<dyn PersistentMetadataStore>,
        topology: crate::topology_control::TopologyBootstrapConfig,
    ) -> Result<Self, UnifiedRuntimeBootstrapError> {
        let topology_authority = mob_spec.definition.id.to_string();
        let topology_controller = match topology.state_path {
            Some(path) => {
                crate::topology_control::TopologyController::load_or_default(topology.policy, path)
            }
            None => crate::topology_control::TopologyController::new(topology.policy),
        }
        .map_err(|error| UnifiedRuntimeBootstrapError::Topology(error.to_string()))?;
        topology_controller
            .bind_authority(topology_authority)
            .await
            .map_err(|error| UnifiedRuntimeBootstrapError::Topology(error.to_string()))?;
        let mob_runtime = MobRuntime::bootstrap(mob_spec)
            .await
            .map_err(UnifiedRuntimeBootstrapError::Mob)?;
        let runtime_options = options.clone();
        let module_start_result = std::thread::spawn(move || {
            start_mobkit_runtime_with_options(module_config, module_agent_events, timeout, options)
        })
        .join();

        match module_start_result {
            Ok(Ok(module_runtime)) => {
                let mut runtime =
                    Self::from_parts(mob_runtime, module_runtime, persistent_metadata).await;
                runtime.topology_controller = topology_controller;
                runtime
                    .configure_implicit_delegate_retirement(&runtime_options)
                    .await;
                if runtime.edge_discovery.is_some()
                    || runtime.topology_controller.revision().await > 0
                    || runtime.topology_controller.has_pending().await
                {
                    let report = runtime.reconcile_edges().await;
                    *runtime.bootstrap_edges_report.write().await = Some(report);
                }
                Ok(runtime)
            }
            Ok(Err(error)) => {
                let startup_error = UnifiedRuntimeBootstrapError::Module(error);
                Self::rollback_mob_runtime(mob_runtime, startup_error).await
            }
            Err(_) => {
                let startup_error = UnifiedRuntimeBootstrapError::ModuleStartupThreadPanicked;
                Self::rollback_mob_runtime(mob_runtime, startup_error).await
            }
        }
    }

    /// Bootstrap edge reconciliation report, if edge discovery was configured.
    ///
    /// Inspect after `build()` to detect incomplete startup topology.
    /// Returns `None` if no edge discovery was configured.
    pub async fn bootstrap_edges_report(&self) -> Option<UnifiedRuntimeReconcileEdgesReport> {
        self.bootstrap_edges_report.read().await.clone()
    }

    /// Register an error hook after construction. Useful when the runtime
    /// is built via `bootstrap()` rather than the builder.
    pub fn set_error_hook(&mut self, hook: ErrorHook) {
        *self
            .error_hook
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(hook.clone());
        if let Some(identity_runtime) = self.identity_runtime() {
            identity_runtime.set_error_hook(Some(hook));
        }
    }

    /// Start the event log ingestion engine. Must be called after
    /// construction (the builder calls this automatically when event_log
    /// config is provided).
    pub fn start_event_log(&mut self, config: EventLogConfig) {
        let handle = event_log::start_event_log(config, current_error_hook(&self.error_hook));
        self.event_log = Some(handle);
    }

    pub(crate) fn console_events(&self) -> ConsoleEventStore {
        self.console_events.clone()
    }

    /// A §9.3 memory-event sink projecting typed memory-plane events onto
    /// the console timeline (standard `ConsoleIdentityEventEnvelope`,
    /// `event_type = "memory.*"`). Must be called from async context — the
    /// sink captures the current runtime handle so sync emitters
    /// (store/taint/guard code) can fire-and-forget.
    pub fn memory_event_sink(&self) -> Arc<dyn crate::memory::events::MemoryEventSink> {
        Arc::new(ConsoleMemoryEventSink {
            store: self.console_events(),
            handle: tokio::runtime::Handle::current(),
        })
    }

    /// Register an observer for gating pending-entry resolutions
    /// (decisions and timeout fallbacks) — the seam the memory steward's
    /// gated promotions commit through (§10.2).
    pub async fn register_gating_resolution_observer(
        &self,
        observer: Arc<dyn crate::runtime::GatingResolutionObserver>,
    ) {
        self.module_runtime
            .lock()
            .await
            .register_gating_resolution_observer(observer);
    }

    /// Internal accessor used by console-facing RPC routers to share the
    /// in-memory structural mob events store without holding a full
    /// runtime reference.
    pub(crate) fn mob_events_store(&self) -> MobEventsStore {
        self.mob_events.clone()
    }

    pub fn binary_blob_store(&self) -> Option<Arc<dyn crate::blob_store::BinaryBlobStore>> {
        self.mob_runtime.binary_blob_store()
    }

    /// Publish the latest rebuildable detached-job observability projection.
    ///
    /// Lifecycle remains owned by Meerkat's generated job machine; this slot
    /// exists only so status, capability, console, and health surfaces can
    /// expose the host-owned projection without a parallel semantic store.
    pub fn set_job_health_projection(&self, projection: Option<serde_json::Value>) {
        *self
            .job_health_projection
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = projection;
    }

    pub fn job_health_projection(&self) -> Option<serde_json::Value> {
        self.job_health_projection
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }

    pub(crate) fn module_runtime_handle(&self) -> Arc<tokio::sync::Mutex<MobkitRuntimeHandle>> {
        Arc::clone(&self.module_runtime)
    }

    pub(crate) fn mobpack_runtime_catalog_state_snapshot(
        &self,
    ) -> crate::mobpack::MobpackRuntimeCatalogState {
        let loaded_modules = self
            .module_runtime
            .try_lock()
            .map(|runtime| runtime.loaded_modules())
            .unwrap_or_default();
        let has_peer_mob_handles = self
            .peer_mob_handles
            .try_read()
            .map(|handles| !handles.is_empty())
            .unwrap_or(false);
        let mut runtime_methods = vec![
            "mobkit/capabilities".to_string(),
            "mobkit/models/catalog".to_string(),
            "mobkit/spawn_member".to_string(),
            "mobkit/list_members".to_string(),
            "mobkit/get_member".to_string(),
            "mobkit/run_flow".to_string(),
            "mobkit/list_flows".to_string(),
            "mobkit/list_runs".to_string(),
        ];
        runtime_methods.extend(
            crate::rpc::MOBPACK_AUTHORING_METHODS
                .iter()
                .map(std::string::ToString::to_string),
        );
        if self.has_contact_directory() {
            runtime_methods.push("mobkit/cross_mob/directory".to_string());
        }
        if (has_peer_mob_handles && self.has_inproc_contacts()) || self.has_remote_contacts() {
            runtime_methods.extend([
                "mobkit/cross_mob/wire".to_string(),
                "mobkit/cross_mob/unwire".to_string(),
                "mobkit/cross_mob/send".to_string(),
            ]);
        }
        crate::mobpack::MobpackRuntimeCatalogState {
            loaded_modules,
            runtime_methods,
            has_contact_directory: self.has_contact_directory(),
            has_peer_mob_handles,
            has_inproc_contacts: self.has_inproc_contacts(),
            runtime_flow_rows: crate::mobpack::runtime_flow_registry_rows_from_definition(
                self.mob_handle().definition(),
            ),
            runtime_agent_definition_sources:
                crate::mobpack::runtime_agent_definition_sources_from_definition(
                    self.mob_handle().definition(),
                ),
            runtime_skill_realms: crate::mobpack::runtime_skill_realms_from_definition(
                self.mob_handle().definition(),
            ),
        }
    }

    /// Return the session bridge for identity-first operations, if configured.
    pub fn session_bridge(&self) -> Option<&Arc<dyn crate::identity_first::bridge::SessionBridge>> {
        self.session_bridge.as_ref()
    }

    pub fn identity_first_context(
        &self,
    ) -> Option<&Arc<crate::identity_first::IdentityFirstRuntimeContext>> {
        self.identity_first_context.as_ref()
    }

    pub fn identity_runtime(&self) -> Option<&Arc<crate::identity_first::IdentityRuntime>> {
        self.identity_first_context.as_ref().map(|ctx| &ctx.runtime)
    }

    pub async fn remember_agent_memory(
        &self,
        realm: &str,
        identity: &crate::identity_first::AgentIdentity,
        memory: crate::identity_first::NewAgentMemory,
    ) -> Result<crate::identity_first::AgentMemoryRecord, crate::identity_first::AgentMemoryError>
    {
        let runtime = self.identity_runtime().ok_or_else(|| {
            crate::identity_first::AgentMemoryError::InvalidConfig(
                "identity-first runtime is not configured".to_string(),
            )
        })?;
        runtime.remember_agent_memory(realm, identity, memory).await
    }

    pub async fn recall_agent_memory(
        &self,
        request: crate::identity_first::AgentMemoryRecallRequest,
    ) -> Result<
        Vec<crate::identity_first::AgentMemoryRecord>,
        crate::identity_first::AgentMemoryError,
    > {
        let runtime = self.identity_runtime().ok_or_else(|| {
            crate::identity_first::AgentMemoryError::InvalidConfig(
                "identity-first runtime is not configured".to_string(),
            )
        })?;
        runtime.recall_agent_memory(request).await
    }

    pub async fn forget_agent_memory(
        &self,
        realm: &str,
        identity: &crate::identity_first::AgentIdentity,
        memory_id: &str,
    ) -> Result<
        crate::identity_first::AgentMemoryForgetResult,
        crate::identity_first::AgentMemoryError,
    > {
        let runtime = self.identity_runtime().ok_or_else(|| {
            crate::identity_first::AgentMemoryError::InvalidConfig(
                "identity-first runtime is not configured".to_string(),
            )
        })?;
        runtime
            .forget_agent_memory(realm, identity, memory_id)
            .await
    }

    pub fn attach_identity_first_context(
        &mut self,
        context: Arc<crate::identity_first::IdentityFirstRuntimeContext>,
    ) {
        self.install_identity_first_context_authority(context);
        self.start_identity_first_supervisors();
    }

    /// Install identity authority before applying the initial roster.
    ///
    /// The gateway builds its base [`UnifiedRuntime`] before callback-backed
    /// identity providers are available. Identity bootstrap can partially
    /// materialize a roster before a later member fails, so the context must be
    /// visible to [`Self::shutdown`] before bootstrap starts. On failure this
    /// method drives the complete runtime shutdown order before returning the
    /// error; on success it starts the long-lived lease and repair supervisors.
    pub async fn install_and_bootstrap_identity_first_context(
        &mut self,
        context: Arc<crate::identity_first::IdentityFirstRuntimeContext>,
        roster: &[crate::identity_first::DurableAgentSpec],
    ) -> Result<crate::identity_first::RestoreFlowResult, crate::identity_first::IdentityRuntimeError>
    {
        self.install_identity_first_context_authority(Arc::clone(&context));
        match context.bootstrap_roster(roster).await {
            Ok(result) => {
                self.start_identity_first_supervisors();
                Ok(result)
            }
            Err(error) => {
                self.shutdown().await;
                Err(error)
            }
        }
    }

    fn install_identity_first_context_authority(
        &mut self,
        context: Arc<crate::identity_first::IdentityFirstRuntimeContext>,
    ) {
        self.install_identity_first_flow_target_provisioner(&context.runtime);
        self.mob_runtime
            .install_identity_runtime_authority(Arc::clone(&context.runtime));
        *self
            .implicit_delegate_identity_runtime
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) =
            Some(Arc::clone(&context.runtime));
        self.identity_first_context = Some(context);
    }

    pub(crate) fn install_identity_first_flow_target_provisioner(
        &self,
        runtime: &Arc<crate::identity_first::IdentityRuntime>,
    ) {
        let identity_runtime = Arc::downgrade(runtime);
        self.mob_runtime
            .handle()
            .install_flow_target_provisioner(Arc::new(move || {
                let identity_runtime = identity_runtime.clone();
                Box::pin(async move {
                    let runtime = identity_runtime.upgrade().ok_or_else(|| {
                        MobError::Internal(
                            "identity-first flow provisioner is no longer available".to_string(),
                        )
                    })?;
                    runtime
                        .materialize_all_required_tracked()
                        .await
                        .map(|_| ())
                        .map_err(|error| {
                            MobError::Internal(format!(
                                "identity-first flow materialization failed: {error}"
                            ))
                        })
                })
            }));
    }

    fn start_identity_first_supervisors(&mut self) {
        let Some(context) = self.identity_first_context.clone() else {
            return;
        };
        // Gateways attach identity-first after the base UnifiedRuntime has
        // been built, so they do not pass through the builder's supervisor
        // installation below. Active callback/local-provider leases need the
        // same proactive renewal regardless of which construction path is
        // used.
        let lease_task = context.runtime.clone().spawn_tracked_lease_renewal_task();
        if let Some(previous) = self
            .identity_lease_renewal_task
            .get_mut()
            .replace(lease_task)
        {
            previous.cancel();
            self.retain_retired_supervisor_cleanup(
                types::RetiredSupervisorKind::LeaseRenewal,
                previous.cancel_and_join(),
            );
        }
        // Broken identities must self-heal: a rejected resume parks the
        // identity "pending reconcile retry", and this task is what runs
        // that retry in a live process (delivery and materialize both
        // refuse the Broken state by design).
        let repair_task = context.spawn_tracked_broken_identity_repair_task(Default::default());
        if let Some(previous) = self
            .identity_continuity_repair_task
            .get_mut()
            .replace(repair_task)
        {
            previous.cancel();
            self.retain_retired_supervisor_cleanup(
                types::RetiredSupervisorKind::ContinuityRepair,
                previous.cancel_and_join(),
            );
        }
    }

    /// Retain a replacement cleanup so `shutdown()` can join it.
    ///
    /// Stays synchronous on purpose: `attach_identity_first_context` is a sync
    /// public API, so this uses `Mutex::get_mut` (the same `&mut self` idiom the
    /// supervisor slots above use) rather than becoming async. `JoinSet::spawn`
    /// needs a runtime context exactly as the previous `tokio::spawn` did, so
    /// this adds no new requirement on callers.
    fn retain_retired_supervisor_cleanup(
        &mut self,
        kind: types::RetiredSupervisorKind,
        cleanup: impl std::future::Future<Output = ()> + Send + 'static,
    ) {
        let retired = self.retired_supervisor_cleanups.get_mut();
        // A JoinSet does not reap on its own: finished tasks sit in it until
        // something polls them. Without this, a process that re-attaches
        // repeatedly accumulates completed entries for its whole lifetime, and
        // the shutdown count would measure total replacements instead of
        // outstanding work. Non-blocking, so the sync API is preserved.
        while retired.try_join_next().is_some() {}
        retired.spawn(async move {
            cleanup.await;
            kind
        });
    }

    pub async fn refresh_desired_topology(
        &self,
    ) -> Result<
        Option<crate::identity_first::RestoreFlowResult>,
        crate::identity_first::IdentityRuntimeError,
    > {
        match self.identity_first_context.as_ref() {
            Some(ctx) => ctx.refresh_desired_topology_tracked().await.map(Some),
            None => Ok(None),
        }
    }

    /// Hydrate identity-first lazy members before handing control to concrete
    /// mob APIs that operate on already-materialized runtime members.
    pub async fn materialize_identity_first_for_flow(
        &self,
    ) -> Result<
        Vec<crate::identity_first::ContinuityRecord>,
        crate::identity_first::IdentityRuntimeError,
    > {
        match self.identity_runtime() {
            Some(runtime) => runtime.materialize_all_required_tracked().await,
            None => Ok(Vec::new()),
        }
    }

    /// Return the mob/run label sidecar table.
    ///
    /// Mobkit owns this table — meerkat-mob has no concept of mob- or
    /// run-level labels. Apps use it to attach external context (repo,
    /// branch, customer, deployment, environment) to a mob or a flow run.
    pub fn metadata_table(&self) -> &Arc<RuntimeMetadataTable> {
        &self.metadata_table
    }

    /// Install the shared access controller. Console routers built after
    /// this call enforce (and live-serve) the ABAC configuration.
    pub fn set_access_controller(&mut self, controller: crate::access::AccessController) {
        self.access_controller = Some(controller);
    }

    /// Wire the bundled sqlite memory store into the console Memory panel
    /// (§9.3). `&self` deliberately: gateways construct the store next to
    /// the memory subsystem wiring, which may run after the runtime is
    /// `Arc`-shared. Routers built *after* this call serve the panel RPCs.
    pub fn set_console_identity_roster(
        &self,
        roster: Arc<crate::identity_first::MutableRosterProvider>,
    ) {
        *self
            .console_identity_roster
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(roster);
    }

    pub fn console_identity_roster(
        &self,
    ) -> Option<Arc<crate::identity_first::MutableRosterProvider>> {
        self.console_identity_roster
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }

    pub fn set_memory_panel_store(
        &self,
        store: Arc<dyn crate::memory::capabilities::MemoryPanelStore>,
    ) {
        *self
            .memory_panel_store
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(store);
    }

    pub fn memory_panel_store(
        &self,
    ) -> Option<Arc<dyn crate::memory::capabilities::MemoryPanelStore>> {
        self.memory_panel_store
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }

    /// The realm-scoped WorkGraph service backing the `mobkit/workgraph/*`
    /// RPC group and the console experience section, seeded from the
    /// bootstrap spec. There is deliberately NO late setter (round-5 S3): the
    /// admission guards — the cross-process sidecar and the agent tool-plane
    /// slots — freeze at `MobRuntime::bootstrap`, so a service wired in
    /// after the fact would silently run guard-degraded. Wire workgraph
    /// through `MobBootstrapSpec::with_workgraph_service` (plus
    /// `with_workgraph_admission_slot`/`with_workgraph_admission_sidecar`)
    /// or the stock spec constructors, which do all three.
    pub fn workgraph_service(&self) -> Option<meerkat::WorkGraphService> {
        self.workgraph_service.clone()
    }

    /// Clone the runtime's lossy WorkGraph fact hub when WorkGraph is
    /// configured. Every subscriber must begin with a durable pull; the hub
    /// has no replay or state authority.
    pub fn workgraph_fact_hub(&self) -> Option<crate::workgraph_events::WorkGraphFactHub> {
        self.workgraph_fact_hub.clone()
    }

    /// Composition-time storage durability resolution (H1/H2) carried from
    /// the bootstrap spec, reported by `mobkit/status` /
    /// `mobkit/capabilities`. `None` when the spec was composed externally
    /// without a declaration.
    pub fn resolved_storage(&self) -> Option<crate::storage_health::ResolvedStorageSummary> {
        self.mob_runtime.resolved_storage()
    }

    /// The runtime-wide admission authority serializing the workgraph
    /// duplicate-binding guards' check-then-act windows (RPC arms + agent
    /// tool plane). Lives on the mob runtime so console routers (which
    /// capture the mob runtime by value) and the unified stdin dispatch
    /// reach the SAME instance, frozen at bootstrap alongside the service.
    pub(crate) fn workgraph_admission(
        &self,
    ) -> std::sync::Arc<crate::workgraph_admission::WorkGraphAdmission> {
        self.mob_runtime.workgraph_admission()
    }

    /// Wire the §16 Q1 console-principal operator resolver (set by the
    /// gateway's memory wiring when `operator_scope = "provisional"`); the
    /// console send path notes authenticated interactions through it.
    pub fn set_console_operator_resolver(
        &self,
        resolver: Arc<crate::memory::coordinator::ConsolePrincipalOperatorResolver>,
    ) {
        *self
            .console_operator_resolver
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(resolver);
    }

    pub fn console_operator_resolver(
        &self,
    ) -> Option<Arc<crate::memory::coordinator::ConsolePrincipalOperatorResolver>> {
        self.console_operator_resolver
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }

    /// Borrow the shared access controller if one was installed.
    pub fn access_controller(&self) -> Option<&crate::access::AccessController> {
        self.access_controller.as_ref()
    }

    /// Optional topology control-plane policy and durable intent store.
    pub fn topology_controller(&self) -> &crate::topology_control::TopologyController {
        &self.topology_controller
    }

    /// Cloneable topology seam for HTTP/RPC routers.
    pub fn topology_runtime_handle(&self) -> crate::topology_control::TopologyRuntimeHandle {
        crate::topology_control::TopologyRuntimeHandle::new(
            self.mob_handle(),
            self.edge_discovery.clone(),
            Arc::clone(&self.managed_dynamic_edges),
            self.topology_controller.clone(),
            self.identity_first_context.clone(),
        )
    }

    /// Replace the topology policy at runtime. Existing additions and
    /// suppression tombstones remain authoritative; disabling hides/denies
    /// mutation rather than silently discarding desired state.
    pub fn set_topology_control_policy(
        &self,
        policy: crate::topology_control::TopologyControlPolicy,
    ) -> Result<(), crate::topology_control::TopologyControlError> {
        self.topology_controller.set_policy(policy)
    }

    /// Return the persistent metadata adapter — used by the
    /// structural-events subscription to checkpoint its last-projected
    /// cursor. Tests and integration code that need to inspect the
    /// persisted cursor reach through this accessor.
    pub fn persistent_metadata(&self) -> &Arc<dyn PersistentMetadataStore> {
        &self.persistent_metadata
    }

    /// Replace the label set associated with this mob.
    ///
    /// An empty `labels` map clears the entry. Replacement is wholesale —
    /// existing labels not present in `labels` are dropped. To merge,
    /// read first via [`Self::get_mob_labels`] and combine.
    pub async fn set_mob_labels(&self, labels: BTreeMap<String, String>) {
        self.metadata_table
            .set_labels(MetadataScope::Mob(self.mob_id()), labels)
            .await;
    }

    /// Return the label set associated with this mob, or an empty map.
    pub async fn get_mob_labels(&self) -> BTreeMap<String, String> {
        self.metadata_table
            .get_labels(&MetadataScope::Mob(self.mob_id()))
            .await
    }

    /// Remove the label set associated with this mob.
    pub async fn delete_mob_labels(&self) {
        let _ = self
            .metadata_table
            .delete_labels(&MetadataScope::Mob(self.mob_id()))
            .await;
    }

    /// Replace the label set for `run_id` under this mob.
    pub async fn set_run_labels(&self, run_id: &str, labels: BTreeMap<String, String>) {
        self.metadata_table
            .set_labels(
                MetadataScope::Run(self.mob_id(), run_id.to_string()),
                labels,
            )
            .await;
    }

    /// Return the label set for `run_id` under this mob, or an empty map.
    pub async fn get_run_labels(&self, run_id: &str) -> BTreeMap<String, String> {
        self.metadata_table
            .get_labels(&MetadataScope::Run(self.mob_id(), run_id.to_string()))
            .await
    }

    /// Remove the label set for `run_id` under this mob.
    pub async fn delete_run_labels(&self, run_id: &str) {
        let _ = self
            .metadata_table
            .delete_labels(&MetadataScope::Run(self.mob_id(), run_id.to_string()))
            .await;
    }

    /// Return the underlying event log store if one is configured.
    ///
    /// Used to share the store with sub-handlers (e.g. console RPC) that
    /// don't hold a full `UnifiedRuntime` reference.
    pub fn event_log_store(&self) -> Option<std::sync::Arc<dyn event_log::EventLogStore>> {
        self.event_log
            .as_ref()
            .map(event_log::EventLogHandle::store)
    }

    pub fn console_log_store(&self) -> Arc<dyn ConsoleLogStore> {
        self.console_log_store.clone()
    }

    pub fn set_console_log_store(&mut self, store: Arc<dyn ConsoleLogStore>) {
        self.console_log_store = store;
    }

    /// Query structural mob events from the meerkat ledger.
    ///
    /// Returns events filtered by [`EventQuery`] in cursor-ascending
    /// order. `EventQuery::after_seq` acts as the pagination cursor: the
    /// caller passes the highest `cursor` seen so far to receive only
    /// strictly-newer events. Without `after_seq` the call returns the
    /// **latest** matching events up to `limit` (default 256), scanning
    /// the ledger backwards from `latest_cursor`.
    ///
    /// Errors propagate the typed [`mob_events::MobEventsQueryError`]
    /// so the JSON-RPC handler can surface `StaleEventCursor` as code
    /// `-32010`.
    pub async fn query_mob_events(
        &self,
        query: &EventQuery,
    ) -> Result<Vec<MobStructuralEventEnvelope>, mob_events::MobEventsQueryError> {
        let events = self.mob_runtime.handle().events();
        mob_events::query_ledger_with_filter(&events, &self.mob_events, query).await
    }

    /// Subscribe to live structural mob events. Returns a broadcast
    /// receiver that yields each newly-projected envelope. The receiver
    /// will report `RecvError::Lagged` if it falls behind the in-memory
    /// channel cap.
    pub fn subscribe_mob_events(
        &self,
    ) -> tokio::sync::broadcast::Receiver<MobStructuralEventEnvelope> {
        self.mob_events.subscribe()
    }

    /// Ingest an event into the event log (if configured). Non-blocking.
    pub(crate) fn ingest_event(&self, event: &EventEnvelope<UnifiedEvent>) {
        if let Some(ref log) = self.event_log {
            log.ingest(event.clone());
        }
    }

    pub(crate) async fn record_console_lifecycle(
        &self,
        identity: &str,
        event_type: &str,
        data: serde_json::Value,
    ) {
        self.console_events
            .record_lifecycle(identity, event_type, data)
            .await;
    }

    pub async fn reserve_identity_interaction(
        &self,
        identity: &str,
        runtime_member_id: Option<&str>,
        interaction_id: &str,
        origin: &str,
        content: serde_json::Value,
    ) -> Result<(), &'static str> {
        self.console_events
            .reserve_interaction_value(identity, runtime_member_id, interaction_id, origin, content)
            .await
    }

    pub(crate) async fn project_console_event_from_unified(
        &self,
        event: &EventEnvelope<UnifiedEvent>,
    ) {
        self.console_events.project_unified_event(event).await;
    }

    /// Fire an error event to the registered hook, if any.
    /// Truly fire-and-forget — spawns a detached task so slow hooks
    /// (HTTP to Slack, PagerDuty) never block the runtime operation.
    pub(crate) fn fire_error(&self, event: ErrorEvent) {
        fire_error_hook(&self.error_hook, event);
    }

    fn create_event_ingress(
        mob_handle: MobHandle,
        agent_mob_mcp_state: Option<Arc<meerkat_mob_mcp::MobMcpState>>,
        mob_events: MobEventsStore,
        identity_runtime: Arc<
            std::sync::RwLock<Option<Arc<crate::identity_first::IdentityRuntime>>>,
        >,
    ) -> MobEventIngress {
        // Keep forwarding bounded to avoid unbounded memory growth under sustained ingress.
        let (event_tx, event_rx) = tokio::sync::mpsc::channel(256);
        // Identity lifecycle repair must remain live even when an embedding
        // application does not drain the bounded console/event channel.
        // A dedicated subscription monitor drains its streams independently
        // and owns only permanent-loss detection; the ordinary forwarder
        // retains lossless backpressure for user-visible events.
        let identity_stream_health_task = tokio::spawn(run_identity_stream_health_monitor(
            mob_handle.clone(),
            agent_mob_mcp_state.clone(),
            identity_runtime,
        ));
        let task = tokio::spawn(run_resilient_mob_agent_event_forwarder(
            mob_handle,
            agent_mob_mcp_state,
            event_tx,
            mob_events,
        ));
        MobEventIngress::Forwarder(MobEventForwarder {
            event_rx,
            task,
            identity_stream_health_task,
        })
    }

    /// Test seam: replace the live ingress with a caller-owned channel so
    /// tests can push forwarded member events through the real drain path.
    #[cfg(test)]
    async fn install_test_event_ingress(&self) -> Sender<ForwardedMemberEvent> {
        let (event_tx, event_rx) = tokio::sync::mpsc::channel(16);
        let replaced = self
            .mob_event_ingress
            .lock()
            .await
            .replace(MobEventIngress::Forwarder(MobEventForwarder {
                event_rx,
                task: tokio::spawn(async {}),
                identity_stream_health_task: tokio::spawn(async {}),
            }));
        if let Some(MobEventIngress::Forwarder(forwarder)) = replaced {
            forwarder.task.abort();
            forwarder.identity_stream_health_task.abort();
        }
        event_tx
    }

    async fn rollback_mob_runtime(
        mob_runtime: MobRuntime,
        startup_error: UnifiedRuntimeBootstrapError,
    ) -> Result<Self, UnifiedRuntimeBootstrapError> {
        match mob_runtime.handle().stop().await {
            Ok(()) => Err(startup_error),
            Err(err) => Err(UnifiedRuntimeBootstrapError::ModuleStartupRollbackFailed {
                startup_error: Box::new(startup_error),
                rollback_error: MobRuntimeError::from(err),
            }),
        }
    }
}

// The trailing `Option<Arc<str>>` is the member's durable identity label,
// present only for identity-first owned members of the primary mob. The
// identity health monitor needs it to attribute a run completion to the
// durable identity; the console forwarder ignores it.
type TaggedAgentEvent = (
    AgentRuntimeId,
    FenceToken,
    ProfileName,
    meerkat_core::event::EventEnvelope<AgentEvent>,
    Option<Arc<str>>,
);

enum ForwardedAgentEvent {
    Event(Box<TaggedAgentEvent>),
    Closed(TrackedAgentEventStream),
}

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
struct TrackedAgentEventStream {
    mob_id: String,
    /// Trusted durable identity stamped by the identity-first spawn bridge.
    /// Ordinary/child mobs do not carry this label and are never handed to
    /// the primary identity repair authority.
    durable_identity: Option<String>,
    /// Concrete roster identity used to subscribe to Meerkat events.
    member_identity: AgentIdentity,
    runtime_id: AgentRuntimeId,
    /// Identity-authority fencing token captured when the health subscription
    /// was established. This is deliberately distinct from `fence_token`,
    /// which belongs to Meerkat Mob's member-binding fencing domain.
    identity_fencing_token: Option<u64>,
    fence_token: FenceToken,
}
type TaggedAgentEventStream = BoxStream<'static, ForwardedAgentEvent>;

/// Per-member subscribe-failure backoff for agent-event subscriptions.
/// The forwarder and independent identity-health monitor reconcile on
/// machine-state/mob-set change signals (plus a slow safety tick); without
/// backoff a member that keeps failing `subscribe_agent_events` is retried
/// on every wake indefinitely and floods the log (observed: ~49k "failed to
/// subscribe" warnings over 3.4h on a single wedged-retiring alias). Both
/// retry transient failures with exponential backoff; only the independent
/// health monitor hands persistent loss to identity repair.
struct SubscribeBackoff {
    next_attempt: tokio::time::Instant,
    consecutive_failures: u32,
}

/// First retry waits one backoff quantum; subsequent retries double up to a
/// cap so a persistently-unsubscribable member costs at most ~1 attempt per
/// `SUBSCRIBE_BACKOFF_MAX`.
const SUBSCRIBE_BACKOFF_BASE: Duration = Duration::from_millis(250);
const SUBSCRIBE_BACKOFF_MAX: Duration = Duration::from_secs(30);
const PERMANENT_STREAM_FAILURE_THRESHOLD: u32 = 4;

/// Upper bound between reconcile passes when no change signal fires. The
/// reconcilers are event-driven (machine-state watches + managed-mob-set
/// epoch + stream closures + backoff deadlines); this tick only bounds drift
/// from signals they cannot observe — identity-lease fencing-token motion
/// without a machine transition, and membership changes that land inside the
/// unwatched window while a brand-new mob's watcher is being bound. It must
/// stay slow: the historical 250ms tick made every pass's full member
/// projection an idle-CPU driver on restore-scale mobs.
const RECONCILE_SAFETY_INTERVAL: Duration = Duration::from_secs(30);

/// Wakes the stream reconcilers when membership/binding truth may have moved,
/// replacing the historical 250ms polling tick.
///
/// Wake sources, in no priority order:
/// - any tracked mob's [`meerkat_mob::MobMachineStateChanges`] firing (the
///   mob actor publishes on every applied machine input),
/// - the managed mob-set epoch changing (child mob created/removed),
/// - the earliest pending subscribe-backoff deadline,
/// - the [`RECONCILE_SAFETY_INTERVAL`] safety tick.
///
/// Watchers are keyed by mob id and RETAINED across rebinds so their
/// internally-tracked seen-version survives: a state change landing while a
/// reconcile pass runs still wakes the next wait. Only a mob first seen by
/// the previous pass starts a fresh watcher (its pre-bind changes are covered
/// by that same pass's subscription attempt and by the safety tick). Closed
/// watchers (actor gone) are dropped on rebind and on wake so a destroyed mob
/// cannot busy-wake the loop.
struct ReconcileCadence {
    machine_watchers: BTreeMap<String, meerkat_mob::MobMachineStateChanges>,
    mob_set_changes: Option<tokio::sync::watch::Receiver<u64>>,
    /// Absolute deadline for the next safety reconcile, anchored at the last
    /// completed reconcile pass ([`Self::rebind`]). Persisting it here is
    /// load-bearing: the callers' outer `select!` drops and recreates the
    /// [`Self::wait`] future on every forwarded member event, so a deadline
    /// computed inside `wait` would reset under sustained event traffic and
    /// the safety reconcile would never fire.
    next_safety_deadline: tokio::time::Instant,
}

impl ReconcileCadence {
    fn new(agent_mob_mcp_state: &Option<Arc<meerkat_mob_mcp::MobMcpState>>) -> Self {
        Self {
            machine_watchers: BTreeMap::new(),
            mob_set_changes: agent_mob_mcp_state
                .as_ref()
                .map(|state| state.mob_set_changes()),
            next_safety_deadline: tokio::time::Instant::now() + RECONCILE_SAFETY_INTERVAL,
        }
    }

    /// Rebind the watcher set to the exact handles the reconcile pass just
    /// enumerated, keeping existing watchers (and their seen-versions) alive.
    /// Every reconcile pass ends here, so this is also where the safety
    /// deadline is re-armed: drift from watch-invisible signals is bounded
    /// relative to the last reconcile, not the last wake attempt.
    fn rebind(&mut self, handles: &[MobHandle]) {
        let mut next = BTreeMap::new();
        for handle in handles {
            let key = handle.mob_id().to_string();
            let watcher = self
                .machine_watchers
                .remove(&key)
                .unwrap_or_else(|| handle.machine_state_changes());
            if !watcher.is_closed() {
                next.insert(key, watcher);
            }
        }
        self.machine_watchers = next;
        self.next_safety_deadline = tokio::time::Instant::now() + RECONCILE_SAFETY_INTERVAL;
    }

    /// Wait for the next reconcile trigger. `next_backoff_attempt` is the
    /// earliest pending subscribe retry, if any.
    async fn wait(&mut self, next_backoff_attempt: Option<tokio::time::Instant>) {
        let now = tokio::time::Instant::now();
        let mut deadline = self.next_safety_deadline;
        if let Some(attempt) = next_backoff_attempt {
            deadline = deadline.min(attempt.max(now));
        }

        let Self {
            machine_watchers,
            mob_set_changes,
            ..
        } = self;

        // Await "any machine watcher fired". A closed watcher is removed
        // in-place so it cannot immediately re-wake the caller.
        let machine_change = async {
            if machine_watchers.is_empty() {
                std::future::pending::<()>().await;
                return;
            }
            let keys: Vec<String> = machine_watchers.keys().cloned().collect();
            let closed_key = {
                let futures: Vec<_> = machine_watchers
                    .values_mut()
                    .map(|watcher| Box::pin(watcher.changed()))
                    .collect();
                let (result, index, rest) = futures::future::select_all(futures).await;
                drop(rest);
                result.is_err().then(|| keys[index].clone())
            };
            if let Some(key) = closed_key {
                machine_watchers.remove(&key);
            }
        };

        let mob_set_change = async {
            match mob_set_changes.as_mut() {
                Some(rx) => rx.changed().await,
                None => std::future::pending().await,
            }
        };

        let mob_set_closed = tokio::select! {
            () = machine_change => false,
            result = mob_set_change => result.is_err(),
            () = tokio::time::sleep_until(deadline) => false,
        };
        if mob_set_closed {
            // The dispatcher state is gone; a closed watch completes
            // immediately, so it must not stay selectable.
            self.mob_set_changes = None;
        }
    }
}

/// Earliest pending subscribe-backoff deadline, if any member is waiting.
fn earliest_backoff_attempt(
    subscribe_failures: &HashMap<TrackedAgentEventStream, SubscribeBackoff>,
) -> Option<tokio::time::Instant> {
    subscribe_failures
        .values()
        .map(|backoff| backoff.next_attempt)
        .min()
}

fn subscribe_backoff_delay(consecutive_failures: u32) -> Duration {
    SUBSCRIBE_BACKOFF_BASE
        .saturating_mul(1u32 << consecutive_failures.min(7))
        .min(SUBSCRIBE_BACKOFF_MAX)
}

/// Whether the console forwarder should hold a live agent-event subscription
/// for a member in this lifecycle state. Only `Active` members have a live
/// runtime delta stream; subscribing a `Retiring`/`Broken`/`Completed` member
/// (which can still carry stale binding atoms) fails every reconcile tick.
fn forwarder_should_subscribe(status: MobMemberStatus) -> bool {
    matches!(status, MobMemberStatus::Active)
}

fn durable_identity_label(labels: &BTreeMap<String, String>) -> Option<String> {
    labels.get("agent_identity").cloned()
}

async fn current_identity_fencing_token(
    primary_mob_id: &str,
    mob_id: &str,
    durable_identity: Option<&str>,
    identity_runtime: Option<
        &Arc<std::sync::RwLock<Option<Arc<crate::identity_first::IdentityRuntime>>>>,
    >,
) -> Option<u64> {
    if mob_id != primary_mob_id {
        return None;
    }
    let durable_identity = durable_identity?;
    let identity_runtime = identity_runtime?;
    let authority = identity_runtime
        .read()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone()?;
    let identity = crate::identity_first::AgentIdentity::parse(durable_identity).ok()?;
    authority
        .status(&identity)
        .await
        .ok()?
        .lease
        .map(|lease| lease.fencing_token.get())
}

/// Advance an identity's completion cursor when meerkat reports that one of
/// its turns finished.
///
/// This is the only place the cursor moves in production, and it is driven by
/// the run-completion EVENT rather than by polling a projection on purpose: a
/// poll cannot distinguish "new turn, byte-identical output" from "no new
/// turn", which is exactly the defect the cursor closes. The identity health
/// monitor is the right host because it drains its own subscription set — a
/// full console channel cannot starve it.
///
/// Losing the subscription mid-turn means a missed completion, so the cursor
/// under-counts rather than over-counts: a waiter times out instead of being
/// told a turn finished that did not.
async fn record_identity_turn_completion(
    identity_runtime: &Arc<std::sync::RwLock<Option<Arc<crate::identity_first::IdentityRuntime>>>>,
    durable_identity: Option<&str>,
    envelope: &meerkat_core::event::EventEnvelope<AgentEvent>,
) {
    if !matches!(envelope.payload, AgentEvent::RunCompleted { .. }) {
        return;
    }
    let Some(durable_identity) = durable_identity else {
        return;
    };
    let authority = identity_runtime
        .read()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone();
    let Some(authority) = authority else {
        return;
    };
    let identity = match crate::identity_first::AgentIdentity::parse(durable_identity) {
        Ok(identity) => identity,
        Err(error) => {
            tracing::debug!(
                identity = %durable_identity,
                error = %error,
                "mobkit identity health monitor: run completion carried an unparseable durable identity"
            );
            return;
        }
    };
    authority.record_turn_completed(&identity).await;
}

async fn trigger_identity_stream_repair(
    primary_mob_id: &str,
    tracked_key: &TrackedAgentEventStream,
    identity_runtime: &Arc<std::sync::RwLock<Option<Arc<crate::identity_first::IdentityRuntime>>>>,
    detail: &str,
) {
    if tracked_key.mob_id != primary_mob_id {
        return;
    }
    let Some(durable_identity) = tracked_key.durable_identity.as_deref() else {
        return;
    };
    let Some(identity_fencing_token) = tracked_key.identity_fencing_token else {
        return;
    };
    let runtime_alias =
        crate::member_comms_id::runtime_alias_str(tracked_key.runtime_id.identity.as_str())
            .into_owned();
    let authority = identity_runtime
        .read()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone();
    let Some(authority) = authority else {
        return;
    };
    let identity = match crate::identity_first::AgentIdentity::parse(durable_identity) {
        Ok(identity) => identity,
        Err(error) => {
            tracing::warn!(
                identity = %durable_identity,
                error = %error,
                "mobkit agent event forwarder: roster identity cannot be mapped to identity authority"
            );
            return;
        }
    };
    if let Err(error) = authority
        .mark_active_runtime_broken(&identity, &runtime_alias, identity_fencing_token, detail)
        .await
    {
        tracing::warn!(
            identity = %identity,
            runtime_id = %tracked_key.runtime_id,
            error = %error,
            "mobkit agent event forwarder: failed to trigger identity repair after permanent stream loss"
        );
    }
}

async fn run_resilient_mob_agent_event_forwarder(
    handle: MobHandle,
    agent_mob_mcp_state: Option<Arc<meerkat_mob_mcp::MobMcpState>>,
    event_tx: Sender<ForwardedMemberEvent>,
    mob_events: MobEventsStore,
) {
    let mut streams: SelectAll<TaggedAgentEventStream> = SelectAll::new();
    let mut tracked = HashSet::new();
    let mut subscribe_failures: HashMap<TrackedAgentEventStream, SubscribeBackoff> = HashMap::new();
    let mut cadence = ReconcileCadence::new(&agent_mob_mcp_state);

    let handles = Box::pin(reconcile_agent_event_streams(
        &handle,
        &agent_mob_mcp_state,
        &mut tracked,
        &mut subscribe_failures,
        &mut streams,
        None,
    ))
    .await;
    cadence.rebind(&handles);

    loop {
        tokio::select! {
            Some(forwarded) = streams.next() => {
                match forwarded {
                    ForwardedAgentEvent::Event(event) => {
                        let (source, source_fence_token, role, envelope, _durable_identity) = *event;
                        let attributed_event = AttributedEvent {
                            source,
                            source_fence_token,
                            role,
                            envelope,
                        };
                        // Fan out to the structural mob events store. Today this is a
                        // no-op for attributed agent events (they don't carry mob/run/
                        // step fields), but the projection seam keeps the surface
                        // symmetric with the structural `MobEvent` subscriber and lets
                        // future code add attribution without touching this shape.
                        let _ = mob_events.project_attributed_event(&attributed_event).await;
                        if event_tx
                            .send(forwarded_member_event(attributed_event))
                            .await
                            .is_err()
                        {
                            break;
                        }
                    }
                    ForwardedAgentEvent::Closed(tracked_key) => {
                        tracked.remove(&tracked_key);
                        subscribe_failures.remove(&tracked_key);
                        // A closure is itself the re-subscribe trigger: the
                        // member may still be live (stream lag/teardown race),
                        // and no machine transition is guaranteed to follow.
                        let handles = Box::pin(reconcile_agent_event_streams(&handle, &agent_mob_mcp_state, &mut tracked, &mut subscribe_failures, &mut streams, None)).await;
                        cadence.rebind(&handles);
                    }
                }
            }
            () = cadence.wait(earliest_backoff_attempt(&subscribe_failures)) => {
                let handles = Box::pin(reconcile_agent_event_streams(&handle, &agent_mob_mcp_state, &mut tracked, &mut subscribe_failures, &mut streams, None)).await;
                cadence.rebind(&handles);
            }
        }
    }
}

/// Drain a second subscription set dedicated to identity health. Keeping this
/// task separate from console/event projection ensures a full user-facing
/// output channel cannot suppress permanent stream-loss detection.
async fn run_identity_stream_health_monitor(
    handle: MobHandle,
    agent_mob_mcp_state: Option<Arc<meerkat_mob_mcp::MobMcpState>>,
    identity_runtime: Arc<std::sync::RwLock<Option<Arc<crate::identity_first::IdentityRuntime>>>>,
) {
    let mut streams: SelectAll<TaggedAgentEventStream> = SelectAll::new();
    let mut tracked = HashSet::new();
    let mut subscribe_failures: HashMap<TrackedAgentEventStream, SubscribeBackoff> = HashMap::new();
    let mut cadence = ReconcileCadence::new(&agent_mob_mcp_state);

    let handles = Box::pin(reconcile_agent_event_streams(
        &handle,
        &agent_mob_mcp_state,
        &mut tracked,
        &mut subscribe_failures,
        &mut streams,
        Some(&identity_runtime),
    ))
    .await;
    cadence.rebind(&handles);

    loop {
        tokio::select! {
            Some(forwarded) = streams.next() => {
                match forwarded {
                    ForwardedAgentEvent::Event(event) => {
                        let (_, _, _, envelope, durable_identity) = *event;
                        record_identity_turn_completion(
                            &identity_runtime,
                            durable_identity.as_deref(),
                            &envelope,
                        ).await;
                    }
                    ForwardedAgentEvent::Closed(tracked_key) => {
                        tracked.remove(&tracked_key);
                        subscribe_failures.remove(&tracked_key);
                        trigger_identity_stream_repair(
                            handle.mob_id().as_str(),
                            &tracked_key,
                            &identity_runtime,
                            "live agent event stream closed permanently",
                        ).await;
                        // Re-attach promptly after a closure; repair latency
                        // must not wait for an unrelated machine transition.
                        let handles = Box::pin(reconcile_agent_event_streams(
                            &handle,
                            &agent_mob_mcp_state,
                            &mut tracked,
                            &mut subscribe_failures,
                            &mut streams,
                            Some(&identity_runtime),
                        )).await;
                        cadence.rebind(&handles);
                    }
                }
            }
            () = cadence.wait(earliest_backoff_attempt(&subscribe_failures)) => {
                let handles = Box::pin(reconcile_agent_event_streams(
                    &handle,
                    &agent_mob_mcp_state,
                    &mut tracked,
                    &mut subscribe_failures,
                    &mut streams,
                    Some(&identity_runtime),
                )).await;
                cadence.rebind(&handles);
            }
        }
    }
}

/// Returns the handles it enumerated (primary + child mobs) so the caller can
/// rebind its [`ReconcileCadence`] watchers to the same set.
async fn reconcile_agent_event_streams(
    handle: &MobHandle,
    agent_mob_mcp_state: &Option<Arc<meerkat_mob_mcp::MobMcpState>>,
    tracked: &mut HashSet<TrackedAgentEventStream>,
    subscribe_failures: &mut HashMap<TrackedAgentEventStream, SubscribeBackoff>,
    streams: &mut SelectAll<TaggedAgentEventStream>,
    identity_runtime: Option<
        &Arc<std::sync::RwLock<Option<Arc<crate::identity_first::IdentityRuntime>>>>,
    >,
) -> Vec<MobHandle> {
    let primary_mob_id = handle.mob_id().to_string();
    let mut handles = vec![handle.clone()];
    if let Some(state) = agent_mob_mcp_state {
        handles.extend(
            Box::pin(state.mob_handles_snapshot())
                .await
                .unwrap_or_default()
                .into_iter()
                .filter_map(|(mob_id, child_handle)| {
                    if mob_id.as_str() == primary_mob_id {
                        None
                    } else {
                        Some(child_handle)
                    }
                }),
        );
    }

    let mut current: HashSet<TrackedAgentEventStream> = HashSet::new();
    for handle in &handles {
        let mob_id = handle.mob_id().to_string();
        for entry in handle.list_members_including_retiring().await {
            // Members without current machine-supplied binding atoms have no
            // live runtime stream to track; their stale streams age out.
            let Some((runtime_id, fence_token)) = entry.binding_atoms() else {
                continue;
            };
            let durable_identity = durable_identity_label(&entry.labels);
            let identity_fencing_token = current_identity_fencing_token(
                &primary_mob_id,
                &mob_id,
                durable_identity.as_deref(),
                identity_runtime,
            )
            .await;
            // The health monitor exists solely for identity-first repair.
            // Avoid duplicating every ordinary/child-mob event stream.
            if identity_runtime.is_some() && identity_fencing_token.is_none() {
                continue;
            }
            current.insert(TrackedAgentEventStream {
                mob_id: mob_id.clone(),
                durable_identity,
                member_identity: entry.agent_identity.clone(),
                runtime_id,
                identity_fencing_token,
                fence_token,
            });
        }
    }

    tracked.retain(|tracked_key| current.contains(tracked_key));
    // Drop backoff bookkeeping for members that have left the roster so the
    // map can't grow without bound across the runtime's lifetime.
    subscribe_failures.retain(|key, _| current.contains(key));

    for handle in &handles {
        let mob_id = handle.mob_id().to_string();
        for entry in handle.list_members_including_retiring().await {
            let identity = entry.agent_identity.clone();
            // No binding atoms means no live runtime to subscribe to.
            let Some((runtime_id, fence_token)) = entry.binding_atoms() else {
                continue;
            };
            let durable_identity = durable_identity_label(&entry.labels);
            let identity_fencing_token = current_identity_fencing_token(
                &primary_mob_id,
                &mob_id,
                durable_identity.as_deref(),
                identity_runtime,
            )
            .await;
            if identity_runtime.is_some() && identity_fencing_token.is_none() {
                continue;
            }
            let tracked_key = TrackedAgentEventStream {
                mob_id: mob_id.clone(),
                durable_identity,
                member_identity: identity.clone(),
                runtime_id: runtime_id.clone(),
                identity_fencing_token,
                fence_token,
            };
            if tracked.contains(&tracked_key) {
                continue;
            }

            // Only Active members have a live runtime delta stream to attach
            // to. A Retiring/Broken/Completed member can still carry stale
            // binding atoms (so `binding_atoms()` is Some) while its session
            // injector is already gone, which makes `subscribe_agent_events`
            // fail every reconcile tick — the source of the 4×/s forwarder
            // hot-loop. Such members are skipped here; their final events
            // arrive via the structural ledger / session-history backfill and
            // their streams age out through `tracked.retain`.
            if !forwarder_should_subscribe(entry.status) {
                subscribe_failures.remove(&tracked_key);
                continue;
            }

            // Back off an Active member that keeps failing to subscribe (a
            // genuinely stuck injector), so even that case can't spin the log.
            let now = tokio::time::Instant::now();
            if let Some(backoff) = subscribe_failures.get(&tracked_key)
                && now < backoff.next_attempt
            {
                continue;
            }

            let role = entry.role.clone();

            match subscribe_agent_events_for_console_forwarder(handle, &tracked_key.member_identity)
                .await
            {
                Ok(stream) => {
                    let close_key = tracked_key.clone();
                    let durable_identity: Option<Arc<str>> = tracked_key
                        .durable_identity
                        .as_deref()
                        .map(Arc::<str>::from);
                    subscribe_failures.remove(&tracked_key);
                    tracked.insert(tracked_key);
                    let mapped = stream
                        .map(move |envelope| {
                            ForwardedAgentEvent::Event(Box::new((
                                runtime_id.clone(),
                                fence_token,
                                role.clone(),
                                envelope,
                                durable_identity.clone(),
                            )))
                        })
                        .chain(futures::stream::once(async move {
                            ForwardedAgentEvent::Closed(close_key)
                        }))
                        .boxed();
                    streams.push(mapped);
                }
                Err(error) => {
                    // Usually a short-lived spawn/resume race while Meerkat
                    // finishes installing the session event injector. Retry
                    // with exponential backoff and warn only on the first
                    // failure. A bounded number of misses remains a spawn
                    // race; persistent loss breaks the exact identity binding
                    // and lets the continuity supervisor rebuild it.
                    let repair_key = tracked_key.clone();
                    let backoff =
                        subscribe_failures
                            .entry(tracked_key)
                            .or_insert(SubscribeBackoff {
                                next_attempt: now,
                                consecutive_failures: 0,
                            });
                    if identity_runtime.is_none() && backoff.consecutive_failures == 0 {
                        tracing::warn!(
                            mob_id = %mob_id,
                            identity = %identity,
                            error = %error,
                            "mobkit agent event forwarder: failed to subscribe; will retry with backoff"
                        );
                    } else if identity_runtime.is_none() {
                        tracing::debug!(
                            mob_id = %mob_id,
                            identity = %identity,
                            error = %error,
                            consecutive_failures = backoff.consecutive_failures,
                            "mobkit agent event forwarder: subscribe still failing; backing off"
                        );
                    }
                    backoff.next_attempt =
                        now + subscribe_backoff_delay(backoff.consecutive_failures);
                    backoff.consecutive_failures = backoff.consecutive_failures.saturating_add(1);
                    let stream_is_permanently_lost =
                        backoff.consecutive_failures >= PERMANENT_STREAM_FAILURE_THRESHOLD;
                    if stream_is_permanently_lost && let Some(identity_runtime) = identity_runtime {
                        trigger_identity_stream_repair(
                            &primary_mob_id,
                            &repair_key,
                            identity_runtime,
                            "live agent event stream remained unavailable after bounded retries",
                        )
                        .await;
                    }
                }
            }
        }
    }
    handles
}

async fn subscribe_agent_events_for_console_forwarder(
    handle: &MobHandle,
    identity: &AgentIdentity,
) -> Result<EventStream, meerkat_mob::MobError> {
    // Keep the console forwarder on the same authoritative subscription path
    // as `/agents/{id}/events`. The observation shortcut can lag the actor's
    // runtime-member projection in identity-first/runtime-backed packs, which
    // leaves the console with only session-history backfill while direct agent
    // SSE streams live deltas correctly.
    handle.subscribe_agent_events(identity).await
}

/// Streaming subscription against the meerkat mob event ledger. Each
/// projected envelope's cursor is the upstream `MobEvent.cursor`; after
/// projection the cursor is checkpointed via `persistent_metadata` so
/// the next runtime instance can resume from where this one left off.
///
/// Resume semantics on startup:
/// - persisted cursor present → `subscribe_after(cursor)`. On
///   `MobError::StaleEventCursor` (the ledger has been truncated past
///   our checkpoint) the task logs a warning and falls through to a
///   fresh `subscribe()` at the current latest.
/// - no persisted cursor → `subscribe()` (latest, no replay).
///
/// Exits when the upstream `event_rx` closes (machine destroyed) or
/// when subscription setup fails after a stale-cursor fallback.
async fn run_mob_events_subscription(
    handle: MobHandle,
    store: MobEventsStore,
    persistent_metadata: Arc<dyn PersistentMetadataStore>,
) {
    let mob_id = handle.mob_id().as_str().to_string();
    let resume_cursor = match persistent_metadata.get_subscription_cursor(&mob_id).await {
        Ok(value) => value,
        Err(err) => {
            tracing::warn!(
                mob_id = %mob_id,
                error = %err,
                "mob_events subscription: failed to read persisted cursor; resuming from latest"
            );
            None
        }
    };

    let events = handle.events();
    let mut subscription = match resume_cursor {
        Some(cursor) => match events.subscribe_after(cursor).await {
            Ok(sub) => sub,
            Err(MobError::StaleEventCursor {
                after_cursor,
                latest_cursor,
            }) => {
                tracing::warn!(
                    mob_id = %mob_id,
                    after_cursor,
                    latest_cursor,
                    "mob_events subscription: persisted cursor is past ledger frontier; resuming at latest"
                );
                match events.subscribe().await {
                    Ok(sub) => sub,
                    Err(err) => {
                        tracing::warn!(
                            mob_id = %mob_id,
                            error = %err,
                            "mob_events subscription: failed to subscribe at latest after stale-cursor recovery"
                        );
                        return;
                    }
                }
            }
            Err(err) => {
                tracing::warn!(
                    mob_id = %mob_id,
                    error = %err,
                    "mob_events subscription: failed to resume from persisted cursor"
                );
                return;
            }
        },
        None => match events.subscribe().await {
            Ok(sub) => sub,
            Err(err) => {
                tracing::warn!(
                    mob_id = %mob_id,
                    error = %err,
                    "mob_events subscription: initial subscribe failed"
                );
                return;
            }
        },
    };

    while let Some(event) = subscription.event_rx.recv().await {
        let envelope = store.project_mob_event(&event).await;
        if let Err(err) = persistent_metadata
            .set_subscription_cursor(&mob_id, envelope.cursor)
            .await
        {
            tracing::warn!(
                mob_id = %mob_id,
                cursor = envelope.cursor,
                error = %err,
                "mob_events subscription: failed to persist cursor; continuing"
            );
        }
    }
}

/// How often the actor-loop probe sends its round trip.
const ACTOR_LOOP_PROBE_INTERVAL: Duration = Duration::from_mins(1);

/// How long one probe round trip may go unanswered before the stall pages.
///
/// Unlike the delivery path's admission budget (`identity_first/bridge.rs`,
/// 600s), which must stay wide because firing it drops a delivery, a probe
/// timeout drops nothing — it only names the stall — so it can afford to be
/// aggressive. The probe's handler is a pure in-memory phase read whose
/// healthy latency is microseconds; 30s therefore cannot fire because the
/// READ is slow, only because the read is queued behind a handler that has
/// not yet RETURNED.
///
/// That is deliberately weaker than "blocked". A handler that has not
/// returned may be wedged, or may be doing legitimate long work — a member
/// revival, a large replay, a compaction, a storage migration. The probe
/// cannot tell those apart, because `QueryPhase` rides the same serialized
/// loop it is watching, so its round trip measures "time to drain everything
/// queued ahead", not health. Note the scale this sits at: the same system
/// treats a 600s in-flight admission as normal (`BRIDGE_ACTOR_ADMISSION_BUDGET`),
/// which is 20x this budget, so a busy loop can cross 30s without anything
/// being wrong. Widening the budget is NOT the fix — it would only move the
/// same ambiguity — and one such long command has already been suppressed by
/// hand (`lifecycle.rs` aborts the probe on shutdown so an intentional
/// shutdown stall cannot page). The real discriminator is whether the loop
/// made PROGRESS while the probe waited, which needs a monotonic
/// command-completion counter on meerkat's mob actor; mobkit cannot observe
/// it from here. Until that exists, the stall is an OPEN INCIDENT rather
/// than a verdict: it is closed by the correlated
/// [`ErrorEvent::ActorLoopRecovered`], and a receiver decides severity from
/// how long the incident stays open on its own clock.
const ACTOR_LOOP_PROBE_BUDGET: Duration = Duration::from_secs(30);

/// Effective probe interval: `MOBKIT_ACTOR_LOOP_PROBE_INTERVAL_SECS`
/// overrides the default, clamped to [1, 3600] seconds (the same knob idiom
/// as `MOBKIT_BRIDGE_ACTOR_ADMISSION_SECS`).
fn actor_loop_probe_interval() -> Duration {
    parse_probe_secs(
        std::env::var("MOBKIT_ACTOR_LOOP_PROBE_INTERVAL_SECS")
            .ok()
            .as_deref(),
        ACTOR_LOOP_PROBE_INTERVAL,
    )
}

/// Effective probe budget: `MOBKIT_ACTOR_LOOP_PROBE_BUDGET_SECS` overrides
/// the default, clamped to [1, 3600] seconds.
fn actor_loop_probe_budget() -> Duration {
    parse_probe_secs(
        std::env::var("MOBKIT_ACTOR_LOOP_PROBE_BUDGET_SECS")
            .ok()
            .as_deref(),
        ACTOR_LOOP_PROBE_BUDGET,
    )
}

fn parse_probe_secs(raw: Option<&str>, default: Duration) -> Duration {
    raw.and_then(|value| value.trim().parse::<u64>().ok())
        .map(|secs| Duration::from_secs(secs.clamp(1, 3600)))
        .unwrap_or(default)
}

/// Actor-loop liveness probe.
///
/// meerkat's mob actor is ONE serialized command loop: a handler that blocks
/// freezes every member's dispatch behind it, and the only existing witness
/// is the delivery path's admission budget — which fires only if a delivery
/// happens to be waiting. If nobody waits, nobody learns. This probe is the
/// unconditional waiter: every `interval` it sends the cheapest round trip
/// the handle exposes (`QueryPhase`, an O(1) in-memory phase read) and pages
/// `ErrorEvent::ActorLoopStalled` when the reply does not arrive within
/// `budget`.
///
/// At most ONE probe is ever outstanding: on timeout the task stays parked
/// on the SAME round trip until the loop drains it, rather than stacking
/// further commands onto a stalled actor (uncapped retry loops already
/// amplify channel pressure there; the probe must never join them). When the
/// parked round trip finally resolves, recovery is emitted as
/// [`ErrorEvent::ActorLoopRecovered`], correlated to the stall by `stall_id`.
///
/// That variant amends a previously stated invariant — `ErrorEvent` used to
/// have no recovery precedent, on the reasoning that every variant names a
/// failure and the hook is a paging channel. An open-only paging channel is
/// not decidable: a receiver that pages on a stall can never close the
/// incident it opened, so it can only escalate. The resolution is filtered
/// out of error-level logging by the default sink, and unaware consumers see
/// it fall into the `_` arm `#[non_exhaustive]` already forces.
///
/// THE CORRELATED PAIR PLUS THE RECEIVER'S OWN CLOCK IS THE DISCRIMINATOR.
/// An open `stall_id` with no matching resolution after ten minutes is a
/// wedged loop; one closed after fifty seconds was a busy loop. That is a
/// complete decision procedure using only what this emitter already sends,
/// which is why the missing progress counter is an IMPROVEMENT rather than a
/// missing piece: it would let the probe skip PAGING for the busy case at
/// all, whereas today the receiver pages first and decides after. Better,
/// but strictly an optimization of a decision the receiver can already make.
///
/// Note what the counters can and cannot say. Because the probe parks on the
/// SAME round trip rather than starting a new one, a genuinely wedged loop
/// pages exactly once and never gets another cycle, so no counter can climb;
/// a merely slow loop recovers and stalls again, so it is the slow case that
/// accumulates. `prior_resolved_stalls` is therefore chronic-busyness
/// evidence, and "wedged" is the absence of a resolution, measured by the
/// receiver's own clock — not by any count this emitter could produce.
///
/// The probe treats ANY completion within budget — `Ok` or a typed error —
/// as a live loop: it measures whether the loop drains, not whether the mob
/// is healthy. A terminal phase reply (`Stopped`/`Completed`/`Destroyed`)
/// ends the probe: there is no loop left worth watching.
async fn run_actor_loop_probe<P, F>(
    mut probe: P,
    error_hook: SharedErrorHook,
    interval: Duration,
    budget: Duration,
) where
    P: FnMut() -> F,
    F: Future<Output = Result<MobState, MobError>>,
{
    // Correlates each stall with the resolution that closes it, and counts
    // the stalls that have already resolved. Task-local: one probe per
    // runtime, so no shared counter is needed.
    let mut next_stall_id: u64 = 1;
    let mut resolved_stalls: u64 = 0;
    loop {
        tokio::time::sleep(interval).await;
        let started = tokio::time::Instant::now();
        let round_trip = probe();
        tokio::pin!(round_trip);
        let result = match tokio::time::timeout(budget, &mut round_trip).await {
            Ok(result) => result,
            Err(_) => {
                let stall_id = next_stall_id;
                next_stall_id += 1;
                fire_error_hook(
                    &error_hook,
                    ErrorEvent::ActorLoopStalled {
                        // Measured, not the configured budget echoed back: a
                        // field that looks like data must be data. At page
                        // time this necessarily reads ~= the budget (we page
                        // the instant it expires), so it is a truthfulness
                        // fix rather than new information — the informative
                        // elapsed is `ActorLoopRecovered::stalled_for_secs`.
                        probe_waited_secs: started.elapsed().as_secs(),
                        detail: format!(
                            "QueryPhase probe round trip unanswered after {}s; the mob actor \
                             is one serialized command loop, so every member's dispatch is \
                             queued behind whatever is blocking it; the probe stays parked on \
                             this round trip and will not stack another. THIS STALL PAGES \
                             ONCE: no further stall events will be emitted for it, so silence \
                             is NOT recovery — hold the incident open until an \
                             actor_loop_recovered arrives with stall_id {}",
                            budget.as_secs(),
                            stall_id
                        ),
                        stall_id: Some(stall_id),
                        prior_resolved_stalls: Some(resolved_stalls),
                    },
                );
                // Park on the SAME round trip until the loop drains it.
                let result = round_trip.await;
                resolved_stalls += 1;
                // The receiver opened an incident on the stall above; this is
                // the only thing that lets it close that incident rather than
                // escalate forever.
                fire_error_hook(
                    &error_hook,
                    ErrorEvent::ActorLoopRecovered {
                        stall_id,
                        stalled_for_secs: started.elapsed().as_secs(),
                    },
                );
                result
            }
        };
        if matches!(
            result,
            Ok(MobState::Stopped | MobState::Completed | MobState::Destroyed)
        ) {
            break;
        }
    }
}

/// Project a forwarded member event for the drain: the wire envelope plus
/// any drain-side alert extracted while the member event is still typed.
fn forwarded_member_event(attributed: AttributedEvent) -> ForwardedMemberEvent {
    ForwardedMemberEvent {
        alert: compaction_rejection_alert(&attributed),
        envelope: attributed_event_to_unified(attributed),
    }
}

/// Compaction persistence rejections must page rather than pass as ordinary
/// console traffic — meerkat emits `CompactionFailed` on every rejected
/// compaction commit, and fleets otherwise discover wedged members by
/// silence. Extracted here, where the member event and its session source
/// identity are still typed.
fn compaction_rejection_alert(attributed: &AttributedEvent) -> Option<ErrorEvent> {
    let AgentEvent::CompactionFailed { reason } = &attributed.envelope.payload else {
        return None;
    };
    // Live member session streams stamp session source identity; a
    // non-session source has no session to attribute.
    let session_id = attributed
        .envelope
        .source
        .session_id()
        .map(ToString::to_string)
        .unwrap_or_default();
    // The projection-handoff refusal carries the severity facts typed: the
    // preserved-history fit is the wedged-member (page) vs still-progressing
    // (log line) discriminator, and hosts must not fish it out of the message
    // string. Other compaction failure reasons have no fit verdict to carry.
    let (preserved_history, attempted_entries) = match reason {
        meerkat_core::event::CompactionFailureReason::ProjectionHandoffRefused {
            preserved_history,
            attempted_entries,
            ..
        } => (Some((*preserved_history).into()), Some(*attempted_entries)),
        _ => (None, None),
    };
    Some(ErrorEvent::CompactionPersistenceRejected {
        identity: crate::member_comms_id::runtime_event_alias(&attributed.source),
        session_id,
        error: reason.to_string(),
        preserved_history,
        attempted_entries,
    })
}

fn attributed_event_to_unified(attributed: AttributedEvent) -> EventEnvelope<UnifiedEvent> {
    EventEnvelope {
        event_id: format!("evt-agent-{}", attributed.envelope.event_id),
        source: "agent".to_string(),
        timestamp_ms: attributed.envelope.timestamp_ms,
        event: UnifiedEvent::Agent {
            // The runtime id's member component is the comms-safe roster
            // encoding (meerkat 0.7 `MemberCommsName`); decode back to the
            // public alias space here so console replay resolution, the
            // `mobkit/events/subscribe` buffer, and the event log all key
            // events by the same ids that spawn/reserve paths register.
            agent_id: crate::member_comms_id::runtime_event_alias(&attributed.source),
            event_type: agent_event_type(&attributed.envelope.payload).to_string(),
            // Project through the console wire shape (not the raw 0.7 event)
            // so downstream surfaces — console timeline frames, the
            // `mobkit/events/subscribe` replay buffer, and the event-log
            // query — keep the `result`/`tool_call_id` keys the SDKs parse.
            payload: Some(crate::mob_handle_runtime::console_agent_event_payload(
                &attributed.envelope.payload,
            )),
        },
    }
}

/// Projects [`crate::memory::events::MemoryTimelineEvent`]s onto the
/// console timeline. Sync fire-and-forget: the async append is spawned on
/// the captured runtime handle, so emitters inside mutexes or blocking
/// threads never wait on the event surface.
struct ConsoleMemoryEventSink {
    store: ConsoleEventStore,
    handle: tokio::runtime::Handle,
}

impl crate::memory::events::MemoryEventSink for ConsoleMemoryEventSink {
    fn emit(&self, event: crate::memory::events::MemoryTimelineEvent) {
        let store = self.store.clone();
        let identity = event
            .identity()
            .map(str::to_string)
            .unwrap_or_else(|| crate::console_contracts::SYSTEM_EVENT_IDENTITY.to_string());
        let event_type = event.event_type().to_string();
        let data = event.data();
        self.handle.spawn(async move {
            store.append(identity, None, event_type, data).await;
        });
    }
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::panic)]
mod tests {
    use std::sync::atomic::Ordering;

    use super::*;
    use meerkat_mob::ids::Generation;

    fn attributed_text_delta(member_id: &str, generation: u64) -> AttributedEvent {
        AttributedEvent {
            source: AgentRuntimeId::new(
                AgentIdentity::from(member_id),
                Generation::new(generation),
            ),
            source_fence_token: FenceToken::new(1),
            role: ProfileName::from("worker"),
            envelope: meerkat_core::event::EventEnvelope {
                event_id: Default::default(),
                source: meerkat_core::event::EventSourceIdentity::runtime("test"),
                seq: 0,
                mob_id: None,
                timestamp_ms: 1,
                payload: AgentEvent::TextDelta {
                    delta: "hello".to_string(),
                },
            },
        }
    }

    #[test]
    fn identity_stream_tracking_uses_trusted_durable_identity_label() {
        let labels =
            BTreeMap::from([("agent_identity".to_string(), "review:singleton".to_string())]);
        assert_eq!(
            durable_identity_label(&labels).as_deref(),
            Some("review:singleton")
        );
        assert_eq!(
            durable_identity_label(&BTreeMap::new()),
            None,
            "ordinary mobs must not be guessed into identity authority"
        );
    }

    /// Regression: identity-first members spawn under comms-safe encoded
    /// roster ids (`mk--…`); the agent-event ingest must decode the member
    /// component back to the public alias space before console/SDK
    /// projection, or events project under junk identities and reserved
    /// interactions never complete.
    #[test]
    fn attributed_event_ingest_decodes_encoded_roster_member_ids() {
        let encoded = crate::member_comms_id::mob_member_id_str("rt:review:singleton:0");
        assert!(encoded.starts_with("mk--"), "precondition: alias encodes");
        let unified = attributed_event_to_unified(attributed_text_delta(&encoded, 1));
        let UnifiedEvent::Agent { agent_id, .. } = unified.event else {
            panic!("expected agent event");
        };
        assert_eq!(agent_id, "rt:review:singleton:0:1");
    }

    #[test]
    fn attributed_event_ingest_passes_plain_member_ids_through() {
        let unified = attributed_event_to_unified(attributed_text_delta("worker-one", 0));
        let UnifiedEvent::Agent { agent_id, .. } = unified.event else {
            panic!("expected agent event");
        };
        assert_eq!(agent_id, "worker-one:0");
    }

    /// Regression: the callers' outer `select!` drops and recreates the
    /// `wait()` future on every forwarded member event, so a safety deadline
    /// computed inside `wait` would reset under sustained event traffic and
    /// the safety reconcile would never fire. The deadline must persist in
    /// the cadence and eventually complete a recreated `wait`.
    #[tokio::test(start_paused = true)]
    async fn reconcile_cadence_safety_deadline_survives_recreated_waits() {
        let mut cadence = ReconcileCadence::new(&None);
        let mut fired = false;
        // Seven 5s rounds = 35s of simulated event churn; the 30s deadline
        // anchored at construction must fire within them.
        for _ in 0..7 {
            tokio::select! {
                () = cadence.wait(None) => {
                    fired = true;
                    break;
                }
                () = tokio::time::sleep(Duration::from_secs(5)) => {}
            }
        }
        assert!(
            fired,
            "safety reconcile starved: recreated wait futures reset the deadline"
        );
    }

    /// The safety bound is "at most 30s since the last reconcile pass", not
    /// "since construction": `rebind` (which ends every reconcile pass) must
    /// re-arm the deadline.
    #[tokio::test(start_paused = true)]
    async fn reconcile_cadence_rebind_rearms_safety_deadline() {
        let mut cadence = ReconcileCadence::new(&None);
        tokio::time::sleep(Duration::from_secs(20)).await;
        cadence.rebind(&[]);
        tokio::select! {
            () = cadence.wait(None) => {
                panic!("deadline fired 30s after construction despite rebind re-arm")
            }
            () = tokio::time::sleep(Duration::from_secs(29)) => {}
        }
        tokio::select! {
            () = cadence.wait(None) => {}
            () = tokio::time::sleep(Duration::from_secs(2)) => {
                panic!("re-armed deadline did not fire 30s after rebind")
            }
        }
    }

    /// Regression: the console forwarder must only hold a live subscription
    /// for Active members. A Retiring member can keep stale binding atoms
    /// while its session injector is gone, so subscribing it fails every
    /// 250ms reconcile tick — the 4×/s "failed to subscribe" hot-loop
    /// (observed ~49k warnings over 3.4h on one wedged-retiring alias).
    #[test]
    fn forwarder_only_subscribes_active_members() {
        assert!(forwarder_should_subscribe(MobMemberStatus::Active));
        assert!(!forwarder_should_subscribe(MobMemberStatus::Retiring));
        assert!(!forwarder_should_subscribe(MobMemberStatus::Broken));
        assert!(!forwarder_should_subscribe(MobMemberStatus::Completed));
        assert!(!forwarder_should_subscribe(MobMemberStatus::Unknown));
    }

    /// The backoff for a persistently-failing Active subscribe must grow from
    /// one reconcile tick and cap, so even a genuinely stuck member retries at
    /// most ~once per cap instead of 4×/s.
    #[test]
    fn subscribe_backoff_grows_and_caps() {
        const { assert!(PERMANENT_STREAM_FAILURE_THRESHOLD > 1) };
        assert_eq!(subscribe_backoff_delay(0), SUBSCRIBE_BACKOFF_BASE);
        assert_eq!(subscribe_backoff_delay(1), SUBSCRIBE_BACKOFF_BASE * 2);
        assert_eq!(subscribe_backoff_delay(3), SUBSCRIBE_BACKOFF_BASE * 8);
        assert_eq!(subscribe_backoff_delay(7), SUBSCRIBE_BACKOFF_MAX);
        // Saturates at the cap for arbitrarily many failures (no shift overflow).
        assert_eq!(subscribe_backoff_delay(50), SUBSCRIBE_BACKOFF_MAX);
        assert!(subscribe_backoff_delay(2) > subscribe_backoff_delay(1));
    }

    async fn bootstrap_minimal_runtime(temp_dir: &tempfile::TempDir) -> UnifiedRuntime {
        let session_path = temp_dir.path().join("sessions");
        std::fs::create_dir_all(&session_path).expect("session path");
        let factory = meerkat::AgentFactory::new(&session_path);
        let session_service: Arc<dyn meerkat_mob::MobSessionService> = Arc::new(
            meerkat::build_ephemeral_service(factory, meerkat::Config::default(), 16),
        );

        let definition = meerkat_mob::MobDefinition::from_toml(
            r#"
[mob]
id = "compaction-alert-mob"

[profiles.worker]
model = "gpt-5.5"
"#,
        )
        .expect("parse mob definition");
        let mob_spec = MobBootstrapSpec::new(
            definition,
            meerkat_mob::MobStorage::in_memory(),
            session_service,
        )
        .with_options(crate::mob_handle_runtime::MobBootstrapOptions {
            allow_ephemeral_sessions: true,
            notify_orchestrator_on_resume: true,
            default_llm_client: Some(Arc::new(meerkat_client::TestClient::for_provider(
                meerkat_core::Provider::OpenAI,
            ))),
        });
        let module_config = MobKitConfig {
            modules: vec![],
            discovery: crate::types::DiscoverySpec {
                namespace: "compaction-alert".to_string(),
                modules: vec![],
            },
            pre_spawn: vec![],
        };
        UnifiedRuntime::bootstrap(mob_spec, module_config, Duration::from_secs(2))
            .await
            .expect("bootstrap unified runtime")
    }

    /// Push member `CompactionFailed` events carrying `reasons` through the
    /// real drain path and return the alerts the error hook received.
    ///
    /// Every reason rides ONE bootstrapped runtime. The mapping is only worth
    /// anything if the fields survive the forwarder, the drain, and the
    /// detached hook fire, so this goes the whole way to the hook — but the
    /// lib suite runs ~1500 tests in one process and `bootstrap`'s 2s budget
    /// is a wall-clock allowance, so a second concurrent bootstrap is load
    /// the suite should not have to carry.
    async fn compaction_alerts_through_drain(
        session_id: &meerkat_core::types::SessionId,
        reasons: Vec<meerkat_core::event::CompactionFailureReason>,
    ) -> Vec<ErrorEvent> {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let mut runtime = bootstrap_minimal_runtime(&temp_dir).await;

        let captured: Arc<tokio::sync::Mutex<Vec<ErrorEvent>>> =
            Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let hook_captured = captured.clone();
        let hook: ErrorHook = Arc::new(move |event| {
            let hook_captured = hook_captured.clone();
            Box::pin(async move {
                hook_captured.lock().await.push(event);
            })
        });
        runtime.set_error_hook(hook);

        let event_tx = runtime.install_test_event_ingress().await;
        let expected = reasons.len();
        for (seq, reason) in reasons.into_iter().enumerate() {
            let attributed = AttributedEvent {
                source: AgentRuntimeId::new(
                    AgentIdentity::from("compaction-worker"),
                    Generation::new(0),
                ),
                source_fence_token: FenceToken::new(1),
                role: ProfileName::from("worker"),
                envelope: meerkat_core::event::EventEnvelope {
                    event_id: Default::default(),
                    source: meerkat_core::event::EventSourceIdentity::session(session_id.clone()),
                    seq: seq as u64,
                    mob_id: None,
                    timestamp_ms: 9,
                    payload: AgentEvent::CompactionFailed { reason },
                },
            };
            event_tx
                .send(forwarded_member_event(attributed))
                .await
                .expect("send forwarded event");
        }
        runtime
            .drain_mob_agent_events()
            .await
            .expect("drain member events");

        // `fire_error` spawns a detached task; wait bounded for every hook.
        let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        loop {
            if captured.lock().await.len() >= expected {
                break;
            }
            assert!(
                tokio::time::Instant::now() < deadline,
                "error hook did not receive all {expected} compaction persistence rejections"
            );
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        let alerts = captured.lock().await.clone();
        runtime.shutdown().await;
        alerts
    }

    /// Compaction persistence rejections must page, not pass as ordinary
    /// console traffic: member `CompactionFailed` agent events pushed through
    /// the drain path fire the error hook with the typed
    /// `CompactionPersistenceRejected` alert carrying the member identity,
    /// the emitting session, and the rejection detail.
    ///
    /// meerkat's typed `ProjectionHandoffRefused` reason additionally lands
    /// its severity facts on the hook typed: the preserved-history fit
    /// discriminator (page vs log line) and the attempted entry count arrive
    /// as fields, not as message substrings, while `error` keeps the full
    /// human rendering for catch-all adopters. `StillFits` is the
    /// discriminating value — an "always page" default would say
    /// `OverWindow`. Every other failure reason carries no verdict at all.
    #[tokio::test]
    async fn compaction_failures_page_error_hook_with_typed_fit_through_drain() {
        let session_id = meerkat_core::types::SessionId::new();
        let alerts = compaction_alerts_through_drain(
            &session_id,
            vec![
                meerkat_core::event::CompactionFailureReason::TranscriptRewriteFailed {
                    message: "runtime epoch mismatch".to_string(),
                },
                meerkat_core::event::CompactionFailureReason::ProjectionHandoffRefused {
                    refusal: meerkat_core::memory::CompactionHandoffRefusal::RuntimeEpochRotated,
                    preserved_history:
                        meerkat_core::event::CompactionPreservedHistoryFit::StillFits,
                    attempted_entries: 12,
                    message: "runtime epoch rotated under the coordinator".to_string(),
                },
            ],
        )
        .await;

        // The hook fires from detached tasks, so arrival order is not the
        // send order: pick each alert out by the fact under test.
        let mut untyped = None;
        let mut typed = None;
        for alert in &alerts {
            match alert {
                ErrorEvent::CompactionPersistenceRejected {
                    identity,
                    session_id: rejected_session,
                    error,
                    preserved_history,
                    attempted_entries,
                } => {
                    assert_eq!(identity, "compaction-worker:0");
                    assert_eq!(rejected_session, &session_id.to_string());
                    if preserved_history.is_some() {
                        typed = Some((error.clone(), *preserved_history, *attempted_entries));
                    } else {
                        untyped = Some((error.clone(), *attempted_entries));
                    }
                }
                other => panic!("expected CompactionPersistenceRejected, got {other:?}"),
            }
        }

        let (untyped_error, untyped_entries) =
            untyped.unwrap_or_else(|| panic!("the non-handoff failure must page too: {alerts:?}"));
        assert!(
            untyped_error.contains("runtime epoch mismatch"),
            "error must carry the rejection detail: {untyped_error}"
        );
        assert_eq!(
            untyped_entries, None,
            "a non-handoff compaction failure carries no fit verdict"
        );

        let (typed_error, typed_fit, typed_entries) = typed.unwrap_or_else(|| {
            panic!("the projection-handoff refusal must page with its fit: {alerts:?}")
        });
        assert_eq!(
            typed_fit,
            Some(CompactionPreservedHistoryFit::StillFits),
            "the wedged/progressing discriminator must cross typed"
        );
        assert_eq!(typed_entries, Some(12));
        assert!(
            typed_error.contains("runtime epoch rotated under the coordinator"),
            "error must keep the human rendering: {typed_error}"
        );
    }

    /// Captures formatted tracing output so the default-sink tests can assert
    /// on the emitted records. `with_default` is thread-local, so everything
    /// asserted here must be logged synchronously on the calling thread.
    #[derive(Clone, Default)]
    struct CaptureWriter(Arc<std::sync::Mutex<Vec<u8>>>);

    impl CaptureWriter {
        fn contents(&self) -> String {
            String::from_utf8_lossy(
                &self
                    .0
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner),
            )
            .into_owned()
        }
    }

    impl std::io::Write for CaptureWriter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .extend_from_slice(buf);
            Ok(buf.len())
        }

        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CaptureWriter {
        type Writer = CaptureWriter;

        fn make_writer(&'a self) -> Self::Writer {
            self.clone()
        }
    }

    fn capture_tracing<T>(body: impl FnOnce() -> T) -> (T, String) {
        let writer = CaptureWriter::default();
        let subscriber = tracing_subscriber::fmt()
            .with_writer(writer.clone())
            // INFO rather than WARN because the default sink deliberately
            // drops the resolution variant to INFO, and a test that capped at
            // WARN could not tell "logged at INFO" from "not logged at all".
            .with_max_level(tracing::Level::INFO)
            // Without this the formatter wraps every field name and `=` in
            // ANSI escapes, so `contains("hook_registered=false")` reads a
            // string that is never literally present.
            .with_ansi(false)
            .finish();
        let value = tracing::subscriber::with_default(subscriber, body);
        (value, writer.contents())
    }

    fn sample_alert() -> ErrorEvent {
        ErrorEvent::CompactionPersistenceRejected {
            identity: "compaction-worker:0".to_string(),
            session_id: "sess-1".to_string(),
            error: "runtime refused the durable compaction projection handoff".to_string(),
            preserved_history: Some(CompactionPreservedHistoryFit::OverWindow),
            attempted_entries: Some(12),
        }
    }

    /// An `ErrorEvent` fired with NO hook registered must still reach the log,
    /// with its typed variant and fields — a paging channel that discards when
    /// nobody wired it is indistinguishable from a healthy fleet. Fired
    /// through `fire_error_hook` (not the log helper directly) so the test
    /// covers the branch that used to drop the event on the floor.
    #[test]
    fn error_event_without_hook_still_reaches_the_log() {
        let slot: SharedErrorHook = Arc::new(std::sync::RwLock::new(None));
        // No hook means no `tokio::spawn`, so this needs no runtime — which is
        // also what keeps the record on this thread where the capture sees it.
        let ((), logged) = capture_tracing(|| fire_error_hook(&slot, sample_alert()));

        assert!(
            logged.contains("CompactionPersistenceRejected"),
            "the typed variant must be named in the record: {logged}"
        );
        assert!(
            logged.contains("OverWindow") && logged.contains("compaction-worker:0"),
            "typed fields must ride the record, not just the Display string: {logged}"
        );
        assert!(
            logged.contains("hook_registered=false"),
            "the record must say nobody was listening: {logged}"
        );
    }

    /// A wired host loses nothing: the record is still emitted, marked as
    /// delivered. Split from the delivery assertion below so that a single
    /// mutation of either behaviour fails its own test.
    #[test]
    fn error_event_with_hook_is_logged_as_delivered() {
        let hook: ErrorHook = Arc::new(move |_event| Box::pin(async move {}));
        let slot: SharedErrorHook = Arc::new(std::sync::RwLock::new(Some(hook)));

        // The dispatch spawns, so this needs a runtime; the log record itself
        // is still written synchronously on this thread.
        let runtime = tokio::runtime::Runtime::new().expect("tokio runtime");
        let _guard = runtime.enter();
        let ((), logged) = capture_tracing(|| fire_error_hook(&slot, sample_alert()));

        assert!(
            logged.contains("hook_registered=true"),
            "a wired host still gets the record, marked as delivered: {logged}"
        );
    }

    /// The default sink is an addition to the hook, not a second delivery
    /// path through it: a registered hook must fire EXACTLY once per event.
    #[tokio::test]
    async fn registered_error_hook_fires_exactly_once() {
        let captured: Arc<tokio::sync::Mutex<Vec<ErrorEvent>>> =
            Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let hook_captured = captured.clone();
        let hook: ErrorHook = Arc::new(move |event| {
            let hook_captured = hook_captured.clone();
            Box::pin(async move {
                hook_captured.lock().await.push(event);
            })
        });
        let slot: SharedErrorHook = Arc::new(std::sync::RwLock::new(Some(hook)));

        fire_error_hook(&slot, sample_alert());

        let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
        loop {
            if !captured.lock().await.is_empty() {
                break;
            }
            assert!(
                tokio::time::Instant::now() < deadline,
                "registered hook never received the event"
            );
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        // Settle any second delivery a double-dispatch bug would produce.
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert_eq!(
            captured.lock().await.len(),
            1,
            "the hook must fire exactly once per event"
        );
    }

    /// The unwired notice must name the condition AND the fix. This is the
    /// same function `UnifiedRuntimeBuilder::build` calls when a host never
    /// registered a hook, so the text is pinned in one place.
    #[test]
    fn error_hook_absent_notice_points_at_the_registration_call() {
        let ((), logged) = capture_tracing(emit_error_hook_absent_notice);

        assert!(
            logged.contains("no error hook is registered"),
            "the notice must state the condition plainly: {logged}"
        );
        assert!(
            logged.contains("on_error"),
            "the notice must point at the registration call: {logged}"
        );
    }

    /// The additive fields must not move the wire shape under an adopter
    /// that predates them: a rejection with no fit verdict serializes
    /// without the new keys, and a payload written before the fields
    /// existed still deserializes.
    #[test]
    fn compaction_rejection_wire_shape_stays_additive() {
        let legacy_json = serde_json::json!({
            "category": "compaction_persistence_rejected",
            "identity": "compaction-worker:0",
            "session_id": "sess-1",
            "error": "compaction curator failed: no summary",
        });

        let alert: ErrorEvent =
            serde_json::from_value(legacy_json.clone()).expect("pre-field payload must load");
        assert_eq!(
            alert,
            ErrorEvent::CompactionPersistenceRejected {
                identity: "compaction-worker:0".to_string(),
                session_id: "sess-1".to_string(),
                error: "compaction curator failed: no summary".to_string(),
                preserved_history: None,
                attempted_entries: None,
            }
        );
        assert_eq!(
            serde_json::to_value(&alert).expect("serialize"),
            legacy_json,
            "an alert with no fit verdict must not emit the new keys"
        );

        let typed = ErrorEvent::CompactionPersistenceRejected {
            identity: "compaction-worker:0".to_string(),
            session_id: "sess-1".to_string(),
            error: "runtime refused the durable compaction projection handoff".to_string(),
            preserved_history: Some(CompactionPreservedHistoryFit::OverWindow),
            attempted_entries: Some(12),
        };
        let encoded = serde_json::to_value(&typed).expect("serialize typed");
        assert_eq!(
            encoded["preserved_history"],
            serde_json::json!("over_window")
        );
        assert_eq!(encoded["attempted_entries"], serde_json::json!(12));
        assert_eq!(
            serde_json::from_value::<ErrorEvent>(encoded).expect("round trip"),
            typed
        );
    }

    fn capturing_error_hook_slot() -> (SharedErrorHook, Arc<tokio::sync::Mutex<Vec<ErrorEvent>>>) {
        let captured: Arc<tokio::sync::Mutex<Vec<ErrorEvent>>> =
            Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let hook_captured = captured.clone();
        let hook: ErrorHook = Arc::new(move |event| {
            let hook_captured = hook_captured.clone();
            Box::pin(async move {
                hook_captured.lock().await.push(event);
            })
        });
        let slot: SharedErrorHook = Arc::new(std::sync::RwLock::new(Some(hook)));
        (slot, captured)
    }

    /// A probe round trip that goes unanswered past its budget must page
    /// `ActorLoopStalled` through the error hook with the waited budget.
    #[tokio::test(start_paused = true)]
    async fn stalled_actor_loop_pages_error_hook() {
        let (slot, captured) = capturing_error_hook_slot();
        let probe_task = tokio::spawn(run_actor_loop_probe(
            std::future::pending::<Result<MobState, MobError>>,
            slot,
            Duration::from_mins(1),
            Duration::from_secs(30),
        ));

        // One interval (60s) + one budget (30s), plus slack for the
        // detached hook task.
        tokio::time::sleep(Duration::from_secs(95)).await;
        tokio::task::yield_now().await;

        let events = captured.lock().await;
        assert_eq!(events.len(), 1, "exactly one stall page: {events:?}");
        match &events[0] {
            ErrorEvent::ActorLoopStalled {
                probe_waited_secs,
                detail,
                stall_id,
                prior_resolved_stalls,
            } => {
                assert_eq!(*probe_waited_secs, 30);
                assert!(
                    detail.contains("QueryPhase"),
                    "detail must name the probe round trip: {detail}"
                );
                // A human reading this page must not conclude "it stopped
                // complaining, so it recovered" — the probe pages once per
                // stall by design, so only the correlated resolution can
                // close it.
                assert!(
                    detail.contains("PAGES ONCE") && detail.contains("silence is NOT recovery"),
                    "detail must say the absence of further pages is not recovery: {detail}"
                );
                assert!(
                    detail.contains("actor_loop_recovered") && detail.contains("stall_id 1"),
                    "detail must name the resolution to wait for, by id: {detail}"
                );
                assert_eq!(
                    *stall_id,
                    Some(1),
                    "the stall must carry the id its resolution will echo"
                );
                assert_eq!(*prior_resolved_stalls, Some(0));
            }
            other => panic!("expected ActorLoopStalled, got {other:?}"),
        }
        drop(events);
        probe_task.abort();
    }

    /// The wedged case, and the reason no counter can name it: the probe
    /// parks on the SAME round trip, so a loop that never answers pages
    /// exactly once and then goes silent — `prior_resolved_stalls` cannot
    /// climb, and "wedged" is the ABSENCE of a resolution on the receiver's
    /// clock. This pins the direction so a future change cannot quietly turn
    /// the count into a wedged proxy.
    #[tokio::test(start_paused = true)]
    async fn wedged_actor_loop_pages_once_and_never_resolves() {
        let (slot, captured) = capturing_error_hook_slot();
        let probe_task = tokio::spawn(run_actor_loop_probe(
            std::future::pending::<Result<MobState, MobError>>,
            slot,
            Duration::from_mins(1),
            Duration::from_secs(30),
        ));

        // Long enough for several more probe cycles had any been possible.
        tokio::time::sleep(Duration::from_mins(10)).await;
        tokio::task::yield_now().await;

        let events = captured.lock().await;
        assert_eq!(
            events.len(),
            1,
            "a wedged loop pages once and cannot page again: {events:?}"
        );
        assert!(
            !events
                .iter()
                .any(|event| matches!(event, ErrorEvent::ActorLoopRecovered { .. })),
            "a wedged loop must never emit a resolution: {events:?}"
        );
        drop(events);
        probe_task.abort();
    }

    /// A stall that later drains must emit a resolution the receiver can
    /// PAIR with the incident it opened. The correlation is the point: an
    /// unpaired "something recovered" cannot close a specific incident, so
    /// the assertion is on matching ids, not merely on both events firing.
    #[tokio::test(start_paused = true)]
    async fn resolved_stall_emits_recovery_correlated_by_stall_id() {
        let (slot, captured) = capturing_error_hook_slot();
        // Answers after 50s: past the 30s budget, so it stalls, then drains.
        let probe_task = tokio::spawn(run_actor_loop_probe(
            || async {
                tokio::time::sleep(Duration::from_secs(50)).await;
                Ok(MobState::Running)
            },
            slot,
            Duration::from_mins(1),
            Duration::from_secs(30),
        ));

        // One interval (60s) + the 50s answer, plus slack for the hook task.
        tokio::time::sleep(Duration::from_mins(2)).await;
        tokio::task::yield_now().await;

        let events = captured.lock().await;
        let stall_id = events
            .iter()
            .find_map(|event| match event {
                ErrorEvent::ActorLoopStalled { stall_id, .. } => Some(*stall_id),
                _ => None,
            })
            .unwrap_or_else(|| panic!("expected a stall page: {events:?}"));
        let (recovered_id, stalled_for_secs) = events
            .iter()
            .find_map(|event| match event {
                ErrorEvent::ActorLoopRecovered {
                    stall_id,
                    stalled_for_secs,
                } => Some((*stall_id, *stalled_for_secs)),
                _ => None,
            })
            .unwrap_or_else(|| panic!("expected a resolution: {events:?}"));

        assert_eq!(
            Some(recovered_id),
            stall_id,
            "the resolution must name the stall it closes, or a receiver \
             cannot close the incident it opened: {events:?}"
        );
        assert!(
            stalled_for_secs >= 50,
            "the resolution must carry how long the loop was stalled, got \
             {stalled_for_secs}s"
        );
        drop(events);
        probe_task.abort();
    }

    /// The resolution is the one variant that reports a failure ending, so
    /// the default sink must not log it at ERROR — a recovery rendered as an
    /// error is a lie about severity and doubles the scary lines per stall.
    #[test]
    fn resolved_stall_is_not_logged_as_an_error() {
        let recovered = ErrorEvent::ActorLoopRecovered {
            stall_id: 7,
            stalled_for_secs: 42,
        };
        let ((), logged) = capture_tracing(|| log_error_event(&recovered, true));

        assert!(
            logged.contains("INFO"),
            "a resolution must log at INFO: {logged}"
        );
        assert!(
            !logged.contains("ERROR"),
            "a resolution must not log at ERROR: {logged}"
        );
        assert!(
            logged.contains("actor_loop_recovered") && logged.contains("42"),
            "the record must still carry the resolution facts: {logged}"
        );
    }

    /// Additivity: an `ActorLoopStalled` payload written before the
    /// correlation fields existed must still load, and a stall carrying them
    /// must round-trip. OB3 pages on this enum today.
    #[test]
    fn actor_loop_stall_wire_shape_stays_additive() {
        let legacy_json = serde_json::json!({
            "category": "actor_loop_stalled",
            "probe_waited_secs": 30,
            "detail": "QueryPhase probe round trip unanswered",
        });

        let alert: ErrorEvent =
            serde_json::from_value(legacy_json.clone()).expect("pre-field payload must load");
        assert_eq!(
            alert,
            ErrorEvent::ActorLoopStalled {
                probe_waited_secs: 30,
                detail: "QueryPhase probe round trip unanswered".to_string(),
                stall_id: None,
                prior_resolved_stalls: None,
            }
        );
        assert_eq!(
            serde_json::to_value(&alert).expect("serialize"),
            legacy_json,
            "a stall with no correlation must not emit the new keys"
        );

        let correlated = ErrorEvent::ActorLoopStalled {
            probe_waited_secs: 30,
            detail: "stalled".to_string(),
            stall_id: Some(3),
            prior_resolved_stalls: Some(2),
        };
        let encoded = serde_json::to_value(&correlated).expect("serialize correlated");
        assert_eq!(encoded["stall_id"], serde_json::json!(3));
        assert_eq!(
            serde_json::from_value::<ErrorEvent>(encoded).expect("round trip"),
            correlated
        );
    }

    /// A responsive command loop must never page across many probe cycles.
    #[tokio::test(start_paused = true)]
    async fn healthy_actor_loop_never_pages() {
        let (slot, captured) = capturing_error_hook_slot();
        let probes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let counted = probes.clone();
        let probe_task = tokio::spawn(run_actor_loop_probe(
            move || {
                counted.fetch_add(1, Ordering::SeqCst);
                std::future::ready(Ok(MobState::Running))
            },
            slot,
            Duration::from_mins(1),
            Duration::from_secs(30),
        ));

        tokio::time::sleep(Duration::from_secs(60 * 5 + 5)).await;
        tokio::task::yield_now().await;

        assert!(
            probes.load(Ordering::SeqCst) >= 4,
            "probe must keep its cadence on a healthy loop: {}",
            probes.load(Ordering::SeqCst)
        );
        assert!(
            captured.lock().await.is_empty(),
            "a healthy loop must not page"
        );
        probe_task.abort();
    }

    /// A stalled probe must never stack another round trip onto the blocked
    /// actor: one page, one parked command, no matter how many cycles pass.
    /// When the parked round trip finally drains, the cadence resumes and
    /// recovery does not page.
    #[tokio::test(start_paused = true)]
    async fn stalled_probe_never_stacks_and_resumes_after_recovery() {
        let (slot, captured) = capturing_error_hook_slot();
        let probes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let released = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let gate = Arc::new(tokio::sync::Notify::new());
        let counted = probes.clone();
        let released_probe = released.clone();
        let gate_probe = gate.clone();
        let probe_task = tokio::spawn(run_actor_loop_probe(
            move || {
                counted.fetch_add(1, Ordering::SeqCst);
                let released = released_probe.clone();
                let gate = gate_probe.clone();
                async move {
                    if !released.load(Ordering::SeqCst) {
                        gate.notified().await;
                    }
                    Ok(MobState::Running)
                }
            },
            slot,
            Duration::from_mins(1),
            Duration::from_secs(30),
        ));

        // Many full interval+budget cycles while the first round trip is
        // parked: no second probe, no second page.
        tokio::time::sleep(Duration::from_mins(10)).await;
        tokio::task::yield_now().await;
        assert_eq!(
            probes.load(Ordering::SeqCst),
            1,
            "a stalled probe must not stack another round trip"
        );
        assert_eq!(
            captured.lock().await.len(),
            1,
            "a persisting stall pages exactly once"
        );

        // Drain the parked round trip; the probe resumes its cadence.
        released.store(true, Ordering::SeqCst);
        gate.notify_one();
        tokio::time::sleep(Duration::from_secs(61)).await;
        tokio::task::yield_now().await;
        assert!(
            probes.load(Ordering::SeqCst) >= 2,
            "probe must resume after the stalled round trip drains: {}",
            probes.load(Ordering::SeqCst)
        );
        // Recovery used to be an info log only. It is now an event, because a
        // receiver that can open an incident but never close it can only
        // escalate — so the drained round trip must produce exactly one
        // resolution, correlated to the stall it closes.
        let events = captured.lock().await;
        assert_eq!(
            events.len(),
            2,
            "the drained stall must produce its resolution: {events:?}"
        );
        let stall_id = match &events[0] {
            ErrorEvent::ActorLoopStalled { stall_id, .. } => *stall_id,
            other => panic!("expected the stall first, got {other:?}"),
        };
        match &events[1] {
            ErrorEvent::ActorLoopRecovered {
                stall_id: closed, ..
            } => {
                assert_eq!(
                    Some(*closed),
                    stall_id,
                    "the resolution must close the stall that opened: {events:?}"
                );
            }
            other => panic!("expected the resolution second, got {other:?}"),
        }
        drop(events);
        probe_task.abort();
    }

    #[test]
    fn probe_env_knob_parses_and_clamps() {
        assert_eq!(
            parse_probe_secs(None, ACTOR_LOOP_PROBE_INTERVAL),
            Duration::from_mins(1)
        );
        assert_eq!(
            parse_probe_secs(Some("120"), ACTOR_LOOP_PROBE_INTERVAL),
            Duration::from_mins(2)
        );
        assert_eq!(
            parse_probe_secs(Some(" 15 "), ACTOR_LOOP_PROBE_BUDGET),
            Duration::from_secs(15)
        );
        assert_eq!(
            parse_probe_secs(Some("0"), ACTOR_LOOP_PROBE_BUDGET),
            Duration::from_secs(1)
        );
        assert_eq!(
            parse_probe_secs(Some("999999"), ACTOR_LOOP_PROBE_BUDGET),
            Duration::from_hours(1)
        );
        assert_eq!(
            parse_probe_secs(Some("junk"), ACTOR_LOOP_PROBE_BUDGET),
            ACTOR_LOOP_PROBE_BUDGET
        );
    }
}