QUALITY-928 M2: Pane path, transcript, OVM, ACM, and hydration simplification

Wires M1's tracker into the client pane group and conversation model:

- OrchestrationViewerModel: uses AgentConversationsModel as fetch authority;
  pending_task_ids_for_discovery drain path; drain_pending_task_discoveries.
- Unified child pane materialization: apply_child_pane_materialization,
  attach_ambient_orchestration_child_session, hydrate_child_transcript, and
  process_pending_child_hydrations replace the parallel owner/viewer paths.
  ChildPaneOrigin dropped from the pending map; stale-session guard universal.
- new_for_ambient_orchestration_child: TerminalManager constructor combining
  is_ambient_agent=true with orchestration_child_conversation_id=Some for
  FailedToJoin recovery routing on all child panes.
- TaskOwnership/TaskScope removed; simple creator.uid check used instead.
- Dead code removed: create_orchestration_child_shared_session_viewer,
  new_for_orchestration_child, ChildPaneOrigin enum.
- EnsureSharedSessionViewerChildPane handler simplified (flag-ON branch removed).
- Orchestration tracker/streamer tests updated for M2 behavioral changes:
  Started inserts TrackedChild immediately (children non-empty after observe_child).

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

Author: cephalonaut

app/src/ai/agent_conversations_model.rs

@@ -1273,6 +1273,8 @@ impl AgentConversationsModel {
         self.tasks.values()
     }
 
