[QUALITY-772] Use ancestor streams for large orchestrators (#12209)

## Description

### What
- Adds a dogfood-gated `OwnerOrchestrationAncestorStreamer` feature
flag.
- Switches owner-side orchestrator event delivery to a parent-family
`ancestor_run_id&include_self=true` stream when the flag is enabled.
- Keeps child-only conversations on `RunIds(self)` and viewer-mode
streams on `include_self=false`.
- Prevents oversized legacy `run_ids[]` streams from retrying forever
when the flag is disabled.

### Why
Large orchestrations can exceed the server's 100 explicit-run-id SSE
limit, causing parents to miss child lifecycle/message events. The
parent-family ancestor stream keeps delivery to one ordered stream while
preserving the existing cursor model.

### How
- Extends `AgentEventFilter::AncestorRunId` with an `include_self`
field.
- Adds client URL support for `include_self=true` only when requested.
- Stores the connected filter shape and reconnects only when the desired
filter is stale.
- Adds regression coverage for large parent streams, limit-crossing
behavior, restore, and child-only filtering.

## Linked Issue
- [ ] The linked issue is labeled `ready-to-spec` or
`ready-to-implement`.
- [ ] Where appropriate, screenshots or a short video of the
implementation are included below (especially for user-visible or UI
changes).

No linked issue; see companion server TECH spec at
`warp-server/specs/QUALITY-790/TECH.md`.

## Testing

- [x] `./script/format`
- [x] `PATH=/usr/local/bin:$PATH cargo clippy --workspace --all-targets
--all-features --tests -- -D warnings`
- [x] `cargo test -p warp --lib run_id_limit_without_flag`
- [x] `cargo test -p warp --lib
parent_with_many_children_opens_one_ancestor_include_self_stream`
- [x] `cargo test -p warp --lib
restored_parent_with_children_opens_ancestor_include_self_stream`

- [ ] I have manually tested my changes locally with `./script/run`

## Agent Mode
- [x] Warp Agent Mode - This PR was created via Warp's AI Agent Mode

CHANGELOG-NONE

Co-Authored-By: Oz <oz-agent@warp.dev>

---------

Co-authored-by: Oz <oz-agent@warp.dev>

Author: cephalonaut

app/Cargo.toml

@@ -959,6 +959,7 @@ active_conversation_requires_interaction = []
 incremental_auto_reload = []
 run_agents_tool = []
 orchestration_viewer_streamer = []
+owner_orchestration_ancestor_streamer = []
 pending_user_query_indicator = []
 queue_slash_command = []
 queued_prompts_v2 = ["queue_slash_command"]

app/src/ai/agent_events/driver.rs

@@ -23,19 +23,25 @@ pub(crate) const DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG: usize = 5;
 /// when opening a stream.
 ///
 /// `RunIds` maps to the `?run_ids[]=` query parameter on the SSE endpoint
-/// and is used by the orchestrator-owner per-conversation stream and the
-/// dormant Claude wake listener. `AncestorRunId` maps to the
-/// `?ancestor_run_id=` shape and streams events for every direct child of
-/// the supplied parent run; today only the shared-session viewer's
-/// pill bar consumes it.
+/// and is used by child-only per-conversation streams and the dormant
+/// Claude wake listener. `AncestorRunId` maps to the `?ancestor_run_id=`
+/// shape: with `include_self=false` it streams events for every direct
+/// child of the supplied parent run (the shared-session viewer's pill bar),
+/// and with `include_self=true` it additionally streams the parent run's
+/// own events so an owner-side orchestrator can receive child lifecycle
+/// events plus its own inbox on one ordered stream.
 #[derive(Clone, Debug)]
 pub(crate) enum AgentEventFilter {
     /// One stream per multiplexed set of run IDs. Matches today's
     /// `?run_ids[]=` endpoint.
     RunIds(Vec<String>),
-    /// Stream events for every direct child of the supplied parent run.
-    /// Matches the `?ancestor_run_id=` endpoint.
-    AncestorRunId(String),
+    /// Stream events for every direct child of the supplied parent run, and
+    /// (when `include_self` is true) the parent run itself. Matches the
+    /// `?ancestor_run_id=` endpoint.
+    AncestorRunId {
+        ancestor_run_id: String,
+        include_self: bool,
+    },
 }
 
 impl AgentEventFilter {
@@ -44,7 +50,10 @@ impl AgentEventFilter {
     pub(crate) fn log_label(&self) -> String {
         match self {
             AgentEventFilter::RunIds(ids) => format!("run_ids={ids:?}"),
-            AgentEventFilter::AncestorRunId(id) => format!("ancestor_run_id={id}"),
+            AgentEventFilter::AncestorRunId {
+                ancestor_run_id,
+                include_self,
+            } => format!("ancestor_run_id={ancestor_run_id} include_self={include_self}"),
         }
     }
 }
@@ -195,9 +204,16 @@ impl AgentEventSource for ServerApiAgentEventSource {
                     .stream_agent_events(run_ids, since_sequence)
                     .await?
             }
-            AgentEventFilter::AncestorRunId(ancestor_run_id) => {
+            AgentEventFilter::AncestorRunId {
+                ancestor_run_id,
+                include_self,
+            } => {
                 self.server_api
-                    .stream_agent_events_for_ancestor(ancestor_run_id, since_sequence)
+                    .stream_agent_events_for_ancestor(
+                        ancestor_run_id,
+                        *include_self,
+                        since_sequence,
+                    )
                     .await?
             }
         };

app/src/ai/blocklist/orchestration_event_streamer.rs

@@ -7,6 +7,7 @@ use async_trait::async_trait;
 use futures::channel::mpsc;
 use uuid::Uuid;
 use warp_cli::agent::Harness;
+use warp_core::features::FeatureFlag;
 use warp_multi_agent_api as api;
 use warpui::r#async::{SpawnedFutureHandle, Timer};
 use warpui::{
@@ -41,6 +42,8 @@ const RESTORE_FETCH_PERMANENT_BACKOFF_STEPS: &[u64] = &[30];
 const SSE_DRAIN_INTERVAL_MS: u64 = 500;
 /// Cap killed-run tombstones while keeping normal sessions well below the limit.
 const MAX_KILLED_RUN_IDS: usize = 1024;
+/// Maximum number of explicit run IDs the server accepts on a `run_ids[]` SSE stream.
+const MAX_RUN_ID_STREAM_FILTER: usize = 100;
 /// Max child runs fetched per cold-start `?ancestor_run_id=` REST seed in
 /// viewer mode. Matches the legacy `OrchestrationViewerModel` poller's value
 /// (the server caps at 100 anyway).
@@ -60,9 +63,8 @@ struct SseConnectionState {
     generation: u64,
     /// Abort handle for the spawned SSE driver task, used to cancel on teardown.
     abort_handle: futures::future::AbortHandle,
-    /// Snapshot of `watched_run_ids` at the time this connection was opened;
-    /// compared in `reevaluate_eligibility` to skip same-set reconnects.
-    connected_run_ids: HashSet<String>,
+    /// Wire filter this connection was opened with.
+    connected_filter: AgentEventFilter,
 }
 
 struct SseForwardingConsumer {
@@ -316,6 +318,19 @@ pub enum OrchestrationEventStreamerEvent {
     },
 }
 
+/// Outcome of selecting the SSE wire filter for an owner-side conversation.
+enum DesiredSseFilter {
+    /// Open (or keep) a stream with this filter.
+    Filter(AgentEventFilter),
+    /// Nothing to watch yet (no watched run IDs); do not open a stream.
+    NoFilter,
+    /// The conversation is a parent with more watched children than the
+    /// explicit `run_ids[]` stream allows and parent-family ancestor
+    /// streaming is unavailable. The payload is the watched-run-id total,
+    /// used only for diagnostics.
+    UnsupportedRunIdCount(usize),
+}
+
 impl OrchestrationEventStreamer {
     fn message_hydrator_for_run_id(&self, run_id: &str) -> MessageHydrator {
         match run_id.parse::<AmbientAgentTaskId>() {
@@ -802,7 +817,13 @@ impl OrchestrationEventStreamer {
              (gen={generation}, since={cursor})"
         );
 
-        let filter = AgentEventFilter::AncestorRunId(parent_task_id.to_string());
+        // Viewer mode subscribes to direct children only: it surfaces child
+        // lifecycle in the pill bar and never needs the orchestrator's inbox,
+        // so `include_self` stays false to preserve the existing contract.
+        let filter = AgentEventFilter::AncestorRunId {
+            ancestor_run_id: parent_task_id.to_string(),
+            include_self: false,
+        };
         let config = AgentEventDriverConfig::retry_forever(filter, cursor);
         let source = ServerApiAgentEventSource::new(server_api);
 
@@ -1345,8 +1366,6 @@ impl OrchestrationEventStreamer {
                 // in-flight, the removal handler already cleaned up all
                 // streamer state. Return early to avoid recreating
                 // state for a deleted conversation.
-                let had_sse;
-                let any_new_children;
                 {
                     let Some(stream) = self.streams.get_mut(&conv_id) else {
                         return;
@@ -1362,25 +1381,12 @@ impl OrchestrationEventStreamer {
                     let server_seq = task.last_event_sequence.unwrap_or(0);
                     stream.event_cursor = sqlite_cursor.max(server_seq);
 
-                    // Insert any new children. If new run_ids were added
-                    // and an SSE connection is already open (e.g. a
-                    // status race opened SSE with only the parent's own
-                    // run_id), reconnect so the new run_ids are included
-                    // in the filter; otherwise re-evaluate eligibility.
-                    had_sse = stream.sse_connection.is_some();
-                    let mut added = false;
+                    // Server-reported children may be absent from local history.
                     for child in task.children {
-                        if stream.watched_run_ids.insert(child) {
-                            added = true;
-                        }
+                        stream.watched_run_ids.insert(child);
                     }
-                    any_new_children = added;
-                }
-                if any_new_children && had_sse {
-                    self.reconnect_sse(conv_id, ctx);
-                } else {
-                    self.reevaluate_eligibility(conv_id, ctx);
                 }
+                self.reevaluate_eligibility(conv_id, ctx);
             }
             Err(err) => {
                 // If the conversation was removed mid-flight, drop the
@@ -1564,6 +1570,32 @@ impl OrchestrationEventStreamer {
             .unwrap_or_default()
     }
 
+    /// Selects the owner-side event stream filter for a conversation.
+    fn desired_sse_filter(
+        &self,
+        conversation_id: AIConversationId,
+        ctx: &warpui::AppContext,
+    ) -> DesiredSseFilter {
+        let is_parent = self.is_parent_agent_conversation(conversation_id, ctx);
+        if is_parent && FeatureFlag::OwnerOrchestrationAncestorStreamer.is_enabled() {
+            if let Some(self_run_id) = self.self_run_id(conversation_id, ctx) {
+                return DesiredSseFilter::Filter(AgentEventFilter::AncestorRunId {
+                    ancestor_run_id: self_run_id,
+                    include_self: true,
+                });
+            }
+        }
+
+        let run_ids = self.run_ids_for_sse(conversation_id);
+        if run_ids.is_empty() {
+            return DesiredSseFilter::NoFilter;
+        }
+        if is_parent && run_ids.len() > MAX_RUN_ID_STREAM_FILTER {
+            return DesiredSseFilter::UnsupportedRunIdCount(run_ids.len());
+        }
+        DesiredSseFilter::Filter(AgentEventFilter::RunIds(run_ids))
+    }
+
     /// Re-evaluates eligibility and either opens / reconnects or tears
     /// down the SSE connection for the given conversation.
     fn reevaluate_eligibility(
@@ -1581,9 +1613,11 @@ impl OrchestrationEventStreamer {
             (true, false) => self.start_sse_connection(conversation_id, ctx),
             (true, true) => {
                 // Status / metadata updates fire `reevaluate_eligibility` on
-                // every exchange transition; only reconnect when the run-id
-                // filter actually changed.
-                if self.watched_run_ids_differ_from_connected(conversation_id) {
+                // every exchange transition; only reconnect when the desired
+                // filter shape actually changed. Registering more children
+                // while a parent-family ancestor stream is connected leaves
+                // the filter unchanged, so it does not reconnect.
+                if self.stream_filter_stale(conversation_id, ctx) {
                     self.reconnect_sse(conversation_id, ctx);
                 }
             }
@@ -1740,10 +1774,19 @@ impl OrchestrationEventStreamer {
         conversation_id: AIConversationId,
         ctx: &mut ModelContext<Self>,
     ) {
-        let run_ids = self.run_ids_for_sse(conversation_id);
-        if run_ids.is_empty() {
-            return;
-        }
+        let filter = match self.desired_sse_filter(conversation_id, ctx) {
+            DesiredSseFilter::Filter(filter) => filter,
+            DesiredSseFilter::NoFilter => return,
+            DesiredSseFilter::UnsupportedRunIdCount(count) => {
+                log::error!(
+                    "Owner-side SSE delivery blocked for {conversation_id:?}: {count} watched \
+                     run IDs exceed the {MAX_RUN_ID_STREAM_FILTER} explicit-run-id limit and \
+                     parent-family ancestor streaming is disabled; enable \
+                     OwnerOrchestrationAncestorStreamer to deliver events for large orchestrators"
+                );
+                return;
+            }
+        };
 
         let cursor = self
             .streams
@@ -1761,10 +1804,11 @@ impl OrchestrationEventStreamer {
 
         log::info!(
             "Opening SSE stream for {conversation_id:?} (gen={generation}, \
-             run_ids={run_ids:?}, since={cursor})"
+             filter={}, since={cursor})",
+            filter.log_label()
         );
 
-        let config = AgentEventDriverConfig::retry_forever_run_ids(run_ids.clone(), cursor);
+        let config = AgentEventDriverConfig::retry_forever(filter.clone(), cursor);
         let source = ServerApiAgentEventSource::new(server_api);
         let hydrator = self.message_hydrator_for_run_id(&self_run_id);
 
@@ -1799,29 +1843,42 @@ impl OrchestrationEventStreamer {
             },
         );
 
-        let connected_run_ids: HashSet<String> = run_ids.iter().cloned().collect();
         let stream = self.streams.entry(conversation_id).or_default();
         stream.sse_connection = Some(SseConnectionState {
             event_receiver: rx,
             generation,
             abort_handle: handle.abort_handle(),
-            connected_run_ids,
+            connected_filter: filter,
         });
 
         // Start periodic event drain.
         self.start_sse_drain_timer(conversation_id, generation, ctx);
     }
 
-    /// True iff the open SSE's recorded run-id set is stale relative to the
-    /// conversation's current `watched_run_ids`.
-    fn watched_run_ids_differ_from_connected(&self, conversation_id: AIConversationId) -> bool {
+    /// True iff the open SSE's connected filter is stale relative to the
+    /// filter the conversation should currently use. Compares the desired
+    /// filter shape (run-id set or parent-family ancestor scope) rather than
+    /// the raw `watched_run_ids` set, so a parent-family stream is not
+    /// reconnected just because additional child IDs were registered.
+    fn stream_filter_stale(
+        &self,
+        conversation_id: AIConversationId,
+        ctx: &warpui::AppContext,
+    ) -> bool {
         let Some(stream) = self.streams.get(&conversation_id) else {
             return false;
         };
         let Some(connection) = stream.sse_connection.as_ref() else {
             return false;
         };
-        stream.watched_run_ids != connection.connected_run_ids
+        match self.desired_sse_filter(conversation_id, ctx) {
+            DesiredSseFilter::Filter(desired) => {
+                !agent_event_filters_equivalent(&desired, &connection.connected_filter)
+            }
+            // Nothing watchable tears down through the eligibility predicate.
+            DesiredSseFilter::NoFilter => false,
+            DesiredSseFilter::UnsupportedRunIdCount(_) => true,
+        }
     }
 
     /// Periodically fires to drain buffered SSE events into the event
@@ -2013,6 +2070,28 @@ async fn resolve_dormant_claude_wake_cursor(
     }
 }
 
+fn agent_event_filters_equivalent(a: &AgentEventFilter, b: &AgentEventFilter) -> bool {
+    match (a, b) {
+        (AgentEventFilter::RunIds(a), AgentEventFilter::RunIds(b)) => {
+            a.len() == b.len() && {
+                let set: HashSet<&String> = a.iter().collect();
+                b.iter().all(|id| set.contains(id))
+            }
+        }
+        (
+            AgentEventFilter::AncestorRunId {
+                ancestor_run_id: a_run,
+                include_self: a_self,
+            },
+            AgentEventFilter::AncestorRunId {
+                ancestor_run_id: b_run,
+                include_self: b_self,
+            },
+        ) => a_run == b_run && a_self == b_self,
+        _ => false,
+    }
+}
+
 fn agent_task_harness(task: &crate::ai::ambient_agents::task::AmbientAgentTask) -> Option<Harness> {
     task.agent_config_snapshot
         .as_ref()