public/galaxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'ryserve/master'

Browse Source 
# Conflicts:
#	app/src/ai/llms.rs
This commit is contained in:
rkw6086
2026-10-25 01:32:18 -05:00
parent 596b7266a4 7dceca51cf
commit 6655fe7822
126 changed files with 7937 additions and 1147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/ai/runtime/event_translator.rs
+85 -57
View File
@@ -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
+59
View File
@@ -82,6 +82,65 @@ 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 {
images: Vec::new(),
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
+3
View File
@@ -4,6 +4,8 @@ mod provider_run_coordinator;
mod rig;
mod rig_request;
mod rig_tool;
pub(crate) use rig_tool::action_from_tool_call;
mod task_progress;
pub(crate) use child_context::add_parent_context;
pub(crate) use event_translator::{
@@ -19,3 +21,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
+44 -3
View File
@@ -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)
@@ -251,6 +253,34 @@ impl ProviderRunCoordinator {
self.profiles.get(profile).map(|profile| &profile.request)
}
/// The pending model profile owns the context budget, including after a
/// preflight overflow has returned the run to its compaction boundary.
pub(crate) fn current_context_limit(&self) -> Option<u32> {
self.profiles
.get(self.run.profile().as_str())
.and_then(|profile| profile.max_context_tokens)
.or(self.max_context_tokens)
}
/// Leave enough history space for a command-monitor turn before switching
/// profiles, accounting for each profile's own fixed prompt and output budget.
pub(crate) fn progressive_summary_context_budget(&self) -> Option<(u32, u32)> {
[self.run.profile().as_str(), CLI_MONITOR_PROVIDER_PROFILE]
.into_iter()
.filter_map(|name| {
let profile = self.profiles.get(name)?;
let limit = profile.max_context_tokens.or(self.max_context_tokens)?;
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))
})
.min_by_key(|(limit, fixed)| limit.saturating_sub(*fixed))
}
/// Includes the current transcript, fixed request context, tools, and output reservation.
/// Read at the model boundary so newly committed tool output is included.
pub(crate) fn estimated_context_tokens(&self) -> Option<(u32, u32)> {
@@ -258,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((
@@ -405,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();
}
@@ -473,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;
@@ -558,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();
@@ -1149,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();
@@ -1157,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
+208
View File
@@ -461,6 +461,7 @@ async fn coordinator_awaiting_tools(
fn tool_result(call_id: &str, content: &str, status: ToolResultStatus) -> ToolEvent {
ToolEvent::Completed {
result: ToolResult {
images: Vec::new(),
call_id: call_id.to_string(),
content: content.to_string(),
status,
@@ -514,6 +515,7 @@ async fn one_run_drives_model_tool_and_followup_turns_with_atomic_history() {
.complete_tool(
&batch.work_id,
ToolResult {
images: Vec::new(),
call_id: "read-1".to_string(),
content: "manifest contents".to_string(),
status: ToolResultStatus::Success,
@@ -945,6 +947,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");
@@ -1558,6 +1604,168 @@ fn context_estimate_tracks_live_transcript_and_reserves_fixed_context() {
assert_eq!(fixed_before, fixed_after);
}
#[tokio::test]
async fn compaction_boundary_keeps_the_monitor_context_limit() {
let base_runtime = Arc::new(ScriptedRuntime::new(vec![]));
let cli_runtime = Arc::new(ScriptedRuntime::new(vec![]));
let run = ProviderRun::new(
"monitor-compaction-budget",
vec![ConversationMessage {
role: MessageRole::User,
content: MessageContent::Text("history".repeat(5_000)),
}],
CLI_MONITOR_PROVIDER_PROFILE,
ProviderRunLimits::default(),
);
let mut profiles = BTreeMap::new();
profiles.insert(
BASE_PROVIDER_PROFILE.to_string(),
ProviderRunProfile::new(base_runtime.clone(), request()),
);
profiles.insert(
CLI_MONITOR_PROVIDER_PROFILE.to_string(),
ProviderRunProfile::new(cli_runtime.clone(), request())
.with_max_context_tokens(Some(8_000)),
);
let mut coordinator = ProviderRunCoordinator::new(run, profiles).unwrap();
coordinator.set_max_context_tokens(Some(100_000));
let (_sender, control) = turn_control();
let block = coordinator
.drive_until_blocked(control, |_| Ok(()))
.await
.unwrap();
assert_eq!(block, ProviderRunBlock::ContextWindowExceeded);
assert_eq!(coordinator.run().state(), &ProviderRunState::ReadyToCallModel);
assert_eq!(coordinator.current_context_limit(), Some(8_000));
assert!(base_runtime.requests().is_empty());
assert!(cli_runtime.requests().is_empty());
let work_id = coordinator.run().ready_work_id().unwrap();
coordinator
.run_mut()
.continue_ready_with_observation(
&work_id,
MessageContent::Text("Command finished".to_string()),
BASE_PROVIDER_PROFILE,
)
.unwrap();
assert_eq!(coordinator.current_context_limit(), Some(100_000));
}
#[test]
fn progressive_summary_budget_anticipates_switching_to_a_smaller_monitor() {
let runtime = Arc::new(ScriptedRuntime::new(vec![]));
let mut coordinator = coordinator(runtime.clone());
coordinator.set_max_context_tokens(Some(500_000));
coordinator
.insert_profile(
CLI_MONITOR_PROVIDER_PROFILE,
ProviderRunProfile::new(runtime, request()).with_max_context_tokens(Some(275_000)),
)
.unwrap();
assert_eq!(coordinator.current_context_limit(), Some(500_000));
assert_eq!(
coordinator
.progressive_summary_context_budget()
.map(|(limit, _)| limit),
Some(275_000)
);
coordinator
.run_mut()
.set_ready_profile(CLI_MONITOR_PROVIDER_PROFILE)
.unwrap();
assert_eq!(coordinator.current_context_limit(), Some(275_000));
assert_eq!(
coordinator
.progressive_summary_context_budget()
.map(|(limit, _)| limit),
Some(275_000)
);
coordinator
.run_mut()
.set_ready_profile(BASE_PROVIDER_PROFILE)
.unwrap();
assert_eq!(coordinator.current_context_limit(), Some(500_000));
assert_eq!(
coordinator
.progressive_summary_context_budget()
.map(|(limit, _)| limit),
Some(275_000)
);
}
#[test]
fn progressive_summary_budget_respects_smaller_active_limits_and_profile_fallbacks() {
for (base_limit, monitor_limit, fallback_limit, expected) in [
(
Some(200_000),
Some(Some(275_000)),
Some(500_000),
Some(200_000),
),
(Some(500_000), Some(None), Some(275_000), Some(275_000)),
(Some(500_000), None, Some(275_000), Some(500_000)),
(None, None, None, None),
] {
let runtime = Arc::new(ScriptedRuntime::new(vec![]));
let mut coordinator = ProviderRunCoordinator::from_request(
"progressive-summary-budget",
ProviderRunProfile::new(runtime.clone(), request()).with_max_context_tokens(base_limit),
Vec::new(),
ProviderRunLimits::default(),
)
.unwrap();
coordinator.set_max_context_tokens(fallback_limit);
if let Some(monitor_limit) = monitor_limit {
coordinator
.insert_profile(
CLI_MONITOR_PROVIDER_PROFILE,
ProviderRunProfile::new(runtime, request())
.with_max_context_tokens(monitor_limit),
)
.unwrap();
}
assert_eq!(
coordinator.current_context_limit(),
base_limit.or(fallback_limit)
);
assert_eq!(
coordinator
.progressive_summary_context_budget()
.map(|(limit, _)| limit),
expected
);
}
}
#[test]
fn progressive_summary_budget_accounts_for_output_reservations_when_selecting_a_profile() {
let runtime = Arc::new(ScriptedRuntime::new(vec![]));
let mut coordinator = coordinator(runtime.clone());
coordinator.set_max_context_tokens(Some(275_000));
let mut monitor_request = request();
monitor_request.max_output_tokens = Some(300_000);
coordinator
.insert_profile(
CLI_MONITOR_PROVIDER_PROFILE,
ProviderRunProfile::new(runtime, monitor_request)
.with_max_context_tokens(Some(500_000)),
)
.unwrap();
let (limit, persistent_tokens) = coordinator.progressive_summary_context_budget().unwrap();
assert_eq!(coordinator.current_context_limit(), Some(275_000));
assert_eq!(limit, 500_000);
assert!(persistent_tokens > 300_000);
}
#[tokio::test]
async fn monitor_request_uses_its_profile_context_limit() {
let base_runtime = Arc::new(ScriptedRuntime::new(vec![answer_turn()]));
app/src/ai/runtime/rig.rs
+21 -16
View File
@@ -133,20 +133,25 @@ pub(crate) async fn prepare_provider_run(
let skill_path_origin = params.session_context.skill_path_origin();
let mut cli_params = params.clone();
let computer_use_model = params.computer_use_model.as_str();
let computer_use_profile = if params.computer_use_enabled
&& !computer_use_model.is_empty()
&& computer_use_model != "computer-use-agent-auto"
&& computer_use_model != params.model.as_str()
&& !matches!(
&computer_use_provider_config,
crate::ai::provider::ProviderConfig::None
) {
let computer_use_profile = if supported_tools.contains(&ToolType::RequestComputerUse) {
let mut computer_params = params.clone();
computer_params.model = params.computer_use_model.clone();
let computer_provider = if computer_use_model.is_empty()
|| computer_use_model == "computer-use-agent-auto"
|| computer_use_model == params.model.as_str()
|| matches!(&computer_use_provider_config, crate::ai::provider::ProviderConfig::None)
{
base_provider_config.clone()
} else {
computer_params.model = params.computer_use_model.clone();
computer_use_provider_config
};
let mut computer_tools = supported_tools.clone();
computer_tools.retain(|tool| *tool != ToolType::RequestComputerUse);
computer_tools.push(ToolType::UseComputer);
let (runtime, prepared) = prepare_provider_profile(
computer_use_provider_config,
computer_provider,
computer_params,
supported_tools.clone(),
computer_tools,
supported_cli_agent_tools.clone(),
None,
)
@@ -315,9 +320,8 @@ pub(crate) async fn provider_runtime_for_request(
model,
reasoning_effort: config.reasoning_effort,
max_output_tokens: config.max_output_tokens.map(u64::from),
auth_file: None,
},
)),
).with_credential_resolver(crate::ai::chatgpt_auth::credential_resolver())),
OpenAIProviderKind::Anthropic => {
let credential = config.api_key.unwrap_or_default();
let stored = crate::ai::claude_subscription::is_stored_credential(&credential);
@@ -354,14 +358,15 @@ pub(crate) async fn provider_runtime_for_request(
},
crate::ai::provider::ProviderConfig::Bedrock(config) => {
let max_output_tokens = Some(bedrock_max_output_tokens(&model));
let cross_region_inference = config.cross_region_inference;
let cross_region_inference = config.invocation_model_id.is_none() && config.cross_region_inference;
let caching_config =
CachingConfig::from_external_config(&ExternalBedrockConfig::load());
CachingConfig::from_external_config(&ExternalBedrockConfig::load()).for_model(&model);
let invocation_model = config.invocation_model_id.clone().unwrap_or(model);
let client = BedrockClient::from_config(config).await?;
Arc::new(BedrockRuntime::from_aws_client(
client.runtime_client(),
BedrockRigConfig {
model,
model: invocation_model,
region: client.region().to_string(),
cross_region_inference,
prompt_caching: caching_config.enabled,
app/src/ai/runtime/rig_request.rs
+22 -13
View File
@@ -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,
@@ -382,17 +389,7 @@ fn input_messages(
) -> Vec<ConversationMessage> {
let mut messages = Vec::new();
if !tool_results.is_empty() {
let mut parts = tool_results
.into_iter()
.map(|result| {
let is_error = result.is_error();
ContentPart::ToolResult {
tool_use_id: result.call_id,
content: result.content,
is_error,
}
})
.collect::<Vec<_>>();
let mut parts = ToolResult::into_batch_content_parts(tool_results);
let content = if parts.len() == 1 {
let ContentPart::ToolResult {
tool_use_id,
@@ -692,6 +689,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 +780,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 +850,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;
@@ -1030,6 +1033,12 @@ fn build_system_prompt(
{
prompt.push_str(CODING_DELEGATION_GUIDANCE);
}
if tools.iter().any(|tool| tool.name == "request_computer_use") {
prompt.push_str("## Desktop Computer Use\nUse request_computer_use for tasks that require seeing or interacting with the user's desktop. This controls the machine running Galaxy, including when the terminal is connected to another host. Request the specific authorized task and wait for the screenshot. A successful request switches to the configured computer-use model; rejected or failed requests do not grant control.\n\n");
}
if tools.iter().any(|tool| tool.name == "use_computer") {
prompt.push_str("## Desktop Computer Use\nInspect the latest returned screenshot before acting. Use use_computer with small action batches, then inspect the new screenshot. Use original physical pixel coordinates: multiply displayed image x by original_width/width and y by original_height/height. Every button/key press requires both down and up actions. For window targeting, select an exact window_id/pid from the latest windows list; never invent a window. Stop on errors or denied permission and report the actual cause. Galaxy's Full Disk Access is separate from Screen Recording and Accessibility permissions. When desktop work is complete and terminal/file work remains, call the appropriate normal tool and Galaxy resumes the base model. Only finish the response when the whole user's task is complete.\n\n");
}
prompt.push_str("## Available Tools\n");
if tools.is_empty() {
prompt.push_str("No tools are available. Do not invent tool calls.\n");
app/src/ai/runtime/rig_request_tests.rs
+71 -2
View File
@@ -242,6 +242,58 @@ fn plan_turn_requires_a_plan_tool_result() {
assert!(prompt.contains("Do not return the plan only as prose"));
}
#[test]
fn normal_and_plan_turns_advertise_clarifications_only_for_interactive_permission_modes() {
use crate::ai::execution_profiles::{
AIExecutionProfile, AskUserQuestionPermission, PermissionMode,
};
let _flag = galaxy_core::features::FeatureFlag::AskUserQuestion.override_enabled(true);
for (permission_mode, expected) in [
(PermissionMode::AskFirst, true),
(PermissionMode::AllowAll, true),
(PermissionMode::GalaxyDecides, false),
(PermissionMode::ReadOnly, false),
] {
for plan_mode in [false, true] {
let mut profile = AIExecutionProfile {
permission_mode,
..Default::default()
};
profile.project_permission_mode();
let mut params = RequestParams::new_for_test();
params.permission_mode = permission_mode;
params.ask_user_question_enabled =
profile.ask_user_question != AskUserQuestionPermission::Never;
let mut input = user_query("Clarify the requested destination before proceeding");
if plan_mode {
let AIAgentInput::UserQuery {
user_query_mode, ..
} = &mut input
else {
unreachable!("user_query returns a user query input");
};
*user_query_mode = UserQueryMode::Plan;
}
params.input = vec![input];
let (normal, cli) =
crate::ai::agent::api::prepare_direct_provider_params(&mut params);
assert!(!cli.contains(&ToolType::AskUserQuestion));
let prepared = prepare_rig_turn(&config(), params, normal, cli);
assert_eq!(
prepared
.request
.tools
.iter()
.any(|tool| tool.name == "ask_user_question"),
expected,
"{permission_mode:?}, plan_mode={plan_mode}"
);
}
}
}
#[test]
fn no_tools_turn_flattens_historical_tool_protocol_messages() {
let mut params = RequestParams::new_for_test();
@@ -270,8 +322,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 +552,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]
@@ -518,6 +585,7 @@ fn lrc_snapshot_follow_up_uses_the_cli_monitor_prompt_and_tools() {
),
};
let snapshot_tool_result = ToolResult {
images: Vec::new(),
call_id: "run-call".to_string(),
content: result.result.model_content(),
status: ToolResultStatus::Success,
@@ -827,6 +895,7 @@ fn normalized_tool_outcomes_are_the_only_action_results_sent_to_rig() {
let tool_results = statuses
.iter()
.map(|(call_id, status, _)| ToolResult {
images: Vec::new(),
call_id: (*call_id).to_string(),
content: format!("normalized-{call_id}"),
status: *status,
app/src/ai/runtime/rig_tests.rs
+21
View File
@@ -87,6 +87,27 @@ async fn selected_computer_use_model_gets_its_own_profile() {
);
}
#[tokio::test]
async fn automatic_and_same_model_computer_use_keep_a_dedicated_profile() {
for selected in ["computer-use-agent-auto", "base-selection", "unroutable-model"] {
let mut params = RequestParams::new_for_test();
params.model = LLMId::from("base-selection");
params.computer_use_model = LLMId::from(selected);
params.computer_use_enabled = true;
let prepared = prepare_provider_run(
ProviderConfig::OpenAI(openai_config("base-provider-model")),
ProviderConfig::None, ProviderConfig::None, params, Vec::new(),
).await.unwrap();
assert!(prepared.base_profile.request.tools.iter().any(|tool| tool.name == "request_computer_use"));
assert!(!prepared.base_profile.request.tools.iter().any(|tool| tool.name == "use_computer"));
let computer = prepared.computer_use_profile.unwrap();
assert_eq!(computer.request.model.as_str(), "base-provider-model");
assert!(computer.request.tools.iter().any(|tool| tool.name == "use_computer"));
assert!(!computer.request.tools.iter().any(|tool| tool.name == "request_computer_use"));
assert!(computer.request.system_prompt.unwrap().contains("Inspect the latest returned screenshot before acting"));
}
}
#[test]
fn subscription_context_budget_uses_catalog_limit_and_clamps_stale_overrides() {
let mut config = openai_config("subscription-model");
app/src/ai/runtime/rig_tool.rs
+124
View File
@@ -55,6 +55,23 @@ pub(crate) fn action_from_tool_call(
rationale: optional_string(input, "rationale")?,
citations: Vec::new(),
},
"request_computer_use" => AIAgentActionType::RequestComputerUse(
crate::ai::agent::RequestComputerUseRequest {
task_summary: required_nonempty_string(input, "task_summary")?,
screenshot_params: Some(computer_screenshot_params(input)?),
},
),
"use_computer" => AIAgentActionType::UseComputer(
crate::ai::agent::UseComputerRequest {
action_summary: required_nonempty_string(input, "action_summary")?,
actions: computer_actions(input)?,
screenshot_params: if optional_boolean(input, "take_screenshot")?.unwrap_or(true) {
Some(computer_screenshot_params(input)?)
} else {
None
},
},
),
"read_files" => AIAgentActionType::ReadFiles(ReadFilesRequest {
locations: limited_required_array(input, "files", MAX_BATCH_READ_ITEMS)?
.iter()
@@ -472,6 +489,113 @@ fn optional_nonnegative_i32(input: &serde_json::Value, key: &str) -> Result<Opti
.transpose()
}
fn computer_i32(input: &serde_json::Value, key: &str) -> Result<i32, String> {
input.get(key).and_then(serde_json::Value::as_i64)
.and_then(|value| i32::try_from(value).ok())
.ok_or_else(|| format!("invalid field {key:?}: expected a 32-bit integer"))
}
fn computer_target(input: &serde_json::Value) -> Result<computer_use::Target, String> {
let Some(target) = input.get("target") else { return Ok(computer_use::Target::Screen); };
require_object(target, "target")?;
match required_string(target, "type")?.as_str() {
"screen" => Ok(computer_use::Target::Screen),
"window" => {
let window_id = optional_bounded_u64(target, "window_id", u64::from(u32::MAX))?
.filter(|id| *id > 0).ok_or("window target requires a non-zero window_id")? as u32;
let pid = computer_i32(target, "pid")?;
if pid <= 0 { return Err("window target requires a positive pid".to_string()); }
if !crate::features::FeatureFlag::BackgroundComputerUse.is_enabled()
|| !computer_use::background_supported()
{
return Err("Background window targeting is unavailable; use the screen target.".to_string());
}
Ok(computer_use::Target::Window { window_id, pid })
}
_ => Err("target.type must be screen or window".to_string()),
}
}
fn computer_screenshot_params(input: &serde_json::Value) -> Result<computer_use::ScreenshotParams, String> {
let empty = serde_json::json!({});
let params = input.get("screenshot_params").unwrap_or(&empty);
require_object(params, "screenshot_params")?;
let positive = |key, maximum, default| -> Result<usize, String> {
let value = optional_bounded_u64(params, key, maximum)?.unwrap_or(default);
if value == 0 { return Err(format!("{key} must be positive")); }
Ok(value as usize)
};
Ok(computer_use::ScreenshotParams {
max_long_edge_px: Some(positive("max_long_edge_px", 4096, 1568)?),
max_total_px: Some(positive("max_total_px", 4_194_304, 1_150_000)?),
region: None,
target: computer_target(params)?,
})
}
fn computer_actions(input: &serde_json::Value) -> Result<Vec<computer_use::TargetedAction>, String> {
use computer_use::{Action, Key, MouseButton, ScrollDirection, ScrollDistance, TargetedAction, Vector2I};
let mut wait_seconds = 0.0;
limited_required_array(input, "actions", 20)?.iter().map(|input| {
require_object(input, "action")?;
let position = || -> Result<Vector2I, String> {
Ok(Vector2I::new(computer_i32(input, "x")?, computer_i32(input, "y")?))
};
let button = || -> Result<MouseButton, String> {
match required_string(input, "button")?.as_str() {
"left" => Ok(MouseButton::Left), "right" => Ok(MouseButton::Right),
"middle" => Ok(MouseButton::Middle), "back" => Ok(MouseButton::Back),
"forward" => Ok(MouseButton::Forward),
_ => Err("button must be left, right, middle, back, or forward".to_string()),
}
};
let key = || -> Result<Key, String> {
match (input.get("key"), input.get("keycode")) {
(Some(_), None) => {
let text = required_string(input, "key")?;
let mut chars = text.chars();
let key = chars.next().ok_or("key must contain exactly one character")?;
if chars.next().is_some() { return Err("key must contain exactly one character".to_string()); }
Ok(Key::Char(key))
}
(None, Some(_)) => Ok(Key::Keycode(computer_i32(input, "keycode")?)),
_ => Err("provide exactly one of key or keycode".to_string()),
}
};
let action = match required_string(input, "type")?.as_str() {
"mouse_move" => Action::MouseMove { to: position()? },
"mouse_down" => Action::MouseDown { button: button()?, at: position()? },
"mouse_up" => Action::MouseUp { button: button()? },
"mouse_wheel" => Action::MouseWheel {
at: position()?,
direction: match required_string(input, "direction")?.as_str() {
"up" => ScrollDirection::Up, "down" => ScrollDirection::Down,
"left" => ScrollDirection::Left, "right" => ScrollDirection::Right,
_ => return Err("direction must be up, down, left, or right".to_string()),
},
distance: ScrollDistance::Pixels(optional_nonnegative_i32(input, "pixels")?.ok_or("mouse_wheel requires pixels")?),
},
"type_text" => {
let text = required_string(input, "text")?;
if text.len() > 100_000 { return Err("text must be no longer than 100000 bytes".to_string()); }
Action::TypeText { text }
}
"key_down" => Action::KeyDown { key: key()? },
"key_up" => Action::KeyUp { key: key()? },
"wait" => {
let seconds = input.get("seconds").and_then(serde_json::Value::as_f64)
.filter(|seconds| seconds.is_finite() && *seconds >= 0.0 && *seconds <= 10.0)
.ok_or("seconds must be between 0 and 10")?;
wait_seconds += seconds;
if wait_seconds > 10.0 { return Err("a computer-use batch may wait at most 10 seconds".to_string()); }
Action::Wait(Duration::from_secs_f64(seconds))
}
_ => return Err("unknown computer action type".to_string()),
};
Ok(TargetedAction { action, target: computer_target(input)? })
}).collect()
}
fn skill_references(
input: &serde_json::Value,
skill_path_origin: &SkillPathOrigin,
app/src/ai/runtime/rig_tool_tests.rs
+49
View File
@@ -16,6 +16,55 @@ fn call(name: &str, arguments: serde_json::Value) -> ToolCall {
}
}
#[test]
fn computer_use_request_always_captures_a_bounded_initial_screenshot() {
let action = action_from_tool_call("task", &call("request_computer_use", serde_json::json!({
"task_summary": "Inspect the browser",
})), &SkillPathOrigin::Local, &HashMap::new()).unwrap();
let AIAgentActionType::RequestComputerUse(request) = action.action else { panic!("expected computer-use request"); };
let screenshot = request.screenshot_params.unwrap();
assert_eq!(screenshot.max_long_edge_px, Some(1568));
assert_eq!(screenshot.max_total_px, Some(1_150_000));
assert_eq!(screenshot.target, computer_use::Target::Screen);
}
#[test]
fn computer_actions_preserve_coordinates_keys_and_screenshot_choices() {
let action = action_from_tool_call("task", &call("use_computer", serde_json::json!({
"action_summary": "Select the field and type",
"actions": [
{ "type": "mouse_down", "button": "left", "x": 420, "y": 310 },
{ "type": "mouse_up", "button": "left" },
{ "type": "type_text", "text": "hello" },
{ "type": "key_down", "keycode": 36 },
{ "type": "key_up", "keycode": 36 },
],
"take_screenshot": false,
})), &SkillPathOrigin::Local, &HashMap::new()).unwrap();
let AIAgentActionType::UseComputer(request) = action.action else { panic!("expected computer actions"); };
assert_eq!(request.actions.len(), 5);
assert_eq!(request.actions[0].action, computer_use::Action::MouseDown {
button: computer_use::MouseButton::Left, at: computer_use::Vector2I::new(420, 310),
});
assert_eq!(request.actions[3].action, computer_use::Action::KeyDown { key: computer_use::Key::Keycode(36) });
assert_eq!(request.screenshot_params, None);
}
#[test]
fn computer_actions_reject_invalid_targets_keys_and_unbounded_waiting() {
for actions in [
serde_json::json!([{ "type": "key_down", "key": "Return" }]),
serde_json::json!([{ "type": "key_up", "key": "a", "keycode": 0 }]),
serde_json::json!([{ "type": "mouse_move", "x": 1, "y": 2, "target": { "type": "window", "window_id": 0, "pid": 12 } }]),
serde_json::json!([{ "type": "wait", "seconds": 6 }, { "type": "wait", "seconds": 6 }]),
serde_json::json!(vec![serde_json::json!({ "type": "wait", "seconds": 0 }); 21]),
] {
assert!(action_from_tool_call("task", &call("use_computer", serde_json::json!({
"action_summary": "Invalid request", "actions": actions,
})), &SkillPathOrigin::Local, &HashMap::new()).is_err());
}
}
#[test]
fn shell_calls_become_domain_actions_without_a_proto_round_trip() {
let action = action_from_tool_call(
app/src/ai/runtime/task_progress.rs
+292
View File
@@ -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
+117
View File
@@ -0,0 +1,117 @@
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 {
images: Vec::new(),
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());
}