apcore 0.22.0

// APCore Protocol — Event emitter
// Spec reference: Event types and emission

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use tokio::task::JoinHandle;

use super::subscribers::EventSubscriber;
use crate::errors::{ErrorCode, ModuleError};

/// Event type for dead-letter-queue notifications emitted on delivery exhaustion.
const DLQ_EVENT_TYPE: &str = "apcore.event.delivery_failed";

/// An event emitted by the `APCore` system.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ApCoreEvent {
    pub event_type: String,
    /// ISO 8601 timestamp string.
    pub timestamp: String,
    pub data: serde_json::Value,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub module_id: Option<String>,
    pub severity: String,
}

impl ApCoreEvent {
    /// Create a new event with "info" severity.
    pub fn new(event_type: impl Into<String>, data: serde_json::Value) -> Self {
        Self {
            event_type: event_type.into(),
            timestamp: chrono::Utc::now().to_rfc3339(),
            data,
            module_id: None,
            severity: "info".to_string(),
        }
    }

    /// Create a new event with explicit `module_id` and severity.
    pub fn with_module(
        event_type: impl Into<String>,
        data: serde_json::Value,
        module_id: impl Into<String>,
        severity: impl Into<String>,
    ) -> Self {
        Self {
            event_type: event_type.into(),
            timestamp: chrono::Utc::now().to_rfc3339(),
            data,
            module_id: Some(module_id.into()),
            severity: severity.into(),
        }
    }
}

/// Manages event subscribers and dispatches events.
///
/// Subscribers are stored as `Arc<dyn EventSubscriber>` so they can be
/// cheaply cloned into spawned tasks for [`Self::emit_spawn`]'s
/// fire-and-forget dispatch model (sync finding A-D-501).
#[derive(Debug)]
pub struct EventEmitter {
    subscribers: Vec<Arc<dyn EventSubscriber>>,
    pub max_workers: usize,
    /// Set to `true` by [`Self::shutdown`]. Once set, all `emit*` methods
    /// drop incoming events as no-ops (sync finding A-D-502).
    is_shutdown: Arc<AtomicBool>,
    /// In-flight delivery tasks spawned by [`Self::emit`]. [`Self::flush`]
    /// awaits these (up to a timeout) so callers can wait for pending
    /// deliveries to drain (sync findings A-D-024 / A-D-027). Completed
    /// handles are pruned lazily on each `emit`/`flush`.
    pending: Arc<Mutex<Vec<JoinHandle<()>>>>,
}

impl EventEmitter {
    /// Create a new event emitter.
    #[must_use]
    pub fn new() -> Self {
        Self {
            subscribers: vec![],
            max_workers: 4,
            is_shutdown: Arc::new(AtomicBool::new(false)),
            pending: Arc::new(Mutex::new(Vec::new())),
        }
    }

    /// Add a subscriber (matching Python's void return signature).
    pub fn subscribe(&mut self, subscriber: Box<dyn EventSubscriber>) {
        // Convert Box -> Arc so subscribers can be cloned into spawned
        // tasks by `emit_spawn` (sync finding A-D-501).
        self.subscribers.push(Arc::from(subscriber));
    }

    /// Remove the first subscriber whose `subscriber_id()` matches the given
    /// subscriber's ID.
    ///
    /// **Cross-language divergence (A-D-030, accepted Rust constraint):**
    /// apcore-python / apcore-typescript remove by exact *instance* identity.
    /// Rust stores subscribers as `Arc<dyn EventSubscriber>` trait objects, and
    /// pointer identity is not preserved across the `Box`→`Arc` conversion in
    /// [`Self::subscribe`], so instance matching is not feasible without a
    /// handle-returning API. Identity is therefore approximated by
    /// `subscriber_id()`; callers needing independent removal MUST assign each
    /// subscriber a unique id.
    pub fn unsubscribe(&mut self, subscriber: &dyn EventSubscriber) -> bool {
        let target_id = subscriber.subscriber_id();
        self.unsubscribe_by_id(target_id)
    }

    /// Remove the first subscriber whose `subscriber_id()` matches the given ID string.
    pub fn unsubscribe_by_id(&mut self, subscriber_id: &str) -> bool {
        let pos = self
            .subscribers
            .iter()
            .position(|s| s.subscriber_id() == subscriber_id);
        if let Some(i) = pos {
            self.subscribers.remove(i);
            true
        } else {
            false
        }
    }

