feat(api): add socket subscriptions and lifecycle events

2 files changed, 623 insertions, 38 deletions

diff --git a/src/api/mod.rs b/src/api/mod.rs
index 642dc70d..e2c29490 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -4,13 +4,16 @@ use std::fs;
 use std::io::{BufRead, BufReader, Write};
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
+use std::time::Duration;
 
 use tracing::{debug, error, info, warn};
 
 use regex::Regex;
 
 use crate::api::schema::{
-    ErrorBody, ErrorResponse, Method, Request, ResponseResult, SuccessResponse,
+    ErrorBody, ErrorResponse, Method, PaneAgentStateChangedEvent, PaneOutputMatchedEvent, Request,
+    ResponseResult, Subscription, SubscriptionEventData, SubscriptionEventEnvelope,
+    SubscriptionEventKind, SuccessResponse,
 };
 
 pub const SOCKET_PATH_ENV_VAR: &str = "HERDR_SOCKET_PATH";
@@ -20,6 +23,46 @@ pub struct ApiRequestMessage {
     pub respond_to: std::sync::mpsc::Sender<String>,
 }
 
+#[derive(Clone, Default)]
+pub struct EventHub {
+    inner: std::sync::Arc<std::sync::Mutex<EventHubState>>,
+}
+
+#[derive(Default)]
+struct EventHubState {
+    next_sequence: u64,
+    events: Vec<(u64, crate::api::schema::EventEnvelope)>,
+}
+
+impl EventHub {
+    const MAX_EVENTS: usize = 512;
+
+    pub fn push(&self, event: crate::api::schema::EventEnvelope) {
+        let Ok(mut state) = self.inner.lock() else {
+            return;
+        };
+        state.next_sequence += 1;
+        let sequence = state.next_sequence;
+        state.events.push((sequence, event));
+        let overflow = state.events.len().saturating_sub(Self::MAX_EVENTS);
+        if overflow > 0 {
+            state.events.drain(0..overflow);
+        }
+    }
+
+    pub fn events_after(&self, sequence: u64) -> Vec<(u64, crate::api::schema::EventEnvelope)> {
+        let Ok(state) = self.inner.lock() else {
+            return Vec::new();
+        };
+        state
+            .events
+            .iter()
+            .filter(|(event_sequence, _)| *event_sequence > sequence)
+            .cloned()
+            .collect()
+    }
+}
+
 pub fn socket_path() -> PathBuf {
     if let Ok(path) = std::env::var(SOCKET_PATH_ENV_VAR) {
         return PathBuf::from(path);
@@ -57,6 +100,7 @@ impl Drop for ServerHandle {
 
 pub fn start_server(
     api_tx: std::sync::mpsc::Sender<ApiRequestMessage>,
+    event_hub: EventHub,
 ) -> std::io::Result<ServerHandle> {
     let path = socket_path();
     prepare_socket_path(&path)?;
@@ -68,9 +112,13 @@ pub fn start_server(
         for stream in listener.incoming() {
             match stream {
                 Ok(stream) => {
-                    if let Err(err) = handle_connection(stream, &api_tx) {
-                        warn!(err = %err, "api connection failed");
-                    }
+                    let api_tx = api_tx.clone();
+                    let event_hub = event_hub.clone();
+                    std::thread::spawn(move || {
+                        if let Err(err) = handle_connection(stream, &api_tx, &event_hub) {
+                            warn!(err = %err, "api connection failed");
+                        }
+                    });
                 }
                 Err(err) => {
                     error!(err = %err, "api listener accept failed");
@@ -104,6 +152,7 @@ fn prepare_socket_path(path: &Path) -> std::io::Result<()> {
 fn handle_connection(
     mut stream: UnixStream,
     api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
+    event_hub: &EventHub,
 ) -> std::io::Result<()> {
     let mut line = String::new();
     {
@@ -119,21 +168,41 @@ fn handle_connection(
         return Ok(());
     }
 
-    let response = match serde_json::from_str::<Request>(line) {
-        Ok(request) => handle_request(request, api_tx),
-        Err(err) => serde_json::to_string(&ErrorResponse {
-            id: String::new(),
-            error: ErrorBody {
-                code: "invalid_request".into(),
-                message: format!("invalid request: {err}"),
-            },
-        })?,
+    let request = match serde_json::from_str::<Request>(line) {
+        Ok(request) => request,
+        Err(err) => {
+            write_json_line(
+                &mut stream,
+                &ErrorResponse {
+                    id: String::new(),
+                    error: ErrorBody {
+                        code: "invalid_request".into(),
+                        message: format!("invalid request: {err}"),
+                    },
+                },
+            )?;
+            return Ok(());
+        }
     };
 
-    stream.write_all(response.as_bytes())?;
-    stream.write_all(b"\n")?;
-    stream.flush()?;
-    Ok(())
+    match request.method {
+        Method::EventsSubscribe(params) => {
+            stream_subscriptions(stream, request.id, params, api_tx, event_hub)
+        }
+        method => {
+            let response = handle_request(
+                Request {
+                    id: request.id,
+                    method,
+                },
+                api_tx,
+            );
+            stream.write_all(response.as_bytes())?;
+            stream.write_all(b"\n")?;
+            stream.flush()?;
+            Ok(())
+        }
+    }
 }
 
 fn handle_request(request: Request, api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>) -> String {
@@ -213,19 +282,7 @@ fn wait_for_output(
             .unwrap();
         };
 
-        let matched_line = match &params.r#match {
-            crate::api::schema::OutputMatch::Substring { value } => read
-                .text
-                .lines()
-                .find(|line| line.contains(value))
-                .map(|line| line.to_string()),
-            crate::api::schema::OutputMatch::Regex { .. } => regex.as_ref().and_then(|re| {
-                read.text
-                    .lines()
-                    .find(|line| re.is_match(line))
-                    .map(|line| line.to_string())
-            }),
-        };
+        let matched_line = match_output(&read.text, &params.r#match, regex.as_ref());
         if matched_line.is_some() {
             let revision = read.revision;
             return serde_json::to_string(&SuccessResponse {
@@ -255,6 +312,393 @@ fn wait_for_output(
     }
 }
 
+fn stream_subscriptions(
+    mut stream: UnixStream,
+    request_id: String,
+    params: crate::api::schema::EventsSubscribeParams,
+    api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
+    event_hub: &EventHub,
+) -> std::io::Result<()> {
+    let mut subscriptions = Vec::with_capacity(params.subscriptions.len());
+    for (index, subscription) in params.subscriptions.into_iter().enumerate() {
+        let active =
+            match ActiveSubscription::new(subscription, &request_id, index, api_tx, event_hub) {
+                Ok(active) => active,
+                Err(response) => {
+                    write_json_line(&mut stream, &response)?;
+                    return Ok(());
+                }
+            };
+        subscriptions.push(active);
+    }
+
+    write_json_line(
+        &mut stream,
+        &SuccessResponse {
+            id: request_id,
+            result: ResponseResult::SubscriptionStarted {},
+        },
+    )?;
+
+    loop {
+        for subscription in &mut subscriptions {
+            if let Some(event) = subscription.poll(api_tx, event_hub) {
+                write_json_line(&mut stream, &event)?;
+            }
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+}
+
+fn write_json_line<T: serde::Serialize>(stream: &mut UnixStream, value: &T) -> std::io::Result<()> {
+    let encoded = serde_json::to_string(value)
+        .map_err(|err| std::io::Error::other(format!("failed to encode json: {err}")))?;
+    stream.write_all(encoded.as_bytes())?;
+    stream.write_all(b"\n")?;
+    stream.flush()?;
+    Ok(())
+}
+
+fn match_output(
+    text: &str,
+    matcher: &crate::api::schema::OutputMatch,
+    regex: Option<&Regex>,
+) -> Option<String> {
+    match matcher {
+        crate::api::schema::OutputMatch::Substring { value } => text
+            .lines()
+            .find(|line| line.contains(value))
+            .map(|line| line.to_string()),
+        crate::api::schema::OutputMatch::Regex { .. } => regex.and_then(|re| {
+            text.lines()
+                .find(|line| re.is_match(line))
+                .map(|line| line.to_string())
+        }),
+    }
+}
+
+struct ActiveOutputMatchedSubscription {
+    pane_id: String,
+    source: crate::api::schema::ReadSource,
+    lines: Option<u32>,
+    matcher: crate::api::schema::OutputMatch,
+    regex: Option<Regex>,
+    strip_ansi: bool,
+    currently_matching: bool,
+    request_prefix: String,
+}
+
+struct ActiveAgentStateChangedSubscription {
+    pane_id: String,
+    state_filter: Option<crate::api::schema::PaneAgentState>,
+    last_state: Option<crate::api::schema::PaneAgentState>,
+    request_prefix: String,
+}
+
+struct ActiveEventSubscription {
+    event_kind: crate::api::schema::EventKind,
+    last_sequence: u64,
+}
+
+enum ActiveSubscription {
+    Event(ActiveEventSubscription),
+    OutputMatched(ActiveOutputMatchedSubscription),
+    AgentStateChanged(ActiveAgentStateChangedSubscription),
+}
+
+impl ActiveSubscription {
+    fn new(
+        subscription: Subscription,
+        request_id: &str,
+        index: usize,
+        api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
+        _event_hub: &EventHub,
+    ) -> Result<Self, ErrorResponse> {
+        match subscription {
+            Subscription::WorkspaceCreated {} => Ok(Self::Event(ActiveEventSubscription {
+                event_kind: crate::api::schema::EventKind::WorkspaceCreated,
+                last_sequence: 0,
+            })),
+            Subscription::WorkspaceClosed {} => Ok(Self::Event(ActiveEventSubscription {
+                event_kind: crate::api::schema::EventKind::WorkspaceClosed,
+                last_sequence: 0,
+            })),
+            Subscription::WorkspaceFocused {} => Ok(Self::Event(ActiveEventSubscription {
+                event_kind: crate::api::schema::EventKind::WorkspaceFocused,
+                last_sequence: 0,
+            })),
+            Subscription::PaneCreated {} => Ok(Self::Event(ActiveEventSubscription {
+                event_kind: crate::api::schema::EventKind::PaneCreated,
+                last_sequence: 0,
+            })),
+            Subscription::PaneClosed {} => Ok(Self::Event(ActiveEventSubscription {
+                event_kind: crate::api::schema::EventKind::PaneClosed,
+                last_sequence: 0,
+            })),
+            Subscription::PaneFocused {} => Ok(Self::Event(ActiveEventSubscription {
+                event_kind: crate::api::schema::EventKind::PaneFocused,
+                last_sequence: 0,
+            })),
+            Subscription::PaneExited {} => Ok(Self::Event(ActiveEventSubscription {
+                event_kind: crate::api::schema::EventKind::PaneExited,
+                last_sequence: 0,
+            })),
+            Subscription::PaneAgentDetected {} => Ok(Self::Event(ActiveEventSubscription {
+                event_kind: crate::api::schema::EventKind::PaneAgentDetected,
+                last_sequence: 0,
+            })),
+            Subscription::PaneOutputMatched {
+                pane_id,
+                source,
+                lines,
+                r#match,
+                strip_ansi,
+            } => {
+                let regex = match &r#match {
+                    crate::api::schema::OutputMatch::Regex { value } => match Regex::new(value) {
+                        Ok(regex) => Some(regex),
+                        Err(err) => {
+                            return Err(ErrorResponse {
+                                id: request_id.to_string(),
+                                error: ErrorBody {
+                                    code: "invalid_regex".into(),
+                                    message: err.to_string(),
+                                },
+                            });
+                        }
+                    },
+                    crate::api::schema::OutputMatch::Substring { .. } => None,
+                };
+
+                let probe = pane_read(
+                    format!("{request_id}:sub:{index}:probe"),
+                    &pane_id,
+                    source.clone(),
+                    lines,
+                    strip_ansi,
+                    api_tx,
+                );
+                if let Err(error) = probe {
+                    return Err(error);
+                }
+
+                Ok(Self::OutputMatched(ActiveOutputMatchedSubscription {
+                    pane_id,
+                    source,
+                    lines,
+                    matcher: r#match,
+                    regex,
+                    strip_ansi,
+                    currently_matching: false,
+                    request_prefix: format!("{request_id}:sub:{index}"),
+                }))
+            }
+            Subscription::PaneAgentStateChanged { pane_id, state } => {
+                let probe =
+                    match pane_get(format!("{request_id}:sub:{index}:probe"), &pane_id, api_tx) {
+                        Ok(probe) => probe,
+                        Err(error) => return Err(error),
+                    };
+
+                Ok(Self::AgentStateChanged(
+                    ActiveAgentStateChangedSubscription {
+                        pane_id,
+                        state_filter: state,
+                        last_state: Some(probe.agent_state),
+                        request_prefix: format!("{request_id}:sub:{index}"),
+                    },
+                ))
+            }
+        }
+    }
+
+    fn poll(
+        &mut self,
+        api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
+        event_hub: &EventHub,
+    ) -> Option<serde_json::Value> {
+        match self {
+            Self::Event(subscription) => subscription.poll(event_hub),
+            Self::OutputMatched(subscription) => {
+                serde_json::to_value(subscription.poll(api_tx)?).ok()
+            }
+            Self::AgentStateChanged(subscription) => {
+                serde_json::to_value(subscription.poll(api_tx)?).ok()
+            }
+        }
+    }
+}
+
+impl ActiveEventSubscription {
+    fn poll(&mut self, event_hub: &EventHub) -> Option<serde_json::Value> {
+        for (sequence, event) in event_hub.events_after(self.last_sequence) {
+            self.last_sequence = sequence;
+            if event.event == self.event_kind {
+                return serde_json::to_value(event).ok();
+            }
+        }
+        None
+    }
+}
+
+impl ActiveOutputMatchedSubscription {
+    fn poll(
+        &mut self,
+        api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
+    ) -> Option<SubscriptionEventEnvelope> {
+        let read = pane_read(
+            format!("{}:read", self.request_prefix),
+            &self.pane_id,
+            self.source.clone(),
+            self.lines,
+            self.strip_ansi,
+            api_tx,
+        )
+        .ok()?;
+
+        let matched_line = match_output(&read.text, &self.matcher, self.regex.as_ref());
+        match matched_line {
+            Some(matched_line) => {
+                if self.currently_matching {
+                    return None;
+                }
+                self.currently_matching = true;
+                Some(SubscriptionEventEnvelope {
+                    event: SubscriptionEventKind::PaneOutputMatched,
+                    data: SubscriptionEventData::PaneOutputMatched(PaneOutputMatchedEvent {
+                        pane_id: self.pane_id.clone(),
+                        matched_line,
+                        read,
+                    }),
+                })
+            }
+            None => {
+                self.currently_matching = false;
+                None
+            }
+        }
+    }
+}
+
+impl ActiveAgentStateChangedSubscription {
+    fn poll(
+        &mut self,
+        api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
+    ) -> Option<SubscriptionEventEnvelope> {
+        let pane = pane_get(
+            format!("{}:pane", self.request_prefix),
+            &self.pane_id,
+            api_tx,
+        )
+        .ok()?;
+        let current_state = pane.agent_state;
+        let previous_state = self.last_state.replace(current_state);
+        if previous_state.is_none() || previous_state == Some(current_state) {
+            return None;
+        }
+        if self
+            .state_filter
+            .is_some_and(|wanted| wanted != current_state)
+        {
+            return None;
+        }
+
+        Some(SubscriptionEventEnvelope {
+            event: SubscriptionEventKind::PaneAgentStateChanged,
+            data: SubscriptionEventData::PaneAgentStateChanged(PaneAgentStateChangedEvent {
+                pane_id: pane.pane_id,
+                workspace_id: pane.workspace_id,
+                state: current_state,
+                agent: pane.agent,
+            }),
+        })
+    }
+}
+
+fn pane_read(
+    request_id: String,
+    pane_id: &str,
+    source: crate::api::schema::ReadSource,
+    lines: Option<u32>,
+    strip_ansi: bool,
+    api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
+) -> Result<crate::api::schema::PaneReadResult, ErrorResponse> {
+    let response = dispatch_to_app(
+        Request {
+            id: request_id.clone(),
+            method: Method::PaneRead(crate::api::schema::PaneReadParams {
+                pane_id: pane_id.to_string(),
+                source,
+                lines,
+                strip_ansi,
+            }),
+        },
+        api_tx,
+    );
+    let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse {
+        id: request_id.clone(),
+        error: ErrorBody {
+            code: "internal_error".into(),
+            message: "failed to decode pane read response".into(),
+        },
+    })?;
+    if value.get("error").is_some() {
+        return serde_json::from_value(value).map_err(|_| ErrorResponse {
+            id: request_id,
+            error: ErrorBody {
+                code: "internal_error".into(),
+                message: "failed to decode pane read error".into(),
+            },
+        });
+    }
+    serde_json::from_value(value["result"]["read"].clone()).map_err(|_| ErrorResponse {
+        id: request_id,
+        error: ErrorBody {
+            code: "internal_error".into(),
+            message: "failed to decode pane read result".into(),
+        },
+    })
+}
+
+fn pane_get(
+    request_id: String,
+    pane_id: &str,
+    api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
+) -> Result<crate::api::schema::PaneInfo, ErrorResponse> {
+    let response = dispatch_to_app(
+        Request {
+            id: request_id.clone(),
+            method: Method::PaneGet(crate::api::schema::PaneTarget {
+                pane_id: pane_id.to_string(),
+            }),
+        },
+        api_tx,
+    );
+    let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse {
+        id: request_id.clone(),
+        error: ErrorBody {
+            code: "internal_error".into(),
+            message: "failed to decode pane get response".into(),
+        },
+    })?;
+    if value.get("error").is_some() {
+        return serde_json::from_value(value).map_err(|_| ErrorResponse {
+            id: request_id,
+            error: ErrorBody {
+                code: "internal_error".into(),
+                message: "failed to decode pane get error".into(),
+            },
+        });
+    }
+    serde_json::from_value(value["result"]["pane"].clone()).map_err(|_| ErrorResponse {
+        id: request_id,
+        error: ErrorBody {
+            code: "internal_error".into(),
+            message: "failed to decode pane get result".into(),
+        },
+    })
+}
+
 fn dispatch_to_app(
     request: Request,
     api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>,
diff --git a/src/api/schema.rs b/src/api/schema.rs
index 29509211..7de0bbed 100644
--- a/src/api/schema.rs
+++ b/src/api/schema.rs
@@ -132,11 +132,44 @@ pub enum ReadSource {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct EventsSubscribeParams {
-    pub events: Vec<EventKind>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub workspace_id: Option<String>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub pane_id: Option<String>,
+    pub subscriptions: Vec<Subscription>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum Subscription {
+    #[serde(rename = "workspace.created")]
+    WorkspaceCreated {},
+    #[serde(rename = "workspace.closed")]
+    WorkspaceClosed {},
+    #[serde(rename = "workspace.focused")]
+    WorkspaceFocused {},
+    #[serde(rename = "pane.created")]
+    PaneCreated {},
+    #[serde(rename = "pane.closed")]
+    PaneClosed {},
+    #[serde(rename = "pane.focused")]
+    PaneFocused {},
+    #[serde(rename = "pane.exited")]
+    PaneExited {},
+    #[serde(rename = "pane.agent_detected")]
+    PaneAgentDetected {},
+    #[serde(rename = "pane.output_matched")]
+    PaneOutputMatched {
+        pane_id: String,
+        source: ReadSource,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        lines: Option<u32>,
+        r#match: OutputMatch,
+        #[serde(default = "default_true")]
+        strip_ansi: bool,
+    },
+    #[serde(rename = "pane.agent_state_changed")]
+    PaneAgentStateChanged {
+        pane_id: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        state: Option<PaneAgentState>,
+    },
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -270,9 +303,7 @@ pub enum ResponseResult {
     PaneRead {
         read: PaneReadResult,
     },
-    SubscriptionStarted {
-        events: Vec<EventKind>,
-    },
+    SubscriptionStarted {},
     WaitMatched {
         event: EventEnvelope,
     },
@@ -324,6 +355,43 @@ pub struct EventEnvelope {
     pub data: EventData,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum SubscriptionEventKind {
+    #[serde(rename = "pane.output_matched")]
+    PaneOutputMatched,
+    #[serde(rename = "pane.agent_state_changed")]
+    PaneAgentStateChanged,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SubscriptionEventEnvelope {
+    pub event: SubscriptionEventKind,
+    pub data: SubscriptionEventData,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum SubscriptionEventData {
+    PaneOutputMatched(PaneOutputMatchedEvent),
+    PaneAgentStateChanged(PaneAgentStateChangedEvent),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PaneOutputMatchedEvent {
+    pub pane_id: String,
+    pub matched_line: String,
+    pub read: PaneReadResult,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PaneAgentStateChangedEvent {
+    pub pane_id: String,
+    pub workspace_id: String,
+    pub state: PaneAgentState,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub agent: Option<String>,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum EventData {
@@ -477,6 +545,79 @@ mod tests {
     }
 
     #[test]
+    fn subscribe_request_parses_parameterized_subscriptions() {
+        let json = r#"
+        {
+            "id": "sub_1",
+            "method": "events.subscribe",
+            "params": {
+                "subscriptions": [
+                    {
+                        "type": "pane.output_matched",
+                        "pane_id": "p_1_1",
+                        "source": "recent",
+                        "lines": 200,
+                        "match": { "type": "substring", "value": "auth: received" }
+                    },
+                    {
+                        "type": "pane.agent_state_changed",
+                        "pane_id": "p_1_1",
+                        "state": "waiting"
+                    }
+                ]
+            }
+        }
+        "#;
+
+        let request: Request = serde_json::from_str(json).unwrap();
+        let Method::EventsSubscribe(params) = request.method else {
+            panic!("wrong method parsed");
+        };
+        assert_eq!(params.subscriptions.len(), 2);
+        assert!(matches!(
+            &params.subscriptions[0],
+            Subscription::PaneOutputMatched {
+                pane_id,
+                source: ReadSource::Recent,
+                lines: Some(200),
+                r#match: OutputMatch::Substring { value },
+                strip_ansi: true,
+            } if pane_id == "p_1_1" && value == "auth: received"
+        ));
+        assert!(matches!(
+            &params.subscriptions[1],
+            Subscription::PaneAgentStateChanged {
+                pane_id,
+                state: Some(PaneAgentState::Waiting),
+            } if pane_id == "p_1_1"
+        ));
+    }
+
+    #[test]
+    fn subscription_event_envelope_round_trips() {
+        let event = SubscriptionEventEnvelope {
+            event: SubscriptionEventKind::PaneOutputMatched,
+            data: SubscriptionEventData::PaneOutputMatched(PaneOutputMatchedEvent {
+                pane_id: "p_1_1".into(),
+                matched_line: "auth: received".into(),
+                read: PaneReadResult {
+                    pane_id: "p_1_1".into(),
+                    workspace_id: "w_1".into(),
+                    source: ReadSource::Recent,
+                    text: "auth: received\n".into(),
+                    revision: 0,
+                    truncated: false,
+                },
+            }),
+        };
+
+        let json = serde_json::to_string(&event).unwrap();
+        assert!(json.contains("\"event\":\"pane.output_matched\""));
+        let restored: SubscriptionEventEnvelope = serde_json::from_str(&json).unwrap();
+        assert_eq!(restored, event);
+    }
+
+    #[test]
     fn success_response_round_trips() {
         let response = SuccessResponse {
             id: "req_1".into(),