diff --git a/crates/agen/src/engine.rs b/crates/agen/src/engine.rs index bc7cf043..75a2386d 100644 --- a/crates/agen/src/engine.rs +++ b/crates/agen/src/engine.rs @@ -1,9 +1,10 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::{marker::PhantomData, sync::Arc, time::Instant}; use futures::StreamExt; use serde_json::{Value, json}; use tokio::sync::mpsc; +use tokio::time::{Duration, Instant as TokioInstant}; use tracing::{debug, info, trace, warn}; use crate::{ @@ -27,11 +28,14 @@ use crate::{ timeline::{TextBlockCollector, ThinkingBlockCollector, Timeline, ToolCallCollector}, tool::{ ToolCall, ToolDefinition as EngineToolDefinition, ToolError, ToolExecutionContext, - ToolOutputLimits, ToolResult, truncate_content, + ToolOutputLimits, ToolResult, ToolResultDisposition, truncate_content, }, tool_server::{ToolServer, ToolServerHandle}, }; +const TOOL_CANCEL_SIGNAL_TIMEOUT: Duration = Duration::from_millis(100); +const TOOL_CANCEL_GRACE_PERIOD: Duration = Duration::from_millis(500); + /// Engine errors #[derive(Debug, thiserror::Error)] pub enum EngineError { @@ -53,6 +57,9 @@ pub enum EngineError { /// A durable-history observer rejected an item before it entered history. #[error("History append failed: {0}")] HistoryAppend(String), + /// Tool terminalization lost its execution-attempt compare-and-set fence. + #[error("Tool execution attempt fence failed: {0}")] + ToolAttemptFence(String), } /// Tool registration error @@ -183,6 +190,64 @@ enum ToolExecutionResult { Paused, } +#[derive(Debug, Clone)] +struct ToolExecutionAttempt { + attempt_id: String, + terminal: bool, +} + +/// Per-batch compare-and-set fence for terminal ToolResult commits. +/// +/// A completion may commit only when its attempt id still matches the active +/// execution for that call and no prior terminal output has won the fence. +#[derive(Debug, Default)] +struct ToolExecutionAttemptFence { + attempts: HashMap, +} + +impl ToolExecutionAttemptFence { + fn register(&mut self, call_id: String, attempt_id: String) { + self.attempts.insert( + call_id, + ToolExecutionAttempt { + attempt_id, + terminal: false, + }, + ); + } + + fn can_commit(&self, call_id: &str, attempt_id: &str) -> bool { + matches!( + self.attempts.get(call_id), + Some(attempt) if attempt.attempt_id == attempt_id && !attempt.terminal + ) + } + + fn commit_terminal(&mut self, call_id: &str, attempt_id: &str) -> bool { + let Some(attempt) = self.attempts.get_mut(call_id) else { + return false; + }; + if attempt.attempt_id != attempt_id || attempt.terminal { + return false; + } + attempt.terminal = true; + true + } + + fn is_terminal(&self, call_id: &str) -> bool { + self.attempts + .get(call_id) + .is_some_and(|attempt| attempt.terminal) + } + + #[cfg(test)] + fn attempt_id(&self, call_id: &str) -> Option<&str> { + self.attempts + .get(call_id) + .map(|attempt| attempt.attempt_id.as_str()) + } +} + const MAX_STREAM_CONTINUATIONS: u32 = 3; /// Central component for managing LLM interactions @@ -1100,40 +1165,73 @@ impl Engine { context.clone(), ), ); - approved_calls.push((tool_call, context)); + approved_calls.push((tool_call, context, Some(info.tool))); } else { // Unknown tools go into approved list as-is (will error at execution) let context = ToolExecutionContext::new(&tool_call.id, &batch_id, call_index); - approved_calls.push((tool_call, context)); + approved_calls.push((tool_call, context, None)); } } // Phase 2: Execute approved tools in parallel. FuturesUnordered yields // each terminal result as soon as that call completes instead of // holding fast siblings behind the slowest call in the batch. + let started_calls: Vec<_> = approved_calls + .iter() + .map(|(tool_call, context, _)| (tool_call.id.clone(), context.batch_id.clone())) + .collect(); + let mut attempt_fence = ToolExecutionAttemptFence::default(); + for (call_id, attempt_id) in &started_calls { + attempt_fence.register(call_id.clone(), attempt_id.clone()); + } let futures: FuturesUnordered<_> = approved_calls .into_iter() - .map(|(tool_call, context)| { - let tool_server = self.tool_server.clone(); - async move { - let input_json = serde_json::to_string(&tool_call.input).unwrap_or_default(); - match tool_server - .call_tool(&tool_call.name, &input_json, context) - .await - { + .map(|(tool_call, context, tool)| async move { + let attempt_id = context.batch_id.clone(); + let input_json = serde_json::to_string(&tool_call.input).unwrap_or_default(); + let result = match tool { + None => ToolResult::error( + &tool_call.id, + format!("Tool not found: {}", tool_call.name), + ), + Some(tool) => match tool.execute(&input_json, context).await { Ok(output) => ToolResult::from_output(&tool_call.id, output), - Err(e) => ToolResult::error(&tool_call.id, e.to_string()), - } - } + Err(ToolError::Cancelled(output)) => { + ToolResult::from_output_with_disposition( + &tool_call.id, + output, + ToolResultDisposition::Cancelled, + ) + } + Err(ToolError::Interrupted(output)) => { + ToolResult::from_output_with_disposition( + &tool_call.id, + output, + ToolResultDisposition::Interrupted, + ) + } + Err(error) => ToolResult::error(&tool_call.id, error.to_string()), + }, + }; + (attempt_id, result) }) .collect(); // Synthetic results are already terminal and need no execution wait. // Commit them before polling ordinary calls so they obey the same // commit-before-publish boundary. + let mut terminal_call_ids = HashSet::new(); for result in synthetic_results { - self.finalize_and_commit_tool_result(history, annotate, result, &call_info_map) - .await?; + self.finalize_and_commit_tool_result( + history, + annotate, + result, + None, + &call_info_map, + &mut attempt_fence, + &mut terminal_call_ids, + ) + .await?; } let mut futures = futures; @@ -1144,18 +1242,94 @@ impl Engine { // output observed before the cancellation boundary. biased; result = futures.next() => { - let result = result.expect("non-empty FuturesUnordered returns a result"); + let (attempt_id, result) = + result.expect("non-empty FuturesUnordered returns a result"); self.finalize_and_commit_tool_result( history, annotate, result, + Some(&attempt_id), &call_info_map, + &mut attempt_fence, + &mut terminal_call_ids, ).await?; } cancel = self.cancel_rx.recv() => { if cancel.is_some() { - info!("Tool execution cancelled"); + info!("Tool execution cancellation requested"); } + + let cancellation_requests = call_info_map + .iter() + .filter(|(call_id, _)| !terminal_call_ids.contains(*call_id)) + .map(|(call_id, (_, _, tool, _))| { + let call_id = call_id.clone(); + let tool = tool.clone(); + async move { (call_id.clone(), tool.cancel(&call_id).await) } + }); + let cancellation_requests: FuturesUnordered<_> = + cancellation_requests.collect(); + match tokio::time::timeout( + TOOL_CANCEL_SIGNAL_TIMEOUT, + cancellation_requests.collect::>(), + ) + .await + { + Ok(results) => { + for (call_id, result) in results { + if let Err(error) = result { + warn!( + %call_id, + error = %error, + "Tool cooperative cancellation request failed" + ); + } + } + } + Err(_) => warn!("Tool cooperative cancellation request timed out"), + } + + // Keep polling the original execution futures for a bounded + // grace period so cooperative providers can return their + // confirmed terminal output, including bounded progress. + let deadline = TokioInstant::now() + TOOL_CANCEL_GRACE_PERIOD; + while !futures.is_empty() { + tokio::select! { + biased; + result = futures.next() => { + let (attempt_id, result) = + result.expect("non-empty FuturesUnordered returns a result"); + self.finalize_and_commit_tool_result( + history, + annotate, + result, + Some(&attempt_id), + &call_info_map, + &mut attempt_fence, + &mut terminal_call_ids, + ).await?; + } + _ = tokio::time::sleep_until(deadline) => break, + } + } + + // Calls that did not confirm a terminal outcome inside the + // grace period are durably closed as OutcomeUnknown before + // Engine/Worker final status becomes observable. + for (call_id, attempt_id) in &started_calls { + if !attempt_fence.is_terminal(call_id) { + self.finalize_and_commit_tool_result( + history, + annotate, + ToolResult::outcome_unknown(call_id), + Some(attempt_id), + &call_info_map, + &mut attempt_fence, + &mut terminal_call_ids, + ).await?; + } + } + self.timeline.abort_current_block(); return Err(EngineError::Cancelled); } @@ -1172,6 +1346,7 @@ impl Engine { history: &mut History, annotate: &mut impl FnMut(&Item) -> Result, mut tool_result: ToolResult, + execution_attempt_id: Option<&str>, call_info_map: &HashMap< String, ( @@ -1181,7 +1356,24 @@ impl Engine { ToolExecutionContext, ), >, - ) -> Result<(), EngineError> { + attempt_fence: &mut ToolExecutionAttemptFence, + terminal_call_ids: &mut HashSet, + ) -> Result { + let call_id = tool_result.tool_use_id.as_str(); + let may_commit = match execution_attempt_id { + Some(attempt_id) => attempt_fence.can_commit(call_id, attempt_id), + None => !terminal_call_ids.contains(call_id), + }; + if !may_commit { + warn!( + call_id, + execution_attempt_id, + disposition = ?tool_result.disposition, + "Ignoring stale or duplicate tool result after terminal output commit" + ); + return Ok(false); + } + let call_info = call_info_map.get(&tool_result.tool_use_id); if let Some((tool_call, meta, tool, context)) = call_info { let mut info = ToolResultInfo { @@ -1200,6 +1392,10 @@ impl Engine { } tool_result = info.result; } + if tool_result.is_error && tool_result.disposition.is_success() { + tool_result.disposition = ToolResultDisposition::Error; + } + tool_result.is_error = !tool_result.disposition.is_success(); // Cap content only after post_tool_call so interceptors still observe // the full payload and any content they inject is bounded too. @@ -1229,16 +1425,33 @@ impl Engine { } } - let item = Item::tool_result_item_with_attachments( + let item = Item::tool_result_item_with_disposition_and_attachments( &tool_result.tool_use_id, &tool_result.summary, tool_result.content.clone(), - tool_result.is_error, + tool_result.disposition, tool_result.attachments.clone(), ); self.append_history_items(history, std::iter::once(item), annotate)?; + if let Some(attempt_id) = execution_attempt_id + && !attempt_fence.commit_terminal(&tool_result.tool_use_id, attempt_id) + { + return Err(EngineError::ToolAttemptFence( + "tool execution attempt fence changed during terminal commit".to_string(), + )); + } + terminal_call_ids.insert(tool_result.tool_use_id.clone()); + debug!( + tool = call_info + .map(|(call, _, _, _)| call.name.as_str()) + .unwrap_or("unknown"), + call_id = %tool_result.tool_use_id, + execution_attempt_id, + disposition = ?tool_result.disposition, + "Tool execution terminalized" + ); self.emit_tool_result(&tool_result); - Ok(()) + Ok(true) } /// Internal turn execution logic @@ -2378,6 +2591,21 @@ mod tests { use crate::tool::{Attachment, ImageAttachment}; use std::time::Duration; + #[test] + fn tool_execution_attempt_fence_rejects_duplicate_and_stale_results() { + let mut fence = ToolExecutionAttemptFence::default(); + fence.register("call".to_string(), "attempt-1".to_string()); + assert!(fence.can_commit("call", "attempt-1")); + assert!(fence.commit_terminal("call", "attempt-1")); + assert!(!fence.commit_terminal("call", "attempt-1")); + + fence.register("call".to_string(), "attempt-2".to_string()); + assert_eq!(fence.attempt_id("call"), Some("attempt-2")); + assert!(!fence.can_commit("call", "attempt-1")); + assert!(!fence.commit_terminal("call", "attempt-1")); + assert!(fence.commit_terminal("call", "attempt-2")); + } + #[test] fn provider_projection_reorders_results_and_remaps_cache_anchor() { let items = vec![ diff --git a/crates/agen/src/lib.rs b/crates/agen/src/lib.rs index 871af33b..aac1727a 100644 --- a/crates/agen/src/lib.rs +++ b/crates/agen/src/lib.rs @@ -28,7 +28,9 @@ pub use handler::ToolUseBlockStart; pub use history::{History, HistoryEntry}; pub use interceptor::Interceptor; pub use message::{ContentPart, Item, Message, Role}; -pub use tool::{ToolCall, ToolExecutionContext, ToolOutputLimits, ToolResult}; +pub use tool::{ + ToolCall, ToolExecutionContext, ToolOutputLimits, ToolResult, ToolResultDisposition, +}; pub use usage_record::UsageRecord; /// Implementation dependencies used by code generated from `agen` macros. diff --git a/crates/agen/src/llm_client/types.rs b/crates/agen/src/llm_client/types.rs index dd2d3dc9..101e9773 100644 --- a/crates/agen/src/llm_client/types.rs +++ b/crates/agen/src/llm_client/types.rs @@ -9,7 +9,7 @@ use std::{fmt, sync::Arc}; -use crate::tool::Attachment; +use crate::tool::{Attachment, ToolResultDisposition}; use base64::Engine as _; use serde::{Deserialize, Serialize}; @@ -121,6 +121,9 @@ pub enum Item { /// Detailed output (removed by pruning when old enough) #[serde(default, skip_serializing_if = "Option::is_none")] content: Option, + /// Typed terminal state used for replay and recovery. + #[serde(default, skip_serializing_if = "ToolResultDisposition::is_success")] + disposition: ToolResultDisposition, /// Whether the tool result represents an execution error. #[serde(default, skip_serializing_if = "is_false")] is_error: bool, @@ -261,7 +264,17 @@ impl Item { content: Option, is_error: bool, ) -> Self { - Self::tool_result_item_with_attachments(call_id, summary, content, is_error, Vec::new()) + Self::tool_result_item_with_disposition_and_attachments( + call_id, + summary, + content, + if is_error { + ToolResultDisposition::Error + } else { + ToolResultDisposition::Success + }, + Vec::new(), + ) } /// Create a tool result item with durable, prunable structured attachments. @@ -272,11 +285,33 @@ impl Item { is_error: bool, attachments: Vec, ) -> Self { + Self::tool_result_item_with_disposition_and_attachments( + call_id, + summary, + content, + if is_error { + ToolResultDisposition::Error + } else { + ToolResultDisposition::Success + }, + attachments, + ) + } + + pub fn tool_result_item_with_disposition_and_attachments( + call_id: impl Into, + summary: impl Into, + content: Option, + disposition: ToolResultDisposition, + attachments: Vec, + ) -> Self { + let is_error = !disposition.is_success(); Self::ToolResult { id: None, call_id: call_id.into(), summary: summary.into(), content, + disposition, is_error, attachments, } diff --git a/crates/agen/src/tool.rs b/crates/agen/src/tool.rs index 6fe66fa6..0ae5e96d 100644 --- a/crates/agen/src/tool.rs +++ b/crates/agen/src/tool.rs @@ -23,6 +23,12 @@ pub enum ToolError { /// Internal error #[error("Internal error: {0}")] Internal(String), + /// Cooperative cancellation completed with bounded terminal output. + #[error("Tool execution cancelled")] + Cancelled(ToolOutput), + /// Execution was interrupted with a confirmed bounded terminal output. + #[error("Tool execution interrupted")] + Interrupted(ToolOutput), } // ============================================================================= @@ -158,6 +164,28 @@ pub enum Attachment { Image(ImageAttachment), } +/// Terminal disposition of one started tool call. +/// +/// `Cancelled` means the tool confirmed cancellation. `OutcomeUnknown` means +/// execution stopped without confirmation, so neither completion nor side +/// effects may be inferred. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub enum ToolResultDisposition { + #[default] + Success, + Error, + Interrupted, + Cancelled, + OutcomeUnknown, +} + +impl ToolResultDisposition { + pub const fn is_success(&self) -> bool { + matches!(self, Self::Success) + } +} + /// Tool execution result. /// /// Every output has a mandatory `summary` (1-2 lines) that persists in @@ -402,6 +430,17 @@ pub trait Tool: Send + Sync { input_json: &str, ctx: ToolExecutionContext, ) -> Result; + + /// Request cooperative cancellation for one started call. + /// + /// Implementations that own cancellable provider operations should signal + /// the exact execution identified by `call_id`, then let `execute` return + /// the confirmed bounded terminal output. The Engine applies a bounded + /// grace period and falls back to `OutcomeUnknown` when confirmation never + /// arrives. + async fn cancel(&self, _call_id: &str) -> Result<(), ToolError> { + Ok(()) + } } // ============================================================================= @@ -429,6 +468,9 @@ pub struct ToolCall { pub struct ToolResult { /// Corresponding tool call ID pub tool_use_id: String, + /// Typed terminal state. + #[serde(default, skip_serializing_if = "ToolResultDisposition::is_success")] + pub disposition: ToolResultDisposition, /// Short summary (always kept in history) pub summary: String, /// Detailed output (prunable) @@ -445,11 +487,20 @@ pub struct ToolResult { impl ToolResult { /// Create a success result from a [`ToolOutput`]. pub fn from_output(tool_use_id: impl Into, output: ToolOutput) -> Self { + Self::from_output_with_disposition(tool_use_id, output, ToolResultDisposition::Success) + } + + pub fn from_output_with_disposition( + tool_use_id: impl Into, + output: ToolOutput, + disposition: ToolResultDisposition, + ) -> Self { Self { tool_use_id: tool_use_id.into(), + disposition, summary: output.summary, content: output.content, - is_error: false, + is_error: !disposition.is_success(), attachments: output.attachments, } } @@ -458,12 +509,28 @@ impl ToolResult { pub fn error(tool_use_id: impl Into, message: impl Into) -> Self { Self { tool_use_id: tool_use_id.into(), + disposition: ToolResultDisposition::Error, summary: message.into(), content: None, is_error: true, attachments: Vec::new(), } } + + /// Close an execution whose completion and side effects cannot be confirmed. + pub fn outcome_unknown(tool_use_id: impl Into) -> Self { + Self { + tool_use_id: tool_use_id.into(), + disposition: ToolResultDisposition::OutcomeUnknown, + summary: "Tool execution outcome unknown".to_string(), + content: Some( + "Execution was interrupted before completion could be confirmed. Completion and side effects are unknown." + .to_string(), + ), + is_error: true, + attachments: Vec::new(), + } + } } #[cfg(test)] diff --git a/crates/agen/tests/parallel_execution_test.rs b/crates/agen/tests/parallel_execution_test.rs index 2141b55c..c9429682 100644 --- a/crates/agen/tests/parallel_execution_test.rs +++ b/crates/agen/tests/parallel_execution_test.rs @@ -10,6 +10,7 @@ use agen::interceptor::{Interceptor, PostToolAction, PreToolAction, ToolCallInfo use agen::llm_client::event::{Event, ResponseStatus, StatusEvent}; use agen::tool::{ Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput, ToolResult, + ToolResultDisposition, }; use agen::{Engine, History, Item}; use async_trait::async_trait; @@ -112,6 +113,53 @@ impl Tool for FirstAttemptHangsTool { } } +#[derive(Clone)] +struct CooperativeCancelTool { + calls: Arc, + cancelled: Arc, +} + +impl CooperativeCancelTool { + fn new() -> Self { + Self { + calls: Arc::new(AtomicUsize::new(0)), + cancelled: Arc::new(tokio::sync::Notify::new()), + } + } + + fn definition(&self) -> ToolDefinition { + let tool = self.clone(); + Arc::new(move || { + let meta = ToolMeta::new("cooperative") + .description("Returns bounded progress after cancellation") + .input_schema(serde_json::json!({"type": "object"})); + (meta, Arc::new(tool.clone()) as Arc) + }) + } +} + +#[async_trait] +impl Tool for CooperativeCancelTool { + async fn execute( + &self, + _input_json: &str, + _ctx: ToolExecutionContext, + ) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + self.cancelled.notified().await; + Err(ToolError::Cancelled(ToolOutput { + summary: "cooperative command cancelled".to_string(), + content: Some("stdout before cancellation\nstderr before cancellation".to_string()), + attachments: Vec::new(), + })) + } + + async fn cancel(&self, _call_id: &str) -> Result<(), ToolError> { + self.cancelled.notify_one(); + Ok(()) + } +} + #[derive(Clone)] struct ContextRecordingTool { name: String, @@ -269,6 +317,7 @@ async fn completed_results_commit_before_publish_without_waiting_for_siblings() let _ = engine .run_with_annotation(&mut history, "run both", &mut annotate) .await; + observed.lock().unwrap().push("run-returned".to_string()); assert_eq!( observed.lock().unwrap().as_slice(), @@ -277,6 +326,7 @@ async fn completed_results_commit_before_publish_without_waiting_for_siblings() "publish:call_fast", "commit:call_slow", "publish:call_slow", + "run-returned", ] ); @@ -308,9 +358,12 @@ async fn cancellation_preserves_completed_results_and_resume_skips_them() { Event::tool_use_start(0, "call_hang", "hang_once"), Event::tool_input_delta(0, r#"{}"#), Event::tool_use_stop(0), - Event::tool_use_start(1, "call_fast", "fast"), + Event::tool_use_start(1, "call_fast_a", "fast_a"), Event::tool_input_delta(1, r#"{}"#), Event::tool_use_stop(1), + Event::tool_use_start(2, "call_fast_b", "fast_b"), + Event::tool_input_delta(2, r#"{}"#), + Event::tool_use_stop(2), Event::Status(StatusEvent { status: ResponseStatus::Completed, }), @@ -326,9 +379,11 @@ async fn cancellation_preserves_completed_results_and_resume_skips_them() { ]); let mut engine = Engine::new(client); let hanging = FirstAttemptHangsTool::new(); - let fast = SlowTool::new("fast", 1); + let fast_a = SlowTool::new("fast_a", 1); + let fast_b = SlowTool::new("fast_b", 2); engine.register_tool(hanging.definition()); - engine.register_tool(fast.definition()); + engine.register_tool(fast_a.definition()); + engine.register_tool(fast_b.definition()); let cancel = engine.cancel_sender(); let cancel_task = tokio::spawn(async move { @@ -345,36 +400,182 @@ async fn cancellation_preserves_completed_results_and_resume_skips_them() { .filter(|entry| { matches!( &entry.item, - Item::ToolResult { call_id, .. } if call_id == "call_fast" + Item::ToolResult { call_id, .. } + if call_id == "call_fast_a" || call_id == "call_fast_b" ) }) .count(); - assert_eq!(completed_before_resume, 1); - assert_eq!(fast.call_count(), 1); + let unknown_before_resume = history + .iter() + .filter(|entry| { + matches!( + &entry.item, + Item::ToolResult { + call_id, + disposition: ToolResultDisposition::OutcomeUnknown, + .. + } if call_id == "call_hang" + ) + }) + .count(); + assert_eq!(completed_before_resume, 2); + assert_eq!(unknown_before_resume, 1); + assert_eq!(fast_a.call_count(), 1); + assert_eq!(fast_b.call_count(), 1); assert_eq!(hanging.call_count(), 1); let _ = engine.resume(&mut history).await; assert_eq!( - fast.call_count(), + fast_a.call_count(), + 1, + "completed call must not be re-executed" + ); + assert_eq!( + fast_b.call_count(), 1, "completed call must not be re-executed" ); assert_eq!( hanging.call_count(), - 2, - "only the unresolved call is retried" + 1, + "OutcomeUnknown is terminal and must not be re-executed" ); let completed_after_resume = history .iter() .filter(|entry| { matches!( &entry.item, - Item::ToolResult { call_id, .. } if call_id == "call_fast" + Item::ToolResult { call_id, .. } + if call_id == "call_fast_a" || call_id == "call_fast_b" ) }) .count(); - assert_eq!(completed_after_resume, 1); + assert_eq!(completed_after_resume, 2); + assert_eq!( + history + .iter() + .filter(|entry| { + matches!( + &entry.item, + Item::ToolResult { + call_id, + disposition: ToolResultDisposition::OutcomeUnknown, + .. + } if call_id == "call_hang" + ) + }) + .count(), + 1 + ); +} + +#[tokio::test] +async fn cooperative_cancellation_commits_bounded_terminal_output() { + let client = MockLlmClient::with_responses(vec![vec![ + Event::tool_use_start(0, "call_cooperative", "cooperative"), + Event::tool_input_delta(0, r#"{}"#), + Event::tool_use_stop(0), + Event::Status(StatusEvent { + status: ResponseStatus::Completed, + }), + ]]); + let mut engine = Engine::new(client); + let tool = CooperativeCancelTool::new(); + engine.register_tool(tool.definition()); + let observed = Arc::new(Mutex::new(Vec::<&'static str>::new())); + let published = observed.clone(); + engine.on_tool_result(move |_| published.lock().unwrap().push("published")); + let committed = observed.clone(); + let mut annotate = move |item: &Item| { + if matches!(item, Item::ToolResult { .. }) { + committed.lock().unwrap().push("committed"); + } + Ok(()) + }; + + let cancel = engine.cancel_sender(); + let cancel_task = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(30)).await; + cancel.send(()).await.unwrap(); + }); + let mut history = History::new(); + let output = engine + .run_with_annotation(&mut history, "start", &mut annotate) + .await; + observed.lock().unwrap().push("run-returned"); + cancel_task.await.unwrap(); + + assert_eq!( + observed.lock().unwrap().as_slice(), + ["committed", "published", "run-returned"] + ); + assert_eq!(tool.calls.load(Ordering::SeqCst), 1); + let terminal: Vec<_> = history + .iter() + .filter_map(|entry| match &entry.item { + Item::ToolResult { + call_id, + disposition, + content, + .. + } if call_id == "call_cooperative" => Some((*disposition, content.as_deref())), + _ => None, + }) + .collect(); + assert_eq!(terminal.len(), 1); + assert_eq!(terminal[0].0, ToolResultDisposition::Cancelled); + assert_eq!( + terminal[0].1, + Some("stdout before cancellation\nstderr before cancellation") + ); + assert!(matches!( + output.result, + agen::EngineRunExit::Interrupted(agen::StopReason::Cancelled) + )); +} + +#[tokio::test] +async fn cancellation_completion_race_commits_one_terminal_output() { + for iteration in 0..24u64 { + let client = MockLlmClient::with_responses(vec![ + vec![ + Event::tool_use_start(0, "call_racy", "racy"), + Event::tool_input_delta(0, r#"{}"#), + Event::tool_use_stop(0), + Event::Status(StatusEvent { + status: ResponseStatus::Completed, + }), + ], + vec![Event::Status(StatusEvent { + status: ResponseStatus::Completed, + })], + ]); + let mut engine = Engine::new(client); + let delay = 2 + iteration % 3; + let tool = SlowTool::new("racy", delay); + engine.register_tool(tool.definition()); + let cancel = engine.cancel_sender(); + let cancel_task = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(delay)).await; + let _ = cancel.send(()).await; + }); + + let mut history = History::new(); + let _ = engine.run(&mut history, "race").await; + cancel_task.await.unwrap(); + let terminal_count = history + .iter() + .filter(|entry| { + matches!( + &entry.item, + Item::ToolResult { call_id, .. } if call_id == "call_racy" + ) + }) + .count(); + assert_eq!(terminal_count, 1, "iteration {iteration}"); + assert_eq!(tool.call_count(), 1, "iteration {iteration}"); + } } #[tokio::test] diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index a9c3bce6..bfc1a49a 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -352,6 +352,18 @@ pub struct InternalWorkerSnapshot { pub internal_workers: Vec, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum ToolResultDisposition { + #[default] + Success, + Error, + Interrupted, + Cancelled, + OutcomeUnknown, +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[serde(tag = "event", content = "data", rename_all = "snake_case")] @@ -501,6 +513,8 @@ pub enum Event { /// summary-only, or when the result was pruned. #[serde(default, skip_serializing_if = "Option::is_none")] output: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + disposition: Option, #[serde(default)] is_error: bool, }, @@ -1839,6 +1853,7 @@ mod tests { id: "call_1".into(), summary: "Read 128 bytes".into(), output: Some("hello world".into()), + disposition: Some(ToolResultDisposition::Success), is_error: false, }; let json = serde_json::to_string(&event).unwrap(); @@ -1855,11 +1870,13 @@ mod tests { id, summary, output, + disposition, is_error, } => { assert_eq!(id, "call_1"); assert_eq!(summary, "Read 128 bytes"); assert_eq!(output.as_deref(), Some("hello world")); + assert_eq!(disposition, Some(ToolResultDisposition::Success)); assert!(!is_error); } other => panic!("expected ToolResult, got {other:?}"), @@ -1872,6 +1889,7 @@ mod tests { id: "call_2".into(), summary: "ok".into(), output: None, + disposition: Some(ToolResultDisposition::Success), is_error: false, }; let json = serde_json::to_string(&event).unwrap(); @@ -1887,6 +1905,7 @@ mod tests { id: "call_3".into(), summary: "invalid argument".into(), output: None, + disposition: Some(ToolResultDisposition::Error), is_error: true, }; let json = serde_json::to_string(&event).unwrap(); diff --git a/crates/protocol/src/typescript.rs b/crates/protocol/src/typescript.rs index 29dffce8..2ac5446c 100644 --- a/crates/protocol/src/typescript.rs +++ b/crates/protocol/src/typescript.rs @@ -8,7 +8,7 @@ use crate::{ CompletionKind, ErrorCode, Event, Greeting, InFlightBlock, InFlightSnapshot, InFlightToolCallState, InternalWorkerKind, InternalWorkerRef, InternalWorkerSnapshot, InvokeKind, MemoryWorkerEvent, Method, Permission, RewindSummary, RewindTarget, RewindTargetId, - RunResult, ScopeRule, Segment, TurnResult, WorkerEvent, WorkerStatus, + RunResult, ScopeRule, Segment, ToolResultDisposition, TurnResult, WorkerEvent, WorkerStatus, subscription::{ EventSubscriptionSelector, SubscriptionEvent, SubscriptionEventPayload, SubscriptionFrame, SubscriptionFramePayload, SubscriptionId, SubscriptionRejectionCode, SubscriptionRequest, @@ -45,6 +45,7 @@ pub fn generated_protocol_types() -> String { push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); diff --git a/crates/session-store/src/logged_item.rs b/crates/session-store/src/logged_item.rs index f2d499e6..f6547265 100644 --- a/crates/session-store/src/logged_item.rs +++ b/crates/session-store/src/logged_item.rs @@ -14,7 +14,7 @@ use agen::{ llm_client::types::{ContentPart, Item, Role}, - tool::{Attachment, ImageAttachment}, + tool::{Attachment, ImageAttachment, ToolResultDisposition}, }; use base64::{Engine as _, engine::general_purpose::STANDARD}; use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _}; @@ -61,6 +61,8 @@ pub enum LoggedItem { content: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] attachments: Vec, + #[serde(default, skip_serializing_if = "ToolResultDisposition::is_success")] + disposition: ToolResultDisposition, #[serde(default, skip_serializing_if = "is_false")] is_error: bool, }, @@ -128,6 +130,7 @@ impl From<&Item> for LoggedItem { summary, content, attachments, + disposition, is_error, .. } => Self::ToolResult { @@ -135,6 +138,7 @@ impl From<&Item> for LoggedItem { summary: summary.clone(), content: content.clone(), attachments: attachments.iter().map(LoggedAttachment::from).collect(), + disposition: *disposition, is_error: *is_error, }, Item::Reasoning { @@ -184,15 +188,24 @@ impl From for Item { summary, content, attachments, + disposition, is_error, - } => Item::ToolResult { - id: None, - call_id, - summary, - content, - is_error, - attachments: attachments.into_iter().map(Attachment::from).collect(), - }, + } => { + let disposition = if is_error && disposition.is_success() { + ToolResultDisposition::Error + } else { + disposition + }; + Item::ToolResult { + id: None, + call_id, + summary, + content, + disposition, + is_error, + attachments: attachments.into_iter().map(Attachment::from).collect(), + } + } LoggedItem::Reasoning { text, summary, @@ -430,6 +443,42 @@ mod tests { } } + #[test] + fn outcome_unknown_tool_result_round_trips_as_terminal() { + let original = Item::tool_result_item_with_disposition_and_attachments( + "call_unknown", + "outcome unknown", + Some("bounded progress".to_string()), + ToolResultDisposition::OutcomeUnknown, + Vec::new(), + ); + let logged: LoggedItem = (&original).into(); + let json = serde_json::to_string(&logged).unwrap(); + assert!(json.contains(r#""disposition":"outcome_unknown""#)); + match Item::from(serde_json::from_str::(&json).unwrap()) { + Item::ToolResult { + disposition, + is_error, + .. + } => { + assert_eq!(disposition, ToolResultDisposition::OutcomeUnknown); + assert!(is_error); + } + other => panic!("unexpected variant: {other:?}"), + } + } + + #[test] + fn legacy_error_tool_result_infers_error_disposition() { + let legacy = r#"{"kind":"tool_result","call_id":"call_old","summary":"failed","content":null,"is_error":true}"#; + match Item::from(serde_json::from_str::(legacy).unwrap()) { + Item::ToolResult { disposition, .. } => { + assert_eq!(disposition, ToolResultDisposition::Error) + } + other => panic!("unexpected variant: {other:?}"), + } + } + #[test] fn tool_result_persistence_round_trips_binary_attachments() { let original = Item::tool_result_item_with_attachments( diff --git a/crates/tools/src/bash.rs b/crates/tools/src/bash.rs index 8224472b..09cceda7 100644 --- a/crates/tools/src/bash.rs +++ b/crates/tools/src/bash.rs @@ -1,5 +1,6 @@ +use std::collections::{HashMap, HashSet}; use std::path::PathBuf; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; use async_trait::async_trait; @@ -20,15 +21,28 @@ struct BashParams { pub(crate) struct BashTool { session: WorkdirSessionHandle, + state: Arc>, +} + +#[derive(Default)] +struct BashExecutionState { + active: HashMap, + cancellation_requested: HashSet, } struct CommandGuard { session: WorkdirSessionHandle, + state: Arc>, + call_id: String, handle: Option, } impl Drop for CommandGuard { fn drop(&mut self) { + let mut state = self.state.lock().unwrap(); + state.active.remove(&self.call_id); + state.cancellation_requested.remove(&self.call_id); + drop(state); if let Some(handle) = self.handle.take() { let workdir = self.session.clone(); tokio::spawn(async move { @@ -52,20 +66,35 @@ impl Tool for BashTool { .unwrap_or(DEFAULT_TIMEOUT_SECS) .clamp(1, MAX_TIMEOUT_SECS); let cmd_summary = truncate_for_summary(¶ms.command); + let call_id = ctx.call_id; + let mut guard = CommandGuard { + session: self.session.clone(), + state: self.state.clone(), + call_id: call_id.clone(), + handle: None, + }; let handle = self .session .start_command(CommandRequest { command: params.command, timeout_secs, output_limit: INLINE_BYTE_BUDGET, - tool_call_id: Some(ctx.call_id), + tool_call_id: Some(call_id.clone()), }) .await .map_err(crate::ToolsError::from)?; - let mut guard = CommandGuard { - session: self.session.clone(), - handle: Some(handle.clone()), + let cancel_after_start = { + let mut state = self.state.lock().unwrap(); + state.active.insert(call_id.clone(), handle.clone()); + state.cancellation_requested.contains(&call_id) }; + guard.handle = Some(handle.clone()); + if cancel_after_start { + self.session + .cancel_command(handle.clone()) + .await + .map_err(crate::ToolsError::from)?; + } let output = self .session .command_output(CommandOutputRequest { @@ -76,9 +105,17 @@ impl Tool for BashTool { }) .await .map_err(crate::ToolsError::from)?; + let cancellation_requested = { + let mut state = self.state.lock().unwrap(); + state.active.remove(&call_id); + state.cancellation_requested.remove(&call_id) + }; guard.handle = None; - let summary = if output.timed_out { + let timed_out = output.timed_out; + let summary = if cancellation_requested { + format!("$ {cmd_summary} (cancelled)") + } else if output.timed_out { format!("$ {cmd_summary} (timed out after {timeout_secs}s)") } else { match output.exit_code { @@ -97,11 +134,33 @@ impl Tool for BashTool { } else { Some(output.content) }; - Ok(ToolOutput { + let output = ToolOutput { summary, content, attachments: Vec::new(), - }) + }; + if cancellation_requested { + Err(ToolError::Cancelled(output)) + } else if timed_out { + Err(ToolError::Interrupted(output)) + } else { + Ok(output) + } + } + + async fn cancel(&self, call_id: &str) -> Result<(), ToolError> { + let handle = { + let mut state = self.state.lock().unwrap(); + state.cancellation_requested.insert(call_id.to_string()); + state.active.get(call_id).cloned() + }; + if let Some(handle) = handle { + self.session + .cancel_command(handle) + .await + .map_err(crate::ToolsError::from)?; + } + Ok(()) } } @@ -123,6 +182,7 @@ pub fn bash_tool(session: WorkdirSessionHandle, _output_dir: PathBuf) -> ToolDef .input_schema(serde_json::to_value(schema).expect("Bash schema serialization")); let tool: Arc = Arc::new(BashTool { session: session.clone(), + state: Arc::new(Mutex::new(BashExecutionState::default())), }); (meta, tool) }) diff --git a/crates/tools/tests/integration.rs b/crates/tools/tests/integration.rs index f8761aa8..55291d52 100644 --- a/crates/tools/tests/integration.rs +++ b/crates/tools/tests/integration.rs @@ -7,7 +7,7 @@ use std::path::Path; use std::sync::Arc; -use agen::tool::{Tool, ToolDefinition, ToolMeta}; +use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta}; use manifest::{Permission, Scope, ScopeConfig, ScopeRule}; use serde_json::json; use tempfile::TempDir; @@ -401,5 +401,45 @@ async fn bash_provider_output_does_not_expose_internal_paths() { assert_eq!(std::fs::read_dir(spill.path()).unwrap().count(), 0); } +#[tokio::test] +async fn bash_cancellation_returns_bounded_progress_as_terminal_output() { + let (_dir, _spill, reg) = setup(); + let bash = reg.get("Bash"); + let executing = bash.clone(); + let execution = tokio::spawn(async move { + executing + .execute( + r#"{"command":"printf 'before\\n'; printf 'err-before\\n' >&2; sleep 5; printf 'after\\n'"}"#, + Default::default(), + ) + .await + }); + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + bash.cancel("direct").await.expect("signal cancellation"); + let error = tokio::time::timeout(std::time::Duration::from_secs(2), execution) + .await + .expect("cancelled Bash should terminate inside the Engine grace budget") + .expect("Bash task join"); + + let ToolError::Cancelled(output) = error.expect_err("cancelled command is non-success") else { + panic!("expected typed cancellation result"); + }; + let content = output.content.expect("bounded progress output"); + assert!( + content.contains("before"), + "missing pre-cancel stdout: {content}" + ); + assert!( + content.contains("err-before"), + "missing pre-cancel stderr: {content}" + ); + assert!( + !content.contains("after"), + "post-cancel output leaked: {content}" + ); + assert!(content.len() <= 16 * 1024, "output must remain bounded"); +} + // Sanity: unused Path import guard const _: fn() -> &'static Path = || Path::new("/"); diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index 2a8ee1bb..fd8b7bc9 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -1244,6 +1244,7 @@ impl App { id, summary, output, + disposition: _, is_error, } => { self.latest_llm_wait_event = None; diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 7ad5f772..c0e48ead 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -763,6 +763,19 @@ pub(crate) fn wire_event_bridges_on_engine( id: result.tool_use_id.clone(), summary: result.summary.clone(), output: result.content.clone(), + disposition: Some(match result.disposition { + agen::ToolResultDisposition::Success => protocol::ToolResultDisposition::Success, + agen::ToolResultDisposition::Error => protocol::ToolResultDisposition::Error, + agen::ToolResultDisposition::Interrupted => { + protocol::ToolResultDisposition::Interrupted + } + agen::ToolResultDisposition::Cancelled => { + protocol::ToolResultDisposition::Cancelled + } + agen::ToolResultDisposition::OutcomeUnknown => { + protocol::ToolResultDisposition::OutcomeUnknown + } + }), is_error: result.is_error, }); }); diff --git a/crates/worker/src/interrupt_prep.rs b/crates/worker/src/interrupt_prep.rs index 6e359828..b02a7284 100644 --- a/crates/worker/src/interrupt_prep.rs +++ b/crates/worker/src/interrupt_prep.rs @@ -13,7 +13,7 @@ #[cfg(test)] use crate::prompt::catalog::PromptCatalog; -use agen::Item; +use agen::{Item, ToolResultDisposition}; /// Build synthetic `Item::ToolResult` items for every unanswered /// `Item::ToolCall` in `history`, preserving order. @@ -28,7 +28,16 @@ pub(crate) fn orphan_tool_result_closures(history: &[Item], summary: &str) -> Ve for item in history { if let Item::ToolCall { call_id, .. } = item { if !answered.contains(call_id.as_str()) { - out.push(Item::tool_result(call_id.clone(), summary)); + out.push(Item::tool_result_item_with_disposition_and_attachments( + call_id.clone(), + summary, + Some( + "Execution ended before completion could be confirmed. Completion and side effects are unknown." + .to_string(), + ), + ToolResultDisposition::OutcomeUnknown, + Vec::new(), + )); } } } @@ -77,10 +86,12 @@ mod tests { Item::ToolResult { call_id, summary: got, + disposition, .. } => { assert_eq!(call_id, "c1"); assert_eq!(got, &summary); + assert_eq!(*disposition, ToolResultDisposition::OutcomeUnknown); } other => panic!("expected ToolResult, got {other:?}"), } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 8bd5cd25..90d12cdb 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -2953,50 +2953,52 @@ impl Worker { Ok(()) } - /// Stage the post-interruption cleanup at the front of worker - /// history: close every unanswered `Item::ToolCall` with a synthetic - /// `Item::ToolResult` (Anthropic wire-validity), then append a - /// system note so the LLM understands the prior turn was cut - /// short. Called from `Worker::run` when the worker's - /// `last_run_interrupted` flag is set (i.e. the Worker just transitioned - /// out of Paused via a new user input). - fn apply_interrupt_prep(&mut self) -> Result<(), WorkerError> { + /// Durably close every unanswered ToolCall before the interrupted run's + /// final lifecycle record/status is published. + fn terminalize_orphan_tool_calls(&mut self) -> Result<(), WorkerError> { let tool_result_summary = self .prompts() .load_full() .interrupt_tool_result_summary() .map_err(WorkerError::from)?; + let history_items = self.history(); + let closures = crate::interrupt_prep::orphan_tool_result_closures( + &history_items, + &tool_result_summary, + ); + if closures.is_empty() { + return Ok(()); + } + + let subject = worker_subject(self.session.session_id()); + for item in closures { + let entry = HistoryEntry::new( + item, + new_history_metadata( + WorkerHistoryProvenance::ToolOutput { + worker: subject.clone(), + }, + None, + ), + ); + self.commit_entry(LogEntry::AnnotatedToolResult { + ts: segment_log::now_millis(), + entry: to_logged_history_entry(&entry), + })?; + self.session.history_mut().push_entry(entry); + self.session.note_mutation(); + } + Ok(()) + } + + fn apply_interrupt_prep(&mut self) -> Result<(), WorkerError> { + self.terminalize_orphan_tool_calls()?; let system_note = self .prompts() .load_full() .interrupt_system_note() .map_err(WorkerError::from)?; - let history_items = self.history(); - let closures = crate::interrupt_prep::orphan_tool_result_closures( - &history_items, - &tool_result_summary, - ); - if !closures.is_empty() { - let subject = worker_subject(self.session.session_id()); - for item in closures { - let entry = HistoryEntry::new( - item, - new_history_metadata( - WorkerHistoryProvenance::ToolOutput { - worker: subject.clone(), - }, - None, - ), - ); - self.commit_entry(LogEntry::AnnotatedToolResult { - ts: segment_log::now_millis(), - entry: to_logged_history_entry(&entry), - })?; - self.session.history_mut().push_entry(entry); - self.session.note_mutation(); - } - } let interrupt_prompt_provenance = self.prompt_render_provenance("internal.interrupt_system_note"); let interrupt_metadata = new_history_metadata( @@ -3262,6 +3264,12 @@ impl Worker { where St: Clone + 'static, { + if matches!( + &result, + EngineRunExit::Paused | EngineRunExit::Interrupted(_) + ) { + self.terminalize_orphan_tool_calls()?; + } self.persist_turn(history_before, &result).await?; if matches!(result, EngineRunExit::Yielded) { @@ -5928,7 +5936,8 @@ fn stop_reason_error_code(reason: &StopReason) -> ErrorCode { EngineError::Aborted(_) | EngineError::Cancelled | EngineError::ConfigWarnings(_) - | EngineError::HistoryAppend(_), + | EngineError::HistoryAppend(_) + | EngineError::ToolAttemptFence(_), ) => ErrorCode::Internal, } } @@ -7862,6 +7871,7 @@ mod build_summary_prompt_tests { summary: "wrote a file".into(), content: None, attachments: Vec::new(), + disposition: Default::default(), is_error: false, }, }, @@ -7904,6 +7914,7 @@ mod build_summary_prompt_tests { summary: "wrote a file".into(), content: None, attachments: Vec::new(), + disposition: Default::default(), is_error: false, }, }, @@ -7948,6 +7959,7 @@ mod build_summary_prompt_tests { summary: "side effect".into(), content: None, attachments: Vec::new(), + disposition: Default::default(), is_error: false, }, metadata: new_history_metadata( @@ -8047,6 +8059,74 @@ mod build_summary_prompt_tests { assert!(err.contains("session head changed")); } + #[tokio::test] + async fn interrupted_result_terminalizes_orphan_before_run_completed() { + let dir = tempfile::tempdir().unwrap(); + let manifest = minimal_manifest(); + let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap(); + let cwd = dir.path().join("workspace"); + std::fs::create_dir_all(&cwd).unwrap(); + let scope = Scope::writable(&cwd).unwrap(); + let authority = WorkerFilesystemAuthority::local(cwd.clone(), cwd.clone()); + let mut worker = Worker::new( + manifest, + Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient), + store, + WorkerWorkspaceContext::local_filesystem(None), + authority, + scope, + ) + .await + .unwrap(); + + worker.ensure_segment_head().unwrap(); + worker.wire_history_persistence(); + worker.set_history_for_test(vec![Item::tool_call("call-1", "Bash", "{}")]); + let _ = worker + .handle_worker_result( + EngineRunExit::Interrupted(StopReason::Cancelled), + worker.history().len(), + ) + .await + .unwrap(); + + let entries = worker + .store + .read_all( + worker.segment_state.session_id(), + worker.segment_state.segment_id(), + ) + .unwrap(); + let terminal_index = entries + .iter() + .position(|entry| { + matches!( + entry, + LogEntry::AnnotatedToolResult { + entry: session_store::LoggedHistoryEntry { + item: session_store::LoggedItem::ToolResult { + disposition: agen::ToolResultDisposition::OutcomeUnknown, + .. + }, + .. + }, + .. + } + ) + }) + .expect("durable OutcomeUnknown closure"); + let final_index = entries + .iter() + .position(|entry| { + matches!( + entry, + LogEntry::RunCompleted { .. } | LogEntry::RunErrored { .. } + ) + }) + .expect("durable final run status"); + assert!(terminal_index < final_index); + } + #[tokio::test] async fn apply_interrupt_prep_appends_via_callback_and_logs_independent_entries() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index d58cfd04..e4ef5b31 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -2169,9 +2169,13 @@ async fn paused_then_run_closes_orphan_tool_use_for_next_request() { for item in items { match item { agen::Item::ToolResult { - call_id, summary, .. + call_id, + summary, + disposition, + .. } if call_id == "call_orphan" => { - assert_eq!(summary, "[Interrupted by user]"); + assert_eq!(summary, "Tool execution outcome unknown"); + assert_eq!(*disposition, agen::ToolResultDisposition::OutcomeUnknown); saw_synthetic_tool_result = true; } agen::Item::Message { role, content, .. } if *role == agen::Role::System => { @@ -2362,8 +2366,11 @@ async fn paused_cancel_abandons_resume_and_next_input_is_fresh_run() { assert!( items.iter().any(|item| matches!( item, - agen::Item::ToolResult { call_id, summary, .. } - if call_id == "call_cancelled" && summary == "[Interrupted by user]" + agen::Item::ToolResult { + call_id, + disposition: agen::ToolResultDisposition::OutcomeUnknown, + .. + } if call_id == "call_cancelled" )), "paused cancel should close orphan tool_use before future requests: {items:?}" ); diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 1e00a828..6a357acb 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -16,6 +16,8 @@ export type InvokeKind = "user_send" | "notify" | "worker_event" | "system_remin export type RunResult = "finished" | "paused" | "limit_reached" | "rolled_back"; +export type ToolResultDisposition = "success" | "error" | "interrupted" | "cancelled" | "outcome_unknown"; + export type ErrorCode = "already_running" | "not_running" | "not_paused" | "provider_error" | "tool_error" | "invalid_request" | "internal"; export type Permission = "read" | "write"; @@ -191,7 +193,7 @@ summary: string, * Full tool output. Absent when the tool chose to return * summary-only, or when the result was pruned. */ -output?: string | null, is_error: boolean, } } | { "event": "usage", "data": { input_tokens: number | null, output_tokens: number | null, cache_read_input_tokens?: number | null, } } | { "event": "run_end", "data": { result: RunResult, } } | { "event": "error", "data": { code: ErrorCode, message: string, } } | { "event": "snapshot", "data": { entries: Array, greeting: Greeting, status: WorkerStatus, +output?: string | null, disposition?: ToolResultDisposition | null, is_error: boolean, } } | { "event": "usage", "data": { input_tokens: number | null, output_tokens: number | null, cache_read_input_tokens?: number | null, } } | { "event": "run_end", "data": { result: RunResult, } } | { "event": "error", "data": { code: ErrorCode, message: string, } } | { "event": "snapshot", "data": { entries: Array, greeting: Greeting, status: WorkerStatus, /** * Unfinished model output that has already streamed in the current * run but is not yet represented by committed snapshot entries.