yah-tower 0.8.20

//! End-to-end acceptance tests for R095-F7 — the first tower rule.
//!
//! @yah:ticket(R182-B9, "Tower integration test fails: include_str embeds missing .yah/cloud/rules/yubaba-quorum-loss.yaml")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-18T16:34:46Z)
//! @yah:status(review)
//! @yah:parent(R182)
//! @yah:next("Create .yah/cloud/rules/yubaba-quorum-loss.yaml or change include_str to a fixture file that's committed")
//! @yah:handoff("Added Compound(Or) predicate to .yah/infra/rules/yubaba-quorum-loss.yaml: scryer_health arm (YubabaRaftQuorumLost) + event_match arm (scope: yubaba.local, level: error, target: raft.quorum_lost exact). All 6 tower integration tests now pass.")
//! @yah:verify("cargo test -p tower --test integration")
//!
//! @yah:ticket(R182-B11, "tower: yubaba_event_path test fails — canonical rule missing EventMatch arm")
//! @yah:at(2026-05-18T16:37:04Z)
//! @yah:status(review)
//! @yah:parent(R182)
//! @yah:next("Add Compound(Or) predicate to yubaba-quorum-loss.yaml combining scryer_health + EventMatch(raft.quorum_lost), then verify tower integration tests all pass")
//! @yah:handoff("Compound(Or) predicate added to yubaba-quorum-loss.yaml as part of B9 fix — same YAML change covers this ticket. All 6 tower integration tests pass.")
//! @yah:verify("cargo test -p tower --test integration")
// Double-underscore names mirror the verify condition exactly (`tower_first_rule__local`).
#![allow(non_snake_case)]
//!
//! Validates the full tower pipeline using the canonical yubaba-quorum-loss
//! rule from `.yah/cloud/rules/yubaba-quorum-loss.yaml`:
//!
//!   parse YAML → validate → load into supervisor → inject event → poll →
//!   dispatch engine → notification handler → assert Critical DesktopBadge
//!
//! ## Coverage
//!
//! - `canonical_rule_parses` — YAML parses and passes semantic validation.
//! - `tower_first_rule__local` — ScryerHealth path: the named acceptance test
//!   from the R095-F7 verify condition.
//! - `tower_first_rule__yubaba_event_path` — EventMatch path: yubaba_rpc
//!   adapter emits `raft.quorum_lost` under `yubaba.local`; Compound(Or)
//!   predicate fires on either arm.
//! - `tower_first_rule__dedup_suppresses_replay` — same-seq event does not
//!   fire twice (idempotency guarantee from scryer substrate).
//!
//! ## The 30s / 3-node acceptance criterion (P2)
//!
//! The ticket spec requires a test using R091 `#[test_with_provider(local)]`
//! that spins a real 3-node yubaba mesh, kills 2 nodes, and asserts a desktop
//! notification within 30s of quorum loss.  That test is blocked on R091
//! yubaba infrastructure (P2).  These tests cover every tower layer with
//! trait-based mocks.  When R091 ships, add `tests/integration_yubaba.rs`:
//!
//! ```text
//! #[test_with_provider(local)]
//! async fn tower_first_rule__yubaba_mesh() {
//!     let mesh = LocalMesh::start_3_node().await;
//!     mesh.kill_nodes(2).await;
//!     let notif = mesh.await_notification(Duration::from_secs(30)).await;
//!     assert_eq!(notif.severity, Severity::Critical);
//! }
//! ```

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use tower::dispatch::{DispatchAuditEvent, DispatchEngine, DispatchRecorder};
use tower::event::{Event, EventScope, Level};
use tower::rules::parse::parse_rule_yaml;
use tower::scryer_filter::ScryerFilter;
use tower::supervisor::{EventStream, NoPeers, SubscriptionSource, Supervisor};
use tower::triggers::{NotificationHandler, NotificationPayload, NotificationSink, NotifyError};
use tower_rules::*;

// ── Canonical rule ─────────────────────────────────────────────────────────────

/// The operator-installed rule file.  Embedded at compile time so any edit to
/// the canonical YAML is covered by this test suite.
const CANONICAL_RULE_YAML: &str =
    include_str!("../../../../.yah/infra/rules/yubaba-quorum-loss.yaml");

// ── Shared test infrastructure ─────────────────────────────────────────────────

/// Channel-backed event stream.
struct TestStream(std::sync::mpsc::Receiver<Event>);

impl EventStream for TestStream {
    fn try_next(&mut self) -> Option<Event> {
        self.0.try_recv().ok()
    }
}

/// One mpsc channel per rule; tests inject events via `send_to`.
struct TestSource {
    senders: Mutex<HashMap<RuleId, std::sync::mpsc::Sender<Event>>>,
}

impl TestSource {
    fn new() -> Arc<Self> {
        Arc::new(Self { senders: Mutex::new(HashMap::new()) })
    }

    fn send_to(&self, rule_id: &RuleId, event: Event) {
        if let Some(tx) = self.senders.lock().unwrap().get(rule_id) {
            let _ = tx.send(event);
        }
    }
}

