Remove orchestration_event_push feature flag and rename poller to streamer

The orchestration_event_push flag has been on in dogfood and staging long
enough that the SSE event-push path is the only one we want to support.
Polling is gone; the dual-mode `OrchestrationEventPoller` is renamed to
`OrchestrationEventStreamer` and only opens SSE connections.

- Removed the `OrchestrationEventPush` FeatureFlag variant and Cargo
  feature.
- Renamed `orchestration_event_poller.rs` (and its tests) to
  `orchestration_event_streamer.rs`. Renamed the public type to
  `OrchestrationEventStreamer` and `handle_poll_result` (the shared
  event-injection sink) to `handle_event_batch`.
- Removed polling-only state and methods: `poll_backoff_index`,
  `poll_in_flight`, `poll_and_inject`, `start_idle_poll_timer`,
  `POLL_BACKOFF_STEPS`, `EVENT_POLL_BATCH_LIMIT`. `event_cursor`,
  `pending_delivery`, and `conversation_statuses` are kept since they
  are also used by the SSE path.
- Removed the dead `AIClient::poll_agent_events` trait method and impl
  in `server_api/ai.rs`.

The `OrchestrationV2` flag continues to gate streamer instantiation and
watched-run registration; behavior under v2 is unchanged.

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

Author: cephalonaut

app/Cargo.toml

@@ -902,7 +902,6 @@ active_conversation_requires_interaction = []
 incremental_auto_reload = []
 orchestration = []
 orchestration_v2 = ["orchestration"]
-orchestration_event_push = ["orchestration_v2"]
 pending_user_query_indicator = []
 queue_slash_command = []
 inline_menu_headers = []

app/src/ai/blocklist/action_model/execute/start_agent.rs

@@ -6,7 +6,7 @@ use crate::ai::agent::{
     AIAgentAction, AIAgentActionResultType, AIAgentActionType, LifecycleEventType,
     StartAgentExecutionMode, StartAgentResult,
 };
-use crate::ai::blocklist::orchestration_event_poller::OrchestrationEventPoller;
+use crate::ai::blocklist::orchestration_event_streamer::OrchestrationEventStreamer;
 use crate::ai::blocklist::orchestration_events::OrchestrationEventService;
 use crate::ai::blocklist::{BlocklistAIHistoryEvent, BlocklistAIHistoryModel};
 use warp_cli::agent::Harness;
