ai-harness/crates/harness-core/src/event.rs

use serde::Serialize;
use tokio::sync::broadcast;

use crate::types::{Message, Part, PartId, Session, SessionId};

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RunOutcome {
    Stopped,
    Aborted,
    Errored { message: String },
}

#[derive(Debug, Clone, Copy, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Level {
    Info,
    Warn,
    Error,
}

#[derive(Debug, Clone, Serialize)]
pub struct PermissionRequest {
    pub id: String,
    pub session_id: SessionId,
    pub permission: String,
    pub pattern: String,
    pub always_pattern: String,
    pub metadata: serde_json::Value,
}

#[derive(Debug, Clone, Serialize)]
pub struct JobRecordEvent(pub serde_json::Value);

/// Serializable so a future HTTP server can pipe this straight to SSE.
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum AppEvent {
    SessionCreated {
        session: Session,
    },
    SessionUpdated {
        session: Session,
    },
    MessageCreated {
        message: Message,
    },
    MessageUpdated {
        message: Message,
    },
    PartUpdated {
        part: Part,
    },
    PartDelta {
        part_id: PartId,
        message_id: crate::types::MessageId,
        delta: String,
    },
    RunStarted {
        session_id: SessionId,
    },
    RunFinished {
        session_id: SessionId,
        outcome: RunOutcome,
    },
    PermissionAsked {
        request: PermissionRequest,
    },
    PermissionResolved {
        id: String,
    },
    JobUpdated {
        job: JobRecordEvent,
    },
    AuthPrompt {
        provider: String,
        user_code: String,
        verification_uri: String,
    },
    ServerNotice {
        level: Level,
        text: String,
    },
}

const EVENT_BUS_CAPACITY: usize = 1024;

/// Thin wrapper over `broadcast`; subscribers that lag just miss old events (`RecvError::Lagged`).
#[derive(Clone)]
pub struct EventBus(broadcast::Sender<AppEvent>);

impl EventBus {
    pub fn new() -> Self {
        let (tx, _rx) = broadcast::channel(EVENT_BUS_CAPACITY);
        Self(tx)
    }

    pub fn publish(&self, event: AppEvent) {
        // No subscribers is a normal state (e.g. headless `harness run`); ignore the error.
        let _ = self.0.send(event);
    }

    pub fn subscribe(&self) -> broadcast::Receiver<AppEvent> {
        self.0.subscribe()
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn publish_reaches_subscriber() {
        let bus = EventBus::new();
        let mut rx = bus.subscribe();
        bus.publish(AppEvent::RunStarted {
            session_id: SessionId::new(),
        });
        let event = rx.recv().await.unwrap();
        assert!(matches!(event, AppEvent::RunStarted { .. }));
    }

    #[tokio::test]
    async fn publish_with_no_subscribers_does_not_panic() {
        let bus = EventBus::new();
        bus.publish(AppEvent::RunStarted {
            session_id: SessionId::new(),
        });
    }

    #[tokio::test]
    async fn multiple_subscribers_each_get_the_event() {
        let bus = EventBus::new();
        let mut rx1 = bus.subscribe();
        let mut rx2 = bus.subscribe();
        bus.publish(AppEvent::PermissionResolved { id: "abc".into() });
        assert!(matches!(
            rx1.recv().await.unwrap(),
            AppEvent::PermissionResolved { .. }
        ));
        assert!(matches!(
            rx2.recv().await.unwrap(),
            AppEvent::PermissionResolved { .. }
        ));
    }
}