From 57e8e3e9becb327ff4a2e27ee467d50b229cd4e4 Mon Sep 17 00:00:00 2001 From: Matthew Albright Date: Tue, 28 Apr 2026 14:24:54 -0400 Subject: [PATCH] Replay agent events on restore (#9251) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description Restore orchestration event delivery on the client after a Warp restart so that a parent conversation continues to receive lifecycle events and inbox messages from its children — including terminal events that arrived while Warp was not running. See the full design in `specs/replay-agent-events-on-restore/PRODUCT.md` and `specs/replay-agent-events-on-restore/TECH.md`. This is a re-land of warpdotdev/warp-internal#24999, which was reverted in warpdotdev/warp-internal#25055 due to a CI race. This version fixes the test compilation issue (missing fields in `AmbientAgentTask` struct literal in `conversation_ended_tombstone_view_tests.rs`). ### What - Persists the per-conversation event cursor across restarts. - Adds `last_event_sequence: Option` to `AgentConversationData` (SQLite) and `AIConversation`. - New `BlocklistAIHistoryModel::update_event_sequence` helper writes the cursor through `write_updated_conversation_state` after each event batch. - Also persists the cursor to the server (fire-and-forget) so driver / cloud restarts can resume without local SQLite state. - Restores `OrchestrationEventPoller.watched_run_ids` and re-establishes event delivery on `BlocklistAIHistoryEvent::RestoredConversations`. - New `on_restored_conversations` handler issues `GET /agent/runs/{run_id}` for each restored parent and uses the response inline `children` and `last_event_sequence` to populate watched run ids and merge the cursor (`max(SQLite, server)`). - Fetch failures retry with exponential backoff (1s, 2s, 5s, 10s capped) keyed off a per-conversation `restore_fetch_failures` counter; reset on success and on conversation removal. - Gated on `OrchestrationV2`. Shared-session viewers and conversations without children are skipped. - `Success` parents resume delivery immediately; `InProgress` parents defer to the existing `on_conversation_status_updated` path. - Restores V1 lifecycle subscriptions on restart by extending the existing `RestoredConversations` handler in `OrchestrationEventService` to re-register `lifecycle_subscription_routes` for restored child conversations whose parents are present locally. ### Why After a Warp restart, parent conversations were silently receiving no further events from children. The `event_cursor` in `OrchestrationEventPoller` was always initialized to 0, so even if delivery had resumed, every event since the start of the conversation would have replayed and produced duplicate messages. V1 lifecycle subscription routes were also not restored, so V1 parents missed child status transitions. ## Testing - Added unit tests in `app/src/ai/blocklist/orchestration_event_poller_tests.rs` covering: cursor merge from server vs SQLite, retry on `get_ambient_agent_task` failure, V2 gating, shared-session-viewer exclusion, cleanup on delete, and `last_event_sequence` round-trip through `AIConversation::new_restored`. - Added unit test coverage in `app/src/ai/blocklist/orchestration_events_tests.rs` for V1 lifecycle re-registration on restore. - Manual verification per `specs/replay-agent-events-on-restore/TECH.md`. ## Server API dependencies - [x] 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)? - [x] If so, is the use of this API restricted to client channels that rely on the staging server (e.g. WarpDev)? The companion server change adds: - `last_event_sequence` on `Task` (`ai_tasks` column), surfaced inline on `GET /agent/runs/:run_id`. - `PATCH /agent/runs/:run_id/event-sequence` for client cursor writes. - `children` inline on the `GET /agent/runs/:run_id` response. ## Agent Mode - [x] Warp Agent Mode - This PR was created via Warp's AI Agent Mode Co-Authored-By: Oz --------- Co-authored-by: Oz --- Cargo.lock | 1 + app/src/ai/agent/api/convert_conversation.rs | 2 + app/src/ai/agent/conversation.rs | 22 + app/src/ai/agent_conversations_model_tests.rs | 9 + app/src/ai/ambient_agents/spawn_tests.rs | 2 + app/src/ai/ambient_agents/task.rs | 14 + app/src/ai/blocklist/history_model.rs | 23 + app/src/ai/blocklist/history_model_test.rs | 1 + .../blocklist/orchestration_event_poller.rs | 295 ++++++++++- .../orchestration_event_poller_tests.rs | 496 ++++++++++++++++++ app/src/ai/blocklist/orchestration_events.rs | 36 ++ .../blocklist/orchestration_events_tests.rs | 64 +++ .../ai/conversation_details_panel_tests.rs | 4 + app/src/server/server_api.rs | 33 ++ app/src/server/server_api/ai.rs | 26 + app/src/terminal/view/load_ai_conversation.rs | 1 + ...conversation_ended_tombstone_view_tests.rs | 2 + crates/http_client/src/lib.rs | 7 + crates/persistence/Cargo.toml | 3 + crates/persistence/src/model.rs | 62 +++ .../replay-agent-events-on-restore/PRODUCT.md | 37 ++ specs/replay-agent-events-on-restore/TECH.md | 195 +++++++ 22 files changed, 1334 insertions(+), 1 deletion(-) create mode 100644 specs/replay-agent-events-on-restore/PRODUCT.md create mode 100644 specs/replay-agent-events-on-restore/TECH.md diff --git a/Cargo.lock b/Cargo.lock index b723c39e..9f5999d3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9337,6 +9337,7 @@ dependencies = [ "diesel", "diesel_migrations", "serde", + "serde_json", "warp_multi_agent_api", ] diff --git a/app/src/ai/agent/api/convert_conversation.rs b/app/src/ai/agent/api/convert_conversation.rs index 7a987ff8..b77e430c 100644 --- a/app/src/ai/agent/api/convert_conversation.rs +++ b/app/src/ai/agent/api/convert_conversation.rs @@ -84,6 +84,7 @@ pub fn convert_conversation_data_to_ai_conversation( parent_conversation_id: None, run_id: None, autoexecute_override: None, + last_event_sequence: None, }, RestorationMode::Continue => AgentConversationData { server_conversation_token: Some( @@ -102,6 +103,7 @@ pub fn convert_conversation_data_to_ai_conversation( // dispatch time; adding it here would avoid a round-trip to StreamInit. run_id: None, autoexecute_override: None, + last_event_sequence: None, }, }; diff --git a/app/src/ai/agent/conversation.rs b/app/src/ai/agent/conversation.rs index b53e0887..05a2fbbe 100644 --- a/app/src/ai/agent/conversation.rs +++ b/app/src/ai/agent/conversation.rs @@ -223,6 +223,11 @@ pub struct AIConversation { /// these conversations — the remote worker's own client handles status /// reporting. TaskStatusSyncModel skips status updates for these. is_remote_child: bool, + + /// The last event sequence number observed from the v2 orchestration + /// event log. Used on restore to resume event delivery without + /// re-delivering already-processed events. + last_event_sequence: Option, } pub(crate) fn artifact_from_fork_proto( @@ -272,6 +277,7 @@ impl AIConversation { agent_name: None, parent_conversation_id: None, is_remote_child: false, + last_event_sequence: None, } } @@ -351,6 +357,7 @@ impl AIConversation { parent_conversation_id, run_id, autoexecute_override, + last_event_sequence, ) = if let Some(data) = conversation_data { let server_conversation_token = data .server_conversation_token @@ -381,6 +388,7 @@ impl AIConversation { } else { AIConversationAutoexecuteMode::default() }; + let last_event_sequence = data.last_event_sequence; ( server_conversation_token, @@ -393,6 +401,7 @@ impl AIConversation { parent_conversation_id, run_id, autoexecute_override, + last_event_sequence, ) } else { ( @@ -406,6 +415,7 @@ impl AIConversation { None, None, AIConversationAutoexecuteMode::default(), + None, ) }; @@ -448,6 +458,7 @@ impl AIConversation { agent_name, parent_conversation_id, is_remote_child: false, + last_event_sequence, }) } @@ -786,6 +797,16 @@ impl AIConversation { self.parent_conversation_id = Some(id); } + /// Returns the last observed v2 orchestration event sequence number, if any. + pub fn last_event_sequence(&self) -> Option { + self.last_event_sequence + } + + /// Updates the last observed v2 orchestration event sequence number. + pub fn set_last_event_sequence(&mut self, sequence: i64) { + self.last_event_sequence = Some(sequence); + } + /// 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() @@ -2811,6 +2832,7 @@ impl AIConversation { parent_conversation_id: self.parent_conversation_id.map(|id| id.to_string()), run_id: self.task_id.map(|id| id.to_string()), autoexecute_override: Some(self.autoexecute_override.into()), + last_event_sequence: self.last_event_sequence, }, }; ctx.spawn( diff --git a/app/src/ai/agent_conversations_model_tests.rs b/app/src/ai/agent_conversations_model_tests.rs index 253c7cec..3042520a 100644 --- a/app/src/ai/agent_conversations_model_tests.rs +++ b/app/src/ai/agent_conversations_model_tests.rs @@ -60,6 +60,8 @@ fn create_test_task( agent_config_snapshot: None, artifacts: vec![], is_sandbox_running: false, + last_event_sequence: None, + children: vec![], } } @@ -152,6 +154,7 @@ fn test_display_status_uses_matching_conversation_for_in_progress_task() { parent_conversation_id: None, run_id: Some(task_id.clone()), autoexecute_override: None, + last_event_sequence: None, }, ); @@ -203,6 +206,7 @@ fn test_display_status_updates_when_blocked_conversation_resumes() { parent_conversation_id: None, run_id: Some(task_id.clone()), autoexecute_override: None, + last_event_sequence: None, }, ); @@ -278,6 +282,7 @@ fn test_display_status_terminal_task_state_overrides_matching_conversation() { parent_conversation_id: None, run_id: Some(task_id.clone()), autoexecute_override: None, + last_event_sequence: None, }, ); @@ -329,6 +334,7 @@ fn test_status_filter_uses_display_status_for_task_backed_conversations() { parent_conversation_id: None, run_id: Some(task_id.clone()), autoexecute_override: None, + last_event_sequence: None, }, ); @@ -763,6 +769,7 @@ fn test_get_tasks_and_conversations_prefers_task_when_task_id_matches_conversati parent_conversation_id: None, run_id: Some(task_id.clone()), autoexecute_override: None, + last_event_sequence: None, }, ); @@ -819,6 +826,7 @@ fn test_get_tasks_and_conversations_prefers_task_when_server_token_matches() { parent_conversation_id: None, run_id: None, autoexecute_override: None, + last_event_sequence: None, }, ); @@ -874,6 +882,7 @@ fn test_get_tasks_and_conversations_keeps_unrelated_tasks_and_conversations() { parent_conversation_id: None, run_id: None, autoexecute_override: None, + last_event_sequence: None, }, ); diff --git a/app/src/ai/ambient_agents/spawn_tests.rs b/app/src/ai/ambient_agents/spawn_tests.rs index 6f14ec64..12fd3186 100644 --- a/app/src/ai/ambient_agents/spawn_tests.rs +++ b/app/src/ai/ambient_agents/spawn_tests.rs @@ -34,6 +34,8 @@ fn task_with( agent_config_snapshot: None, artifacts: vec![], is_sandbox_running: true, + last_event_sequence: None, + children: vec![], } } diff --git a/app/src/ai/ambient_agents/task.rs b/app/src/ai/ambient_agents/task.rs index 6459570a..375b1d48 100644 --- a/app/src/ai/ambient_agents/task.rs +++ b/app/src/ai/ambient_agents/task.rs @@ -262,6 +262,20 @@ pub struct AmbientAgentTask { pub agent_config_snapshot: Option, #[serde(default, deserialize_with = "deserialize_artifacts")] pub artifacts: Vec, + + /// The last event sequence number recorded for this run by the server. + /// Used by orchestration event delivery to resume from the correct + /// cursor on restart. Populated by `GET /agent/runs/{run_id}` when the + /// server supports it; `None` on older servers. + #[serde(default)] + pub last_event_sequence: Option, + + /// The server-recorded `run_id`s of direct children of this run. Used + /// by orchestration event-delivery restore to discover children whose + /// records may not exist locally (e.g. remote-worker children in the + /// driver case). Empty on older servers. + #[serde(default)] + pub children: Vec, } /// Represents a single attachment input from the client (e.g., file upload) diff --git a/app/src/ai/blocklist/history_model.rs b/app/src/ai/blocklist/history_model.rs index c03ac69e..3beeb8b4 100644 --- a/app/src/ai/blocklist/history_model.rs +++ b/app/src/ai/blocklist/history_model.rs @@ -452,6 +452,23 @@ impl BlocklistAIHistoryModel { .unwrap_or_default() } + /// Updates the persisted `last_event_sequence` for a conversation and + /// writes the updated conversation state to SQLite. Used by the + /// orchestration event poller after draining an event batch to keep the + /// cursor durable across restarts. + pub fn update_event_sequence( + &mut self, + conversation_id: AIConversationId, + sequence: i64, + ctx: &mut ModelContext, + ) { + let Some(conversation) = self.conversations_by_id.get_mut(&conversation_id) else { + return; + }; + conversation.set_last_event_sequence(sequence); + conversation.write_updated_conversation_state(ctx); + } + /// Sets a live conversation's server token, and updates the mapping in the history /// model. pub fn set_server_conversation_token_for_conversation( @@ -1075,6 +1092,9 @@ impl BlocklistAIHistoryModel { parent_conversation_id: None, run_id: None, autoexecute_override: Some(source_conversation.autoexecute_override().into()), + // The event cursor belongs to the source conversation's run; the + // forked conversation will establish its own cursor. + last_event_sequence: None, }; let forked_conversation_id = AIConversationId::new(); if let Err(e) = sqlite_sender.send(ModelEvent::UpdateMultiAgentConversation { @@ -1227,6 +1247,9 @@ impl BlocklistAIHistoryModel { parent_conversation_id: None, run_id: None, autoexecute_override: Some(conversation.autoexecute_override().into()), + // The event cursor belongs to the source conversation's run; the + // forked conversation will establish its own cursor. + last_event_sequence: None, }; let forked_conversation_id = AIConversationId::new(); diff --git a/app/src/ai/blocklist/history_model_test.rs b/app/src/ai/blocklist/history_model_test.rs index 95c4fd2c..6dc2a4f4 100644 --- a/app/src/ai/blocklist/history_model_test.rs +++ b/app/src/ai/blocklist/history_model_test.rs @@ -1132,6 +1132,7 @@ fn test_find_by_token_after_insert_forked_conversation_from_tasks() { parent_conversation_id: None, run_id: None, autoexecute_override: None, + last_event_sequence: None, }; let tasks = vec![warp_multi_agent_api::Task { id: "root-task".to_string(), diff --git a/app/src/ai/blocklist/orchestration_event_poller.rs b/app/src/ai/blocklist/orchestration_event_poller.rs index 9d5a6594..b8ecd819 100644 --- a/app/src/ai/blocklist/orchestration_event_poller.rs +++ b/app/src/ai/blocklist/orchestration_event_poller.rs @@ -100,6 +100,9 @@ pub struct OrchestrationEventPoller { /// Monotonic counter for SSE connection generations. Ensures stale /// callbacks from replaced connections are discarded. next_sse_generation: u64, + /// Consecutive failure count for the post-restore `get_ambient_agent_task` + /// fetch (resets on success). Drives exponential backoff for retries. + restore_fetch_failures: HashMap, } pub enum OrchestrationEventPollerEvent { @@ -126,6 +129,35 @@ impl OrchestrationEventPoller { poll_in_flight: HashSet::new(), sse_connections: HashMap::new(), next_sse_generation: 0, + restore_fetch_failures: HashMap::new(), + } + } + + /// Constructs a poller wired to the supplied (mock) clients instead of + /// looking them up via `ServerApiProvider`. Lets unit tests inject a + /// `MockAIClient` while still subscribing to `BlocklistAIHistoryModel`. + #[cfg(test)] + pub(super) fn new_with_clients_for_test( + ai_client: Arc, + server_api: Arc, + ctx: &mut ModelContext, + ) -> Self { + let history_model = BlocklistAIHistoryModel::handle(ctx); + ctx.subscribe_to_model(&history_model, |me, event, ctx| { + me.handle_history_event(event, ctx); + }); + Self { + ai_client, + server_api, + watched_run_ids: HashMap::new(), + event_cursor: HashMap::new(), + poll_backoff_index: HashMap::new(), + pending_delivery: HashMap::new(), + conversation_statuses: HashMap::new(), + poll_in_flight: HashSet::new(), + sse_connections: HashMap::new(), + next_sse_generation: 0, + restore_fetch_failures: HashMap::new(), } } @@ -187,6 +219,7 @@ impl OrchestrationEventPoller { self.pending_delivery.remove(conversation_id); self.conversation_statuses.remove(conversation_id); self.poll_in_flight.remove(conversation_id); + self.restore_fetch_failures.remove(conversation_id); // SSE cleanup // task's next send to fail, which terminates the task. self.sse_connections.remove(conversation_id); @@ -202,9 +235,235 @@ impl OrchestrationEventPoller { | BlocklistAIHistoryEvent::UpdatedTodoList { .. } | BlocklistAIHistoryEvent::UpdatedAutoexecuteOverride { .. } | BlocklistAIHistoryEvent::SplitConversation { .. } - | BlocklistAIHistoryEvent::RestoredConversations { .. } | BlocklistAIHistoryEvent::UpdatedConversationMetadata { .. } | BlocklistAIHistoryEvent::UpdatedConversationArtifacts { .. } => {} + BlocklistAIHistoryEvent::RestoredConversations { + conversation_ids, .. + } => { + self.on_restored_conversations(conversation_ids.clone(), ctx); + } + } + } + + /// Handles restoration of conversations on startup (or driver re-attach). + /// + /// Re-establishes orchestration event delivery state that is not persisted + /// directly in memory: watched run_ids, the per-conversation event cursor, + /// and — for `Success` parents with watched children — the poll/SSE loop. + fn on_restored_conversations( + &mut self, + conversation_ids: Vec, + ctx: &mut ModelContext, + ) { + // Orchestration v2 owns the events endpoints and the cursor model. + // V1 conversations may carry a run_id but the v2-only event APIs + // would return spurious 4xx responses, so skip restore entirely + // when V2 is disabled. + if !FeatureFlag::OrchestrationV2.is_enabled() { + return; + } + + for conv_id in conversation_ids { + let (run_id, cursor, status, is_viewer) = { + let history_model = BlocklistAIHistoryModel::as_ref(ctx); + let Some(conversation) = history_model.conversation(&conv_id) else { + continue; + }; + let is_viewer = conversation.is_viewing_shared_session(); + let run_id = conversation.run_id(); + let cursor = conversation.last_event_sequence().unwrap_or(0); + let status = conversation.status().clone(); + (run_id, cursor, status, is_viewer) + }; + + // Shared-session viewers receive updates through session sharing; + // polling here would re-inject events the session has already + // processed. + if is_viewer { + continue; + } + + // Initialize the in-memory cursor from the persisted SQLite value. + // A later server `GET /agent/runs/{run_id}` response may advance + // it to `max(SQLite, server)` before delivery starts. + // + // Note: a status transition arriving in the window before + // finish_restore_fetch completes may trigger + // start_event_delivery with only the SQLite cursor. This is + // acceptable — worst case is one extra batch of duplicate + // events. + self.event_cursor.insert(conv_id, cursor); + self.conversation_statuses.insert(conv_id, status.clone()); + + // Register the conversation's own run_id so lifecycle events for + // self are correctly filtered and the SSE/poll loop has a set + // of run_ids to open against. + if let Some(ref own) = run_id { + self.watched_run_ids + .entry(conv_id) + .or_default() + .insert(own.clone()); + } + + // No run_id means we can't query the server for children or for + // the canonical cursor. There's nothing more to do here; if a + // run_id gets assigned later the standard self-registration path + // will pick it up. + let Some(run_id) = run_id else { + self.maybe_start_delivery_after_restore(conv_id, &status, ctx); + continue; + }; + + let Ok(task_id) = run_id.parse::() + else { + log::warn!("could not parse run_id {run_id:?} for {conv_id:?}"); + self.maybe_start_delivery_after_restore(conv_id, &status, ctx); + continue; + }; + + self.spawn_restore_fetch(conv_id, task_id, cursor, ctx); + } + } + + /// Issues `GET /agent/runs/{task_id}` and routes the result through + /// `finish_restore_fetch`. Used both for the initial post-restore fetch + /// and for backoff-driven retries. + fn spawn_restore_fetch( + &mut self, + conv_id: AIConversationId, + task_id: crate::ai::ambient_agents::AmbientAgentTaskId, + sqlite_cursor: i64, + ctx: &mut ModelContext, + ) { + let ai_client = self.ai_client.clone(); + ctx.spawn( + async move { ai_client.get_ambient_agent_task(&task_id).await }, + move |me, run_result, ctx| { + me.finish_restore_fetch(conv_id, task_id, sqlite_cursor, run_result, ctx); + }, + ); + } + + /// Completes the post-restore async fetch by merging the server cursor, + /// installing the server-reported child run_ids, and — if the parent is + /// `Success` — starting event delivery. On a server-fetch failure, + /// schedules a retry with exponential backoff: V2 children always have a + /// server-side `ai_tasks` row, so the server is the authoritative source + /// for the watched run_id set, and any local fallback would be incomplete + /// anyway. Without network connectivity event delivery wouldn't function, + /// so retrying is the right behavior. + fn finish_restore_fetch( + &mut self, + conv_id: AIConversationId, + task_id: crate::ai::ambient_agents::AmbientAgentTaskId, + sqlite_cursor: i64, + run_result: anyhow::Result, + ctx: &mut ModelContext, + ) { + match run_result { + Ok(task) => { + // If the conversation was removed while the fetch was in-flight, + // the removal handler already cleaned up all poller state. Return + // early to avoid recreating watched_run_ids for a deleted conversation. + if !self.event_cursor.contains_key(&conv_id) { + self.restore_fetch_failures.remove(&conv_id); + return; + } + + // Reset the retry counter on success. + self.restore_fetch_failures.remove(&conv_id); + + // Merge the server cursor: use the max of SQLite and server + // values so we don't re-deliver events the client already + // acknowledged locally. + let server_seq = task.last_event_sequence.unwrap_or(0); + let merged = sqlite_cursor.max(server_seq); + self.event_cursor.insert(conv_id, merged); + + // The server response includes `children` inline on + // `AmbientAgentTask`; this is the authoritative set of + // direct child run_ids for the parent. + // + // Insert children and reconnect SSE once if any new run_ids + // were added and a connection is already open (e.g. because a + // status transition raced with this fetch and opened SSE with + // only the parent's own run_id). + let had_sse = self.sse_connections.contains_key(&conv_id); + let watched = self.watched_run_ids.entry(conv_id).or_default(); + let mut any_new_children = false; + for child in task.children { + if watched.insert(child) { + any_new_children = true; + } + } + if any_new_children && had_sse { + self.reconnect_sse(conv_id, ctx); + } + + let status = BlocklistAIHistoryModel::as_ref(ctx) + .conversation(&conv_id) + .map(|c| c.status().clone()) + .unwrap_or(ConversationStatus::Success); + self.maybe_start_delivery_after_restore(conv_id, &status, ctx); + } + Err(err) => { + log::warn!("Restore: get_agent_run failed for {conv_id:?}: {err:#}; will retry"); + self.start_restore_fetch_retry_timer(conv_id, task_id, sqlite_cursor, ctx); + } + } + } + + /// Schedules a retry of the post-restore `get_ambient_agent_task` fetch + /// after an exponential backoff. The backoff schedule reuses + /// `POLL_BACKOFF_STEPS` (1s, 2s, 5s, 10s capped) keyed on a per-conversation + /// failure counter. The counter resets on success. + fn start_restore_fetch_retry_timer( + &mut self, + conv_id: AIConversationId, + task_id: crate::ai::ambient_agents::AmbientAgentTaskId, + sqlite_cursor: i64, + ctx: &mut ModelContext, + ) { + let failures = self + .restore_fetch_failures + .entry(conv_id) + .and_modify(|c| *c += 1) + .or_insert(1); + let step_index = failures.saturating_sub(1).min(POLL_BACKOFF_STEPS.len() - 1); + let backoff = Duration::from_secs(POLL_BACKOFF_STEPS[step_index]); + ctx.spawn( + async move { Timer::after(backoff).await }, + move |me, _, ctx| { + // The conversation may have been removed in the meantime; + // if so, drop the retry. Otherwise re-issue the fetch. + if !me.event_cursor.contains_key(&conv_id) { + me.restore_fetch_failures.remove(&conv_id); + return; + } + me.spawn_restore_fetch(conv_id, task_id, sqlite_cursor, ctx); + }, + ); + } + + /// Starts event delivery for a restored conversation if the parent is + /// currently `Success` and has at least one watched run_id. `InProgress` + /// parents are deferred to `on_conversation_status_updated` once they + /// next transition to `Success`. + fn maybe_start_delivery_after_restore( + &mut self, + conv_id: AIConversationId, + status: &ConversationStatus, + ctx: &mut ModelContext, + ) { + let has_watched = self + .watched_run_ids + .get(&conv_id) + .is_some_and(|w| !w.is_empty()); + if !has_watched { + return; + } + if matches!(status, ConversationStatus::Success) { + self.start_event_delivery(conv_id, ctx); } } @@ -429,6 +688,40 @@ impl OrchestrationEventPoller { .unwrap_or(previous_cursor); self.event_cursor.insert(conversation_id, max_seq); + // Persist the cursor to SQLite so that after a restart we can resume + // event delivery from this sequence number without re-delivering + // events the parent has already acted on. + BlocklistAIHistoryModel::handle(ctx).update(ctx, |model, ctx| { + model.update_event_sequence(conversation_id, max_seq, ctx); + }); + + // Also persist the cursor to the server so driver / cloud restarts + // can resume without local SQLite state. Fire-and-forget: log on + // failure, don't block event delivery. The server persists the + // cursor on `ai_tasks.last_event_sequence`. + let own_run_id = BlocklistAIHistoryModel::as_ref(ctx) + .conversation(&conversation_id) + .and_then(|c| c.run_id()); + if let Some(run_id) = own_run_id { + // TODO: consider debouncing this server write (see + // specs/replay-agent-events-on-restore/TECH.md Risks). + let ai_client = self.ai_client.clone(); + ctx.spawn( + async move { + ai_client + .update_event_sequence_on_server(&run_id, max_seq) + .await + }, + move |_, result, _| { + if let Err(err) = result { + log::warn!( + "Failed to persist event cursor to server for {conversation_id:?}: {err:#}" + ); + } + }, + ); + } + // Track message IDs for server-side mark_delivered calls. let message_ids: Vec = events .iter() diff --git a/app/src/ai/blocklist/orchestration_event_poller_tests.rs b/app/src/ai/blocklist/orchestration_event_poller_tests.rs index 50c651df..06d5a209 100644 --- a/app/src/ai/blocklist/orchestration_event_poller_tests.rs +++ b/app/src/ai/blocklist/orchestration_event_poller_tests.rs @@ -104,3 +104,499 @@ fn convert_lifecycle_events_maps_run_restarted() { Some(api::agent_event::lifecycle_event::Detail::InProgress(..)) )); } + +#[test] +fn ai_conversation_new_restored_preserves_last_event_sequence() { + // Guards against regressions that drop the field when wiring the restore + // path: a conversation restored with `last_event_sequence: Some(N)` + // should expose it via `conversation.last_event_sequence()`. + use crate::ai::agent::conversation::{AIConversation, AIConversationId}; + use crate::persistence::model::AgentConversationData; + + let task = api::Task { + id: "root".to_string(), + messages: vec![api::Message { + id: "m1".to_string(), + task_id: "root".to_string(), + server_message_data: String::new(), + citations: vec![], + message: Some(api::message::Message::AgentOutput( + api::message::AgentOutput { + text: "hi".to_string(), + }, + )), + request_id: String::new(), + timestamp: None, + }], + dependencies: None, + description: String::new(), + summary: String::new(), + server_data: String::new(), + }; + let data = AgentConversationData { + server_conversation_token: None, + conversation_usage_metadata: None, + reverted_action_ids: None, + forked_from_server_conversation_token: None, + artifacts_json: None, + parent_agent_id: None, + agent_name: None, + parent_conversation_id: None, + run_id: None, + autoexecute_override: None, + last_event_sequence: Some(42), + }; + let conversation = + AIConversation::new_restored(AIConversationId::new(), vec![task], Some(data)) + .expect("should restore"); + assert_eq!(conversation.last_event_sequence(), Some(42)); +} + +// ---- Helpers for App-based poller tests ---- + +fn make_ambient_task_with_children( + children: Vec, +) -> crate::ai::ambient_agents::AmbientAgentTask { + let mut task = make_ambient_task_with_event_seq(None); + task.children = children; + task +} + +fn make_ambient_task_with_event_seq( + last_event_sequence: Option, +) -> crate::ai::ambient_agents::AmbientAgentTask { + use chrono::Utc; + crate::ai::ambient_agents::AmbientAgentTask { + task_id: "550e8400-e29b-41d4-a716-446655440000".parse().unwrap(), + parent_run_id: None, + title: "test".to_string(), + state: crate::ai::ambient_agents::AmbientAgentTaskState::Succeeded, + prompt: "prompt".to_string(), + created_at: Utc::now(), + started_at: Some(Utc::now()), + updated_at: Utc::now(), + status_message: None, + source: None, + session_id: None, + session_link: None, + creator: None, + conversation_id: None, + request_usage: None, + agent_config_snapshot: None, + artifacts: vec![], + is_sandbox_running: false, + last_event_sequence, + children: vec![], + } +} + +#[test] +fn finish_restore_fetch_uses_server_cursor_when_sqlite_is_absent() { + use crate::ai::agent::conversation::AIConversation; + use crate::server::server_api::ai::MockAIClient; + use crate::server::server_api::ServerApiProvider; + use std::sync::Arc; + use warpui::App; + + App::test((), |mut app| async move { + let _v2_guard = FeatureFlag::OrchestrationV2.override_enabled(true); + + let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new(vec![], &[])); + + // Restore a conversation with no SQLite cursor (`last_event_sequence: + // None`). After the server fetch completes with `Some(42)` we expect + // the in-memory cursor to be 42 (max(0, 42)). + let conversation = AIConversation::new(false); + let conversation_id = conversation.id(); + let terminal_view_id = warpui::EntityId::new(); + history_model.update(&mut app, |model, ctx| { + model.restore_conversations(terminal_view_id, vec![conversation], ctx); + }); + + let mock = MockAIClient::new(); + let ai_client: Arc = 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) + }); + + // Seed event_cursor as on_restored_conversations would before spawning + // the async fetch. Without this the guard that detects mid-flight + // conversation deletion would fire and return early. + poller.update(&mut app, |me, _| { + me.event_cursor.insert(conversation_id, 0); + }); + + let task_id: crate::ai::ambient_agents::AmbientAgentTaskId = + "550e8400-e29b-41d4-a716-446655440000".parse().unwrap(); + poller.update(&mut app, |me, ctx| { + me.finish_restore_fetch( + conversation_id, + task_id, + /* sqlite_cursor */ 0, + Ok(make_ambient_task_with_event_seq(Some(42))), + ctx, + ); + }); + + poller.read(&app, |me, _| { + assert_eq!(me.event_cursor.get(&conversation_id).copied(), Some(42)); + }); + }); +} + +#[test] +fn restored_inprogress_parent_defers_delivery_until_success() { + use crate::ai::agent::conversation::{AIConversation, AIConversationId, ConversationStatus}; + use crate::server::server_api::ai::MockAIClient; + use crate::server::server_api::ServerApiProvider; + use std::sync::Arc; + use warpui::App; + + App::test((), |mut app| async move { + let _v2_guard = FeatureFlag::OrchestrationV2.override_enabled(true); + + let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new(vec![], &[])); + + let mut conversation = AIConversation::new(false); + // Use a parsable UUID-shaped run_id so the poller can construct + // an `AmbientAgentTaskId` for the (mocked) server fetch. + conversation.set_run_id("550e8400-e29b-41d4-a716-446655440100".to_string()); + let conversation_id: AIConversationId = conversation.id(); + let terminal_view_id = warpui::EntityId::new(); + history_model.update(&mut app, |model, ctx| { + model.restore_conversations(terminal_view_id, vec![conversation], ctx); + // The default status after restore is `InProgress` for live + // conversations, but assert it explicitly to make the test + // self-documenting. + model.update_conversation_status( + terminal_view_id, + conversation_id, + ConversationStatus::InProgress, + ctx, + ); + }); + + let mut mock = MockAIClient::new(); + // The async restore fetch may or may not complete during the test; + // 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 = 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) + }); + + // 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| { + me.on_restored_conversations(vec![conversation_id], ctx); + }); + poller.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). + history_model.update(&mut app, |model, ctx| { + model.update_conversation_status( + terminal_view_id, + conversation_id, + ConversationStatus::Success, + ctx, + ); + }); + poller.read(&app, |me, _| { + assert!( + me.poll_in_flight.contains(&conversation_id), + "Success transition with watched run_ids should start delivery" + ); + }); + }); +} + +#[test] +fn handle_poll_result_persists_max_seq_to_history_model() { + use crate::ai::agent::conversation::{AIConversation, AIConversationId}; + use crate::persistence::ModelEvent; + use crate::server::server_api::ai::MockAIClient; + use crate::server::server_api::ServerApiProvider; + use crate::test_util::settings::initialize_settings_for_tests; + use crate::{GlobalResourceHandles, GlobalResourceHandlesProvider}; + use std::sync::Arc; + use warpui::App; + + App::test((), |mut app| async move { + let _v2_guard = FeatureFlag::OrchestrationV2.override_enabled(true); + + // `update_event_sequence` calls `write_updated_conversation_state`, + // which reads `GeneralSettings`, `AppExecutionMode`, and the global + // resource sender. Wire all of these up so the SQLite write can run. + initialize_settings_for_tests(&mut app); + let (sender, receiver) = std::sync::mpsc::sync_channel::(4); + let mut global_resource_handles = GlobalResourceHandles::mock(&mut app); + global_resource_handles.model_event_sender = Some(sender); + app.add_singleton_model(|_| GlobalResourceHandlesProvider::new(global_resource_handles)); + + let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new(vec![], &[])); + + let mut conversation = AIConversation::new(false); + conversation.set_run_id("550e8400-e29b-41d4-a716-446655440200".to_string()); + let conversation_id: AIConversationId = conversation.id(); + let terminal_view_id = warpui::EntityId::new(); + history_model.update(&mut app, |model, ctx| { + model.restore_conversations(terminal_view_id, vec![conversation], ctx); + }); + + let mut mock = MockAIClient::new(); + // The fire-and-forget server PATCH should be issued; permissive Ok. + mock.expect_update_event_sequence_on_server() + .returning(|_, _| Ok(())); + let ai_client: Arc = 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) + }); + + // Build a poll batch with max sequence = 42. Use an unrecognized + // event_type so `convert_lifecycle_events` returns empty and the + // function early-exits before touching `OrchestrationEventService` + // (which we did not register in this test App). + let events = vec![ + AgentRunEvent { + event_type: "unrecognized_event_type".to_string(), + run_id: "some-other-run".to_string(), + ref_id: None, + execution_id: None, + occurred_at: "2026-01-01T00:00:00Z".to_string(), + sequence: 17, + }, + AgentRunEvent { + event_type: "unrecognized_event_type".to_string(), + run_id: "some-other-run".to_string(), + ref_id: None, + execution_id: None, + occurred_at: "2026-01-01T00:00:00Z".to_string(), + sequence: 42, + }, + ]; + + poller.update(&mut app, |me, ctx| { + me.handle_poll_result( + conversation_id, + /* self_run_id */ "some-other-run", + /* previous_cursor */ 0, + events, + /* messages */ vec![], + ctx, + ); + }); + + history_model.read(&app, |model, _| { + let last_seq = model + .conversation(&conversation_id) + .and_then(|c| c.last_event_sequence()); + assert_eq!( + last_seq, + Some(42), + "BlocklistAIHistoryModel.update_event_sequence must be called with max_seq" + ); + }); + + // Drain at least one persistence event to confirm the SQLite write + // path was triggered (sanity check for the side effect, not the + // primary assertion). + let _ = receiver.recv_timeout(std::time::Duration::from_secs(1)); + }); +} + +#[test] +fn finish_restore_fetch_no_ops_when_conversation_deleted_mid_flight() { + // If the conversation is removed while the async fetch is in-flight, the + // RemoveConversation handler clears event_cursor. finish_restore_fetch + // uses the missing cursor as a sentinel and must not re-populate + // watched_run_ids or event_cursor for the deleted conversation. + use crate::ai::agent::conversation::AIConversation; + use crate::server::server_api::ai::MockAIClient; + use crate::server::server_api::ServerApiProvider; + use std::sync::Arc; + use warpui::App; + + App::test((), |mut app| async move { + let _v2_guard = FeatureFlag::OrchestrationV2.override_enabled(true); + + let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new(vec![], &[])); + + let mut conversation = AIConversation::new(false); + conversation.set_run_id("550e8400-e29b-41d4-a716-446655440300".to_string()); + let conversation_id = conversation.id(); + let terminal_view_id = warpui::EntityId::new(); + history_model.update(&mut app, |model, ctx| { + model.restore_conversations(terminal_view_id, vec![conversation], ctx); + }); + + let mock = MockAIClient::new(); + let ai_client: Arc = 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) + }); + + // Seed cursor as on_restored_conversations would. + poller.update(&mut app, |me, _| { + me.event_cursor.insert(conversation_id, 0); + }); + + // Simulate the RemoveConversation handler firing while the fetch is + // in-flight: it clears event_cursor (and all other state). + poller.update(&mut app, |me, _| { + me.watched_run_ids.remove(&conversation_id); + me.event_cursor.remove(&conversation_id); + }); + + // The in-flight fetch now completes — with children. + let task_id: crate::ai::ambient_agents::AmbientAgentTaskId = + "550e8400-e29b-41d4-a716-446655440000".parse().unwrap(); + poller.update(&mut app, |me, ctx| { + me.finish_restore_fetch( + conversation_id, + task_id, + /* sqlite_cursor */ 0, + Ok(make_ambient_task_with_children(vec![ + "child-run-1".to_string() + ])), + ctx, + ); + }); + + poller.read(&app, |me, _| { + assert!( + !me.watched_run_ids.contains_key(&conversation_id), + "watched_run_ids must not be repopulated for a deleted conversation" + ); + assert!( + !me.event_cursor.contains_key(&conversation_id), + "event_cursor must not be repopulated for a deleted conversation" + ); + }); + }); +} + +#[test] +fn finish_restore_fetch_reconnects_sse_when_children_added_to_open_connection() { + // When a status transition races with the restore fetch and opens SSE + // before children are known, finish_restore_fetch must reconnect SSE + // with the updated run_id set rather than leaving children unwatched. + use crate::ai::agent::conversation::{AIConversation, ConversationStatus}; + use crate::server::server_api::ai::MockAIClient; + use crate::server::server_api::ServerApiProvider; + use std::sync::Arc; + use warpui::App; + + App::test((), |mut app| async move { + let _v2_guard = FeatureFlag::OrchestrationV2.override_enabled(true); + + let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new(vec![], &[])); + + let own_run_id = "550e8400-e29b-41d4-a716-446655440400"; + let mut conversation = AIConversation::new(false); + conversation.set_run_id(own_run_id.to_string()); + let conversation_id = conversation.id(); + let terminal_view_id = warpui::EntityId::new(); + history_model.update(&mut app, |model, ctx| { + model.restore_conversations(terminal_view_id, vec![conversation], ctx); + model.update_conversation_status( + terminal_view_id, + conversation_id, + ConversationStatus::InProgress, + ctx, + ); + }); + + let mock = MockAIClient::new(); + let ai_client: Arc = 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) + }); + + // Seed the state on_restored_conversations would have set up, then + // inject a fake open SSE connection (generation 0) simulating the + // race: a status transition fired before the restore fetch completed. + let (_, rx) = futures::channel::mpsc::unbounded::(); + poller.update(&mut app, |me, _| { + me.event_cursor.insert(conversation_id, 0); + me.watched_run_ids + .entry(conversation_id) + .or_default() + .insert(own_run_id.to_string()); + me.sse_connections.insert( + conversation_id, + SseConnectionState { + event_receiver: rx, + generation: 0, + }, + ); + me.next_sse_generation = 1; + }); + + // The restore fetch returns with a child run_id. + let task_id: crate::ai::ambient_agents::AmbientAgentTaskId = + "550e8400-e29b-41d4-a716-446655440000".parse().unwrap(); + poller.update(&mut app, |me, ctx| { + me.finish_restore_fetch( + conversation_id, + task_id, + /* sqlite_cursor */ 0, + Ok(make_ambient_task_with_children(vec![ + "child-run-1".to_string() + ])), + ctx, + ); + }); + + poller.read(&app, |me, _| { + assert!( + me.watched_run_ids + .get(&conversation_id) + .is_some_and(|w| w.contains("child-run-1")), + "child run_id must be in watched set" + ); + // The old generation-0 connection must have been replaced by a + // new one with a higher generation, proving SSE was reconnected. + let gen = me + .sse_connections + .get(&conversation_id) + .map(|s| s.generation); + assert!( + gen.is_some_and(|g| g > 0), + "SSE must be reconnected (new generation) after children are discovered; got gen={gen:?}" + ); + }); + }); +} diff --git a/app/src/ai/blocklist/orchestration_events.rs b/app/src/ai/blocklist/orchestration_events.rs index 72f66be8..38190619 100644 --- a/app/src/ai/blocklist/orchestration_events.rs +++ b/app/src/ai/blocklist/orchestration_events.rs @@ -357,6 +357,42 @@ impl OrchestrationEventService { } => { for conversation_id in conversation_ids { self.sync_conversation_status(*conversation_id, ctx); + // Under V1 local lifecycle dispatch, child status + // transitions are forwarded to the parent via + // `lifecycle_subscription_routes`. That map is not + // persisted, so re-register subscriptions for each + // restored child whose parent is loaded locally so that + // child status transitions continue to propagate after + // a restart. V2 uses the server event log and does not + // need this. + if !FeatureFlag::OrchestrationV2.is_enabled() { + let parent_agent_id = { + let history_model = BlocklistAIHistoryModel::as_ref(ctx); + let Some(child_conv) = history_model.conversation(conversation_id) + else { + continue; + }; + if !child_conv.is_child_agent_conversation() { + continue; + } + child_conv + .parent_conversation_id() + .and_then(|pid| history_model.conversation(&pid)) + .and_then(|p| p.server_conversation_token()) + .map(|t| t.as_str().to_string()) + }; + if let Some(parent_agent_id) = parent_agent_id { + // `None` event-type filter = subscribe to all + // lifecycle types. The original filter (if any) + // is not persisted; subscribing broader than the + // original is acceptable per the tech spec. + self.register_lifecycle_subscription( + *conversation_id, + parent_agent_id, + None, + ); + } + } } } BlocklistAIHistoryEvent::RemoveConversation { diff --git a/app/src/ai/blocklist/orchestration_events_tests.rs b/app/src/ai/blocklist/orchestration_events_tests.rs index 8ff21ddd..c3ba0a52 100644 --- a/app/src/ai/blocklist/orchestration_events_tests.rs +++ b/app/src/ai/blocklist/orchestration_events_tests.rs @@ -357,3 +357,67 @@ fn test_lifecycle_event_type_from_proto_includes_cancelled_and_blocked() { api::LifecycleEventType::Blocked ); } + +#[test] +fn restored_v1_child_conversation_re_registers_lifecycle_subscription() { + use crate::ai::agent::conversation::AIConversation; + use warp_core::features::FeatureFlag; + use warpui::{App, EntityId}; + + App::test((), |mut app| async move { + // V1 path is gated on `!OrchestrationV2`. + let _v1_guard = FeatureFlag::OrchestrationV2.override_enabled(false); + + let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new(vec![], &[])); + + // Build a parent conversation with a server token; under V1 the + // parent's `server_conversation_token` is the agent identifier the + // child subscribes to. + let parent_token = "parent-token-v1"; + let mut parent_conversation = AIConversation::new(false); + parent_conversation.set_server_conversation_token(parent_token.to_string()); + let parent_conversation_id = parent_conversation.id(); + + // Build a child conversation pointing at the parent. + let mut child_conversation = AIConversation::new(false); + child_conversation.set_parent_conversation_id(parent_conversation_id); + let child_conversation_id = child_conversation.id(); + + let terminal_view_id = EntityId::new(); + history_model.update(&mut app, |model, ctx| { + model.restore_conversations( + terminal_view_id, + vec![parent_conversation, child_conversation], + ctx, + ); + }); + + // Drive the OrchestrationEventService through its standard + // `handle_history_event` entry point; `restore_conversations` already + // emitted `RestoredConversations`, so we replay it explicitly through + // the service to keep this test independent of subscription wiring. + let service = app.add_singleton_model(|_| OrchestrationEventService::default()); + service.update(&mut app, |svc, ctx| { + svc.handle_history_event( + &BlocklistAIHistoryEvent::RestoredConversations { + terminal_view_id, + conversation_ids: vec![parent_conversation_id, child_conversation_id], + }, + ctx, + ); + }); + + service.read(&app, |svc, _| { + let routes = svc + .lifecycle_subscription_routes + .get(&child_conversation_id) + .expect("expected V1 lifecycle route to be registered for the child"); + assert_eq!(routes.len(), 1, "expected exactly one route"); + assert_eq!(routes[0].target_agent_id, parent_token); + assert!( + routes[0].subscribed_event_types.is_none(), + "restore re-registers with `None` (subscribe to all event types)" + ); + }); + }); +} diff --git a/app/src/ai/conversation_details_panel_tests.rs b/app/src/ai/conversation_details_panel_tests.rs index cfc494f7..001898a4 100644 --- a/app/src/ai/conversation_details_panel_tests.rs +++ b/app/src/ai/conversation_details_panel_tests.rs @@ -39,6 +39,8 @@ fn create_test_task(task_id: &str) -> AmbientAgentTask { agent_config_snapshot: None, artifacts: vec![], is_sandbox_running: false, + last_event_sequence: None, + children: vec![], } } @@ -130,6 +132,7 @@ fn test_from_task_includes_linked_directory_when_run_id_matches() { parent_conversation_id: None, run_id: Some(task_id.to_string()), autoexecute_override: None, + last_event_sequence: None, }, ); @@ -244,6 +247,7 @@ fn test_from_task_includes_linked_directory_when_server_token_matches() { parent_conversation_id: None, run_id: None, autoexecute_override: None, + last_event_sequence: None, }, ); diff --git a/app/src/server/server_api.rs b/app/src/server/server_api.rs index 452779c2..7405abb7 100644 --- a/app/src/server/server_api.rs +++ b/app/src/server/server_api.rs @@ -791,6 +791,39 @@ impl ServerApi { Ok(()) } + /// Sends a PATCH request to a public API endpoint that returns no response body. + async fn patch_public_api_unit(&self, path: &str, body: &B) -> Result<()> + where + B: Serialize, + { + let auth_token = self + .get_or_refresh_access_token() + .await + .context("Failed to get access token for API request")?; + + let url = format!("{}/api/v1/{}", ChannelState::server_root_url(), path); + + let mut request = self.client.patch(&url).json(body); + if let Some(token) = auth_token.as_bearer_token() { + request = request.bearer_auth(token); + } + + for (name, value) in self.ambient_agent_headers().await? { + request = request.header(name, value); + } + + let response = request + .send() + .await + .with_context(|| format!("Failed to send API request to {url}"))?; + + if response.status().is_success() { + Ok(()) + } else { + Err(Self::error_from_response(response).await) + } + } + /// Sends an authenticated empty POST request to /client/login, which signals to the server /// that the user is logged in. pub async fn notify_login(&self) { diff --git a/app/src/server/server_api/ai.rs b/app/src/server/server_api/ai.rs index 81448b70..6e859d36 100644 --- a/app/src/server/server_api/ai.rs +++ b/app/src/server/server_api/ai.rs @@ -923,6 +923,16 @@ pub trait AIClient: 'static + Send + Sync { limit: i32, ) -> anyhow::Result, 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 + /// parent has already acted on. + async fn update_event_sequence_on_server( + &self, + run_id: &str, + sequence: i64, + ) -> anyhow::Result<(), anyhow::Error>; + async fn report_agent_event( &self, run_id: &str, @@ -1885,6 +1895,22 @@ impl AIClient for ServerApi { Ok(events) } + async fn update_event_sequence_on_server( + &self, + run_id: &str, + sequence: i64, + ) -> anyhow::Result<(), anyhow::Error> { + #[derive(serde::Serialize)] + struct UpdateBody { + sequence: i64, + } + self.patch_public_api_unit( + &format!("agent/runs/{run_id}/event-sequence"), + &UpdateBody { sequence }, + ) + .await + } + async fn report_agent_event( &self, run_id: &str, diff --git a/app/src/terminal/view/load_ai_conversation.rs b/app/src/terminal/view/load_ai_conversation.rs index a0a143fa..b83b80f4 100644 --- a/app/src/terminal/view/load_ai_conversation.rs +++ b/app/src/terminal/view/load_ai_conversation.rs @@ -988,6 +988,7 @@ impl TerminalView { parent_conversation_id: None, run_id: None, autoexecute_override: None, + last_event_sequence: None, }; match AIConversation::new_restored(conversation_id, tasks, Some(conversation_data)) { diff --git a/app/src/terminal/view/shared_session/conversation_ended_tombstone_view_tests.rs b/app/src/terminal/view/shared_session/conversation_ended_tombstone_view_tests.rs index 9450671a..7a50ade1 100644 --- a/app/src/terminal/view/shared_session/conversation_ended_tombstone_view_tests.rs +++ b/app/src/terminal/view/shared_session/conversation_ended_tombstone_view_tests.rs @@ -44,6 +44,8 @@ fn task_with_run_time_and_credits() -> AmbientAgentTask { agent_config_snapshot: None, artifacts: vec![], is_sandbox_running: false, + last_event_sequence: None, + children: vec![], } } diff --git a/crates/http_client/src/lib.rs b/crates/http_client/src/lib.rs index a2c5a019..f10875d0 100644 --- a/crates/http_client/src/lib.rs +++ b/crates/http_client/src/lib.rs @@ -203,6 +203,13 @@ impl Client { ) } + pub fn patch(&self, url: U) -> RequestBuilder<'_> { + self.builder( + self.wrapped.patch(url.clone()), + Self::include_warp_http_headers(url), + ) + } + pub fn put(&self, url: U) -> RequestBuilder<'_> { self.builder( self.wrapped.put(url.clone()), diff --git a/crates/persistence/Cargo.toml b/crates/persistence/Cargo.toml index e3cd417e..e283522e 100644 --- a/crates/persistence/Cargo.toml +++ b/crates/persistence/Cargo.toml @@ -21,3 +21,6 @@ diesel = { workspace = true, features = [ diesel_migrations = "2.2.0" serde.workspace = true warp_multi_agent_api.workspace = true + +[dev-dependencies] +serde_json.workspace = true diff --git a/crates/persistence/src/model.rs b/crates/persistence/src/model.rs index eabfadb1..f99220c6 100644 --- a/crates/persistence/src/model.rs +++ b/crates/persistence/src/model.rs @@ -1036,6 +1036,11 @@ pub struct AgentConversationData { pub run_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub autoexecute_override: Option, + /// The last event sequence number from the v2 orchestration event log + /// that this conversation has observed. Used on restore to resume event + /// delivery without re-delivering already-processed events. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_event_sequence: Option, } #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] @@ -1327,6 +1332,63 @@ pub struct NewMCPServerInstallation { pub last_modified_at: NaiveDateTime, } +#[cfg(test)] +mod tests { + use super::AgentConversationData; + + #[test] + fn agent_conversation_data_roundtrips_last_event_sequence() { + let data = AgentConversationData { + server_conversation_token: None, + conversation_usage_metadata: None, + reverted_action_ids: None, + forked_from_server_conversation_token: None, + artifacts_json: None, + parent_agent_id: None, + agent_name: None, + parent_conversation_id: None, + run_id: None, + autoexecute_override: None, + last_event_sequence: Some(42), + }; + let json = serde_json::to_string(&data).expect("serialize"); + let roundtripped: AgentConversationData = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(roundtripped.last_event_sequence, Some(42)); + } + + #[test] + fn agent_conversation_data_deserializes_legacy_payload_without_last_event_sequence() { + // Legacy rows persisted before this feature landed omit the field + // entirely. `#[serde(default)]` must accept them as `None`. + let legacy_json = r#"{"server_conversation_token":null}"#; + let data: AgentConversationData = + serde_json::from_str(legacy_json).expect("legacy rows must deserialize"); + assert_eq!(data.last_event_sequence, None); + } + + #[test] + fn agent_conversation_data_skips_serializing_none_last_event_sequence() { + let data = AgentConversationData { + server_conversation_token: None, + conversation_usage_metadata: None, + reverted_action_ids: None, + forked_from_server_conversation_token: None, + artifacts_json: None, + parent_agent_id: None, + agent_name: None, + parent_conversation_id: None, + run_id: None, + autoexecute_override: None, + last_event_sequence: None, + }; + let json = serde_json::to_string(&data).expect("serialize"); + assert!( + !json.contains("last_event_sequence"), + "None should be skipped in serialized output: {json}" + ); + } +} + #[derive(Insertable)] #[diesel(table_name = panels)] pub struct NewPanel { diff --git a/specs/replay-agent-events-on-restore/PRODUCT.md b/specs/replay-agent-events-on-restore/PRODUCT.md new file mode 100644 index 00000000..5d1f27b0 --- /dev/null +++ b/specs/replay-agent-events-on-restore/PRODUCT.md @@ -0,0 +1,37 @@ +# Orchestration Conversation Restore + +## Summary +When Warp restarts — or when a parent conversation is re-attached via `warp agent run --conversation` — an orchestration session involving parent and child agent runs must resume as if the interruption had not occurred. Parents must receive pending events from children, event delivery must not duplicate messages already processed, and parent-child identity links must be re-established automatically. + +## Problem +Orchestration conversations spanning a parent and one or more child agent runs are not durable across Warp restarts today. After a restart: +- The parent's event listeners are not restored, so no new events (lifecycle signals or messages) from children arrive. +- The event cursor is not persisted, so if event delivery is re-established, all events since the beginning of the conversation replay from sequence 0, causing duplicate delivery. +- In the driver case (`warp agent run --conversation`), the parent cannot discover children that ran on remote workers with no local DB record. + +## Behavior + +### Startup restoration (Warp GUI restart) +1. After Warp restarts and a previously active parent conversation is restored, the parent continues to receive lifecycle events (in-progress, succeeded, failed, blocked, cancelled, errored) and inbox messages from any children that were running at the time of the restart. +2. Any lifecycle events from children — including terminal events such as `succeeded` or `failed` — that arrived while Warp was not running are delivered to the parent once event delivery resumes. This is the primary scenario the feature addresses. +3. A parent that was in `Success` status at restart time with watched children resumes event delivery without requiring the user to take any action or for the conversation to transition back through `InProgress`. +4. A parent that was in `InProgress` at restart (i.e., Warp quit while the parent was actively running) resumes event delivery from children once the parent's current exchange completes and the parent becomes idle again. +5. A parent that has never spawned a child agent does not start any event polling after restart. +6. Child conversations whose records have a `parent_conversation_id` pointing at the restored parent are re-linked to that parent, so their status transitions continue to propagate correctly. +7. Event delivery resumes from the last event the parent had confirmed receiving; no event that the parent had already acted on before the restart is delivered again. +8. Under normal operation, each event is delivered to a parent agent at most once. A crash between the parent receiving a batch of events and acknowledging them may result in that batch being retransmitted once on restart; this is the worst-case behavior and does not cascade. +9. Transient network failures during event resumption after restart are retried automatically. The user does not need to restart Warp again to recover from a failed resume attempt. + +### Driver restoration (`warp agent run --conversation`) +10. When a parent conversation is loaded from the server via `--conversation`, all children that were spawned by that conversation — including children that ran on remote workers with no record in the local database — are rediscovered and their events are delivered. +10a. If rediscovery of a specific child fails (e.g. a server error), the parent is still restored and events from any other children continue to be delivered. Partial child-rediscovery failure does not prevent the overall session from resuming. +11. After a driver restoration, event delivery from children behaves identically to invariants 1–9 above. + +### V1 orchestration (local lifecycle dispatch) +12. When the client is running in legacy local lifecycle-dispatch mode, child status transitions (InProgress → Success, InProgress → Error, etc.) are still forwarded to the parent after restart without any action by the user. + +### Invariants that must not regress +13. A conversation that was not part of an orchestration session (no parent, no children) is unaffected by this change — its restoration behavior is identical to before. +14. A conversation that is a shared-session viewer does not begin receiving events from a watched-event stream after restore. Shared-session viewers continue to receive updates through the session-sharing mechanism as before. +15. If a child's parent conversation no longer exists on the local machine (deleted or on another device), the child is restored as a standalone conversation with no parent; no error is surfaced to the user. +16. Removing or deleting a conversation tears down any associated event delivery state immediately — no further events are delivered after a conversation is removed. diff --git a/specs/replay-agent-events-on-restore/TECH.md b/specs/replay-agent-events-on-restore/TECH.md new file mode 100644 index 00000000..7607f9f8 --- /dev/null +++ b/specs/replay-agent-events-on-restore/TECH.md @@ -0,0 +1,195 @@ +# Orchestration Conversation Restore — Tech Spec + +## Context +See `PRODUCT.md` for user-visible behavior. + +### Current restore path +On startup, `AgentConversation` records are read from SQLite and converted to `AIConversation` via `new_restored` (`app/src/ai/agent/conversation.rs (281-451)`). These are placed in the `RestoredAgentConversations` singleton. When a terminal view opens, `restore_conversations_on_view_creation` calls `BlocklistAIHistoryModel::restore_conversations`, which loads conversations into memory and emits `BlocklistAIHistoryEvent::RestoredConversations`. + +For the driver case, `AgentDriver::new` (`app/src/ai/agent_sdk/driver.rs (474-608)`) accepts a `ConversationRestorationInNewPaneType::Historical` conversation pre-loaded from the server. It flows through `restore_conversation_after_view_creation` → `restore_conversations_from_block_params` → `BlocklistAIHistoryModel::restore_conversations`, emitting the same `RestoredConversations` event. + +### What is correctly restored today +- `parent_agent_id`, `parent_conversation_id`, `run_id` — persisted in `AgentConversationData` JSON, restored via `new_restored`. +- `children_by_parent` index — rebuilt at startup from all local DB conversations in `initialize_historical_conversations` (`conversation_loader.rs (430-588)`). +- `agent_id_to_conversation_id` routing index — populated in `restore_conversations` for each loaded conversation. + +### What is NOT restored (the three gaps) +1. **`OrchestrationEventPoller.watched_run_ids`** (`orchestration_event_poller.rs:65`) — the set of child run_ids to poll per parent conversation. Populated only when `start_agent` runs at runtime. `handle_history_event` explicitly ignores `RestoredConversations` (line 183). After restart, no SSE/poll loop opens and no events arrive. +2. **`OrchestrationEventPoller.event_cursor`** (`orchestration_event_poller.rs:66`) — per-conversation i64 sequence number. Always initializes to 0; `AgentConversationData` has no cursor field. All historical events are re-fetched on the first poll. +3. **`OrchestrationEventService.lifecycle_subscription_routes`** (`orchestration_events.rs:101`) — maps child conversation_id → parent agent_id for V1 local lifecycle dispatch. Not restored; V1 parents miss child status transitions. + +## Proposed Changes + +### 1. Persist and restore the event cursor sequence number (covers invariants 7, 8, 9, 10) + +The in-progress branch `katarina/quality-503-driver-owned-parent-bridge` in `wc-pine` establishes the right primitives here. It introduces a shared `AgentEventConsumer` trait with a `persist_cursor(sequence: i64)` callback (in `app/src/ai/agent_events/driver.rs`) and refactors both the Oz SSE path and the non-Oz Claude Code parent bridge to use a common `run_agent_event_driver`. The i64 sequence number is the existing API cursor — no new parameter type is needed. + +The cursor is currently persisted locally only: +- **Non-Oz (parent bridge)**: `ParentBridgeEventConsumer::persist_cursor` writes to a local file (`~/.claude-code/oz-parent-bridge/{session_id}/last-sequence`) and initializes from it via `read_parent_bridge_last_sequence` when the bridge starts. +- **Oz (`SseForwardingConsumer`)**: uses the default no-op `persist_cursor` — the cursor is not persisted today. + +For the driver/cloud-restart case, local-only persistence is insufficient: the session changes between runs, so the previous session's local file is not found, and a server-loaded conversation has no local SQLite state. The fix is to persist the cursor to **both** local storage and server-side conversation metadata, then use whichever source is available on restore. + +**Local persistence for Oz — call site** +The in-memory cursor is advanced in one place in this repo: `handle_poll_result` at `orchestration_event_poller.rs:413-418` (`self.event_cursor.insert(conversation_id, max_seq)`). The SSE path drains through this same function, so a single write here covers both modes. Add `last_event_sequence: Option` to `AgentConversationData` and `AIConversation`; immediately after the `event_cursor.insert` call, invoke a new `BlocklistAIHistoryModel::update_event_sequence(conversation_id, max_seq)` helper that calls `write_updated_conversation_state`. This is per-batch (up to `EVENT_POLL_BATCH_LIMIT = 100` events per batch) — acceptable granularity. + +Note: The WIP branch `katarina/quality-503-driver-owned-parent-bridge` in `wc-pine` refactors this path to introduce an `SseForwardingConsumer` type with a `persist_cursor` callback. If that branch merges before this one, the call site moves to that callback instead. If this feature lands first, the inline write at `handle_poll_result` is sufficient and the WIP migration can adapt it. + +**Server-side persistence (both Oz and non-Oz)** +Add `last_event_sequence: Option` to `Task` in warp-server (on `ai_tasks`). This field is part of this feature's scope; the companion warp-server change adds it to `GET /agent/runs/:run_id` and a new `PATCH /agent/runs/:run_id/event-sequence` endpoint. When `update_event_sequence` fires, call this endpoint fire-and-forget (log on failure). Losing a cursor update is recoverable — next best cursor is used on restore. + +**Restore initialization** +In the `RestoredConversations` handler in the poller, initialize the cursor for each conversation by taking `max` of all available sources: +1. `conversation.last_event_sequence()` — from `AgentConversationData` in local SQLite (present for local restores) +2. The result of `GET /agent/runs/{run_id}` — fetched asynchronously as part of the same async restore query that fetches child run_ids (see change 2). The run response includes `last_event_sequence` directly on the `Task`. + +The async server fetch happens alongside the child-run discovery query. When both return, the cursor is set to `max(SQLite value, server value)` and event delivery starts. + +### 2. Restore watched_run_ids and event delivery on `RestoredConversations` (covers invariants 1–11) + +**`app/src/ai/blocklist/orchestration_event_poller.rs`** + +Add a helper to scan task messages for child run_ids (driver/remote-worker case). Gate on `OrchestrationV2` because in V2 the `agent_id` in `StartAgentV2` results is the child's `run_id`; in V1 it is a different identifier and the scan would produce wrong values: +```rust +fn child_run_ids_from_task_messages(conversation: &AIConversation) -> Vec { + if !FeatureFlag::OrchestrationV2.is_enabled() { + return vec![]; + } + let mut out = Vec::new(); + for task in conversation.all_tasks() { + let Some(api_task) = task.source() else { continue }; + for msg in &api_task.messages { + let Some(api::message::Message::ToolCallResult(tcr)) = msg.message.as_ref() else { continue }; + let Some(api::tool_call_result::Type::StartAgentV2(r)) = tcr.r#type.as_ref() else { continue }; + let Some(api::start_agent_v2_result::Result::Success(s)) = r.result.as_ref() else { continue }; + if !s.agent_id.is_empty() { + out.push(s.agent_id.clone()); + } + } + } + out +} +``` + +Handle `RestoredConversations` in `handle_history_event` (replacing the current no-op at line 183): +- For each `conv_id` in `conversation_ids`, read the conversation from the history model. +- Skip shared-session viewers (`conversation.is_viewing_shared_session()`). (invariant 14) +- Cursor (initial): set `self.event_cursor[conv_id]` to `conversation.last_event_sequence().unwrap_or(0)` from local SQLite. The server value will be merged once the async run fetch completes (see below). +- Own run_id: if `conversation.run_id()` is `Some`, insert into `self.watched_run_ids[conv_id]`. +- Child run_ids and server cursor: spawn two concurrent async calls — `GET /agent/runs/{run_id}/children` and `GET /agent/runs/{run_id}` — and wait for **both** before starting event delivery. Starting delivery before the run response arrives could cause the cursor to be initialized from SQLite only, and a later server cursor merge might not advance it in time to prevent duplicate events. + - **Both succeed**: merge `event_cursor[conv_id] = max(SQLite, run.last_event_sequence)`; insert child run_ids; start event delivery. + - **Run fetch fails, children succeed**: keep SQLite cursor; insert child run_ids; start event delivery. + - **Children fetch fails (with or without run fetch)**: fall back to `child_conversation_ids_of(conv_id)` and `child_run_ids_from_task_messages(conversation)`; start event delivery. +- Start event delivery for any `conv_id` where `watched_run_ids` is non-empty and status is `Success`. If `InProgress`, defer to `on_conversation_status_updated` as usual. (invariants 3, 4) + +The `child_run_ids_from_task_messages` helper is still needed as the failure fallback path. The restored cursor value flows naturally into the existing `poll_and_inject` and `start_sse_connection` paths, which both read `self.event_cursor.get(&conversation_id).copied().unwrap_or(0)`. No plumbing change is needed in those methods. + +**Delete/remove handler update.** `last_event_sequence` lives in `AgentConversationData`, whose SQLite row is deleted alongside the conversation row. No change to the existing `RemoveConversation`/`DeletedConversation` arm in the poller is needed beyond the existing removals of `watched_run_ids`, `event_cursor`, `poll_backoff_index`, etc. + +### 3. Restore V1 lifecycle subscriptions (covers invariant 12) + +**`app/src/ai/blocklist/orchestration_events.rs`** + +Extend the existing `RestoredConversations` handler (currently lines 355-361) to also re-register subscriptions. For each restored conversation that `is_child_agent_conversation()`: +```rust +if !FeatureFlag::OrchestrationV2.is_enabled() { + if let Some(parent_agent_id) = history_model + .conversation(&child_conv.parent_conversation_id()?) + .and_then(|p| p.server_conversation_token()) + .map(|t| t.as_str().to_string()) + { + self.register_lifecycle_subscription(conv_id, parent_agent_id, None); + } +} +``` +This mirrors the runtime registration in `terminal_pane.rs (1158-1165)`. + +## End-to-end flow + +There are two separate entry points for event delivery; both must work after this change: + +``` +[Entry point 1 — restore-time, new] +Warp restarts + → BlocklistAIHistoryModel::restore_conversations() + → conversations inserted into conversations_by_id + → RestoredConversations { conversation_ids } emitted + → OrchestrationEventService (subscription independent, no ordering guarantee) + sync conversation statuses + re-register V1 lifecycle subscriptions for restored children [new] + → OrchestrationEventPoller [new handler] + for each conv_id: + event_cursor[conv_id] ← SQLite last_event_sequence (if present) + watched_run_ids[conv_id] += own run_id + spawn async GET /agent/runs/{run_id} + GET /agent/runs/{run_id}/children + → run response: event_cursor[conv_id] = max(SQLite, server last_event_sequence) + → children response: watched_run_ids[conv_id] += child run_ids + → on failure: fallback to DB index ∪ task message scan + → if Status=Success and watched_run_ids non-empty: + start_event_delivery() → poll/SSE with cursor = event_cursor[conv_id] + → persist_cursor after each event → updates SQLite + server metadata + for InProgress convs: delivery deferred until next Success transition (entry point 2) + +[Entry point 2 — status transition, pre-existing] +Parent conversation transitions to Success (InProgress → Success) + → BlocklistAIHistoryEvent::UpdatedConversationStatus emitted + → OrchestrationEventPoller::on_conversation_status_updated + if watched_run_ids contains entries: start_event_delivery() + (cursor already initialized from restore; persist_cursor keeps it current) +``` + +## Risks and Mitigations + +**Schema additions are backward-compatible**: Adding `last_event_sequence` to `AgentConversationData` (with `#[serde(default)]`) and to `ai_tasks` (with no NOT NULL constraint) is non-breaking. Older clients or missing fields fall back to cursor=0 and re-deliver events — the same behavior as today. + +**V1 lifecycle subscription filter is lost on restart**: At runtime, `register_lifecycle_subscription` is called with the original `request.lifecycle_subscription` (which may be a filtered subset of event types). After restart the filter is not persisted, so re-registration uses `None` (subscribe to all types). The parent will receive lifecycle types it was not originally subscribed to. This is acceptable given V1 is legacy and the behavior is strictly wider coverage, not narrower. + +**`persist_cursor` frequency**: The callback fires after every event. Server writes on every event could be noisy for active orchestrations. The fire-and-forget nature (log-on-failure, no blocking) mitigates latency impact. If server write frequency becomes a concern, the implementation can debounce (e.g., only write to the server if the cursor advanced by more than N or more than T seconds have elapsed since the last server write). + +**Mid-response crash**: `persist_cursor` is called after each event is processed, before the agent's response (echo) is received. If the process crashes after a `persist_cursor` write but before the agent processes the event, the cursor would reflect an event the agent hasn't acted on yet. On restart, those events would not be re-delivered. This is acceptable: the agent already received the event as input in a previous request, and the server-side task messages record that the input was sent. + +**Child run_ids from task messages may include finished children**: A `StartAgentV2 { Success { agent_id } }` result means the child was launched, but it may have long since finished. The poller will open a connection for that run_id and receive zero events (all before the restored cursor), then idle. This is harmless. + +**Startup cost of task message scan**: `child_run_ids_from_task_messages` is an in-memory scan over data already decoded during restoration. The matched message type (`StartAgentV2` results) is rare. O(total messages), near-zero constant, no I/O. + +## Testing and Validation + +Reference `PRODUCT.md` for invariant numbers. + +- **Invariant 1, 2** (parent receives child events after restart — polling path): Unit test in `orchestration_event_poller_tests.rs` — emit `RestoredConversations` with `OrchestrationEventPush` disabled; verify `watched_run_ids` is populated and `poll_and_inject` uses the restored `event_cursor` value (not 0) as its `cursor` argument on the first call. +- **Invariant 1, 2** (parent receives child events after restart — SSE path): Same test with `OrchestrationEventPush` enabled; verify `start_sse_connection` uses the restored `event_cursor` value as `since_sequence` on the first open and continues using the in-memory cursor (updated via `persist_cursor`) on reconnects. +- **Invariant 2** (terminal events during downtime): Unit test — emit `RestoredConversations` for a `Success` parent with watched run_ids; verify that delivery starts immediately without any `InProgress` → `Success` transition. +- **Invariant 4** (delivery deferred for InProgress parent): Unit test — emit `RestoredConversations` for an `InProgress` parent; verify `watched_run_ids` is populated but `start_event_delivery` is not called; then emit `UpdatedConversationStatus` → `Success` and verify delivery starts. +- **Invariant 5** (no children, no polling): Unit test — emit `RestoredConversations` for a conversation with no `run_id` and no children; verify `watched_run_ids` is not populated and no delivery is started. +- **Invariant 7, 8** (no duplicate delivery via SQLite cursor): Unit test — emit `RestoredConversations` for a conversation whose `last_event_sequence` in `AgentConversationData` is 42; verify `event_cursor[conv_id]` is initialized to 42 and the first poll call uses `cursor=42`. +- **Invariant 7, 8** (no duplicate delivery via server run response): Unit test — same setup but `AgentConversationData.last_event_sequence` is absent and the server run response returns `last_event_sequence = 42`; verify `event_cursor[conv_id]` is set to 42 when the async query returns. +- **Invariant 7, 8** (cursor write at `handle_poll_result`): Unit test — call `handle_poll_result` with a batch whose max sequence is 42; verify `BlocklistAIHistoryModel::update_event_sequence` is called with 42. +- **Invariant 6** (children_by_parent index pre-populated): Unit test asserting `BlocklistAIHistoryModel::child_conversation_ids_of(parent)` returns the child IDs before `RestoredConversations` fires (i.e., that `initialize_historical_conversations` at `conversation_loader.rs:466-477` builds the index at startup). +- **Invariant 10** (driver child run_ids from task messages): Unit test for `child_run_ids_from_task_messages` with `OrchestrationV2` enabled — build a conversation with a `StartAgentV2` success result; verify the agent_id is returned. Also test with `OrchestrationV2` disabled; verify an empty vec is returned. +- **Invariant 10** (driver case: no local DB child): Unit test — emit `RestoredConversations` where `child_conversation_ids_of` returns empty but task messages contain a `StartAgentV2` success result; verify the child run_id is still added to `watched_run_ids`. +- **Invariant 11** (umbrella): Covered by invariants 1–9 and 10 above; no separate test. +- **Invariant 12** (V1 lifecycle subscriptions after restart): Unit test in `orchestration_events_tests.rs` with legacy local lifecycle-dispatch mode — emit `RestoredConversations` with a child conversation having `parent_conversation_id` set; verify `lifecycle_subscription_routes` is populated with `None` event-type filter (subscribe-all). Note: `None` is intentional and broader than the original subscription filter, which is not persisted (see Risks). +- **Invariant 13** (non-orchestration conversations unaffected): Existing restore tests must continue to pass without modification. +- **Invariant 14** (shared-session viewers excluded): Unit test — emit `RestoredConversations` for a `is_viewing_shared_session = true` conversation; verify no entry in `watched_run_ids` and `event_cursor` is not initialized for it. +- **Invariant 15** (orphan child restored standalone): Unit test — emit `RestoredConversations` for a child whose `parent_conversation_id` points at an id not present in `conversations_by_id`; verify no lifecycle subscription is registered and no error is surfaced. +- **Invariant 16** (cleanup on delete): Unit test — populate `watched_run_ids` and `event_cursor` for a conversation; emit `DeletedConversation`; verify both are removed. +- **Invariant 9** (transient failures retried): Covered by existing SSE failure/backoff tests in `orchestration_event_poller_tests.rs`; verify those tests still pass. +- **Manual (local child, OrchestrationV2 on)**: Run parent + local child with V2 enabled, quit Warp mid-run, restart; confirm child's final status is shown and no previously seen messages are re-delivered. +- **Manual (local child, OrchestrationV2 off)**: Same scenario with V2 disabled; confirm V1 lifecycle subscriptions propagate the child's status after restart. +- **Manual (driver/remote child)**: `warp agent run --conversation ` where children ran on remote workers; confirm child run_ids are discovered from task messages and events are delivered from the correct resume point. + +## Non-Oz harness considerations + +Non-Oz harnesses cannot currently be parents in an orchestration session, so this feature does not directly affect them. The notes below are forward-looking. + +The WIP branch `katarina/quality-503-driver-owned-parent-bridge` in `wc-pine` introduces `AgentEventConsumer::persist_cursor` and implements it in `ParentBridgeEventConsumer` (for a Claude Code **child** receiving parent messages) by writing to a local `last-sequence` file. That is a different role from a non-Oz conversation acting as a parent, but the `persist_cursor` hook is the same primitive. + +When non-Oz parents become supported, they can reuse the server-side `last_event_sequence` field on `ai_tasks` added by change 1 above: wiring their `persist_cursor` callback to also call `PATCH /agent/runs/{run_id}/event-sequence` gives cloud/driver restore without any harness-specific logic. + +**Why `AgentConversationData` alone is insufficient for cloud/driver non-Oz**: `AgentConversationData` is exclusively local SQLite (never sent to server). The non-Oz parent bridge state directory (`last-sequence` file) is session-scoped, so a new `warp agent run --conversation` invocation won't find the previous session's file. The server-side field is the only path that works for driver restarts. + +**Child run_id discovery for non-Oz parents**: The two synchronous fallback sources used today (`children_by_parent` DB index and `child_run_ids_from_task_messages`) both depend on Oz's structured task messages and will not apply to non-Oz parents. The primary path — `GET /agent/runs/{parent_run_id}/children` — is already harness-agnostic (the `ai_tasks` table stores `parent_run_id` regardless of harness) and will cover non-Oz parents without any additional changes. + +## Follow-ups +- Consider scanning V1 `StartAgent` tool results (`ToolCallResultType::StartAgent`) alongside `StartAgentV2` once V1 orchestration is fully deprecated, for completeness. +- When non-Oz event delivery is fully implemented, wire `persist_cursor` in the non-Oz parent bridge to also call the server-side `last_event_sequence` update, mirroring the Oz path.