fix(cdp): fan out target-owned runtime exceptions

moli-protocol/src/conn/browser_context/target_session_owner.rs

@@ -3403,6 +3403,37 @@ impl CdpConnection {
         .then_some(attachments)
     }
 
+    /// Captures every exact attachment that had enabled the CDP `Runtime`
+    /// domain when one target-owned Runtime fact was ingested.
+    ///
+    /// The renderer publishes asynchronous exceptions once per target, not
+    /// once per Inspector session. Freeze the audience at ingress so a later
+    /// detach, target replacement, or `Runtime.enable` cannot retarget that
+    /// historical fact.
+    pub(crate) fn runtime_event_protocol_attachments_for_session_owner(
+        &self,
+        session_id: Option<&str>,
+    ) -> Option<Vec<crate::conn::TargetPageProtocolAttachmentIdentity>> {
+        let source = self.target_page_protocol_attachment_identity_for_session(session_id)?;
+        let attachments = self
+            .page_event_session_ids_for_session_owner(session_id)
+            .into_iter()
+            .filter(|event_session_id| {
+                self.target_runtime_session_state_for_session(event_session_id.as_deref())
+                    .is_some_and(|state| state.runtime_frontend_enabled)
+            })
+            .map(|event_session_id| {
+                self.target_page_protocol_attachment_identity_for_session(
+                    event_session_id.as_deref(),
+                )
+            })
+            .collect::<Option<Vec<_>>>()?;
+        attachments
+            .iter()
+            .all(|attachment| attachment.page_owner() == source.page_owner())
+            .then_some(attachments)
+    }
+
     pub(crate) fn runtime_session_owner_target_url(
         &self,
         session_id: Option<&str>,
@@ -4813,6 +4844,60 @@ mod tests {
         );
     }
 