    /// Remove all subscribers whose `event_type_filter()` equals `event_type`.
    ///
    /// Returns the number of subscribers removed. Matches Python/TypeScript
    /// `off(event_type)` semantics where passing an event-type string removes
    /// all handlers bound to that type.
    pub fn unsubscribe_by_event_type(&mut self, event_type: &str) -> usize {
        let before = self.subscribers.len();
        self.subscribers
            .retain(|s| s.event_type_filter().is_none_or(|t| t != event_type));
        before - self.subscribers.len()
    }

    /// Emit an event to all subscribers whose pattern matches the event type.
    ///
    /// Canonical delivery path — applies the per-subscriber retry policy and
    /// emits an `apcore.event.delivery_failed` DLQ event on exhaustion, per
    /// spec docs/features/event-system.md §Event Delivery Semantics (#61).
    ///
    /// **Non-blocking (sync finding A-D-024):** each matching subscriber's
    /// retry loop (including backoff sleeps) runs on its own spawned task, so
    /// this method returns immediately without waiting for subscriber
    /// execution — matching apcore-python (`ThreadPoolExecutor.submit`) and
    /// apcore-typescript (fire-and-forget). Use [`Self::flush`] to wait for
    /// pending deliveries to drain.
    ///
    /// **Post-shutdown behaviour:** if [`Self::shutdown`] has been called,
    /// this method returns immediately as a no-op (sync finding A-D-502).
    ///
    /// Spec event-system.md:448 declares `No errors raised`. Subscriber
    /// errors are caught, retried per the per-subscriber `retry` config, and
    /// surfaced via DLQ + `on_failure` only after exhaustion. The return
    /// type is unit (D10-008).
    ///
    /// `async fn` is retained for API parity and so callers may `.await` it;
    /// the body itself does not block on subscriber execution.
    #[allow(clippy::unused_async)]
    pub async fn emit(&self, event: &ApCoreEvent) {
        if self.is_shutdown.load(Ordering::SeqCst) {
            return;
        }
        let all_subscribers: Vec<Arc<dyn EventSubscriber>> = self.subscribers.clone();
        let mut handles: Vec<JoinHandle<()>> = Vec::new();
        for subscriber in &self.subscribers {
            if Self::matches_pattern(subscriber.event_pattern(), &event.event_type) {
                let sub = Arc::clone(subscriber);
                let evt = event.clone();
                let dlq_subs = all_subscribers.clone();
                handles.push(tokio::spawn(async move {
                    Self::deliver_with_dlq(sub, evt, dlq_subs).await;
                }));
            }
        }
        // Track in-flight tasks so flush() can await them; prune finished ones.
        let mut pending = self.pending.lock();
        pending.retain(|h| !h.is_finished());
        pending.extend(handles);
    }

    /// Single-attempt sequential emit — bypasses the per-subscriber retry
    /// policy and DLQ machinery.
    ///
    /// Use this in tests that need strict ordering and deterministic
    /// single-shot delivery semantics. Production code SHOULD call
    /// [`Self::emit`] (canonical retry + DLQ) instead.
    ///
    /// Errors from individual subscribers are logged but not propagated
    /// (error isolation), matching Python's behaviour.
    pub async fn emit_sequential(&self, event: &ApCoreEvent) {
        if self.is_shutdown.load(Ordering::SeqCst) {
            return;
        }
        for subscriber in &self.subscribers {
            if Self::matches_pattern(subscriber.event_pattern(), &event.event_type) {
                if let Err(e) = subscriber.on_event(event).await {
                    tracing::warn!(
                        subscriber_id = %subscriber.subscriber_id(),
                        event_type = %event.event_type,
                        error = %e,
                        "event subscriber failed"
                    );
                }
            }
        }
    }

    /// Emit an event to subscribers matching both the caller's filter pattern
    /// AND the subscriber's own `event_pattern`.
    ///
    /// Applies the same retry + DLQ semantics as [`Self::emit`].
    ///
    /// **Post-shutdown behaviour:** drops the event as a no-op (sync
    /// finding A-D-502).
    pub async fn emit_filtered(
        &self,
        event: &ApCoreEvent,
        pattern: &str,
    ) -> Result<(), ModuleError> {
        if self.is_shutdown.load(Ordering::SeqCst) {
            return Ok(());
        }
        let all_subscribers: Vec<Arc<dyn EventSubscriber>> = self.subscribers.clone();
        for subscriber in &self.subscribers {
            if Self::matches_pattern(pattern, &event.event_type)
                && Self::matches_pattern(subscriber.event_pattern(), &event.event_type)
            {
                Self::deliver_with_dlq(
                    Arc::clone(subscriber),
                    event.clone(),
                    all_subscribers.clone(),
                )
                .await;
            }
        }
        Ok(())
    }