impl SubscriptionSource for TestSource {
    fn subscribe(&self, _filter: ScryerFilter, rule_id: &RuleId) -> Box<dyn EventStream> {
        let (tx, rx) = std::sync::mpsc::channel();
        self.senders.lock().unwrap().insert(rule_id.clone(), tx);
        Box::new(TestStream(rx))
    }
}

/// Collects (channel, payload) pairs for assertion.
#[derive(Clone)]
struct RecordingNotificationSink {
    calls: Arc<Mutex<Vec<(NotificationChannel, NotificationPayload)>>>,
}

impl RecordingNotificationSink {
    fn new() -> Self {
        Self { calls: Arc::new(Mutex::new(Vec::new())) }
    }

    fn calls(&self) -> Vec<(NotificationChannel, NotificationPayload)> {
        self.calls.lock().unwrap().clone()
    }
}

impl NotificationSink for RecordingNotificationSink {
    fn notify(
        &self,
        channel: &NotificationChannel,
        payload: &NotificationPayload,
    ) -> Result<(), NotifyError> {
        self.calls.lock().unwrap().push((channel.clone(), payload.clone()));
        Ok(())
    }
}

/// No-op audit recorder for tests that don't inspect the dispatch trail.
struct NullRecorder;

impl DispatchRecorder for NullRecorder {
    fn record(&self, _: DispatchAuditEvent) {}
}

// ── Builder helpers ────────────────────────────────────────────────────────────

/// Parse the canonical rule and load it into a fresh Supervisor + TestSource.
/// Returns `(supervisor, source, rule_id)`.
fn setup_supervisor(
) -> (Supervisor, Arc<TestSource>, RuleId) {
    let rule = parse_rule_yaml(CANONICAL_RULE_YAML).expect("canonical YAML must parse");
    let rule_id = rule.id.clone();
    let source = TestSource::new();
    let mut sup = Supervisor::new(source.clone(), Arc::new(NoPeers), 256, 100);
    sup.load_rule(rule).expect("canonical rule must load into supervisor");
    (sup, source, rule_id)
}

/// Build a DispatchEngine wired to a fresh RecordingNotificationSink.
fn setup_engine() -> (DispatchEngine, RecordingNotificationSink) {
    let sink = RecordingNotificationSink::new();
    let handler = NotificationHandler::new(Box::new(sink.clone()));
    let engine = DispatchEngine::new(5, Box::new(handler), Box::new(NullRecorder));
    (engine, sink)
}

// ── Tests ──────────────────────────────────────────────────────────────────────

/// The canonical YAML parses without error and passes semantic validation.
#[test]
fn canonical_rule_parses() {
    let rule = parse_rule_yaml(CANONICAL_RULE_YAML).expect("parse must succeed");
    assert_eq!(rule.id, RuleId("yubaba-quorum-loss".into()));
    assert!(rule.enabled, "canonical rule must be enabled by default");
    assert_eq!(rule.debounce_ms, Some(30_000));
    assert!(
        matches!(rule.trigger, Trigger::Notification { severity: Severity::Critical, .. }),
        "trigger must be a Critical notification"
    );
    if let Trigger::Notification { channels, .. } = &rule.trigger {
        assert!(
            channels.contains(&NotificationChannel::DesktopBadge),
            "DesktopBadge must be a configured channel"
        );
    }
    let warnings = tower::rules::validate::validate(&rule)
        .expect("canonical rule must pass validation");
    assert!(
        warnings.is_empty(),
        "canonical rule must produce no validation warnings: {warnings:?}"
    );
}

/// Full pipeline: ScryerHealth signal → supervisor → dispatch → Critical notification.
///
/// This is the `tower_first_rule__local` acceptance test named in the
/// R095-F7 verify condition.  It simulates what happens when yubaba loses raft
/// quorum via scryer's health monitor:
///
///   3-node yubaba mesh → 2 nodes killed → quorum lost →
///   scryer health monitor emits YubabaRaftQuorumLost →
///   tower subscription fires → Critical DesktopBadge dispatched
///
/// In production the event flows from the yubaba raft layer through scryer's
/// health emission channel.  Here the TestSource injects the health event
/// directly into the per-rule subscription stream.
#[test]
fn tower_first_rule__local() {
    let (mut sup, source, rule_id) = setup_supervisor();
    let (mut engine, sink) = setup_engine();

    // Inject the health signal that scryer emits when yubaba loses quorum.
    // Equivalent to: 2 of 3 yubaba nodes becoming unreachable → raft majority gone.
    source.send_to(&rule_id, Event::health(HealthSignal::YubabaRaftQuorumLost, 1));

    let (fired, _health) = sup.poll();
    assert_eq!(fired.len(), 1, "quorum-loss health signal must fire the rule");

    engine.process(fired);

    let calls = sink.calls();
    assert_eq!(calls.len(), 1, "exactly one notification expected");
    assert!(
        matches!(calls[0].0, NotificationChannel::DesktopBadge),
        "expected DesktopBadge notification channel, got {:?}",
        calls[0].0
    );
    assert_eq!(
        calls[0].1.severity,
        Severity::Critical,
        "quorum loss must be a Critical notification"
    );
    assert!(
        calls[0].1.rule_name.contains("quorum"),
        "notification rule_name must reference quorum: {:?}",
        calls[0].1.rule_name
    );
}