+    #[test]
+    fn runtime_event_attachments_include_every_enabled_session_on_the_exact_page() {
+        let mut conn = CdpConnection::default();
+        let mut browser_context = BrowserContext::new("BID-runtime-events".to_owned());
+        browser_context.set_active_target_id("TID-runtime-events".to_owned());
+        browser_context.attach_active_session("SID-runtime-a".to_owned());
+        assert!(
+            browser_context.assign_auxiliary_session_to_target(
+                "TID-runtime-events",
+                "SID-runtime-b".to_owned(),
+            )
+        );
+        assert!(browser_context.assign_auxiliary_session_to_target(
+            "TID-runtime-events",
+            "SID-runtime-disabled".to_owned(),
+        ));
+        browser_context
+            .active_target
+            .runtime_slot
+            .set_page_attachment_id_for_test(41);
+        conn.browser_context = Some(browser_context);
+
+        conn.with_target_devtools_session_state_for_session_mut(Some("SID-runtime-b"), |state| {
+            state.runtime_session_state.runtime_frontend_enabled = true
+        })
+        .expect("Runtime-enabled auxiliary session should be mutable");
+        assert_eq!(
+            conn.runtime_event_protocol_attachments_for_session_owner(Some("SID-runtime-a"))
+                .expect("the current Page should expose its Runtime audience")
+                .into_iter()
+                .map(|attachment| attachment.session_id().map(str::to_owned))
+                .collect::<Vec<_>>(),
+            vec![Some("SID-runtime-b".to_owned())],
+            "a disabled primary must not hide the enabled peer attachment"
+        );
+
+        conn.with_target_devtools_session_state_for_session_mut(Some("SID-runtime-a"), |state| {
+            state.runtime_session_state.runtime_frontend_enabled = true
+        })
+        .expect("Runtime-enabled primary session should be mutable");
+        assert_eq!(
+            conn.runtime_event_protocol_attachments_for_session_owner(Some("SID-runtime-b"))
+                .expect("the current Page should expose its Runtime audience")
+                .into_iter()
+                .map(|attachment| attachment.session_id().map(str::to_owned))
+                .collect::<Vec<_>>(),
+            vec![
+                Some("SID-runtime-a".to_owned()),
+                Some("SID-runtime-b".to_owned()),
+            ],
+            "one target-owned Runtime fact must freeze every enabled attachment"
+        );
+    }
+
     #[test]
     fn runtime_context_identity_includes_service_worker_without_page_owner_identity() {
         let mut conn = CdpConnection::default();

moli-protocol/src/domains/activity/output_ingress/renderer.rs

@@ -193,7 +193,7 @@ async fn project_renderer_output_records_for_route(
                 barriers
                     .route_publication_outputs(
                         conn,
-                        session_id,
+                        action_session_id.as_deref(),
                         renderer_cause.as_ref(),
                         Some(cursor),
                         outputs,

moli-protocol/src/domains/observable_output/emission.rs

@@ -153,8 +153,9 @@ impl ObservableActivityEmissionPlan {
         prepared_outputs: Option<&mut ObservablePreparedOutputs>,
     ) -> Option<Self> {
         if let Some(prepared_outputs) = prepared_outputs
-            && let Some(items) = prepared_outputs.take_runtime_observable_items()
-            && runtime_observable_prepared_items_match_owner(conn, session_id, &items)
+            && let Some(prepared) =
+                prepared_outputs.take_runtime_observable_items_for_session(session_id)
+            && let Some((_event_session_id, items)) = prepared.materialize_for_owner(conn)
         {
             return Some(Self::from_runtime_prepared_items(items));
         }
@@ -312,6 +313,19 @@ pub(crate) async fn emit_pending_observable_activity_background_events_async(
         }
         return;
     }
+    if step == ObservableOutputProjectionStep::RuntimeObservable {
+        let Some(prepared_outputs) = prepared_outputs else {
+            return;
+        };
+        for prepared in prepared_outputs.take_runtime_observable_items() {
+            let Some((event_session_id, items)) = prepared.materialize_for_owner(conn) else {
+                continue;
+            };
+            ObservableActivityEmissionPlan::from_runtime_prepared_items(items)
+                .emit_background_events(conn, out, event_session_id.as_deref());
+        }
+        return;
+    }
     if let Some(plan) =
         ObservableActivityEmissionPlan::prepare_async(step, conn, session_id, prepared_outputs)
             .await
@@ -320,22 +334,6 @@ pub(crate) async fn emit_pending_observable_activity_background_events_async(
     }
 }
 
-fn runtime_observable_prepared_items_match_owner(
-    conn: &CdpConnection,
-    session_id: Option<&str>,
-    items: &ObservableRuntimePreparedItems,
-) -> bool {
-    let Some(url) = conn.runtime_session_owner_target_url(session_id) else {
-        return false;
-    };
-    let Ok(runtime_slot) = conn.runtime_session_owner_slot(session_id) else {
-        return false;
-    };
-    runtime_slot
-        .page_attachment_id()
-        .is_some_and(|attachment_id| items.matches_source_identity(&url, attachment_id))
-}
-
 #[cfg(test)]
 mod tests {
     use std::collections::BTreeMap;

moli-protocol/src/domains/observable_output/output_queue.rs

@@ -6,6 +6,7 @@ use crate::conn::DevToolsConsoleOutputSessionState;
 use crate::conn::TargetRuntimeSlot;
 use crate::conn::{
     BackgroundProtocolEvent, CdpConnection, TargetOwnerState, TargetPageAttachmentId,
+    TargetPageProtocolAttachmentIdentity,
 };
 use crate::domains::activity::ProtocolOutputSink;
 use crate::domains::audits_output_state::TargetAuditsOutputCursor;
@@ -28,7 +29,7 @@ pub(crate) struct ObservablePreparedOutputs {
     audits: Vec<ObservableSessionAuditsPreparedRange>,
     console: Option<ObservableConsoleLogPreparedRange>,
     log: Vec<ObservableSessionLogPreparedRange>,
-    runtime_observable_items: Option<ObservableRuntimePreparedItems>,
+    runtime_observable_items: Vec<ObservableSessionRuntimePreparedItems>,
 }
 
 #[derive(Clone, Debug, Default, Eq, PartialEq)]
@@ -65,6 +66,12 @@ pub(in crate::domains::observable_output) struct ObservableSessionAuditsPrepared
     cursor: TargetAuditsOutputCursor,
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub(in crate::domains::observable_output) struct ObservableSessionRuntimePreparedItems {
+    attachment: TargetPageProtocolAttachmentIdentity,
+    items: ObservableRuntimePreparedItems,
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(in crate::domains::observable_output) struct ObservableConsoleLogEmissionCursor {
     domain: ObservableConsoleLogDomain,
@@ -108,8 +115,8 @@ impl ObservablePreparedOutputs {
         for range in other.log {
             self.push_log(range.session_id.as_deref(), range.range);
         }
-        if let Some(items) = other.runtime_observable_items {
-            self.push_runtime_observable_items(items);
+        for prepared in other.runtime_observable_items {
+            self.push_runtime_observable_items(prepared.attachment, prepared.items);
         }
     }
 
@@ -117,7 +124,7 @@ impl ObservablePreparedOutputs {
         self.audits.is_empty()
             && self.console.is_none()
             && self.log.is_empty()
-            && self.runtime_observable_items.is_none()
+            && self.runtime_observable_items.is_empty()
     }
 
     pub(in crate::domains) fn append_to_output_sink(
@@ -185,8 +192,19 @@ impl ObservablePreparedOutputs {
 
     pub(in crate::domains::observable_output) fn take_runtime_observable_items(
         &mut self,
-    ) -> Option<ObservableRuntimePreparedItems> {
-        self.runtime_observable_items.take()
+    ) -> Vec<ObservableSessionRuntimePreparedItems> {
+        std::mem::take(&mut self.runtime_observable_items)
+    }
+
+    pub(in crate::domains::observable_output) fn take_runtime_observable_items_for_session(
+        &mut self,
+        session_id: Option<&str>,
+    ) -> Option<ObservableSessionRuntimePreparedItems> {
+        let index = self
+            .runtime_observable_items
+            .iter()
+            .position(|prepared| prepared.session_id() == session_id)?;
+        Some(self.runtime_observable_items.remove(index))
     }
 
     pub(in crate::domains::observable_output) fn has_console(&self) -> bool {
@@ -202,7 +220,7 @@ impl ObservablePreparedOutputs {
     }
 
     pub(in crate::domains::observable_output) fn has_runtime_observable(&self) -> bool {
-        self.runtime_observable_items.is_some()
+        !self.runtime_observable_items.is_empty()
     }
 
     pub(in crate::domains::observable_output) fn push_console(
@@ -249,9 +267,18 @@ impl ObservablePreparedOutputs {
 
     pub(in crate::domains::observable_output) fn push_runtime_observable_items(
         &mut self,
+        attachment: TargetPageProtocolAttachmentIdentity,
         items: ObservableRuntimePreparedItems,
     ) {
-        self.runtime_observable_items.get_or_insert(items);
+        if self
+            .runtime_observable_items
+            .iter()
+            .any(|prepared| prepared.attachment == attachment)
+        {
+            return;
+        }
+        self.runtime_observable_items
+            .push(ObservableSessionRuntimePreparedItems { attachment, items });
     }
 }
 
@@ -331,6 +358,30 @@ impl ObservableSessionAuditsPreparedRange {
     }
 }
 
+impl ObservableSessionRuntimePreparedItems {
+    pub(in crate::domains::observable_output) fn session_id(&self) -> Option<&str> {
+        self.attachment.session_id()
+    }
+
+    pub(in crate::domains::observable_output) fn materialize_for_owner(
+        self,
+        conn: &CdpConnection,
+    ) -> Option<(Option<String>, ObservableRuntimePreparedItems)> {
+        if !conn.target_page_protocol_attachment_identity_is_current(&self.attachment) {
+            return None;
+        }
+        let session_id = self.attachment.session_id().map(str::to_owned);
+        let url = conn.runtime_session_owner_target_url(session_id.as_deref())?;
+        let runtime_slot = conn
+            .runtime_session_owner_slot(session_id.as_deref())
+            .ok()?;
+        runtime_slot
+            .page_attachment_id()
+            .is_some_and(|attachment_id| self.items.matches_source_identity(&url, attachment_id))
+            .then_some((session_id, self.items))
+    }
+}
+
 impl ObservablePreparedOutputSlot {
     pub(crate) fn from_outputs(outputs: ObservablePreparedOutputs) -> Self {
         Self { outputs }
@@ -968,13 +1019,30 @@ mod tests {
         ObservableConsoleLogDomain, ObservableConsoleLogPreparedRange, ObservablePreparedOutputs,
         TargetObservableOutputQueue,
     };
-    use crate::conn::{BrowserContext, TargetPageAttachmentId, TargetRuntimeSlot};
+    use crate::conn::{
+        BrowserContext, TargetPageAttachmentId, TargetPageProtocolAttachmentIdentity,
+        TargetPageResidenceIdentity, TargetRuntimeSlot,
+    };
     use crate::domains::log_output_state::TargetLogOutputCursor;
 
     fn page_attachment_id(raw: u64) -> TargetPageAttachmentId {
         TargetPageAttachmentId::from_raw_for_test(raw)
     }
 
+    fn protocol_attachment(
+        session_id: Option<&str>,
+        raw: u64,
+    ) -> TargetPageProtocolAttachmentIdentity {
+        TargetPageProtocolAttachmentIdentity::new(
+            TargetPageResidenceIdentity::new(
+                "BID-observable-output".to_owned(),
+                Some("TID-observable-output".to_owned()),
+                page_attachment_id(raw),
+            ),
+            session_id.map(str::to_owned),
+        )
+    }
+
     fn renderer_source_snapshot(
         source: RendererRuntimeObservableSourceSummary,
     ) -> RendererPageDiagnosticsSnapshot {
@@ -1433,7 +1501,7 @@ mod tests {
         assert!(
             prepared.take_console_range().is_none()
                 && prepared.take_log_range().is_none()
-                && prepared.take_runtime_observable_items().is_none(),
+                && prepared.take_runtime_observable_items().is_empty(),
             "empty captured outputs should not expose projection payload slots"
         );
 
@@ -1483,7 +1551,10 @@ mod tests {
         );
         prepared.push_console(console_range.clone());
         prepared.push_log(None, log_range.clone());
-        prepared.push_runtime_observable_items(runtime_items.clone());
+        prepared.push_runtime_observable_items(
+            protocol_attachment(Some("SID-runtime"), 1),
+            runtime_items.clone(),
+        );
 
         assert_eq!(
             prepared.take_console_range(),
@@ -1496,7 +1567,9 @@ mod tests {
             "captured Log slot should expose only its projection payload"
         );
         assert_eq!(
-            prepared.take_runtime_observable_items(),
+            prepared
+                .take_runtime_observable_items_for_session(Some("SID-runtime"))
+                .map(|prepared| prepared.items),
             Some(runtime_items),
             "captured RuntimeObservable slot should expose only concrete projection items"
         );
@@ -1504,7 +1577,7 @@ mod tests {
         assert!(
             prepared.take_console_range().is_none()
                 && prepared.take_log_range().is_none()
-                && prepared.take_runtime_observable_items().is_none(),
+                && prepared.take_runtime_observable_items().is_empty(),
             "taking prepared slots should clear the payload container"
         );
     }

moli-protocol/src/domains/observable_output/presence.rs

@@ -9,6 +9,12 @@ use super::output_queue::{
 };
 use crate::conn::CdpConnection;
 
+#[derive(Clone, Copy)]
+enum RuntimeObservableAudience {
+    SourceSession,
+    EnabledTargetAttachments,
+}
+
 #[cfg(test)]
 #[derive(Clone, Debug, Default, Eq, PartialEq)]
 pub(in crate::domains) struct ObservableActivityOutputs {
@@ -84,7 +90,14 @@ pub(in crate::domains) fn runtime_console_message_prepared_outputs(
     else {
         return prepared;
     };
-    push_runtime_observable_tail_prepared_outputs(&mut prepared, conn, &source, session_id, true);
+    push_runtime_observable_tail_prepared_outputs(
+        &mut prepared,
+        conn,
+        &source,
+        session_id,
+        true,
+        RuntimeObservableAudience::SourceSession,
+    );
     prepared
 }
 
@@ -103,7 +116,14 @@ pub(in crate::domains) fn runtime_lifecycle_error_prepared_outputs(
     ) else {
         return prepared;
     };
-    push_runtime_observable_tail_prepared_outputs(&mut prepared, conn, &source, session_id, true);
+    push_runtime_observable_tail_prepared_outputs(
+        &mut prepared,
+        conn,
+        &source,
+        session_id,
+        true,
+        RuntimeObservableAudience::EnabledTargetAttachments,
+    );
     prepared
 }
 
@@ -192,7 +212,14 @@ fn push_runtime_observable_source_prepared_outputs(
     else {
         return;
     };
-    push_runtime_observable_tail_prepared_outputs(prepared, conn, &source, session_id, false);
+    push_runtime_observable_tail_prepared_outputs(
+        prepared,
+        conn,
+        &source,
+        session_id,
+        false,
+        RuntimeObservableAudience::SourceSession,
+    );
 }
 
 fn push_runtime_observable_tail_prepared_outputs(
@@ -201,6 +228,7 @@ fn push_runtime_observable_tail_prepared_outputs(
     source: &super::TargetRuntimeObservableSourceOutput,
     session_id: Option<&str>,
     deliver_log_to_enabled_sessions: bool,
+    runtime_audience: RuntimeObservableAudience,
 ) {
     let owner_state = conn
         .target_owner_state_for_session(session_id)
@@ -219,17 +247,31 @@ fn push_runtime_observable_tail_prepared_outputs(
         deliver_log_to_enabled_sessions,
     );
 
-    let include_runtime_console_api_messages =
-        !renderer_runtime_agent_owns_page_console_api_events(conn, session_id);
-    if conn
-        .target_runtime_session_state_for_session(session_id)
-        .is_some_and(|state| state.runtime_frontend_enabled)
-        && let Some(items) = source.source_items_prepared_for_state(
+    let attachments = match runtime_audience {
+        RuntimeObservableAudience::SourceSession => conn
+            .target_page_protocol_attachment_identity_for_session(session_id)
+            .into_iter()
+            .collect(),
+        RuntimeObservableAudience::EnabledTargetAttachments => conn
+            .runtime_event_protocol_attachments_for_session_owner(session_id)
+            .unwrap_or_default(),
+    };
+    for attachment in attachments {
+        let event_session_id = attachment.session_id();
+        if !conn
+            .target_runtime_session_state_for_session(event_session_id)
+            .is_some_and(|state| state.runtime_frontend_enabled)
+        {
+            continue;
+        }
+        let include_runtime_console_api_messages =
+            !renderer_runtime_agent_owns_page_console_api_events(conn, event_session_id);
+        if let Some(items) = source.source_items_prepared_for_state(
             &owner_state.runtime_observable_state,
             include_runtime_console_api_messages,
-        )
-    {
-        prepared.push_runtime_observable_items(items);
+        ) {
+            prepared.push_runtime_observable_items(attachment, items);
+        }
     }
 }
 
@@ -518,9 +560,46 @@ mod tests {
     use super::{
         ObservableOutputProjectionStep,
         observable_backlog_activity_outputs_for_session_owner as observable_backlog_activity_outputs,
-        observable_source_activity_outputs,
+        observable_source_activity_outputs, runtime_lifecycle_error_prepared_outputs,
     };
 
+    fn runtime_lifecycle_error_audience(enabled_session_ids: &[&str]) -> Vec<Option<String>> {
+        let mut conn = crate::conn::CdpConnection::default();
+        let mut bc = BrowserContext::new("BID-runtime-lifecycle".to_owned());
+        bc.set_active_target_id("TID-runtime-lifecycle".to_owned());
+        bc.set_target_url("https://example.test/runtime-lifecycle".to_owned());
+        bc.attach_active_session("SID-runtime-a".to_owned());
+        assert!(bc.assign_auxiliary_session_to_target(
+            "TID-runtime-lifecycle",
+            "SID-runtime-b".to_owned(),
+        ));
+        assert!(bc.assign_auxiliary_session_to_target(
+            "TID-runtime-lifecycle",
+            "SID-runtime-disabled".to_owned(),
+        ));
+        bc.active_target
+            .runtime_slot
+            .set_page_attachment_id_for_test(17);
+        conn.browser_context = Some(bc);
+        for session_id in enabled_session_ids {
+            conn.with_target_devtools_session_state_for_session_mut(Some(session_id), |state| {
+                state.runtime_session_state.runtime_frontend_enabled = true
+            })
+            .expect("Runtime audience session should be mutable");
+        }
+
+        runtime_lifecycle_error_prepared_outputs(
+            &mut conn,
+            "uncaught timer error".to_owned(),
+            Some(7),
+            Some("SID-runtime-a"),
+        )
+        .take_runtime_observable_items()
+        .into_iter()
+        .map(|prepared| prepared.session_id().map(str::to_owned))
+        .collect()
+    }
+
     #[test]
     fn observable_source_outputs_own_runtime_observable_presence() {
         let mut conn = crate::conn::CdpConnection::default();
@@ -551,6 +630,23 @@ mod tests {
         );
     }
 
+    #[test]
+    fn runtime_lifecycle_error_freezes_every_enabled_attachment_only() {
+        assert_eq!(
+            runtime_lifecycle_error_audience(&["SID-runtime-b"]),
+            vec![Some("SID-runtime-b".to_owned())],
+            "a disabled source attachment must not hide its enabled peer"
+        );
+        assert_eq!(
+            runtime_lifecycle_error_audience(&["SID-runtime-a", "SID-runtime-b"]),
+            vec![
+                Some("SID-runtime-a".to_owned()),
+                Some("SID-runtime-b".to_owned()),
+            ],
+            "one target-owned exception must freeze all Runtime-enabled attachments"
+        );
+    }
+
     #[test]
     fn observable_source_sync_is_independent_from_runtime_emission() {
         let mut conn = crate::conn::CdpConnection::default();