    /// Flush all pending event deliveries, waiting up to `timeout_ms`
    /// milliseconds for the tasks spawned by [`Self::emit`] to complete.
    ///
    /// Drains the tracked in-flight delivery tasks (sync findings A-D-024 /
    /// A-D-027), matching apcore-python's `flush()` which waits on each pending
    /// future. If the overall timeout elapses before all tasks finish, the
    /// remaining (still-running) tasks are left in place and `flush` returns
    /// `Ok(())` — delivery continues in the background, mirroring Python which
    /// swallows per-future timeouts. A `timeout_ms` of 0 waits indefinitely.
    pub async fn flush(&self, timeout_ms: u64) -> Result<(), ModuleError> {
        // Take ownership of the current in-flight handles.
        let handles: Vec<JoinHandle<()>> = {
            let mut pending = self.pending.lock();
            std::mem::take(&mut *pending)
        };
        if handles.is_empty() {
            return Ok(());
        }

        let deadline = if timeout_ms == 0 {
            None
        } else {
            Some(tokio::time::Instant::now() + Duration::from_millis(timeout_ms))
        };

        let mut unfinished: Vec<JoinHandle<()>> = Vec::new();
        for handle in handles {
            match deadline {
                None => {
                    let _ = handle.await;
                }
                Some(dl) => {
                    let now = tokio::time::Instant::now();
                    if now >= dl {
                        // Budget exhausted — keep remaining tasks running.
                        unfinished.push(handle);
                        continue;
                    }
                    // Await the task up to the deadline. On either outcome
                    // (completed or per-task timeout) the result is discarded:
                    // a timed-out delivery continues detached in the background
                    // (the JoinHandle is consumed by timeout_at).
                    let _ = tokio::time::timeout_at(dl, handle).await;
                }
            }
        }

        if !unfinished.is_empty() {
            let mut pending = self.pending.lock();
            // Re-track any tasks we did not get to await, plus any new ones.
            unfinished.extend(std::mem::take(&mut *pending));
            *pending = unfinished;
        }
        Ok(())
    }

    /// Fire-and-forget dispatch: spawns one `tokio::task` per matching
    /// subscriber and returns immediately.
    ///
    /// Use this when the caller cannot wait for subscriber completion (the
    /// canonical "fire-and-forget" path called out by the spec, sync finding
    /// A-D-501). Errors from subscribers are logged via `tracing::warn!` and
    /// never propagated. Subscribers run **concurrently** because each is
    /// driven on its own task.
    ///
    /// `emit_spawn` is the preferred dispatch path for runtime modules
    /// (system modules, circuit breaker, etc.); the sequential `emit` is
    /// retained for tests that need deterministic ordering.
    ///
    /// **Post-shutdown behaviour:** drops the event as a no-op once
    /// [`Self::shutdown`] has been called (sync finding A-D-502).
    // Takes `event: ApCoreEvent` by value because each spawned task needs an
    // owned `ApCoreEvent`; passing by reference would force the caller to
    // clone, hiding the cost. The spec's pseudocode signature also
    // matches by-value (sync finding A-D-501).
    #[allow(clippy::needless_pass_by_value)]
    pub fn emit_spawn(&self, event: ApCoreEvent) {
        if self.is_shutdown.load(Ordering::SeqCst) {
            return;
        }
        for subscriber in &self.subscribers {
            if !Self::matches_pattern(subscriber.event_pattern(), &event.event_type) {
                continue;
            }
            let sub = Arc::clone(subscriber);
            let evt = event.clone();
            tokio::spawn(async move {
                if let Err(e) = sub.on_event(&evt).await {
                    tracing::warn!(
                        subscriber_id = %sub.subscriber_id(),
                        event_type = %evt.event_type,
                        error = %e,
                        "emit_spawn: subscriber on_event failed"
                    );
                }
            });
        }
    }

