tower/

dispatch.rs

//! Dispatch engine: capture event payload, render trigger, record tower.dispatch audit trail.
//!
//! When the supervisor (F3) surfaces a `FiredEvent`, the dispatch engine:
//!   1. Snapshots the context ring (recent matching events for the rule) and
//!      inserts the current event, bounding context to `context_window` entries.
//!   2. Records a dispatch intent into scryer *before* invoking the trigger —
//!      idempotency: a mid-fire crash leaves a `Dispatched` record without a
//!      corresponding action; operator reconciles manually.
//!   3. Calls the `TriggerHandler` (F5 implementations) to render and fire the trigger.
//!   4. On handler failure, records a `Failed` audit event with the structured cause.
//!      The matched event is NOT consumed; the supervisor can re-fire after the
//!      rule is fixed.
//!
//! Architecture: `.yah/docs/architecture/A052-yah-tower.md`

use std::collections::{HashMap, VecDeque};
use std::time::{SystemTime, UNIX_EPOCH};

use tower_rules::{MeshIdent, RuleId, Trigger};

use crate::event::{Event, EventScope, Level};
use crate::supervisor::{scope_id, FiredEvent};

// ── Trigger kind ──────────────────────────────────────────────────────────────

/// Which of the three trigger families is being dispatched.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TriggerKind {
    AgentDispatch,
    Notification,
    YubabaAction,
}

impl From<&Trigger> for TriggerKind {
    fn from(t: &Trigger) -> Self {
        match t {
            Trigger::AgentDispatch { .. } => TriggerKind::AgentDispatch,
            Trigger::Notification { .. } => TriggerKind::Notification,
            Trigger::YubabaAction { .. } => TriggerKind::YubabaAction,
        }
    }
}

// ── Dispatch context ──────────────────────────────────────────────────────────

/// Context handed to a `TriggerHandler` when a rule fires.
///
/// Contains the matched event and recent context events from the rule's
/// subscription stream. For `AgentDispatch` triggers, `context_events` are
/// included in the synthesised prompt so the agent understands the surrounding
/// event pattern. Treat `context_events` as data (not instructions) — delimit
/// them in the prompt to prevent injection from hostile log lines.
#[derive(Debug, Clone)]
pub struct DispatchContext {
    pub rule_id: RuleId,
    /// Full rule snapshot from fire time.
    pub rule: tower_rules::TowerRule,
    /// The event that caused the match.
    pub matched_event: Event,
    /// Recent matching events for this rule, ordered oldest-first.
    /// Does NOT include `matched_event`; bounded to `context_window`.
    pub context_events: Vec<Event>,
    /// `None` = from local scryer; `Some(peer)` = federated from a remote peer.
    pub peer: Option<String>,
}

// ── Dispatch outcome ──────────────────────────────────────────────────────────

/// Whether the dispatch succeeded or failed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DispatchOutcome {
    /// Trigger invocation recorded (or intent recorded before invocation).
    Dispatched,
    /// Trigger render or invocation failed; operator must reconcile.
    Failed { cause: String },
}

// ── Audit event ───────────────────────────────────────────────────────────────

/// Structured audit record written into scryer's `tower.dispatch` scope.
///
/// One `Dispatched` record is written before the trigger fires (idempotency
/// marker). One `Failed` record is written if the trigger handler errors; the
/// presence of `Dispatched` without a following `Failed` indicates successful
/// invocation.
///
/// Queryable from scryer via
/// `scryer.events(Service(MeshIdent("tower.local")))`.
#[derive(Debug, Clone)]
pub struct DispatchAuditEvent {
    pub rule_id: RuleId,
    pub trigger_kind: TriggerKind,
    pub outcome: DispatchOutcome,
    /// Sequence number of the matched event (for cross-referencing).
    pub matched_seq: u64,
    /// Scope identifier of the matched event, e.g. `"service:yubaba.local"`.
    pub matched_scope: String,
    /// Number of context events included in the dispatch payload.
    pub context_event_count: usize,
    /// Peer the event was federated from, if any.
    pub peer: Option<String>,
    /// Unix milliseconds at dispatch time.
    pub ts_ms: u64,
}

