Wake up remote Claude Code agents on new events (warpdotdev#9399)

## Description
<!-- Please remember to add your design buddy onto the PR for review, if
it contains any UI changes! -->


Fixes orchestration v2 parent/child agent wake-up and messaging behavior
for remote child agents.

This PR updates the client-side orchestration flow so that incoming
parent-agent messages wake remote child agents through the server run
follow-up path instead of trying to treat them like local dormant Claude
harnesses. Previously, a remote child could receive the parent’s
message, but it would not be restarted correctly in a harness and could
fail or hang when trying to send a message back to the parent.

Main changes:

•  Adds a remote-child wake path in the blocklist AI controller:
◦ detects remote child conversations with pending parent-agent message
events
  ◦  submits a run follow-up to agent/runs/{run_id}/followups
◦ removes delivered pending message events after successful follow-up
submission
  ◦  retries/logs failures instead of silently hanging
• Keeps local dormant Claude wake behavior separate from remote child
wake behavior.
• Restores remote hidden child panes as cloud/ambient agent panes
instead of local terminal-backed child panes.
• Ensures restored remote child panes enter the existing ambient session
in AgentRunning state.
• Persists and restores remote-child conversation metadata so the client
can distinguish local children from remote children across reloads.
•  Improves orchestration v2 message sending:
◦  uses task-scoped server APIs when available
◦  adds bounded timeouts and error logging for send failures
◦ surfaces failures instead of leaving action execution indefinitely
pending
•  Adds regression coverage for:
◦  remote child conversation restoration
◦  remote child pane/session state
◦  task-scoped ambient agent messaging
◦  orchestration v2 message/error behavior

## Testing
<!--
How did you test this change? What automated tests did you add? If you
didn't add any new tests, what's your justification for not adding any?

If you're not sure whether you should add a test, check our testing
policy:
https://www.notion.so/warpdev/How-We-Code-at-Warp-257fe43d556e4b3c8dfd42f70004cc72#1f97825450504baa9c5fd87a737daa09
-->

## Server API dependencies
<!-- You may remove this section if your PR does not have any server
dependencies. -->
- [ ] Is this change necessary to make the client compatible with a
desired [server API breaking
change](https://www.notion.so/warpdev/How-to-safely-introduce-server-API-breaking-changes-0aa805ff5d5d41fd8834f3c95caba0b4?pvs=4#d55ecf8aea3449949d3c33b0e67f6800)?
- [ ] Does this change rely on a [new server
API](https://www.notion.so/warpdev/How-to-add-a-new-full-stack-feature-8412cede405a4ec194b32bdd4b951035?pvs=4#04da1e6a493542d68b3e998c7d339640)?
- [ ] If so, is the use of this API restricted to client channels that
rely on the staging server (e.g. WarpDev)?
- [ ] Is this change enabling the use of a server API on client channels
that rely on the production server (e.g. WarpStable)?
- [ ] If so, has the new server API been stable on production for at
least one server release cycle? See
[here](https://www.notion.so/warpdev/How-to-add-a-new-full-stack-feature-8412cede405a4ec194b32bdd4b951035?pvs=4#73b202f939834b97ab1fbdf7fc82cd53)
for more details.

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

## Changelog Entries for Stable
<!--
The entries below will be used when constructing a soft-copy of the
stable release changelog. Leave blank or remove the lines if no entry in
the stable changelog is needed. Entries should be on the same line,
without the `{{` `}}` brackets. You can use multiple lines, even of the
same type. The valid suffixes are:

* NEW-FEATURE: for new, relatively sizable features. Features listed
here will likely have docs / social media posts / marketing launches
associated with them, so use sparingly.
* IMPROVEMENT: for new functionality of existing features.
* BUG-FIX: for fixes related to known bugs or regressions.
* IMAGE: the image specified by the URL (hosted on GCP) will be added to
Dev & Preview releases. For Stable releases, see the pinned doc in the
#release Slack channel.
* OZ: Oz-related updates. Use `CHANGELOG-OZ`. At most 4 Oz updates are
shown in-app per release.
-->

CHANGELOG-NEW-FEATURE: {{text goes here...}}
CHANGELOG-IMPROVEMENT: {{text goes here...}}
CHANGELOG-BUG-FIX: {{text goes here...}}
CHANGELOG-BUG-FIX: {{more text goes here...}}
CHANGELOG-IMAGE: {{GCP-hosted URL goes here...}}
CHANGELOG-OZ: {{text goes here...}}

---------

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

Author: kjankov

app/src/ai/agent/api/convert_conversation.rs

@@ -82,6 +82,7 @@ pub fn convert_conversation_data_to_ai_conversation(
             parent_agent_id: None,
             agent_name: None,
             parent_conversation_id: None,
+            is_remote_child: false,
             run_id: None,
             autoexecute_override: None,
             last_event_sequence: None,
@@ -97,6 +98,7 @@ pub fn convert_conversation_data_to_ai_conversation(
             parent_agent_id: None,
             agent_name: None,
             parent_conversation_id: None,
+            is_remote_child: false,
             // TODO: Populate run_id from server metadata once it is exposed
             // in ServerAIConversationMetadata. For cloud conversations that
             // were spawned via the server API, the run_id is created at task

app/src/ai/agent/conversation.rs

@@ -355,6 +355,7 @@ impl AIConversation {
             parent_agent_id,
             agent_name,
             parent_conversation_id,
+            is_remote_child,
             run_id,
             autoexecute_override,
             last_event_sequence,
@@ -380,6 +381,7 @@ impl AIConversation {
             let parent_conversation_id = data
                 .parent_conversation_id
                 .and_then(|id| AIConversationId::try_from(id).ok());
+            let is_remote_child = data.is_remote_child;
             let run_id = data.run_id;
             let autoexecute_override = if FeatureFlag::RememberFastForwardState.is_enabled() {
                 data.autoexecute_override
@@ -399,6 +401,7 @@ impl AIConversation {
                 parent_agent_id,
                 agent_name,
                 parent_conversation_id,
+                is_remote_child,
                 run_id,
                 autoexecute_override,
                 last_event_sequence,
@@ -413,6 +416,7 @@ impl AIConversation {
                 None,
                 None,
                 None,
+                false,
                 None,
                 AIConversationAutoexecuteMode::default(),
                 None,
@@ -457,7 +461,7 @@ impl AIConversation {
             parent_agent_id,
             agent_name,
             parent_conversation_id,
-            is_remote_child: false,
+            is_remote_child,
             last_event_sequence,
         })
     }
@@ -809,7 +813,7 @@ impl AIConversation {
 
     /// Returns true if this conversation was spawned by a parent orchestrator agent.
     pub fn is_child_agent_conversation(&self) -> bool {
-        self.parent_conversation_id.is_some()
+        self.parent_conversation_id.is_some() || self.parent_agent_id.is_some()
     }
 
     /// True iff this conversation knows about a parent agent — either via a
@@ -2838,6 +2842,7 @@ impl AIConversation {
                 parent_agent_id: self.parent_agent_id.clone(),
                 agent_name: self.agent_name.clone(),
                 parent_conversation_id: self.parent_conversation_id.map(|id| id.to_string()),
+                is_remote_child: self.is_remote_child,
                 run_id: self.task_id.map(|id| id.to_string()),
                 autoexecute_override: Some(self.autoexecute_override.into()),
                 last_event_sequence: self.last_event_sequence,

app/src/ai/agent/conversation_tests.rs

@@ -121,6 +121,41 @@ fn restored_conversation_defaults_autoexecute_override_when_not_persisted() {
     );
 }
 
+#[test]
+fn restored_conversation_uses_persisted_last_event_sequence() {
+    let conversation_data: AgentConversationData =
+        serde_json::from_str(r#"{"server_conversation_token":null,"last_event_sequence":42}"#)
+            .unwrap();
+
+    let conversation = restored_conversation(Some(conversation_data));
+
+    assert_eq!(conversation.last_event_sequence(), Some(42));
+}
+
+#[test]
+fn restored_conversation_uses_persisted_remote_child_marker() {
+    let conversation_data: AgentConversationData =
+        serde_json::from_str(r#"{"server_conversation_token":null,"is_remote_child":true}"#)
+            .unwrap();
+
+    let conversation = restored_conversation(Some(conversation_data));
+
+    assert!(conversation.is_remote_child());
+}
+
+#[test]
+fn child_conversation_detection_uses_parent_agent_id() {
+    let conversation_data: AgentConversationData = serde_json::from_str(
+        r#"{"server_conversation_token":null,"parent_agent_id":"parent-run-id"}"#,
+    )
+    .unwrap();
+
+    let conversation = restored_conversation(Some(conversation_data));
+
+    assert!(conversation.is_child_agent_conversation());
+    assert_eq!(conversation.parent_conversation_id(), None);
+}
+
 #[test]
 fn restored_conversation_defaults_unknown_persisted_autoexecute_override() {
     let _flag = FeatureFlag::RememberFastForwardState.override_enabled(true);

app/src/ai/agent_conversations_model_tests.rs

@@ -152,6 +152,7 @@ fn test_display_status_uses_matching_conversation_for_in_progress_task() {
                 parent_agent_id: None,
                 agent_name: None,
                 parent_conversation_id: None,
+                is_remote_child: false,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
                 last_event_sequence: None,
@@ -204,6 +205,7 @@ fn test_display_status_updates_when_blocked_conversation_resumes() {
                 parent_agent_id: None,
                 agent_name: None,
                 parent_conversation_id: None,
+                is_remote_child: false,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
                 last_event_sequence: None,
@@ -280,6 +282,7 @@ fn test_display_status_terminal_task_state_overrides_matching_conversation() {
                 parent_agent_id: None,
                 agent_name: None,
                 parent_conversation_id: None,
+                is_remote_child: false,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
                 last_event_sequence: None,
@@ -332,6 +335,7 @@ fn test_status_filter_uses_display_status_for_task_backed_conversations() {
                 parent_agent_id: None,
                 agent_name: None,
                 parent_conversation_id: None,
+                is_remote_child: false,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
                 last_event_sequence: None,
@@ -766,6 +770,7 @@ fn test_get_tasks_and_conversations_prefers_task_when_task_id_matches_conversati
                 parent_agent_id: None,
                 agent_name: None,
                 parent_conversation_id: None,
+                is_remote_child: false,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
                 last_event_sequence: None,
@@ -823,6 +828,7 @@ fn test_get_tasks_and_conversations_prefers_task_when_server_token_matches() {
                 parent_agent_id: None,
                 agent_name: None,
                 parent_conversation_id: None,
+                is_remote_child: false,
                 run_id: None,
                 autoexecute_override: None,
                 last_event_sequence: None,
@@ -879,6 +885,7 @@ fn test_get_tasks_and_conversations_keeps_unrelated_tasks_and_conversations() {
                 parent_agent_id: None,
                 agent_name: None,
                 parent_conversation_id: None,
+                is_remote_child: false,
                 run_id: None,
                 autoexecute_override: None,
                 last_event_sequence: None,

app/src/ai/agent_events/message_hydrator.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use crate::ai::ambient_agents::AmbientAgentTaskId;
 use anyhow::{anyhow, Context, Result};
 #[cfg(not(target_family = "wasm"))]
 use futures::future::Either;
@@ -9,6 +10,7 @@ use warpui::r#async::Timer;
 
 use crate::ai::agent::ReceivedMessageInput;
 use crate::server::server_api::ai::{AIClient, AgentRunEvent, ReadAgentMessageResponse};
+use crate::server::server_api::ServerApi;
 
 pub(crate) const DEFAULT_AGENT_MESSAGE_FETCH_TIMEOUT: Duration = Duration::from_secs(5);
 
@@ -17,6 +19,8 @@ pub(crate) const DEFAULT_AGENT_MESSAGE_FETCH_TIMEOUT: Duration = Duration::from_
 #[derive(Clone)]
 pub(crate) struct MessageHydrator {
     ai_client: Arc<dyn AIClient>,
+    task_scoped_server_api: Option<Arc<ServerApi>>,
+    task_id: Option<AmbientAgentTaskId>,
     #[cfg_attr(target_family = "wasm", allow(dead_code))]
     fetch_timeout: Duration,
 }
@@ -26,16 +30,40 @@ impl MessageHydrator {
         Self::with_fetch_timeout(ai_client, DEFAULT_AGENT_MESSAGE_FETCH_TIMEOUT)
     }
 
+    pub(crate) fn for_task(server_api: Arc<ServerApi>, task_id: AmbientAgentTaskId) -> Self {
+        let ai_client: Arc<dyn AIClient> = server_api.clone();
+        Self {
+            ai_client,
+            task_scoped_server_api: Some(server_api),
+            task_id: Some(task_id),
+            fetch_timeout: DEFAULT_AGENT_MESSAGE_FETCH_TIMEOUT,
+        }
+    }
+
     pub(crate) fn with_fetch_timeout(
         ai_client: Arc<dyn AIClient>,
         fetch_timeout: Duration,
     ) -> Self {
         Self {
             ai_client,
+            task_scoped_server_api: None,
+            task_id: None,
             fetch_timeout,
         }
     }
 
+    async fn read_message(&self, message_id: &str) -> Result<ReadAgentMessageResponse> {
+        match (self.task_scoped_server_api.as_ref(), self.task_id) {
+            (Some(server_api), Some(task_id)) => {
+                server_api
+                    .read_agent_message_for_task(&task_id, message_id)
+                    .await
+            }
+            _ => self.ai_client.read_agent_message(message_id).await,
+        }
+        .with_context(|| format!("Failed to read agent message {message_id}"))
+    }
+
     pub(crate) async fn hydrate_event_for_recipient(
         &self,
         event: &AgentRunEvent,
@@ -55,6 +83,17 @@ impl MessageHydrator {
                 return None;
             }
         };
+        if message.body.is_empty() {
+            log::warn!(
+                "Hydrated empty-body agent message: message_id={} event_sequence={} recipient_run_id={} sender_run_id={} subject={:?} task_id={:?}",
+                message.message_id,
+                event.sequence,
+                recipient_run_id,
+                message.sender_run_id,
+                message.subject,
+                self.task_id.map(|task_id| task_id.to_string())
+            );
+        }
 
         Some(ReceivedMessageInput {
             message_id: message.message_id,
@@ -70,15 +109,13 @@ impl MessageHydrator {
         &self,
         message_id: &str,
     ) -> Result<ReadAgentMessageResponse> {
-        let read_message = self.ai_client.read_agent_message(message_id);
+        let read_message = self.read_message(message_id);
         let timeout = Timer::after(self.fetch_timeout);
         futures::pin_mut!(read_message);
         futures::pin_mut!(timeout);
 
         match futures::future::select(read_message, timeout).await {
-            Either::Left((result, _)) => {
-                result.with_context(|| format!("Failed to read agent message {message_id}"))
-            }
+            Either::Left((result, _)) => result,
             Either::Right(_) => Err(anyhow!("Timed out reading agent message {message_id}")),
         }
     }
@@ -88,10 +125,7 @@ impl MessageHydrator {
         &self,
         message_id: &str,
     ) -> Result<ReadAgentMessageResponse> {
-        self.ai_client
-            .read_agent_message(message_id)
-            .await
-            .with_context(|| format!("Failed to read agent message {message_id}"))
+        self.read_message(message_id).await
     }
 
     pub(crate) async fn read_message_from_event_with_timeout(
@@ -105,10 +139,15 @@ impl MessageHydrator {
     }
 
     pub(crate) async fn mark_message_delivered(&self, message_id: &str) -> Result<()> {
-        self.ai_client
-            .mark_message_delivered(message_id)
-            .await
-            .with_context(|| format!("Failed to mark agent message {message_id} as delivered"))
+        match (self.task_scoped_server_api.as_ref(), self.task_id) {
+            (Some(server_api), Some(task_id)) => {
+                server_api
+                    .mark_message_delivered_for_task(&task_id, message_id)
+                    .await
+            }
+            _ => self.ai_client.mark_message_delivered(message_id).await,
+        }
+        .with_context(|| format!("Failed to mark agent message {message_id} as delivered"))
     }
 
     pub(crate) async fn mark_messages_delivered_best_effort<'a, I>(