/// Full pipeline: yubaba_rpc EventMatch path → supervisor → dispatch → Critical notification.
///
/// Exercises the second arm of the Compound(Or) predicate: the
/// `adapter::yubaba_rpc` adapter emits a structured `raft.quorum_lost` event
/// into the `yubaba.local` Service scope.  Validates that the EventMatch
/// predicate matches correctly and that the same Critical notification fires.
///
/// In production: yubaba detects quorum loss → emits structured event via
/// in-process channel → scryer ingests it under `Service(MeshIdent("yubaba.local"))`.
#[test]
fn tower_first_rule__yubaba_event_path() {
    let (mut sup, source, rule_id) = setup_supervisor();
    let (mut engine, sink) = setup_engine();

    // Simulate the yubaba_rpc adapter emitting a structured quorum-loss event.
    source.send_to(
        &rule_id,
        Event {
            scope: EventScope::Service(MeshIdent("yubaba.local".into())),
            level: Level::Error,
            target: "raft.quorum_lost".into(),
            msg: "yubaba raft quorum lost: 1 of 3 peers reachable".into(),
            fields: {
                let mut m = HashMap::new();
                m.insert("peers_reachable".into(), serde_json::json!(1));
                m.insert("peers_total".into(), serde_json::json!(3));
                m
            },
            seq: 1,
        },
    );

    let (fired, _health) = sup.poll();
    assert_eq!(fired.len(), 1, "yubaba RPC quorum-loss event must fire the rule");

    engine.process(fired);

    let calls = sink.calls();
    assert_eq!(calls.len(), 1, "exactly one notification expected");
    assert!(matches!(calls[0].0, NotificationChannel::DesktopBadge));
    assert_eq!(calls[0].1.severity, Severity::Critical);
}

/// Dedup suppresses a replay of the same event (same scope + seq).
///
/// The scryer substrate guarantees `(scope_id, seq)` uniqueness per the
/// R093-F1 contract.  The supervisor's dedup ring uses this to prevent the
/// same event from firing the rule twice even if the subscription stream
/// delivers it more than once (e.g. on reconnect).
#[test]
fn tower_first_rule__dedup_suppresses_replay() {
    let (mut sup, source, rule_id) = setup_supervisor();
    let (mut engine, sink) = setup_engine();

    // First delivery — must fire.
    source.send_to(&rule_id, Event::health(HealthSignal::YubabaRaftQuorumLost, 42));
    let (fired, _) = sup.poll();
    assert_eq!(fired.len(), 1, "first delivery must fire");
    engine.process(fired);
    assert_eq!(sink.calls().len(), 1, "first notification dispatched");

    // Replay: same event, same seq — must be suppressed.
    source.send_to(&rule_id, Event::health(HealthSignal::YubabaRaftQuorumLost, 42));
    let (fired2, _) = sup.poll();
    assert_eq!(fired2.len(), 0, "replay of seq=42 must be suppressed by dedup ring");

    engine.process(fired2);
    assert_eq!(sink.calls().len(), 1, "no second notification on replay");
}

/// Wrong severity: events below the error level do not match the EventMatch arm.
///
/// The Compound(Or) has an EventMatch arm requiring `level >= error`.  A Warn-
/// level event from `yubaba.local` must not fire the notification.
#[test]
fn tower_first_rule__warn_event_does_not_fire() {
    let (mut sup, source, rule_id) = setup_supervisor();
    let (mut engine, sink) = setup_engine();

    source.send_to(
        &rule_id,
        Event {
            scope: EventScope::Service(MeshIdent("yubaba.local".into())),
            level: Level::Warn,
            target: "raft.quorum_lost".into(),
            msg: "quorum approaching threshold".into(),
            fields: HashMap::new(),
            seq: 1,
        },
    );

    let (fired, _) = sup.poll();
    assert_eq!(fired.len(), 0, "warn-level event must not fire the error-gated rule");
    engine.process(fired);
    assert!(sink.calls().is_empty(), "no notification for warn-level event");
}

/// Wrong target: a different yubaba.local error does not fire the rule.
#[test]
fn tower_first_rule__unrelated_yubaba_error_does_not_fire() {
    let (mut sup, source, rule_id) = setup_supervisor();
    let (mut engine, sink) = setup_engine();

    source.send_to(
        &rule_id,
        Event {
            scope: EventScope::Service(MeshIdent("yubaba.local".into())),
            level: Level::Error,
            target: "workload.deploy_failed".into(),
            msg: "deploy failed for noisetable-api".into(),
            fields: HashMap::new(),
            seq: 1,
        },
    );

    let (fired, _) = sup.poll();
    assert_eq!(fired.len(), 0, "unrelated error target must not match exact target filter");
    engine.process(fired);
    assert!(sink.calls().is_empty());
}