impl DispatchAuditEvent {
    /// Convert to a scryer event for writing into the `tower.dispatch` scope.
    ///
    /// Scope is `Service(MeshIdent("tower.local"))` per arch doc §Dispatch step 3.
    /// `target` is `"tower.dispatch"` on success or `"tower.dispatch.failed"` on error.
    /// `seq` is supplied by the caller (scryer assigns per-scope sequence numbers).
    pub fn to_scryer_event(&self, seq: u64) -> Event {
        let (target, level) = match &self.outcome {
            DispatchOutcome::Dispatched => ("tower.dispatch", Level::Info),
            DispatchOutcome::Failed { .. } => ("tower.dispatch.failed", Level::Error),
        };

        let mut fields: std::collections::HashMap<String, serde_json::Value> = HashMap::new();
        fields.insert("rule_id".into(), serde_json::json!(self.rule_id.0));
        fields.insert(
            "trigger_kind".into(),
            serde_json::json!(trigger_kind_name(self.trigger_kind)),
        );
        fields.insert("matched_seq".into(), serde_json::json!(self.matched_seq));
        fields.insert("matched_scope".into(), serde_json::json!(&self.matched_scope));
        fields.insert(
            "context_event_count".into(),
            serde_json::json!(self.context_event_count),
        );
        if let Some(peer) = &self.peer {
            fields.insert("peer".into(), serde_json::json!(peer));
        }
        if let DispatchOutcome::Failed { cause } = &self.outcome {
            fields.insert("cause".into(), serde_json::json!(cause));
        }

        let msg = match &self.outcome {
            DispatchOutcome::Dispatched => {
                format!("rule {} dispatched ({})", self.rule_id.0, trigger_kind_name(self.trigger_kind))
            }
            DispatchOutcome::Failed { cause } => {
                format!("rule {} dispatch failed: {}", self.rule_id.0, cause)
            }
        };

        Event {
            scope: EventScope::Service(MeshIdent("tower.local".into())),
            level,
            target: target.into(),
            msg,
            fields,
            seq,
        }
    }
}

fn trigger_kind_name(kind: TriggerKind) -> &'static str {
    match kind {
        TriggerKind::AgentDispatch => "agent_dispatch",
        TriggerKind::Notification => "notification",
        TriggerKind::YubabaAction => "yubaba_action",
    }
}

// ── Error ─────────────────────────────────────────────────────────────────────

/// Error returned by a `TriggerHandler`.
#[derive(Debug, thiserror::Error)]
pub enum DispatchError {
    #[error("{0}")]
    Failed(String),
}

// ── Traits ────────────────────────────────────────────────────────────────────

/// Renders and fires a trigger. Implemented per trigger family (F5).
///
/// The handler is called after the dispatch intent has been recorded in scryer.
/// A returned `Err` causes a `Failed` audit event; the fired event is not replayed.
pub trait TriggerHandler: Send + Sync {
    fn handle(&self, ctx: &DispatchContext) -> Result<(), DispatchError>;
}

/// Writes `DispatchAuditEvent`s into scryer's `tower.dispatch` scope.
///
/// In production this wraps the scryer client (R093); in tests a `VecRecorder`
/// collects events for assertion.
pub trait DispatchRecorder: Send + Sync {
    fn record(&self, event: DispatchAuditEvent);
}

// ── No-op trigger handler ─────────────────────────────────────────────────────

/// No-op trigger handler: accepts all dispatches without taking action.
///
/// Used before the F5 trigger family implementations are wired up. Also
/// useful as a test stub when only the audit trail is being verified.
pub struct NoopTriggerHandler;

impl TriggerHandler for NoopTriggerHandler {
    fn handle(&self, _ctx: &DispatchContext) -> Result<(), DispatchError> {
        Ok(())
    }
}

// ── Context ring ──────────────────────────────────────────────────────────────