+    /// Seeds the task cache so tests can exercise cache-hit paths without a
+    /// server round trip.
     #[cfg(test)]
     pub(crate) fn insert_task_for_test(&mut self, task: AmbientAgentTask) {
         self.tasks.insert(task.task_id, task);
@@ -1719,6 +1721,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/blocklist/agent_view/controller.rs

@@ -14,6 +14,7 @@ use crate::ai::agent::conversation::AIConversationId;
 use crate::ai::blocklist::orchestration_topology::{
     OrchestrationNavigationDirection, adjacent_orchestration_child_conversation_id,
 };
+use crate::features::FeatureFlag;
 use crate::terminal::TerminalModel;
 use crate::terminal::input::message_bar::{Message, MessageItem};
 use crate::terminal::input::slash_commands::SlashCommandTrigger;
@@ -804,22 +805,29 @@ impl AgentViewController {
         }
 
         let history_model = BlocklistAIHistoryModel::handle(ctx);
-        let (conversation_id, exchange_count) = if let Some(conversation) =
-            conversation_id.and_then(|id| history_model.as_ref(ctx).conversation(&id))
-        {
-            (conversation.id(), conversation.exchange_count())
-        } else {
-            let id = history_model.update(ctx, |history_model, ctx| {
-                history_model.start_new_conversation(
-                    self.terminal_view_id,
-                    false,
-                    matches!(&origin, AgentViewEntryOrigin::CloudAgent),
-                    matches!(&origin, AgentViewEntryOrigin::ThirdPartyCloudAgent),
-                    ctx,
+        let (conversation_id, exchange_count, is_existing_child_placeholder) =
+            if let Some(conversation) =
+                conversation_id.and_then(|id| history_model.as_ref(ctx).conversation(&id))
+            {
+                (
+                    conversation.id(),
+                    conversation.exchange_count(),
+                    conversation.is_remote_child()
+                        || (conversation.is_viewing_shared_session()
+                            && conversation.parent_conversation_id().is_some()),
                 )
-            });
-            (id, 0)
-        };
+            } else {
+                let id = history_model.update(ctx, |history_model, ctx| {
+                    history_model.start_new_conversation(
+                        self.terminal_view_id,
+                        false,
+                        matches!(&origin, AgentViewEntryOrigin::CloudAgent),
+                        matches!(&origin, AgentViewEntryOrigin::ThirdPartyCloudAgent),
+                        ctx,
+                    )
+                });
+                (id, 0, false)
+            };
         history_model.update(ctx, |history_model, ctx| {
             history_model.set_active_conversation_id(conversation_id, self.terminal_view_id, ctx)
         });
@@ -841,9 +849,17 @@ impl AgentViewController {
             .block_list_mut()
             .enter_conversation_context(conversation_id, display_mode.is_inline(), is_cloud);
 
+        // An empty child placeholder is still an existing run, not a brand-new
+        // cloud conversation. This applies to owner-side remote children and
+        // viewer-side shared-session children. Preserve that distinction so
+        // TerminalView does not insert cloud composition UI while the child is
+        // restoring or waiting for its first streamed exchange.
+        let is_new = exchange_count == 0
+            && !(FeatureFlag::OrchestrationUnifiedStack.is_enabled()
+                && is_existing_child_placeholder);
         ctx.emit(AgentViewControllerEvent::EnteredAgentView {
             conversation_id,
-            is_new: exchange_count == 0,
+            is_new,
             origin,
             display_mode,
         });

app/src/ai/blocklist/history_model.rs

@@ -568,6 +568,49 @@ impl BlocklistAIHistoryModel {
         conversation_id
     }
 
+    /// Returns the existing run-id mapping for a remote child, creating one
+    /// from the supplied task metadata if none exists yet. Idempotent: racing
+    /// `ChildStarted`, lifecycle, and viewer metadata callbacks all converge
+    /// on the same entry.
+    #[allow(clippy::too_many_arguments)]
+    pub fn ensure_remote_child_conversation(
+        &mut self,
+        terminal_surface_id: EntityId,
+        parent_conversation_id: AIConversationId,
+        run_id: String,
+        task_id: crate::ai::ambient_agents::AmbientAgentTaskId,
+        name: String,
+        fallback_title: String,
+        orchestration_harness: Option<Harness>,
+        ctx: &mut ModelContext<Self>,
+    ) -> AIConversationId {
+        if let Some(conversation_id) = self.conversation_id_for_agent_id(&run_id) {
+            return conversation_id;
+        }
+
+        let conversation_id = self.start_new_child_conversation(
+            terminal_surface_id,
+            name,
+            parent_conversation_id,
+            orchestration_harness,
+            ctx,
+        );
+        self.mark_conversation_as_remote_child(conversation_id, ctx);
+        if !fallback_title.is_empty()
+            && let Some(conversation) = self.conversation_mut(&conversation_id)
+        {
+            conversation.set_fallback_display_title(fallback_title);
+        }
+        self.assign_run_id_for_conversation(
+            conversation_id,
+            run_id,
+            Some(task_id),
+            terminal_surface_id,
+            ctx,
+        );
+        conversation_id
+    }
+
     /// Sets the parent conversation ID on a child conversation and updates
     /// the `children_by_parent` index.  All parent-child relationships should
     /// be established through this method so the index stays in sync.

app/src/ai/blocklist/history_model_tests.rs

@@ -70,6 +70,67 @@ fn create_persisted_query(
     }
 }
 
+#[test]
+fn ensure_remote_child_conversation_creates_one_named_run_mapping() {
+    App::test((), |mut app| async move {
+        initialize_history_persistence_for_tests(&mut app);
+        let terminal_view_id = EntityId::new();
+        let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new_for_test());
+        let parent_run_id = "11111111-1111-1111-1111-111111111111";
+        let child_task_id: AmbientAgentTaskId =
+            "22222222-2222-2222-2222-222222222222".parse().unwrap();
+
+        let (parent_id, first, second) = history_model.update(&mut app, |history, ctx| {
+            let parent_id =
+                history.start_new_conversation(terminal_view_id, false, true, false, ctx);
+            history.assign_run_id_for_conversation(
+                parent_id,
+                parent_run_id.to_string(),
+                parent_run_id.parse().ok(),
+                terminal_view_id,
+                ctx,
+            );
+            let first = history.ensure_remote_child_conversation(
+                terminal_view_id,
+                parent_id,
+                child_task_id.to_string(),
+                child_task_id,
+                "Researcher".to_string(),
+                "Investigate observer restore".to_string(),
+                Some(Harness::Codex),
+                ctx,
+            );
+            let second = history.ensure_remote_child_conversation(
+                terminal_view_id,
+                parent_id,
+                child_task_id.to_string(),
+                child_task_id,
+                "Duplicate".to_string(),
+                String::new(),
+                Some(Harness::Oz),
+                ctx,
+            );
+            (parent_id, first, second)
+        });
+
+        assert_eq!(first, second);
+        history_model.read(&app, |history, _| {
+            assert_eq!(
+                history.conversation_id_for_agent_id(&child_task_id.to_string()),
+                Some(first),
+                "message sender attribution must resolve through the run-id index",
+            );
+            assert_eq!(history.child_conversation_ids_of(&parent_id), &[first]);
+            let child = history.conversation(&first).unwrap();
+            assert_eq!(child.agent_name(), Some("Researcher"));
+            assert_eq!(child.parent_conversation_id(), Some(parent_id));
+            assert!(child.is_remote_child());
+            assert!(!child.is_viewing_shared_session());
+            assert_eq!(child.orchestration_harness(), Some(Harness::Codex));
+        });
+    });
+}
+
 fn create_user_query_message(
     id: &str,
     task_id: &str,

app/src/ai/blocklist/mod.rs

@@ -12,7 +12,6 @@ pub(crate) mod diff_types;
 pub(crate) mod handoff;
 
 pub(crate) mod local_agent_task_sync_model;
-#[allow(dead_code)]
 pub(crate) mod orchestration_child_tracker;
 pub(crate) mod orchestration_event_streamer;
 pub(crate) mod orchestration_events;

app/src/ai/blocklist/orchestration_child_tracker.rs

@@ -7,6 +7,11 @@
 //! session links, REST seed rows, and in-band registrations — enters through
 //! the single [`OrchestrationChildTracker::observe_child`] entry point.
 //!
+//! `FamilyDrainMode` captures the one behavioral axis between orchestrator
+//! and shared-session observer: who pushes the server cursor and who receives
+//! the parent's own inbox events. It says nothing about authenticated
+//! ownership, permissions, or pane capability.
+//!
 //! Pill-bar broadcasts (`ChildSpawned` / `ChildStatusChanged`) are emitted
 //! via the `ctx` so downstream views can react without polling.
 //!
@@ -50,9 +55,11 @@ pub enum ChildSignal {
     Lifecycle(api::LifecycleEventType),
     /// A REST seed row (cold-start seed / restore fetch). Boxed because the
     /// task row dwarfs the other variants.
+    #[allow(dead_code)]
     Seeded(Box<AmbientAgentTask>),
     /// A child created in this process, already backed by a local
     /// conversation that its executor hydrates.
+    #[allow(dead_code)]
     Registered,
 }
 
@@ -67,6 +74,7 @@ pub struct TrackedChild {
     /// `true` for every placeholder the tracker materializes on behalf of a
     /// run hosted elsewhere. `false` only for in-band children, which already
     /// own a real local conversation and are tracked for status only.
+    #[allow(dead_code)]
     pub is_remote_child: bool,
 }
 
@@ -160,8 +168,9 @@ impl OrchestrationChildTracker {
 
     /// Discovery via `child_agent_started`. Idempotent: an already-known child
     /// only continues hydrating, while the first sighting of a genuinely new
-    /// out-of-band run kicks off a single metadata fetch to create its
-    /// placeholder.
+    /// out-of-band run inserts a pending `TrackedChild` immediately — before
+    /// the async metadata fetch completes — so later `Lifecycle` and
+    /// `SessionLinked` signals see a known child and are processed.
     fn apply_started(
         &mut self,
         task_id: AmbientAgentTaskId,
@@ -174,9 +183,24 @@ impl OrchestrationChildTracker {
             self.maybe_request_pane_materialization(task_id, ctx);
             return;
         }
-        // New out-of-band child: start (or dedupe) discovery. The placeholder
-        // is created when the fetch completes (a cache hit resolves inline; an
-        // in-flight fetch resolves on a later re-drive).
+        // Insert a placeholder TrackedChild immediately so lifecycle and
+        // session-linked signals that arrive before the async metadata fetch
+        // completes see tracker_known=true. Any session_id that arrived
+        // before this signal is also applied now.
+        let session_id = self.pending_session_ids.remove(&task_id);
+        self.insert_child(
+            task_id,
+            run_id,
+            TrackedChild {
+                session_id,
+                last_state: None,
+                pane_materialized: false,
+                is_remote_child: true,
+            },
+            ctx,
+        );
+        // Also kick the metadata fetch to get real task state, session_id,
+        // and conversation token for transcript / live-attach decisions.
         self.spawn_metadata_fetch(task_id, run_id, ctx);
     }
 
@@ -192,7 +216,8 @@ impl OrchestrationChildTracker {
         kind: api::LifecycleEventType,
         ctx: &mut ModelContext<OrchestrationEventStreamer>,
     ) {
-        if self.children.contains_key(&task_id) {
+        let tracker_known = self.children.contains_key(&task_id);
+        if tracker_known {
             let status = conversation_status_from_lifecycle_event_type(kind);
             // Write status through immediately so the pill bar badge reflects
             // the lifecycle transition without waiting for a redraw cycle.
@@ -228,11 +253,17 @@ impl OrchestrationChildTracker {
             self.maybe_request_pane_materialization(task_id, ctx);
             return;
         }
-        // Lifecycle for an unknown run: only self-heal a real discovery miss,
-        // not a run whose fetch is already in flight.
-        if !self.metadata_fetches.contains(run_id) {
-            self.spawn_metadata_fetch(task_id, run_id, ctx);
-        }
+        // Lifecycle for an unknown run is a complete discovery backstop:
+        // insert once (emitting ChildSpawned), start/dedupe metadata hydration,
+        // and publish the status immediately. This handles a missed or
+        // reordered child_agent_started event without a tracker-only ghost.
+        self.apply_started(task_id, run_id, ctx);
+        let status = conversation_status_from_lifecycle_event_type(kind);
+        ctx.emit(OrchestrationEventStreamerEvent::ChildStatusChanged {
+            parent_task_id: self.parent_task_id,
+            run_id: run_id.to_string(),
+            status,
+        });
     }
 
     /// Registers a child created in this process against its existing local