    /// Mark this emitter as shut down and flush any pending work.
    ///
    /// After `shutdown` returns:
    /// - Subsequent calls to `emit`, `emit_filtered`, and `emit_spawn` are
    ///   no-ops — the event is dropped.
    /// - Subsequent calls to `shutdown` return `Ok(())` immediately
    ///   (idempotent).
    ///
    /// `timeout_ms` bounds the wait for in-flight deliveries to drain via
    /// [`Self::flush`]. After the flush completes (or times out), no new
    /// events are accepted (sync finding A-D-502). Kept for API parity with
    /// apcore-python and apcore-typescript.
    pub async fn shutdown(&mut self, timeout_ms: u64) -> Result<(), ModuleError> {
        if self.is_shutdown.swap(true, Ordering::SeqCst) {
            // Already shut down — idempotent.
            return Ok(());
        }
        self.flush(timeout_ms).await
    }

    /// Returns `true` if this emitter has been shut down.
    #[must_use]
    pub fn is_shutdown(&self) -> bool {
        self.is_shutdown.load(Ordering::SeqCst)
    }

    /// Fire-and-forget dispatch with full delivery semantics (retry + DLQ).
    ///
    /// For each matching subscriber, spawns a `tokio::task` running the full
    /// per-subscriber retry loop. On exhaustion (regardless of `max_attempts`):
    /// - a `apcore.event.delivery_failed` DLQ event is delivered to any
    ///   subscriber whose pattern matches that event type, EXCEPT catch-all
    ///   `'*'` subscribers (sync findings A-D-025 / A-D-026).
    /// - `on_failure` is called.
    ///
    /// Spawned tasks are tracked so [`Self::flush`] can await them.
    ///
    /// **Post-shutdown:** drops the event as a no-op.
    #[allow(clippy::needless_pass_by_value)]
    pub fn emit_delivery_semantics(&self, event: ApCoreEvent) {
        if self.is_shutdown.load(Ordering::SeqCst) {
            return;
        }
        // Capture a snapshot for both the delivery loop and DLQ delivery.
        let all_subscribers: Vec<Arc<dyn EventSubscriber>> = self.subscribers.clone();

        let mut handles: Vec<JoinHandle<()>> = Vec::new();
        for subscriber in &self.subscribers {
            if !Self::matches_pattern(subscriber.event_pattern(), &event.event_type) {
                continue;
            }
            let sub = Arc::clone(subscriber);
            let evt = event.clone();
            let dlq_subs = all_subscribers.clone();
            handles.push(tokio::spawn(async move {
                Self::deliver_with_dlq(sub, evt, dlq_subs).await;
            }));
        }
        let mut pending = self.pending.lock();
        pending.retain(|h| !h.is_finished());
        pending.extend(handles);
    }

    /// Deliver an event to one subscriber with retry + optional DLQ emission.
    async fn deliver_with_dlq(
        subscriber: Arc<dyn EventSubscriber>,
        event: ApCoreEvent,
        all_subscribers: Vec<Arc<dyn EventSubscriber>>,
    ) {
        let retry = subscriber.retry();
        let mut last_error: Option<ModuleError> = None;

        for attempt in 0..retry.max_attempts {
            match subscriber.on_event(&event).await {
                Ok(()) => return,
                Err(e) => {
                    last_error = Some(e);
                    if attempt + 1 < retry.max_attempts {
                        tokio::time::sleep(Duration::from_millis(retry.compute_delay_ms(attempt)))
                            .await;
                    }
                }
            }
        }

        let err = last_error.unwrap_or_else(|| {
            ModuleError::new(ErrorCode::GeneralInternalError, "unknown delivery failure")
        });

        // A-D-025: always emit the DLQ event on exhaustion, regardless of
        // max_attempts (including single-attempt subscribers). Matches
        // apcore-python / apcore-typescript, which emit the DLQ whenever a
        // subscriber's delivery attempts are exhausted.
        let sub_id = subscriber.subscriber_id().to_string();
        // A-D-029: use the subscriber's DECLARED type rather than parsing the id.
        let subscriber_type = subscriber.subscriber_type().to_string();
        let dlq_event = ApCoreEvent::new(
            DLQ_EVENT_TYPE,
            serde_json::json!({
                "subscriber_type": subscriber_type,
                "subscriber_id": sub_id,
                // A-D-009: original_event MUST use the spec wire keys
                // name/payload/metadata (event-system.md, conformance fixture
                // event_delivery_semantics.json). The original event's
                // module_id/timestamp are preserved under metadata so no
                // information is lost. Mirrors Python emitter.py original_event.
                "original_event": {
                    "name": event.event_type,
                    "payload": event.data,
                    "metadata": {
                        "module_id": event.module_id,
                        "timestamp": event.timestamp,
                    },
                },
                "error": {
                    "type": format!("{:?}", err.code),
                    "message": err.message,
                },
                "attempt_count": retry.max_attempts,
                "timestamp": chrono::Utc::now().to_rfc3339(),
            }),
        );
        for dlq_sub in &all_subscribers {
            // A-D-026: exclude catch-all '*' subscribers from DLQ delivery to
            // prevent cascading failures where every wildcard subscriber would
            // recursively receive a DLQ about itself. Matches apcore-python.
            if dlq_sub.event_pattern() == "*" {
                continue;
            }
            if Self::matches_pattern(dlq_sub.event_pattern(), DLQ_EVENT_TYPE) {
                // DLQ delivery is single-attempt (no retry, no second-order DLQ).
                if let Err(e) = dlq_sub.on_event(&dlq_event).await {
                    tracing::error!(
                        subscriber_id = %dlq_sub.subscriber_id(),
                        error = %e,
                        "DLQ subscriber on_event failed (discarded, not retried)"
                    );
                }
            }
        }

        subscriber
            .on_failure(&event, &err, retry.max_attempts)
            .await;
    }