@@ -110,8 +110,8 @@ impl StartAgentExecutor {
                             agent_id: id.clone(),
                         });
                         if FeatureFlag::OrchestrationV2.is_enabled() {
-                            OrchestrationEventPoller::handle(ctx).update(ctx, |poller, ctx| {
-                                poller.register_watched_run_id(
+                            OrchestrationEventStreamer::handle(ctx).update(ctx, |streamer, ctx| {
+                                streamer.register_watched_run_id(
                                     pending.parent_conversation_id,
                                     id,
                                     ctx,

app/src/ai/blocklist/mod.rs

@@ -5,7 +5,7 @@ pub mod block;
 pub mod code_block;
 mod context_model;
 mod controller;
-pub(crate) mod orchestration_event_poller;
+pub(crate) mod orchestration_event_streamer;
 pub(crate) mod orchestration_events;
 mod passive_suggestions;
 pub(crate) mod task_status_sync_model;

…/blocklist/orchestration_event_poller.rs …locklist/orchestration_event_streamer.rsapp/src/ai/blocklist/orchestration_event_poller.rs renamed to app/src/ai/blocklist/orchestration_event_streamer.rs app/src/ai/blocklist/orchestration_event_poller.rs renamed to app/src/ai/blocklist/orchestration_event_streamer.rs

@@ -218,7 +218,7 @@ fn finish_restore_fetch_uses_server_cursor_when_sqlite_is_absent() {
         let server_api = ServerApiProvider::new_for_test().get();
 
         let poller = app.add_singleton_model(|ctx| {
-            OrchestrationEventPoller::new_with_clients_for_test(ai_client, server_api, ctx)
+            OrchestrationEventStreamer::new_with_clients_for_test(ai_client, server_api, ctx)
         });
 
         // Seed event_cursor as on_restored_conversations would before spawning
@@ -283,43 +283,37 @@ fn restored_inprogress_parent_defers_delivery_until_success() {
         // a permissive expectation prevents spurious panics either way.
         mock.expect_get_ambient_agent_task()
             .returning(|_| Ok(make_ambient_task_with_event_seq(None)));
-        mock.expect_poll_agent_events()
-            .returning(|_, _, _| Ok(vec![]));
         mock.expect_update_event_sequence_on_server()
             .returning(|_, _| Ok(()));
         let ai_client: Arc<dyn AIClient> = Arc::new(mock);
         let server_api = ServerApiProvider::new_for_test().get();
 
-        let poller = app.add_singleton_model(|ctx| {
-            OrchestrationEventPoller::new_with_clients_for_test(ai_client, server_api, ctx)
+        let streamer = app.add_singleton_model(|ctx| {
+            OrchestrationEventStreamer::new_with_clients_for_test(ai_client, server_api, ctx)
         });
 
         // Synchronous part of `on_restored_conversations`: cursor seeded,
         // own run_id watched. No event delivery yet because parent is
         // InProgress.
-        poller.update(&mut app, |me, ctx| {
+        streamer.update(&mut app, |me, ctx| {
             me.on_restored_conversations(vec![conversation_id], ctx);
         });
-        poller.read(&app, |me, _| {
+        streamer.read(&app, |me, _| {
             assert_eq!(me.event_cursor.get(&conversation_id).copied(), Some(0));
             assert!(
                 me.watched_run_ids
                     .get(&conversation_id)
                     .is_some_and(|w| !w.is_empty()),
                 "own run_id should have been registered as watched"
             );
-            assert!(
-                !me.poll_in_flight.contains(&conversation_id),
-                "InProgress parent must not start polling"
-            );
             assert!(
                 me.sse_connections.is_empty(),
                 "InProgress parent must not open SSE"
             );
         });
 
-        // Transitioning the conversation to Success should trigger event
-        // delivery (poll_and_inject in the non-SSE default path).
+        // Transitioning the conversation to Success should open an SSE
+        // connection for event delivery.
         history_model.update(&mut app, |model, ctx| {
             model.update_conversation_status(
                 terminal_view_id,
@@ -328,17 +322,17 @@ fn restored_inprogress_parent_defers_delivery_until_success() {
                 ctx,
             );
         });
-        poller.read(&app, |me, _| {
+        streamer.read(&app, |me, _| {
             assert!(
-                me.poll_in_flight.contains(&conversation_id),
-                "Success transition with watched run_ids should start delivery"
+                me.sse_connections.contains_key(&conversation_id),
+                "Success transition with watched run_ids should open an SSE connection"
             );
         });
     });
 }
 
 #[test]
-fn handle_poll_result_persists_max_seq_to_history_model() {
+fn handle_event_batch_persists_max_seq_to_history_model() {
     use crate::ai::agent::conversation::{AIConversation, AIConversationId};
     use crate::persistence::ModelEvent;
     use crate::server::server_api::ai::MockAIClient;
@@ -378,7 +372,7 @@ fn handle_poll_result_persists_max_seq_to_history_model() {
         let server_api = ServerApiProvider::new_for_test().get();
 
         let poller = app.add_singleton_model(|ctx| {
-            OrchestrationEventPoller::new_with_clients_for_test(ai_client, server_api, ctx)
+            OrchestrationEventStreamer::new_with_clients_for_test(ai_client, server_api, ctx)
         });
 
         // Build a poll batch with max sequence = 42. Use an unrecognized
@@ -405,7 +399,7 @@ fn handle_poll_result_persists_max_seq_to_history_model() {
         ];
 
         poller.update(&mut app, |me, ctx| {
-            me.handle_poll_result(
+            me.handle_event_batch(
                 conversation_id,
                 /* self_run_id */ "some-other-run",
                 /* previous_cursor */ 0,
@@ -463,7 +457,7 @@ fn finish_restore_fetch_no_ops_when_conversation_deleted_mid_flight() {
         let server_api = ServerApiProvider::new_for_test().get();
 
         let poller = app.add_singleton_model(|ctx| {
-            OrchestrationEventPoller::new_with_clients_for_test(ai_client, server_api, ctx)
+            OrchestrationEventStreamer::new_with_clients_for_test(ai_client, server_api, ctx)
         });
 
         // Seed cursor as on_restored_conversations would.
@@ -542,7 +536,7 @@ fn finish_restore_fetch_reconnects_sse_when_children_added_to_open_connection()
         let server_api = ServerApiProvider::new_for_test().get();
 
         let poller = app.add_singleton_model(|ctx| {
-            OrchestrationEventPoller::new_with_clients_for_test(ai_client, server_api, ctx)
+            OrchestrationEventStreamer::new_with_clients_for_test(ai_client, server_api, ctx)
         });
 
         // Seed the state on_restored_conversations would have set up, then

…list/orchestration_event_poller_tests.rs …st/orchestration_event_streamer_tests.rsapp/src/ai/blocklist/orchestration_event_poller_tests.rs renamed to app/src/ai/blocklist/orchestration_event_streamer_tests.rs app/src/ai/blocklist/orchestration_event_poller_tests.rs renamed to app/src/ai/blocklist/orchestration_event_streamer_tests.rs

@@ -1562,7 +1562,7 @@ fn initialize_app(
     ctx.add_singleton_model(ai::blocklist::task_status_sync_model::TaskStatusSyncModel::new);
     if warp_core::features::FeatureFlag::OrchestrationV2.is_enabled() {
         ctx.add_singleton_model(
-            ai::blocklist::orchestration_event_poller::OrchestrationEventPoller::new,
+            ai::blocklist::orchestration_event_streamer::OrchestrationEventStreamer::new,
         );
     }
 
@@ -2718,8 +2718,6 @@ pub fn enabled_features() -> HashSet<FeatureFlag> {
         FeatureFlag::Orchestration,
         #[cfg(feature = "orchestration_v2")]
         FeatureFlag::OrchestrationV2,
-        #[cfg(feature = "orchestration_event_push")]
-        FeatureFlag::OrchestrationEventPush,
         #[cfg(feature = "pending_user_query_indicator")]
         FeatureFlag::PendingUserQueryIndicator,
         #[cfg(feature = "queue_slash_command")]

app/src/lib.rs

@@ -916,13 +916,6 @@ pub trait AIClient: 'static + Send + Sync {
         request: ListAgentMessagesRequest,
     ) -> anyhow::Result<Vec<AgentMessageHeader>, anyhow::Error>;
 
-    async fn poll_agent_events(
-        &self,
-        run_ids: &[String],
-        since_sequence: i64,
-        limit: i32,
-    ) -> anyhow::Result<Vec<AgentRunEvent>, anyhow::Error>;
-
     /// Persists the latest observed event sequence number for a run on the
     /// server. Used to keep the server-side cursor in sync with the client so
     /// that driver/cloud restores can resume without replaying events the
@@ -1879,22 +1872,6 @@ impl AIClient for ServerApi {
         Ok(response)
     }
 
-    async fn poll_agent_events(
-        &self,
-        run_ids: &[String],
-        since_sequence: i64,
-        limit: i32,
-    ) -> anyhow::Result<Vec<AgentRunEvent>, anyhow::Error> {
-        let run_ids_param: String = run_ids
-            .iter()
-            .map(|id| format!("run_ids={}", urlencoding::encode(id)))
-            .collect::<Vec<_>>()
-            .join("&");
-        let url = format!("agent/events?{run_ids_param}&since={since_sequence}&limit={limit}");
-        let events: Vec<AgentRunEvent> = self.get_public_api(&url).await?;
-        Ok(events)
-    }
-
     async fn update_event_sequence_on_server(
         &self,
         run_id: &str,

app/src/server/server_api/ai.rs

@@ -710,15 +710,11 @@ pub enum FeatureFlag {
     Orchestration,
 
     /// Enables server-side durable messaging for orchestration (v2).
-    /// When enabled, messages and events are stored in Postgres and discovered
-    /// via server polling instead of client-local conversation history.
+    /// When enabled, messages and events are stored in Postgres and the client
+    /// opens a persistent SSE connection to the server to receive events in
+    /// real time.
     OrchestrationV2,
 
-    /// Enables SSE-based event push for orchestration instead of polling.
-    /// When enabled the client opens a persistent SSE connection to the server
-    /// and receives events in real time instead of short-polling.
-    OrchestrationEventPush,
-
     /// Shows a pending user query indicator during summarization when a follow-up
     /// prompt is queued via `/fork-and-compact` or `/compact-and`.
     PendingUserQueryIndicator,
@@ -906,7 +902,6 @@ pub const DOGFOOD_FLAGS: &[FeatureFlag] = &[
     FeatureFlag::RememberFastForwardState,
     FeatureFlag::HOANotifications,
     FeatureFlag::OrchestrationV2,
-    FeatureFlag::OrchestrationEventPush,
     FeatureFlag::GeminiNotifications,
     FeatureFlag::LocalDockerSandbox,
     FeatureFlag::VerticalTabsSummaryMode,