Track task progress and organize Galaxy Drive plans

app/src/ai/runtime/event_translator.rs

@@ -2,13 +2,14 @@ use std::collections::HashMap;
 
 use galaxy_agent_core::{
     AgentEvent, ProviderRunOutcome, RuntimeActivity, RuntimeActivityStatus, RuntimeCapabilities,
-    StopReason, Usage,
+    StopReason, TaskListSnapshot, Usage,
 };
 use uuid::Uuid;
 use warp_multi_agent_api::response_event::stream_finished;
 use warp_multi_agent_api::{self as api, ClientAction, ResponseEvent};
 
 use super::provider_run_coordinator::ProviderRunProjection;
+use super::task_progress::{explicit_task_updates, task_list_from_plan_batch, todo_operations};
 use crate::ai::agent::runtime_activity;
 use crate::ai::openai::response_translator::{build_stream_finished, StreamUsage};
 use crate::ai::provider::response_translator::{
@@ -54,6 +55,8 @@ pub(crate) struct ProviderRunResponseProjector {
     translator: RuntimeResponseTranslator,
     has_started_model_turn: bool,
     plan_tasks_projected: bool,
+    task_list: Option<TaskListSnapshot>,
+    pending_task_list: Option<TaskListSnapshot>,
     finished: bool,
 }
 
@@ -63,6 +66,8 @@ impl ProviderRunResponseProjector {
             translator: RuntimeResponseTranslator::new(config),
             has_started_model_turn: false,
             plan_tasks_projected: false,
+            task_list: None,
+            pending_task_list: None,
             finished: false,
         }
     }
@@ -75,10 +80,45 @@ impl ProviderRunResponseProjector {
             translator: RuntimeResponseTranslator::restored(config, projection_was_initialized),
             has_started_model_turn: false,
             plan_tasks_projected: true,
+            task_list: None,
+            pending_task_list: None,
             finished: false,
         }
     }
 