    /// Simple glob-style pattern matching with `*` wildcard.
    ///
    /// - `"*"` matches everything.
    /// - `"foo.*"` matches `"foo.bar"`, `"foo.baz"`, etc.
    /// - An exact string matches only itself.
    fn matches_pattern(pattern: &str, event_type: &str) -> bool {
        if pattern == "*" {
            return true;
        }
        // Split pattern by '*' and check that all parts appear in order.
        let parts: Vec<&str> = pattern.split('*').collect();
        let mut remaining = event_type;
        for (i, part) in parts.iter().enumerate() {
            if part.is_empty() {
                continue;
            }
            if i == 0 {
                // First part must be a prefix.
                if let Some(rest) = remaining.strip_prefix(part) {
                    remaining = rest;
                } else {
                    return false;
                }
            } else if let Some(pos) = remaining.find(part) {
                remaining = &remaining[pos + part.len()..];
            } else {
                return false;
            }
        }
        // If pattern doesn't end with *, remaining must be empty.
        if !pattern.ends_with('*') && !remaining.is_empty() {
            return false;
        }
        true
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use parking_lot::Mutex;
    use serde_json::json;
    use std::sync::Arc;

    #[derive(Debug, Clone)]
    struct RecordingSubscriber {
        id: String,
        pattern: String,
        received: Arc<Mutex<Vec<String>>>,
    }

    impl RecordingSubscriber {
        fn new(id: &str, pattern: &str) -> Self {
            Self {
                id: id.to_string(),
                pattern: pattern.to_string(),
                received: Arc::new(Mutex::new(Vec::new())),
            }
        }
    }

    #[async_trait]
    impl EventSubscriber for RecordingSubscriber {
        fn subscriber_id(&self) -> &str {
            &self.id
        }
        fn event_pattern(&self) -> &str {
            &self.pattern
        }
        async fn on_event(&self, event: &ApCoreEvent) -> Result<(), ModuleError> {
            self.received.lock().push(event.event_type.clone());
            Ok(())
        }
    }

    #[test]
    fn test_event_new_defaults() {
        let event = ApCoreEvent::new("test.event", json!({"key": "val"}));
        assert_eq!(event.event_type, "test.event");
        assert_eq!(event.severity, "info");
        assert!(event.module_id.is_none());
        assert!(!event.timestamp.is_empty());
    }

    #[test]
    fn test_event_with_module() {
        let event = ApCoreEvent::with_module("err.event", json!({}), "mod.a", "error");
        assert_eq!(event.event_type, "err.event");
        assert_eq!(event.severity, "error");
        assert_eq!(event.module_id.as_deref(), Some("mod.a"));
    }

    #[test]
    fn test_event_serialization_skips_none_module_id() {
        let event = ApCoreEvent::new("test", json!(null));
        let serialized = serde_json::to_value(&event).unwrap();
        assert!(serialized.get("module_id").is_none());
    }

    #[test]
    fn test_emitter_default_max_workers() {
        let emitter = EventEmitter::new();
        assert_eq!(emitter.max_workers, 4);
    }

    #[tokio::test]
    async fn test_emit_to_matching_subscriber() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "test.*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub));

        let event = ApCoreEvent::new("test.hello", json!({}));
        emitter.emit(&event).await;
        emitter.flush(5000).await.unwrap();
        assert_eq!(received.lock().len(), 1);
        assert_eq!(received.lock()[0], "test.hello");
    }

