QUALITY-928: Orchestration unified stack — observer-parent restore (M3)

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

@@ -85,6 +85,7 @@ pub fn convert_conversation_data_to_ai_conversation(
             orchestration_harness_type: None,
             parent_conversation_id: None,
             is_remote_child: false,
+            is_durable_observer_parent: false,
             root_task_is_optimistic: None,
             run_id: None,
             autoexecute_override: None,
@@ -104,6 +105,7 @@ pub fn convert_conversation_data_to_ai_conversation(
             orchestration_harness_type: None,
             parent_conversation_id: None,
             is_remote_child: false,
+            is_durable_observer_parent: false,
             root_task_is_optimistic: None,
             run_id: metadata
                 .ambient_agent_task_id

app/src/ai/agent/conversation.rs

@@ -346,6 +346,10 @@ pub struct AIConversation {
     /// these conversations — the remote worker's own client handles status
     /// reporting.
     is_remote_child: bool,
+    /// True when this is an owned cloud parent hosted by a remote driver and
+    /// observed locally. Unlike `is_viewing_shared_session`, this marker is
+    /// durable so the local observer cursor and hierarchy can be restored.
+    is_durable_observer_parent: bool,
 
     /// The last event sequence number observed from the v2 orchestration
     /// event log. Used on restore to resume event delivery without
@@ -411,6 +415,7 @@ impl AIConversation {
             orchestration_harness_type: None,
             parent_conversation_id: None,
             is_remote_child: false,
+            is_durable_observer_parent: false,
             last_event_sequence: None,
             orchestration_configs: HashMap::new(),
             pinned: false,
@@ -545,6 +550,7 @@ impl AIConversation {
             orchestration_harness_type,
             parent_conversation_id,
             is_remote_child,
+            is_durable_observer_parent,
             run_id,
             autoexecute_override,
             last_event_sequence,
@@ -596,6 +602,7 @@ impl AIConversation {
                 data.orchestration_harness_type,
                 parent_conversation_id,
                 data.is_remote_child,
+                data.is_durable_observer_parent,
                 data.run_id,
                 autoexecute_override,
                 data.last_event_sequence,
@@ -613,6 +620,7 @@ impl AIConversation {
                 None,
                 None,
                 false,
+                false,
                 None,
                 AIConversationAutoexecuteMode::default(),
                 None,
@@ -622,7 +630,7 @@ impl AIConversation {
 
         Ok(Self {
             id,
-            is_viewing_shared_session: false,
+            is_viewing_shared_session: is_durable_observer_parent,
             is_cli_agent_transcript: false,
             task_store,
             status,
@@ -653,6 +661,7 @@ impl AIConversation {
             orchestration_harness_type,
             parent_conversation_id,
             is_remote_child,
+            is_durable_observer_parent,
             last_event_sequence,
             orchestration_configs: HashMap::new(),
             pinned,
@@ -683,6 +692,14 @@ impl AIConversation {
         self.is_viewing_shared_session = is_viewing_shared_session;
     }
 
+    pub fn is_durable_observer_parent(&self) -> bool {
+        self.is_durable_observer_parent
+    }
+
+    pub fn set_is_durable_observer_parent(&mut self, durable: bool) {
+        self.is_durable_observer_parent = durable;
+    }
+
     pub fn is_cli_agent_transcript(&self) -> bool {
         self.is_cli_agent_transcript
     }
@@ -3477,8 +3494,10 @@ impl AIConversation {
         &mut self,
         ctx: &mut ModelContext<BlocklistAIHistoryModel>,
     ) {
-        // We should not persist non-local conversations (e.g. shared sessions).
-        if self.is_viewing_shared_session {
+        // Passive shared-session views remain ephemeral. Owned cloud parents
+        // are the narrow exception: their local observer cursor and child
+        // hierarchy must survive restart.
+        if self.is_viewing_shared_session && !self.is_durable_observer_parent {
             return;
         }
 
@@ -3546,6 +3565,7 @@ impl AIConversation {
                 orchestration_harness_type: self.orchestration_harness_type.clone(),
                 parent_conversation_id: self.parent_conversation_id.map(|id| id.to_string()),
                 is_remote_child: self.is_remote_child,
+                is_durable_observer_parent: self.is_durable_observer_parent,
                 // Legacy field; retained for backward-compatible
                 // deserialization but no longer written. The optimistic-root
                 // case is now handled by `Task::source_for_persistence`

app/src/ai/agent_conversations_model.rs

@@ -1719,6 +1719,59 @@ impl AgentConversationsModel {
         }
     }
 
+    /// Updates a cached task to reflect that execution has started and its
+    /// session is now known (from a `run_session_linked` wire event). If the
+    /// task is not yet cached, starts a fetch to retrieve it.
+    ///
+    /// Mutating the cache entry directly avoids a full round-trip while still
+    /// giving `decide_child_pane_materialization` the `InProgress` +
+    /// `is_sandbox_running=true` + `session_id` it needs to return `AttachLive`
+    /// on the next pill click. `TasksUpdated` is emitted so any pending
+    /// re-drives fire immediately.
+    pub fn update_task_as_running_with_session(
+        &mut self,
+        task_id: &AmbientAgentTaskId,
+        session_id_str: String,
+        ctx: &mut ModelContext<Self>,
+    ) {
+        use crate::ai::ambient_agents::AmbientAgentTaskState;
+        if let Some(task) = self.tasks.get_mut(task_id) {
+            task.session_id = Some(session_id_str);
+            task.is_sandbox_running = true;
+            // Only promote to InProgress if still in a queued/pending state;
+            // never downgrade a terminal state that may have arrived concurrently.
+            match task.state {
+                AmbientAgentTaskState::Queued
+                | AmbientAgentTaskState::Pending
+                | AmbientAgentTaskState::Claimed => {
+                    task.state = AmbientAgentTaskState::InProgress;
+                }
+                _ => {}
+            }
+            ctx.emit(AgentConversationsModelEvent::TasksUpdated);
+        } else {
+            // Task not cached yet; start a fetch.
+            self.async_fetch_task(task_id, ctx);
+        }
+    }
+
+    /// Evicts a task from the cache and immediately starts a fresh
+    /// `GET /agent/runs/{id}` fetch. Used by the family drain when a terminal
+    /// lifecycle event arrives for a child whose cached state is stale (e.g.
+    /// still shows `Queued` from the initial discovery fetch). The refreshed
+    /// data — including the server conversation token and terminal state —
+    /// enables `decide_child_pane_materialization` to return `LoadTranscript`
+    /// so subsequent pill clicks load the cloud transcript.
+    pub fn evict_and_refetch_task(
+        &mut self,
+        task_id: &AmbientAgentTaskId,
+        ctx: &mut ModelContext<Self>,
+    ) {
+        self.tasks.remove(task_id);
+        self.task_fetch_state.remove(task_id);
+        self.async_fetch_task(task_id, ctx);
+    }
+
     /// Get raw task data by task ID, fetching from server if not in memory.
     /// If the task is already in memory, returns it immediately.
     /// If not, spawns an async task to fetch it from the server, stores it in memory,

app/src/ai/agent_conversations_model_tests.rs

@@ -70,6 +70,7 @@ fn create_test_task(
             display_name: Some(format!("User {creator_uid}")),
         }),
         executor: None,
+        scope: None,
         conversation_id: None,
         request_usage: None,
         agent_config_snapshot: None,
@@ -234,6 +235,7 @@ fn test_title_update_refreshes_shadowing_task_title() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: None,
                 autoexecute_override: None,
@@ -341,6 +343,7 @@ fn test_display_status_uses_matching_conversation_for_in_progress_task() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
@@ -398,6 +401,7 @@ fn test_display_status_uses_active_execution_over_previous_conversation_status()
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
@@ -462,6 +466,7 @@ fn test_display_status_updates_when_blocked_conversation_resumes() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
@@ -542,6 +547,7 @@ fn test_display_status_terminal_task_state_overrides_matching_conversation() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
@@ -597,6 +603,7 @@ fn test_status_filter_uses_display_status_for_task_backed_conversations() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
@@ -1052,6 +1059,7 @@ fn test_get_entries_excludes_conversation_shadowed_by_child_task() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: None,
                 autoexecute_override: None,
@@ -1202,6 +1210,7 @@ fn test_get_entries_merges_task_and_local_conversation_by_run_id() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
@@ -1257,6 +1266,7 @@ fn test_get_entries_merges_task_and_local_conversation_by_server_token() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: None,
                 autoexecute_override: None,
@@ -1467,6 +1477,7 @@ fn test_resolve_open_action_returns_none_for_active_unattachable_session() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
@@ -1753,6 +1764,7 @@ fn test_server_token_assignment_updates_copy_link_resolution() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: None,
                 autoexecute_override: None,
@@ -1915,6 +1927,7 @@ fn test_resolve_copy_link_uses_attached_synced_conversation_for_task_without_tok
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
@@ -2244,6 +2257,7 @@ fn test_get_entries_prefers_task_when_task_id_matches_conversation_run_id() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: Some(task_id.clone()),
                 autoexecute_override: None,
@@ -2305,6 +2319,7 @@ fn test_get_entries_prefers_task_when_server_token_matches() {
                 orchestration_harness_type: None,
                 parent_conversation_id: None,
                 is_remote_child: false,
+                is_durable_observer_parent: false,
                 root_task_is_optimistic: None,
                 run_id: None,
                 autoexecute_override: None,

app/src/ai/ambient_agents/mod.rs

@@ -18,7 +18,8 @@ pub mod telemetry;
 
 pub use task::{
     AgentConfigSnapshot, AgentSource, AmbientAgentLiveSessionState, AmbientAgentTask,
-    AmbientAgentTaskState, TaskStatusMessage, cancel_task_silently, cancel_task_with_toast,
+    AmbientAgentTaskState, TaskOwnership, TaskStatusMessage, cancel_task_silently,
+    cancel_task_with_toast,
 };
 pub const OUT_OF_CREDITS_TASK_FAILURE_MESSAGE: &str =
     "Out of credits. Upgrade your Warp plan to continue running cloud agents.";

app/src/ai/ambient_agents/spawn_tests.rs

@@ -34,6 +34,7 @@ fn task_with(
         session_link,
         creator: None,
         executor: None,
+        scope: None,
         conversation_id: None,
         request_usage: None,
         agent_config_snapshot: None,