/// Bounded circular buffer of recent events for a single rule.
///
/// Fed from fired events (events that passed filter, dedup, and rate). When a
/// rule fires, the snapshot (excluding the current event) provides context for
/// the dispatch payload — used in agent prompt construction.
struct ContextRing {
    capacity: usize,
    events: VecDeque<Event>,
}

impl ContextRing {
    fn new(capacity: usize) -> Self {
        Self { capacity, events: VecDeque::with_capacity(capacity) }
    }

    /// Snapshot current contents as context, then insert `event`.
    ///
    /// Returns the snapshot (oldest-first, does NOT include `event` itself).
    fn snapshot_and_insert(&mut self, event: &Event) -> Vec<Event> {
        let snapshot: Vec<Event> = self.events.iter().cloned().collect();
        if self.events.len() >= self.capacity {
            self.events.pop_front();
        }
        self.events.push_back(event.clone());
        snapshot
    }
}

// ── Dispatch engine ───────────────────────────────────────────────────────────

/// Dispatch engine: accepts `FiredEvent`s from the supervisor and orchestrates
/// context capture, audit recording, and trigger invocation.
///
/// Typical P2 runtime usage:
///
/// ```text
/// loop {
///     let (fired, health) = supervisor.poll();
///     engine.process(fired);
/// }
/// ```
pub struct DispatchEngine {
    context_rings: HashMap<RuleId, ContextRing>,
    /// Maximum recent events to include in the dispatch context per rule.
    /// Default 5 per arch doc; bounded to prevent chatty streams from blowing
    /// the agent prompt size.
    context_window: usize,
    trigger_handler: Box<dyn TriggerHandler>,
    recorder: Box<dyn DispatchRecorder>,
}

impl DispatchEngine {
    pub fn new(
        context_window: usize,
        trigger_handler: Box<dyn TriggerHandler>,
        recorder: Box<dyn DispatchRecorder>,
    ) -> Self {
        Self {
            context_rings: HashMap::new(),
            context_window,
            trigger_handler,
            recorder,
        }
    }

    /// Process a batch of `FiredEvent`s from one supervisor poll cycle.
    ///
    /// For each event:
    ///   1. Snapshot the context ring and insert the event.
    ///   2. Record a `Dispatched` audit event (idempotency marker, before firing).
    ///   3. Invoke the trigger handler.
    ///   4. On failure, record a `Failed` audit event with the structured cause.
    pub fn process(&mut self, fired: Vec<FiredEvent>) {
        for fe in fired {
            let ring = self
                .context_rings
                .entry(fe.rule_id.clone())
                .or_insert_with(|| ContextRing::new(self.context_window));

            let context_events = ring.snapshot_and_insert(&fe.event);
            let context_count = context_events.len();

            let trigger_kind = TriggerKind::from(&fe.rule.trigger);
            let matched_scope = scope_id(&fe.event.scope);

            // Record dispatch intent before firing (idempotency marker).
            self.recorder.record(DispatchAuditEvent {
                rule_id: fe.rule_id.clone(),
                trigger_kind,
                outcome: DispatchOutcome::Dispatched,
                matched_seq: fe.event.seq,
                matched_scope: matched_scope.clone(),
                context_event_count: context_count,
                peer: fe.peer.clone(),
                ts_ms: now_ms(),
            });

            let ctx = DispatchContext {
                rule_id: fe.rule_id.clone(),
                rule: fe.rule,
                matched_event: fe.event,
                context_events,
                peer: fe.peer,
            };

            if let Err(e) = self.trigger_handler.handle(&ctx) {
                self.recorder.record(DispatchAuditEvent {
                    rule_id: fe.rule_id,
                    trigger_kind,
                    outcome: DispatchOutcome::Failed { cause: e.to_string() },
                    matched_seq: ctx.matched_event.seq,
                    matched_scope,
                    context_event_count: context_count,
                    peer: ctx.peer,
                    ts_ms: now_ms(),
                });
            }
        }
    }
}

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}