    #[tokio::test]
    async fn test_emit_skips_non_matching_subscriber() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "other.*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub));

        let event = ApCoreEvent::new("test.hello", json!({}));
        emitter.emit(&event).await;
        assert!(received.lock().is_empty());
    }

    #[tokio::test]
    async fn test_emit_wildcard_matches_all() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub));

        let event = ApCoreEvent::new("anything.at.all", json!({}));
        emitter.emit(&event).await;
        emitter.flush(5000).await.unwrap();
        assert_eq!(received.lock().len(), 1);
    }

    #[tokio::test]
    async fn test_unsubscribe_by_id() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "*");
        emitter.subscribe(Box::new(sub));
        assert!(emitter.unsubscribe_by_id("sub1"));
        assert!(!emitter.unsubscribe_by_id("sub1"));
    }

    #[tokio::test]
    async fn test_unsubscribe_removes_subscriber() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub.clone()));
        emitter.unsubscribe(&sub);

        let event = ApCoreEvent::new("test", json!({}));
        emitter.emit(&event).await;
        assert!(received.lock().is_empty());
    }

    #[tokio::test]
    async fn test_emit_filtered() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "test.*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub));

        let event = ApCoreEvent::new("test.hello", json!({}));
        emitter.emit_filtered(&event, "test.*").await.unwrap();
        assert_eq!(received.lock().len(), 1);

        emitter.emit_filtered(&event, "other.*").await.unwrap();
        assert_eq!(received.lock().len(), 1);
    }

    #[tokio::test]
    async fn test_flush_succeeds() {
        let emitter = EventEmitter::new();
        emitter.flush(1000).await.unwrap();
    }

    #[test]
    fn test_matches_pattern_wildcard() {
        assert!(EventEmitter::matches_pattern("*", "anything"));
    }

    #[test]
    fn test_matches_pattern_exact() {
        assert!(EventEmitter::matches_pattern("test.event", "test.event"));
        assert!(!EventEmitter::matches_pattern("test.event", "test.other"));
    }

    #[test]
    fn test_matches_pattern_prefix_wildcard() {
        assert!(EventEmitter::matches_pattern("test.*", "test.hello"));
        assert!(EventEmitter::matches_pattern("test.*", "test."));
        assert!(!EventEmitter::matches_pattern("test.*", "other.hello"));
    }

    #[test]
    fn test_matches_pattern_suffix_wildcard() {
        assert!(EventEmitter::matches_pattern("*.event", "test.event"));
        assert!(!EventEmitter::matches_pattern("*.event", "test.other"));
    }

    #[test]
    fn test_matches_pattern_middle_wildcard() {
        assert!(EventEmitter::matches_pattern("a.*.z", "a.b.z"));
        assert!(EventEmitter::matches_pattern("a.*.z", "a.anything.z"));
        assert!(!EventEmitter::matches_pattern("a.*.z", "a.b.c"));
    }

    #[tokio::test]
    async fn test_emit_error_isolation() {
        #[derive(Debug)]
        struct FailingSub;

        #[async_trait]
        impl EventSubscriber for FailingSub {
            fn subscriber_id(&self) -> &'static str {
                "fail"
            }
            fn event_pattern(&self) -> &'static str {
                "*"
            }
            async fn on_event(&self, _event: &ApCoreEvent) -> Result<(), ModuleError> {
                Err(ModuleError::new(
                    crate::errors::ErrorCode::GeneralInternalError,
                    "boom",
                ))
            }
        }

        let mut emitter = EventEmitter::new();
        emitter.subscribe(Box::new(FailingSub));
        let good_sub = RecordingSubscriber::new("good", "*");
        let received = good_sub.received.clone();
        emitter.subscribe(Box::new(good_sub));

        let event = ApCoreEvent::new("test", json!({}));
        // Use the sequential single-attempt path so this test focuses on
        // error isolation without invoking the retry+DLQ machinery (which
        // is exercised by tests/test_v022_event_delivery_semantics.rs).
        emitter.emit_sequential(&event).await;
        assert_eq!(received.lock().len(), 1);
    }
}