Reapply "Replay agent events on restore" (#25055)

Reverts warpdotdev/warp-internal#25055

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

Author: cephalonaut

Cargo.lock

@@ -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,
         },
     };
 

app/src/ai/agent/api/convert_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<i64>,
 }
 
 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<i64> {
+        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(

app/src/ai/agent/conversation.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,
             },
         );
 

app/src/ai/agent_conversations_model_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![],
     }
 }
 

app/src/ai/ambient_agents/spawn_tests.rs

@@ -262,6 +262,20 @@ pub struct AmbientAgentTask {
     pub agent_config_snapshot: Option<AgentConfigSnapshot>,
     #[serde(default, deserialize_with = "deserialize_artifacts")]
     pub artifacts: Vec<Artifact>,
+
+    /// 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<i64>,
+
+    /// 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<String>,
 }
 
 /// Represents a single attachment input from the client (e.g., file upload)

app/src/ai/ambient_agents/task.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<Self>,
+    ) {
+        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();

app/src/ai/blocklist/history_model.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(),