+    pub(crate) fn with_task_list(mut self, task_list: Option<TaskListSnapshot>) -> Self {
+        self.plan_tasks_projected |= task_list.is_some();
+        self.task_list = task_list;
+        self
+    }
+
+    pub(crate) fn with_restored_task_lists(
+        mut self,
+        visible: Option<TaskListSnapshot>,
+        desired: Option<TaskListSnapshot>,
+    ) -> Self {
+        self.plan_tasks_projected = visible.is_some() || desired.is_some();
+        self.task_list = visible;
+        self.pending_task_list =
+            desired.filter(|snapshot| self.task_list.as_ref() != Some(snapshot));
+        self
+    }
+
+    fn reconcile_restored_tasks(&mut self) -> Vec<ResponseEvent> {
+        let Some(snapshot) = self.pending_task_list.take() else {
+            return Vec::new();
+        };
+        let mut events = Vec::new();
+        self.translator.initialize(&mut events);
+        events.extend(
+            todo_operations(self.task_list.as_ref(), &snapshot)
+                .into_iter()
+                .map(|operation| build_todo_update(&self.translator.config.task_id, operation)),
+        );
+        self.task_list = Some(snapshot);
+        events
+    }
+
     pub(crate) fn project(
         &mut self,
         projection: ProviderRunProjection,
@@ -86,7 +126,8 @@ impl ProviderRunResponseProjector {
         if self.finished {
             return Err("provider run projection is already finished".to_string());
         }
-        match projection {
+        let mut events = self.reconcile_restored_tasks();
+        let projected = match projection {
             ProviderRunProjection::ModelTurnStarted { .. } => {
                 if self.has_started_model_turn {
                     self.translator.begin_followup_turn();
@@ -111,23 +152,28 @@ impl ProviderRunResponseProjector {
             }
             ProviderRunProjection::ModelTurnFinished { .. } => Ok(Vec::new()),
             ProviderRunProjection::ToolBatchReady { batch } => {
-                if !self.plan_tasks_projected {
-                    if let Some(todos) = todos_from_plan_batch(&batch) {
-                        self.plan_tasks_projected = true;
-                        return Ok(vec![build_todo_update(
-                            &self.translator.config.task_id,
-                            api::message::update_todos::Operation::CreateTodoList(
-                                api::CreateTodoList {
-                                    initial_todos: todos,
-                                },
-                            ),
-                        )]);
-                    }
+                let mut snapshots = explicit_task_updates(&batch);
+                if snapshots.is_empty() && !self.plan_tasks_projected {
+                    snapshots.extend(task_list_from_plan_batch(&batch));
+                }
+                let mut events = Vec::new();
+                for snapshot in snapshots {
+                    events.extend(
+                        todo_operations(self.task_list.as_ref(), &snapshot)
+                            .into_iter()
+                            .map(|operation| {
+                                build_todo_update(&self.translator.config.task_id, operation)
+                            }),
+                    );
+                    self.task_list = Some(snapshot);
+                    self.plan_tasks_projected = true;
                 }
                 // Tool batches indicate runtime activity, not completion of planned work.
-                Ok(Vec::new())
+                Ok(events)
             }
-        }
+        }?;
+        events.extend(projected);
+        Ok(events)
     }
 
     pub(crate) fn set_task_id(&mut self, task_id: impl Into<String>) {
@@ -143,19 +189,22 @@ impl ProviderRunResponseProjector {
             return Err("provider run projection is already finished".to_string());
         }
         self.finished = true;
-        match outcome {
-            ProviderRunOutcome::Completed(completion) => Ok(self
+        let mut events = self.reconcile_restored_tasks();
+        let terminal = match outcome {
+            ProviderRunOutcome::Completed(completion) => self
                 .translator
-                .finish_provider_run(completion.stop_reason.clone(), aggregate_usage)),
-            ProviderRunOutcome::Failed(failure) => Ok(self.translator.provider_failure(
+                .finish_provider_run(completion.stop_reason.clone(), aggregate_usage),
+            ProviderRunOutcome::Failed(failure) => self.translator.provider_failure(
                 &failure.message,
                 failure.source.as_ref(),
                 aggregate_usage,
-            )),
-            ProviderRunOutcome::Cancelled { .. } => Ok(self
+            ),
+            ProviderRunOutcome::Cancelled { .. } => self
                 .translator
-                .finish_provider_run(StopReason::Cancelled, aggregate_usage)),
-        }
+                .finish_provider_run(StopReason::Cancelled, aggregate_usage),
+        };
+        events.extend(terminal);
+        Ok(events)
     }
 }
 
@@ -174,42 +223,21 @@ fn authentication_provider_for_runtime(runtime_id: &str) -> api::LlmProvider {
 
 // Only explicit multi-step checklists in a plan become tasks. Other document bullets
 // (risks, examples, requirements) are not executable tasks.
+#[cfg(test)]
 fn todos_from_plan_batch(
     batch: &galaxy_agent_core::PendingToolBatch,
 ) -> Option<Vec<api::TodoItem>> {
-    let plan_call = batch.calls.iter().find(|pending| {
-        matches!(
-            pending.call.name.as_str(),
-            "create_plan" | "create_documents"
-        )
-    })?;
-    let content = plan_call
-        .call
-        .arguments
-        .get("documents")?
-        .as_array()?
-        .first()?
-        .get("content")?
-        .as_str()?;
-    let mut lines = content.lines();
-    lines.find(|line| matches!(line.trim(), "## Tasks" | "## Implementation Tasks"))?;
-    let todos = lines
-        .take_while(|line| !line.trim().starts_with("## "))
-        .filter_map(|line| line.trim().strip_prefix("- [ ]").map(str::trim))
-        .filter(|item| !item.is_empty())
-        .take(50)
-        .enumerate()
-        .map(|(index, item)| api::TodoItem {
-            id: format!("plan-{}-{index}", plan_call.call.id),
-            title: item
-                .split_once(" - ")
-                .map_or(item, |(title, _)| title)
-                .trim()
-                .to_owned(),
-            description: item.to_owned(),
-        })
-        .collect::<Vec<_>>();
-    (todos.len() > 1).then_some(todos)
+    task_list_from_plan_batch(batch).map(|snapshot| {
+        snapshot
+            .tasks
+            .into_iter()
+            .map(|task| api::TodoItem {
+                id: task.id,
+                description: task.title.clone(),
+                title: task.title,
+            })
+            .collect()
+    })
 }
 
 fn build_todo_update(

app/src/ai/runtime/event_translator_tests.rs

@@ -82,6 +82,64 @@ fn restored_provider_projection_skips_stream_initialization() {
     ));
 }
 
+#[test]
+fn restored_task_progress_replay_does_not_duplicate_the_persisted_checklist() {
+    use galaxy_agent_core::{TaskListItem, TaskListSnapshot, TaskListStatus};
+    let snapshot = TaskListSnapshot { tasks: vec![
+        TaskListItem { id: "inspect".to_owned(), title: "Inspect".to_owned(), status: TaskListStatus::Completed },
+        TaskListItem { id: "verify".to_owned(), title: "Verify".to_owned(), status: TaskListStatus::InProgress },
+    ] };
+    let mut projector = ProviderRunResponseProjector::restored(RuntimeResponseConfig {
+        task_id: "task".to_owned(), conversation_id: "conversation".to_owned(), needs_create_task: false,
+        user_query: None, model_id: "model".to_owned(), max_context_tokens: Some(1000),
+        capabilities: RuntimeCapabilities::provider(), empty_output_message: None,
+    }, true).with_task_list(Some(snapshot.clone()));
+    let batch = galaxy_agent_core::PendingToolBatch {
+        work_id: galaxy_agent_core::ExternalWorkId {
+            run_id: galaxy_agent_core::ProviderRunId::new("run"), epoch: galaxy_agent_core::RunEpoch::new(1),
+        },
+        calls: vec![galaxy_agent_core::PendingToolCall {
+            call: galaxy_agent_core::ToolCall {
+                id: "progress".to_owned(), name: galaxy_agent_core::UPDATE_TASK_LIST_NAME.to_owned(),
+                arguments: serde_json::to_value(&snapshot).unwrap(),
+            },
+            state: galaxy_agent_core::PendingToolCallState::Resolved { result: galaxy_agent_core::ToolResult {
+                call_id: "progress".to_owned(), status: galaxy_agent_core::ToolResultStatus::Success,
+                content: snapshot.to_tool_result_content(),
+            }},
+        }],
+    };
+    assert!(projector.project(ProviderRunProjection::ToolBatchReady { batch }).unwrap().is_empty());
+}
+
+#[test]
+fn restored_committed_task_snapshot_repairs_missing_projection_even_on_terminal_cleanup() {
+    use galaxy_agent_core::{TaskListItem, TaskListSnapshot, TaskListStatus};
+    let snapshot = TaskListSnapshot { tasks: vec![TaskListItem {
+        id: "verify".to_owned(), title: "Verify".to_owned(), status: TaskListStatus::Completed,
+    }] };
+    let mut projector = ProviderRunResponseProjector::restored(RuntimeResponseConfig {
+        task_id: "task".to_owned(), conversation_id: "conversation".to_owned(), needs_create_task: false,
+        user_query: None, model_id: "model".to_owned(), max_context_tokens: Some(1000),
+        capabilities: RuntimeCapabilities::provider(), empty_output_message: None,
+    }, true).with_restored_task_lists(None, Some(snapshot));
+    let events = projector.finish(&galaxy_agent_core::ProviderRunOutcome::Cancelled {
+        reason: "restored cancellation".to_owned(),
+    }, &Usage::default()).unwrap();
+    let operations = events.into_iter().filter_map(|event| {
+        let Some(response_event::Type::ClientActions(actions)) = event.r#type else { return None };
+        Some(actions.actions.into_iter().filter_map(|action| {
+            let Some(client_action::Action::AddMessagesToTask(add)) = action.action else { return None };
+            Some(add.messages.into_iter().filter_map(|message| {
+                let Some(message::Message::UpdateTodos(update)) = message.message else { return None };
+                update.operation
+            }).collect::<Vec<_>>())
+        }).flatten().collect::<Vec<_>>())
+    }).flatten().collect::<Vec<_>>();
+    assert!(matches!(&operations[0], message::update_todos::Operation::CreateTodoList(_)));
+    assert!(matches!(&operations[1], message::update_todos::Operation::MarkTodosCompleted(mark) if mark.todo_ids == ["verify"]));
+}
+
 #[test]
 fn provider_activity_does_not_create_or_complete_task_lists() {
     let mut projector = ProviderRunResponseProjector::new(RuntimeResponseConfig {

app/src/ai/runtime/mod.rs

@@ -4,6 +4,7 @@ mod provider_run_coordinator;
 mod rig;
 mod rig_request;
 mod rig_tool;
+mod task_progress;
 
 pub(crate) use child_context::add_parent_context;
 pub(crate) use event_translator::{
@@ -19,3 +20,4 @@ pub(crate) use rig::{
     PreparedProviderRun, ProviderActionContext,
 };
 pub(crate) use rig_request::OrchestrationModelOption;
+pub(crate) use task_progress::{restored_task_list, task_list_from_api_tasks};

app/src/ai/runtime/provider_run_coordinator.rs

@@ -12,11 +12,12 @@ use galaxy_agent_core::{
     ModelFailureDisposition, PendingToolBatch, ProviderModelCall, ProviderRequestProfile,
     ProviderRun, ProviderRunFailureKind, ProviderRunId, ProviderRunLimits, ProviderRunOutcome,
     ProviderRunPhase, ProviderRunProtocolError, ProviderRunState, ProviderRunStep, RunEpoch,
-    RuntimeKind, StopReason, ToolEvent, TurnCommand, TurnControl, TurnRequest, Usage,
+    RuntimeKind, StopReason, TaskListSnapshot, ToolEvent, TurnCommand, TurnControl, TurnRequest, Usage,
 };
 use instant::Instant;
 use warpui::r#async::Timer;
 
+use super::task_progress::{add_task_list_context, initial_task_list, task_list_from_plan_batch};
 use crate::ai::agent::conversation::AIConversationId;
 #[cfg(not(target_family = "wasm"))]
 use crate::ai::model_output_logging;
@@ -205,6 +206,7 @@ impl ProviderRunCoordinator {
             limits,
         );
         run.replace_tool_result_archive(tool_result_archive);
+        run.set_task_list(initial_task_list(&base_profile.request))?;
         let mut profiles = BTreeMap::new();
         profiles.insert(BASE_PROVIDER_PROFILE.to_string(), base_profile);
         Self::new(run, profiles)
@@ -271,6 +273,7 @@ impl ProviderRunCoordinator {
                 let mut request = profile.request.clone();
                 request.prompt = None;
                 request.messages.clear();
+                add_task_list_context(&mut request, self.run.task_list());
                 let fixed = u32::try_from(estimate_turn_request_tokens(&request))
                     .unwrap_or(u32::MAX);
                 Some((limit, fixed))
@@ -285,6 +288,7 @@ impl ProviderRunCoordinator {
         let mut request = profile.request.clone();
         request.prompt = None;
         request.messages.clear();
+        add_task_list_context(&mut request, self.run.task_list());
         let persistent_tokens = estimate_turn_request_tokens(&request);
         request.messages = self.run.transcript().to_vec();
         Some((
@@ -432,6 +436,9 @@ impl ProviderRunCoordinator {
                     }
                 }
                 Some(ProviderRunStep::DispatchTools(batch)) => {
+                    if self.run.task_list().is_none() {
+                        self.run.set_task_list(task_list_from_plan_batch(&batch))?;
+                    }
                     if !self.checkpoint_or_fail(&mut checkpoint).await? {
                         return self.terminal_block();
                     }
@@ -500,8 +507,12 @@ impl ProviderRunCoordinator {
         else {
             return false;
         };
-        let request =
-            request_for_model_call(profile.request.clone(), call, Some(max_context_tokens));
+        let request = request_for_model_call(
+            profile.request.clone(),
+            call,
+            Some(max_context_tokens),
+            self.run.task_list(),
+        );
         // Provider tokenizers and request envelopes differ. Compact before the
         // estimate reaches the advertised limit, leaving room for that uncertainty.
         let safe_context_tokens = u64::from(max_context_tokens) * 9 / 10;
@@ -585,6 +596,7 @@ impl ProviderRunCoordinator {
             profile.request.clone(),
             &call,
             profile.max_context_tokens.or(self.max_context_tokens),
+            self.run.task_list(),
         );
         let started_at = Instant::now();
         let (attempt_sender, attempt_control) = turn_control();
@@ -1176,6 +1188,7 @@ fn request_for_model_call(
     mut template: TurnRequest,
     call: &ProviderModelCall,
     max_context_tokens: Option<u32>,
+    task_list: Option<&TaskListSnapshot>,
 ) -> TurnRequest {
     template.prompt = None;
     template.messages = call.messages.clone();
@@ -1184,6 +1197,7 @@ fn request_for_model_call(
         message
             .truncate_tagged_terminal_output_for_provider_request_with_context(max_context_tokens);
     }
+    add_task_list_context(&mut template, task_list);
     add_exploration_reminder(&mut template);
     template
 }

app/src/ai/runtime/provider_run_coordinator_tests.rs

@@ -945,6 +945,50 @@ async fn inline_tool_batches_commit_before_the_next_model_boundary() {
     ));
 }
 
+#[tokio::test]
+async fn task_progress_reaches_the_next_model_after_compaction_and_restore() {
+    use galaxy_agent_core::{TaskListItem, TaskListSnapshot, TaskListStatus};
+    use crate::ai::runtime::task_progress::{add_task_list_context, initial_task_list, task_list_tool};
+
+    let initial = TaskListSnapshot { tasks: vec![TaskListItem {
+        id: "verify".to_owned(), title: "Build the change".to_owned(), status: TaskListStatus::InProgress,
+    }] };
+    let current = TaskListSnapshot { tasks: vec![TaskListItem {
+        status: TaskListStatus::Completed, ..initial.tasks[0].clone()
+    }] };
+    let runtime = Arc::new(ScriptedRuntime::new(vec![
+        tool_turn_with_calls(vec![ToolCall {
+            id: "progress".to_owned(),
+            name: galaxy_agent_core::UPDATE_TASK_LIST_NAME.to_owned(),
+            arguments: serde_json::to_value(&current).unwrap(),
+        }]),
+        answer_turn(),
+    ]));
+    let mut template = request();
+    template.tools.push(task_list_tool());
+    add_task_list_context(&mut template, Some(&initial));
+    let mut coordinator = ProviderRunCoordinator::from_request(
+        "run", ProviderRunProfile::new(runtime.clone(), template.clone()), Vec::new(), ProviderRunLimits::default(),
+    ).unwrap();
+    let (_sender, control) = turn_control();
+    assert!(matches!(coordinator.drive_until_blocked(control, |_| Ok(())).await.unwrap(), ProviderRunBlock::ReadyToCallModel));
+    assert_eq!(coordinator.run().task_list(), Some(&current));
+    let length = coordinator.run().transcript().len();
+    coordinator.run_mut().compact_transcript_at_model_boundary(length, vec![ConversationMessage {
+        role: MessageRole::User, content: MessageContent::Text("Prior work summarized.".to_owned()),
+    }]).unwrap();
+    coordinator.run_mut().replace_tool_result_archive(Vec::new());
+    let restored: ProviderRun = serde_json::from_str(&serde_json::to_string(coordinator.run()).unwrap()).unwrap();
+    restored.validate_restored_state().unwrap();
+    let mut coordinator = ProviderRunCoordinator::new(restored, BTreeMap::from([
+        (BASE_PROVIDER_PROFILE.to_owned(), ProviderRunProfile::new(runtime.clone(), template)),
+    ])).unwrap();
+    let (_sender, control) = turn_control();
+    coordinator.drive_until_blocked(control, |_| Ok(())).await.unwrap();
+    assert_eq!(initial_task_list(&runtime.requests()[1]), Some(current));
+    assert_eq!(runtime.requests()[1].messages.len(), 1);
+}
+
 #[tokio::test]
 async fn recoverable_start_failure_retries_the_same_work_identity() {
     let mut recoverable = AgentError::new(AgentErrorKind::Transport, "temporary network error");

app/src/ai/runtime/rig_request.rs

@@ -15,6 +15,9 @@ use uuid::Uuid;
 use warp_multi_agent_api as api;
 use warp_multi_agent_api::ToolType;
 
+use super::task_progress::{
+    add_task_list_context, task_list_from_api_tasks, task_list_tool, TASK_PROGRESS_GUIDANCE,
+};
 use crate::ai::agent::api::RequestParams;
 use crate::ai::agent::{AIAgentContext, AIAgentInput, MCPContext, UserQueryMode};
 use crate::ai::openai::client::OpenAIClientConfig;
@@ -277,7 +280,9 @@ fn prepare_rig_turn_for_provider(
             // History recall cannot advance a running command. Keeping it in the CLI tool list lets
             // the model spend its monitor turn recalling a prior snapshot instead of scheduling
             // `read_shell_command_output`, so make polling the only inspection path here.
-            tools.retain(|tool| tool.name != "recall_tool_history");
+            tools.retain(|tool| {
+                !matches!(tool.name.as_str(), "recall_tool_history" | "update_task_list")
+            });
         }
         RigRequestMode::CompletedCommandAssessment => {
             // The caller deliberately disables tools for the final assessment. The inline history
@@ -348,6 +353,8 @@ fn prepare_rig_turn_for_provider(
     request.system_prompt = Some(system_prompt);
     request.tools = tools;
     request.max_output_tokens = max_output_tokens;
+    let task_list = task_list_from_api_tasks(&tasks, &task_id);
+    add_task_list_context(&mut request, task_list.as_ref());
 
     PreparedRigTurn {
         max_context_tokens,
@@ -692,6 +699,7 @@ fn tool_definitions(
         .into_iter()
         .filter(|tool| tool_name_is_supported(&tool.name, &supported))
         .collect::<Vec<_>>();
+    tools.push(task_list_tool());
     let mut mcp_tool_aliases = HashMap::new();
 
     if !supported.contains(&ToolType::CallMcpTool) {
@@ -782,7 +790,9 @@ fn add_tool_title_parameter(schema: &mut serde_json::Value) -> String {
 }
 
 fn tools_are_inline_only(tools: &[ToolDefinition]) -> bool {
-    tools.iter().all(|tool| tool.name == "recall_tool_history")
+    tools.iter().all(|tool| {
+        matches!(tool.name.as_str(), "recall_tool_history" | "update_task_list")
+    })
 }
 
 const MAX_PROVIDER_TOOL_NAME_BYTES: usize = 64;
@@ -850,6 +860,9 @@ fn build_system_prompt(
     );
     prompt.push_str(SHELL_EXECUTION_GUIDANCE);
     prompt.push_str(PROGRESS_COMMUNICATION_GUIDANCE);
+    if tools.iter().any(|tool| tool.name == "update_task_list") {
+        prompt.push_str(TASK_PROGRESS_GUIDANCE);
+    }
     let contexts = inputs.iter().filter_map(AIAgentInput::context).flatten();
     let mut environment = Vec::new();
     let mut request_time = None;

app/src/ai/runtime/rig_request_tests.rs

@@ -270,8 +270,9 @@ fn no_tools_turn_flattens_historical_tool_protocol_messages() {
 
     let prepared = prepare_rig_turn(&config(), params, Vec::new(), Vec::new());
 
-    assert_eq!(prepared.request.tools.len(), 1);
-    assert_eq!(prepared.request.tools[0].name, "recall_tool_history");
+    assert_eq!(prepared.request.tools.len(), 2);
+    assert!(prepared.request.tools.iter().any(|tool| tool.name == "recall_tool_history"));
+    assert!(prepared.request.tools.iter().any(|tool| tool.name == "update_task_list"));
     assert!(prepared
         .request
         .messages
@@ -499,6 +500,20 @@ fn forced_cli_profile_uses_monitor_prompt_and_tools_for_an_initial_query() {
         .tools
         .iter()
         .any(|tool| tool.name == "recall_tool_history"));
+    assert!(!prepared.request.tools.iter().any(|tool| tool.name == "update_task_list"));
+}
+
+#[test]
+fn task_progress_is_available_with_explicit_status_guidance_and_ui_title() {
+    let mut params = RequestParams::new_for_test();
+    params.input = vec![user_query("Implement and verify the fix")];
+    let prepared = prepare_rig_turn(&config(), params, vec![ToolType::ReadFiles], Vec::new());
+    let tool = prepared.request.tools.iter().find(|tool| tool.name == "update_task_list").unwrap();
+    assert!(tool.input_schema["required"].as_array().unwrap().iter().any(|value| value == "_galaxy_tool_title"));
+    let prompt = prepared.request.system_prompt.unwrap();
+    assert!(prompt.contains("## Task List Progress"));
+    assert!(prompt.contains("reconcile the list before the final response"));
+    assert!(prompt.contains("Planning or returning a draft does not complete"));
 }
 
 #[test]

app/src/ai/runtime/task_progress.rs

@@ -0,0 +1,292 @@
+//! Task state is kept outside the compactable transcript and projected only from
+//! explicit, validated model updates. Finishing a response never finishes tasks.
+
+use galaxy_agent_core::{
+    PendingToolBatch, PendingToolCallState, ProviderRun, ProviderRunState, TaskListItem,
+    TaskListSnapshot, TaskListStatus, ToolDefinition, ToolResultStatus, TurnRequest,
+    UPDATE_TASK_LIST_NAME,
+};
+use warp_multi_agent_api as api;
+
+const SNAPSHOT_START: &str = "\n<galaxy-task-list>\n";
+const SNAPSHOT_END: &str = "\n</galaxy-task-list>\n";
+const TASK_LIST_REMINDER: &str = "\nKeep the current task list synchronized with the work at meaningful milestones and before finishing. If a step is now done or work has moved to another step, call update_task_list with the evidence-based status. Pending work must remain visible.\n";
+
+pub(super) const TASK_PROGRESS_GUIDANCE: &str = "\n## Task List Progress\nUse `update_task_list` to keep the visible checklist current throughout multi-step work. It accepts the complete list, with stable task IDs, titles, and pending/in_progress/completed statuses. Keep at most one task in progress and place it before pending tasks. After creating a plan, use its checklist and IDs from the current task list; do not create a competing list. Update when starting the next step, completing a meaningful step, changing scope, or discovering a blocker, and reconcile the list before the final response. Preserve unfinished tasks when blocked or waiting for a decision; explain the blocker in the task title if needed. Mark completed only when the work and required verification actually succeeded. Planning or returning a draft does not complete its implementation tasks. Do not recreate an unchanged list, update after every poll, or spend turns just restating progress. Tool panes show work; task updates do not require prose announcements. CLI command monitoring must continue using its advertised monitoring tools.\n";
+
+pub(super) fn task_list_tool() -> ToolDefinition {
+    ToolDefinition {
+        name: UPDATE_TASK_LIST_NAME.to_owned(),
+        description: "Create or update the visible task list for this conversation. Send the complete list with stable IDs, retaining completed and unfinished work. Update at meaningful milestones, when the current step changes, and before finishing; mark completed only after the work and required checks succeed. This records task progress only and does not execute the tasks or edit the plan document.".to_owned(),
+        input_schema: serde_json::json!({
+            "type": "object",
+            "properties": {
+                "tasks": {
+                    "type": "array", "minItems": 1, "maxItems": 50,
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "id": {"type": "string", "minLength": 1, "maxLength": 200},
+                            "title": {"type": "string", "minLength": 1, "maxLength": 1000},
+                            "status": {"type": "string", "enum": ["pending", "in_progress", "completed"]}
+                        },
+                        "required": ["id", "title", "status"],
+                        "additionalProperties": false
+                    }
+                }
+            },
+            "required": ["tasks"],
+            "additionalProperties": false
+        }),
+    }
+}
+
+pub(crate) fn task_list_from_api_tasks(
+    tasks: &[api::Task],
+    root_task_id: &str,
+) -> Option<TaskListSnapshot> {
+    use api::message::update_todos::Operation;
+
+    let task = tasks.iter().find(|task| task.id == root_task_id)?;
+    let mut pending = Vec::<api::TodoItem>::new();
+    let mut completed = Vec::<api::TodoItem>::new();
+    for message in &task.messages {
+        let Some(api::message::Message::UpdateTodos(update)) = &message.message else {
+            continue;
+        };
+        let Some(operation) = update.operation.as_ref() else {
+            continue;
+        };
+        match operation {
+            Operation::CreateTodoList(create) => {
+                pending = create.initial_todos.clone();
+                completed.clear();
+            }
+            Operation::UpdatePendingTodos(update) => {
+                pending = update.updated_pending_todos.clone();
+            }
+            Operation::MarkTodosCompleted(mark) => {
+                for id in &mark.todo_ids {
+                    if let Some(index) = pending.iter().position(|task| task.id == *id) {
+                        let item = pending.remove(index);
+                        if let Some(old) = completed.iter_mut().find(|task| task.id == *id) {
+                            *old = item;
+                        } else {
+                            completed.push(item);
+                        }
+                    }
+                }
+            }
+        }
+    }
+    let snapshot = TaskListSnapshot {
+        tasks: completed
+            .into_iter()
+            .map(|task| task_item(task, TaskListStatus::Completed))
+            .chain(pending.into_iter().enumerate().map(|(index, task)| {
+                task_item(
+                    task,
+                    if index == 0 {
+                        TaskListStatus::InProgress
+                    } else {
+                        TaskListStatus::Pending
+                    },
+                )
+            }))
+            .collect(),
+    };
+    snapshot.validate().ok()?;
+    Some(snapshot)
+}
+
+fn task_item(task: api::TodoItem, status: TaskListStatus) -> TaskListItem {
+    TaskListItem {
+        id: task.id,
+        title: task.title,
+        status,
+    }
+}
+
+pub(super) fn initial_task_list(request: &TurnRequest) -> Option<TaskListSnapshot> {
+    let prompt = request.system_prompt.as_deref()?;
+    let (_, snapshot) = prompt.rsplit_once(SNAPSHOT_START)?;
+    let (snapshot, _) = snapshot.split_once(SNAPSHOT_END)?;
+    let snapshot: TaskListSnapshot = serde_json::from_str(snapshot).ok()?;
+    snapshot.validate().ok()?;
+    Some(snapshot)
+}
+
+pub(super) fn add_task_list_context(request: &mut TurnRequest, snapshot: Option<&TaskListSnapshot>) {
+    let Some(prompt) = request.system_prompt.as_mut() else {
+        return;
+    };
+    if prompt.ends_with(TASK_LIST_REMINDER) {
+        prompt.truncate(prompt.len() - TASK_LIST_REMINDER.len());
+    }
+    if let Some(start) = prompt.rfind(SNAPSHOT_START) {
+        if let Some(end) = prompt[start..].find(SNAPSHOT_END) {
+            prompt.replace_range(start..start + end + SNAPSHOT_END.len(), "");
+        }
+    }
+    if let Some(snapshot) = snapshot {
+        prompt.push_str(SNAPSHOT_START);
+        prompt.push_str(&serde_json::to_string(snapshot).expect("serializable task list"));
+        prompt.push_str(SNAPSHOT_END);
+    }
+    if request
+        .tools
+        .iter()
+        .any(|tool| tool.name == UPDATE_TASK_LIST_NAME)
+    {
+        prompt.push_str(TASK_LIST_REMINDER);
+    }
+}
+
+/// Preserve the existing plan checklist behavior, with stable IDs shared by the
+/// UI and the next provider request. Rejected tools cannot create task lists.
+pub(super) fn task_list_from_plan_batch(batch: &PendingToolBatch) -> Option<TaskListSnapshot> {
+    let plan_call = batch.calls.iter().find(|pending| {
+        matches!(pending.call.name.as_str(), "create_plan" | "create_documents")
+            && !matches!(
+                &pending.state,
+                PendingToolCallState::Resolved { result }
+                    if result.status != galaxy_agent_core::ToolResultStatus::Success
+            )
+    })?;
+    let content = plan_call
+        .call
+        .arguments
+        .get("documents")?
+        .as_array()?
+        .first()?
+        .get("content")?
+        .as_str()?;
+    let mut lines = content.lines();
+    lines.find(|line| matches!(line.trim(), "## Tasks" | "## Implementation Tasks"))?;
+    let tasks = lines
+        .take_while(|line| !line.trim().starts_with("## "))
+        .filter_map(|line| line.trim().strip_prefix("- [ ]").map(str::trim))
+        .filter(|item| !item.is_empty())
+        .take(50)
+        .enumerate()
+        .map(|(index, item)| TaskListItem {
+            id: format!("plan-{}-{index}", plan_call.call.id),
+            title: item
+                .split_once(" - ")
+                .map_or(item, |(title, _)| title)
+                .trim()
+                .chars()
+                .take(1_000)
+                .collect(),
+            status: TaskListStatus::Pending,
+        })
+        .collect::<Vec<_>>();
+    if tasks.len() < 2 {
+        return None;
+    }
+    let snapshot = TaskListSnapshot { tasks };
+    snapshot.validate().ok()?;
+    Some(snapshot)
+}
+
+fn todo_item(task: &TaskListItem) -> api::TodoItem {
+    api::TodoItem {
+        id: task.id.clone(),
+        title: task.title.clone(),
+        description: task.title.clone(),
+    }
+}
+
+pub(super) fn explicit_task_updates(batch: &PendingToolBatch) -> Vec<TaskListSnapshot> {
+    batch
+        .calls
+        .iter()
+        .filter_map(|pending| {
+            if pending.call.name != UPDATE_TASK_LIST_NAME {
+                return None;
+            }
+            let PendingToolCallState::Resolved { result } = &pending.state else {
+                return None;
+            };
+            if result.status != ToolResultStatus::Success {
+                return None;
+            }
+            galaxy_agent_core::task_list_from_tool_result(&result.content)
+        })
+        .collect()
+}
+
+pub(crate) fn restored_task_list(run: &ProviderRun) -> Option<TaskListSnapshot> {
+    // A mixed batch can have projected an inline update before the other tools
+    // finish. Conversely an inline-only batch can commit during normalization
+    // before its UI projection was ever persisted. Reconcile both crash windows.
+    if let ProviderRunState::AwaitingTools { batch } = run.state() {
+        if let Some(snapshot) = explicit_task_updates(batch).pop() {
+            return Some(snapshot);
+        }
+    }
+    run.task_list().cloned()
+}
+
+pub(super) fn todo_operations(
+    previous: Option<&TaskListSnapshot>,
+    snapshot: &TaskListSnapshot,
+) -> Vec<api::message::update_todos::Operation> {
+    use api::message::update_todos::Operation;
+
+    if previous == Some(snapshot) {
+        return Vec::new();
+    }
+    // Existing completed items are immutable in the transcript protocol. A new
+    // scope or reopening a completed item therefore starts a replacement list.
+    let replace = previous.is_none_or(|previous| {
+        previous.tasks.iter().any(|old| {
+            old.status == TaskListStatus::Completed
+                && !snapshot.tasks.iter().any(|task| {
+                    task.id == old.id
+                        && task.title == old.title
+                        && task.status == TaskListStatus::Completed
+                })
+        })
+    });
+    let already_completed = |id: &str| {
+        !replace
+            && previous.is_some_and(|previous| {
+                previous
+                    .tasks
+                    .iter()
+                    .any(|task| task.id == id && task.status == TaskListStatus::Completed)
+            })
+    };
+    let todos = snapshot
+        .tasks
+        .iter()
+        .filter(|task| !already_completed(&task.id))
+        .map(todo_item)
+        .collect::<Vec<_>>();
+    let mut operations = vec![if replace {
+        Operation::CreateTodoList(api::CreateTodoList {
+            initial_todos: todos,
+        })
+    } else {
+        Operation::UpdatePendingTodos(api::UpdatePendingTodos {
+            updated_pending_todos: todos,
+        })
+    }];
+    let todo_ids = snapshot
+        .tasks
+        .iter()
+        .filter(|task| task.status == TaskListStatus::Completed && !already_completed(&task.id))
+        .map(|task| task.id.clone())
+        .collect::<Vec<_>>();
+    if !todo_ids.is_empty() {
+        operations.push(Operation::MarkTodosCompleted(api::MarkTodosCompleted {
+            todo_ids,
+        }));
+    }
+    operations
+}
+
+#[cfg(test)]
+#[path = "task_progress_tests.rs"]
+mod tests;

app/src/ai/runtime/task_progress_tests.rs

@@ -0,0 +1,116 @@
+use super::*;
+
+fn snapshot(tasks: &[(&str, TaskListStatus)]) -> TaskListSnapshot {
+    TaskListSnapshot {
+        tasks: tasks.iter().map(|(id, status)| TaskListItem {
+            id: (*id).to_owned(),
+            title: format!("Task {id}"),
+            status: *status,
+        }).collect(),
+    }
+}
+
+fn task_with_operations(operations: Vec<api::message::update_todos::Operation>) -> api::Task {
+    api::Task {
+        id: "root".to_owned(),
+        messages: operations.into_iter().map(|operation| api::Message {
+            message: Some(api::message::Message::UpdateTodos(api::message::UpdateTodos {
+                operation: Some(operation),
+            })),
+            ..Default::default()
+        }).collect(),
+        ..Default::default()
+    }
+}
+
+#[test]
+fn task_status_updates_reuse_the_list_and_survive_a_followup_request() {
+    use TaskListStatus::{Completed, InProgress, Pending};
+    use api::message::update_todos::Operation;
+
+    let initial = snapshot(&[("inspect", InProgress), ("implement", Pending), ("verify", Pending)]);
+    let next = snapshot(&[("inspect", Completed), ("implement", InProgress), ("verify", Pending)]);
+    let updates = todo_operations(Some(&initial), &next);
+    assert!(matches!(&updates[0], Operation::UpdatePendingTodos(_)));
+    assert!(matches!(&updates[1], Operation::MarkTodosCompleted(mark) if mark.todo_ids == ["inspect"]));
+    let mut operations = todo_operations(None, &initial);
+    operations.extend(updates);
+    let root = task_with_operations(operations);
+    let unrelated_child = api::Task { id: "child".to_owned(), ..Default::default() };
+    assert_eq!(task_list_from_api_tasks(&[unrelated_child, root], "root"), Some(next.clone()));
+    assert!(todo_operations(Some(&next), &next).is_empty());
+
+    let finished = snapshot(&[("inspect", Completed), ("implement", Completed), ("verify", Completed)]);
+    let operations = todo_operations(Some(&next), &finished);
+    let Operation::UpdatePendingTodos(update) = &operations[0] else {
+        panic!("normal progress preserves the existing checklist");
+    };
+    assert_eq!(update.updated_pending_todos.len(), 2);
+    assert!(!update.updated_pending_todos.iter().any(|task| task.id == "inspect"));
+}
+
+#[test]
+fn reopening_completed_work_replaces_the_protocol_list_without_losing_ids() {
+    use TaskListStatus::{Completed, InProgress};
+    let old = snapshot(&[("inspect", Completed), ("verify", Completed)]);
+    let reopened = snapshot(&[("inspect", Completed), ("verify", InProgress)]);
+    let operations = todo_operations(Some(&old), &reopened);
+    assert!(matches!(&operations[0], api::message::update_todos::Operation::CreateTodoList(_)));
+    let root = task_with_operations(operations);
+    assert_eq!(task_list_from_api_tasks(&[root], "root"), Some(reopened));
+}
+
+#[test]
+fn redelivered_progress_cannot_duplicate_completed_items_in_followup_state() {
+    use TaskListStatus::{Completed, InProgress, Pending};
+    let initial = snapshot(&[("inspect", InProgress), ("verify", Pending)]);
+    let next = snapshot(&[("inspect", Completed), ("verify", InProgress)]);
+    let update = todo_operations(Some(&initial), &next);
+    let mut operations = todo_operations(None, &initial);
+    operations.extend(update.clone());
+    operations.extend(update);
+    assert_eq!(task_list_from_api_tasks(&[task_with_operations(operations)], "root"), Some(next));
+}
+
+#[test]
+fn current_snapshot_replaces_stale_prompt_state_without_accumulating_reminders() {
+    let initial = snapshot(&[("inspect", TaskListStatus::InProgress)]);
+    let current = snapshot(&[("inspect", TaskListStatus::Completed)]);
+    let mut request = TurnRequest::new("model", Vec::new());
+    request.system_prompt = Some("Follow user instructions.\n".to_owned());
+    request.tools = vec![task_list_tool()];
+    add_task_list_context(&mut request, Some(&initial));
+    assert_eq!(initial_task_list(&request), Some(initial));
+    add_task_list_context(&mut request, Some(&current));
+    let prompt = request.system_prompt.clone();
+    add_task_list_context(&mut request, Some(&current));
+    assert_eq!(request.system_prompt, prompt);
+    assert_eq!(initial_task_list(&request), Some(current));
+    let prompt = request.system_prompt.unwrap();
+    assert_eq!(prompt.matches(SNAPSHOT_START).count(), 1);
+    assert_eq!(prompt.matches(TASK_LIST_REMINDER).count(), 1);
+    assert!(prompt.starts_with("Follow user instructions."));
+}
+
+#[test]
+fn rejected_plan_creation_cannot_seed_visible_or_durable_tasks() {
+    let batch = PendingToolBatch {
+        work_id: galaxy_agent_core::ExternalWorkId {
+            run_id: galaxy_agent_core::ProviderRunId::new("run"),
+            epoch: galaxy_agent_core::RunEpoch::new(1),
+        },
+        calls: vec![galaxy_agent_core::PendingToolCall {
+            call: galaxy_agent_core::ToolCall {
+                id: "plan".to_owned(),
+                name: "create_plan".to_owned(),
+                arguments: serde_json::json!({"documents": [{"content": "## Tasks\n- [ ] Inspect\n- [ ] Verify"}]}),
+            },
+            state: PendingToolCallState::Resolved { result: galaxy_agent_core::ToolResult {
+                call_id: "plan".to_owned(),
+                content: "Not advertised".to_owned(),
+                status: galaxy_agent_core::ToolResultStatus::Error,
+            }},
+        }],
+    };
+    assert!(task_list_from_plan_batch(&batch).is_none());
+}