From 1515a2fb86f30f60ee8597d10f103a1b252b80d6 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 27 Aug 2026 15:12:05 +0900 Subject: [PATCH] fix: reconcile typed history with run exits --- crates/agen/README.md | 4 +- crates/agen/examples/engine_cancel_demo.rs | 20 +-- crates/agen/examples/engine_cli.rs | 32 ++-- crates/agen/tests/annotated_history_test.rs | 5 +- crates/agen/tests/callback_test.rs | 36 ++-- crates/agen/tests/engine_fixtures.rs | 8 +- crates/agen/tests/engine_state_test.rs | 168 +++++++----------- crates/agen/tests/parallel_execution_test.rs | 5 +- .../agen/tests/reasoning_round_trip_test.rs | 8 +- crates/session-store/tests/session_test.rs | 2 +- crates/worker/src/worker.rs | 51 +++--- 11 files changed, 144 insertions(+), 195 deletions(-) diff --git a/crates/agen/README.md b/crates/agen/README.md index f9dba8c9..13478848 100644 --- a/crates/agen/README.md +++ b/crates/agen/README.md @@ -32,10 +32,10 @@ async fn conversation(client: C) -> Result<(), EngineError> { let output = Engine::new(client) .system_prompt("You are a concise assistant.") .run(&mut history, "Explain typed state in one sentence.") - .await?; + .await; let mut engine = output.engine; - let _result = engine.run(&mut history, "Give a Rust example.").await?; + let _result = engine.run(&mut history, "Give a Rust example.").await; Ok(()) } ``` diff --git a/crates/agen/examples/engine_cancel_demo.rs b/crates/agen/examples/engine_cancel_demo.rs index 5f9d9874..0ee6f0dc 100644 --- a/crates/agen/examples/engine_cancel_demo.rs +++ b/crates/agen/examples/engine_cancel_demo.rs @@ -4,7 +4,7 @@ use agen::llm_client::scheme::{Scheme, anthropic::AnthropicScheme}; use agen::llm_client::transport::{HttpTransport, ResolvedAuth}; -use agen::{Engine, EngineResult, History}; +use agen::{Engine, EngineRunExit, StopReason}; use std::time::Duration; #[tokio::main] @@ -29,7 +29,7 @@ async fn main() -> Result<(), Box> { let base_url = scheme.default_base_url().to_string(); let client = HttpTransport::new(scheme, model, base_url, ResolvedAuth::ApiKey(api_key), cap); let engine = Engine::new(client); - let mut history = History::new(); + let mut history = agen::History::new(); println!("πŸš€ Starting Engine..."); println!("πŸ’‘ Will cancel after 2 seconds\n"); @@ -46,15 +46,13 @@ async fn main() -> Result<(), Box> { println!("πŸ“‘ Sending request to LLM..."); - match engine.run(&mut history, "Tell me a very long story about a brave knight. Make it as detailed as possible with many paragraphs.").await { - Ok(out) => match out.result { - EngineResult::Finished => println!("βœ… Task completed normally"), - EngineResult::Paused => println!("⏸️ Task paused"), - EngineResult::LimitReached => println!("πŸ”’ Turn limit reached"), - EngineResult::Yielded => println!("↩️ Task yielded"), - }, - Err(e) => { - println!("❌ Task error: {}", e); + let output = engine.run(&mut history, "Tell me a very long story about a brave knight. Make it as detailed as possible with many paragraphs.").await; + match output.result { + EngineRunExit::Finished => println!("βœ… Task completed normally"), + EngineRunExit::Paused => println!("⏸️ Task paused"), + EngineRunExit::Yielded => println!("↩️ Task yielded"), + EngineRunExit::Interrupted(StopReason::LimitReached) => { + println!("πŸ”’ Turn limit reached") } EngineRunExit::Interrupted(reason) => println!("❌ Task interrupted: {reason:?}"), } diff --git a/crates/agen/examples/engine_cli.rs b/crates/agen/examples/engine_cli.rs index 1c9abfd2..9fa2e143 100644 --- a/crates/agen/examples/engine_cli.rs +++ b/crates/agen/examples/engine_cli.rs @@ -39,7 +39,7 @@ use tracing::info; use tracing_subscriber::EnvFilter; use agen::{ - Engine, History, + Engine, EngineRunExit, StopReason, interceptor::{Interceptor, PostToolAction, ToolResultInfo}, llm_client::{ LlmClient, @@ -451,6 +451,7 @@ async fn main() -> Result<(), Box> { // Create Engine let mut engine = Engine::new(client); + let mut history = agen::History::new(); let tool_call_names = Arc::new(Mutex::new(HashMap::new())); @@ -474,16 +475,11 @@ async fn main() -> Result<(), Box> { engine.set_interceptor(ToolResultPrinterPolicy::new(tool_call_names)); - let mut history = History::new(); - // One-shot mode if let Some(prompt) = args.prompt { - match engine.run(&mut history, &prompt).await { - Ok(_) => {} - Err(e) => { - eprintln!("\n❌ Error: {}", e); - std::process::exit(1); - } + let output = engine.run(&mut history, &prompt).await; + if let EngineRunExit::Interrupted(StopReason::Unexpected(error)) = output.result { + eprintln!("\n❌ Error: {error}"); } return Ok(()); @@ -502,13 +498,8 @@ async fn main() -> Result<(), Box> { return Ok(()); } - let mut locked = match engine.run(&mut history, first_input).await { - Ok(out) => out.engine, - Err(e) => { - eprintln!("\n❌ Error: {}", e); - return Ok(()); - } - }; + let output = engine.run(&mut history, first_input).await; + let mut locked = output.engine; loop { print!("\nπŸ‘€ You: "); @@ -527,11 +518,10 @@ async fn main() -> Result<(), Box> { break; } - match locked.run(&mut history, input).await { - Ok(_) => {} - Err(e) => { - eprintln!("\n❌ Error: {}", e); - } + if let EngineRunExit::Interrupted(StopReason::Unexpected(error)) = + locked.run(&mut history, input).await + { + eprintln!("\n❌ Error: {error}"); } } diff --git a/crates/agen/tests/annotated_history_test.rs b/crates/agen/tests/annotated_history_test.rs index bb0872e9..09a5019d 100644 --- a/crates/agen/tests/annotated_history_test.rs +++ b/crates/agen/tests/annotated_history_test.rs @@ -38,10 +38,9 @@ async fn run_preserves_item_annotations_without_projecting_them() { let output = engine .run_with_annotation(&mut history, "hello", &mut annotate) - .await - .unwrap(); + .await; - assert!(matches!(output.result, agen::EngineResult::Finished)); + assert!(matches!(output.result, agen::EngineRunExit::Finished)); assert_eq!(history.len(), 2); assert_eq!(history.entries()[0].annotation, "1:user"); assert_eq!(history.entries()[1].annotation, "2:assistant"); diff --git a/crates/agen/tests/callback_test.rs b/crates/agen/tests/callback_test.rs index 43d91053..1c3535f8 100644 --- a/crates/agen/tests/callback_test.rs +++ b/crates/agen/tests/callback_test.rs @@ -8,11 +8,11 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; +use agen::Engine; use agen::llm_client::event::{Event, ResponseStatus, StatusEvent as ClientStatusEvent}; use agen::llm_client::retry::RetryPolicy; use agen::llm_client::{ClientError, LlmClient, Request, ResponseStream}; use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; -use agen::{Engine, History}; use async_trait::async_trait; use common::MockLlmClient; @@ -58,7 +58,7 @@ async fn test_callback_llm_retry_event() { max_attempts: 2, total_timeout: Duration::from_secs(1), }); - let mut history: History = History::new(); + let mut history = agen::History::new(); let notices = Arc::new(Mutex::new(Vec::new())); let sink = notices.clone(); @@ -67,7 +67,10 @@ async fn test_callback_llm_retry_event() { }); let result = engine.run(&mut history, "retry once").await; - assert!(result.is_ok(), "engine should succeed after one retry"); + assert!( + matches!(result.result, agen::EngineRunExit::Finished), + "engine should succeed after one retry" + ); let notices = notices.lock().unwrap(); assert_eq!(notices.len(), 1); @@ -92,7 +95,7 @@ async fn test_callback_text_block_events() { let client = MockLlmClient::new(events); let mut engine = Engine::new(client); - let mut history: History = History::new(); + let mut history = agen::History::new(); let text_deltas = Arc::new(Mutex::new(Vec::new())); let text_completes = Arc::new(Mutex::new(Vec::new())); @@ -110,9 +113,12 @@ async fn test_callback_text_block_events() { }); }); - // Mutable::run consumes self, returns (Locked, EngineResult) + // Mutable::run consumes self, returns (Locked, EngineRunExit) let result = engine.run(&mut history, "Greet me").await; - assert!(result.is_ok(), "Engine should complete"); + assert!( + matches!(result.result, agen::EngineRunExit::Finished), + "Engine should complete" + ); let deltas = text_deltas.lock().unwrap(); assert_eq!(deltas.len(), 2); @@ -139,7 +145,7 @@ async fn test_callback_tool_call_complete() { let client = MockLlmClient::new(events); let mut engine = Engine::new(client); - let mut history: History = History::new(); + let mut history = agen::History::new(); let tool_starts = Arc::new(Mutex::new(Vec::<(String, String)>::new())); let tool_completes = Arc::new(Mutex::new(Vec::new())); @@ -157,7 +163,7 @@ async fn test_callback_tool_call_complete() { }); }); - // Mutable::run consumes self, returns (Locked, EngineResult) + // Mutable::run consumes self, returns (Locked, EngineRunExit) let _ = engine.run(&mut history, "Weather please").await; let starts = tool_starts.lock().unwrap(); @@ -186,7 +192,7 @@ async fn test_callback_turn_events() { let client = MockLlmClient::new(events); let mut engine = Engine::new(client); - let mut history: History = History::new(); + let mut history = agen::History::new(); let turn_starts = Arc::new(Mutex::new(Vec::new())); let turn_ends = Arc::new(Mutex::new(Vec::new())); @@ -201,9 +207,9 @@ async fn test_callback_turn_events() { ends.lock().unwrap().push(turn); }); - // Mutable::run consumes self, returns (Locked, EngineResult) + // Mutable::run consumes self, returns (Locked, EngineRunExit) let result = engine.run(&mut history, "Do something").await; - assert!(result.is_ok()); + assert!(matches!(result.result, agen::EngineRunExit::Finished)); let starts = turn_starts.lock().unwrap(); let ends = turn_ends.lock().unwrap(); @@ -258,7 +264,7 @@ async fn test_callback_tool_result_events() { let client = MockLlmClient::new(events); let mut engine = Engine::new(client); - let mut history: History = History::new(); + let mut history = agen::History::new(); engine.register_tool(fixed_tool( "fixed", @@ -335,7 +341,7 @@ async fn test_callback_tool_result_error_path() { let client = MockLlmClient::new(events); let mut engine = Engine::new(client); - let mut history: History = History::new(); + let mut history = agen::History::new(); engine.register_tool(erroring_tool("erroring", "boom")); @@ -380,7 +386,7 @@ async fn test_callback_usage_events() { let client = MockLlmClient::new(events); let mut engine = Engine::new(client); - let mut history: History = History::new(); + let mut history = agen::History::new(); let usage_events = Arc::new(Mutex::new(Vec::new())); @@ -389,7 +395,7 @@ async fn test_callback_usage_events() { usages.lock().unwrap().push(event.clone()); }); - // Mutable::run consumes self, returns (Locked, EngineResult) + // Mutable::run consumes self, returns (Locked, EngineRunExit) let _ = engine.run(&mut history, "Hello").await; let usages = usage_events.lock().unwrap(); diff --git a/crates/agen/tests/engine_fixtures.rs b/crates/agen/tests/engine_fixtures.rs index c5471beb..e3f63e97 100644 --- a/crates/agen/tests/engine_fixtures.rs +++ b/crates/agen/tests/engine_fixtures.rs @@ -9,8 +9,8 @@ use std::path::Path; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; +use agen::Engine; use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; -use agen::{Engine, History}; use async_trait::async_trait; use common::MockLlmClient; @@ -134,7 +134,7 @@ async fn test_engine_simple_text_response() { let client = MockLlmClient::from_fixture(&fixture_path).unwrap(); let engine = Engine::new(client); - let mut history: History = History::new(); + let mut history = agen::History::new(); // Send a simple message (Mutable::run consumes self, returns tuple) let result = engine.run(&mut history, "Hello").await; @@ -160,7 +160,7 @@ async fn test_engine_tool_call() { let client = MockLlmClient::from_fixture(&fixture_path).unwrap(); let mut engine = Engine::new(client); - let mut history: History = History::new(); + let mut history = agen::History::new(); // Register tool let weather_tool = MockWeatherTool::new(); @@ -202,7 +202,7 @@ async fn test_engine_with_programmatic_events() { let client = MockLlmClient::new(events); let engine = Engine::new(client); - let mut history: History = History::new(); + let mut history = agen::History::new(); // Mutable::run consumes self, returns tuple let result = engine.run(&mut history, "Greet me").await; diff --git a/crates/agen/tests/engine_state_test.rs b/crates/agen/tests/engine_state_test.rs index c72f136d..c1bfe966 100644 --- a/crates/agen/tests/engine_state_test.rs +++ b/crates/agen/tests/engine_state_test.rs @@ -12,45 +12,12 @@ use agen::Item; use agen::interceptor::{ Interceptor, PreRequestAction, PreToolAction, ToolCallInfo, TurnEndAction, }; -use agen::llm_client::ClientError; use agen::llm_client::event::{Event, ResponseStatus, StatusEvent}; use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; -use agen::{Engine, EngineError, EngineResult, History}; +use agen::{Engine, EngineError, EngineRunExit, History, StopReason}; use async_trait::async_trait; use common::MockLlmClient; -#[test] -fn engine_source_has_no_worker_owned_interruption_marker() { - let source = include_str!("../src/engine.rs"); - assert!(!source.contains("last_run_interrupted")); - assert!(!source.contains("set_last_run_interrupted")); -} - -#[test] -fn run_exit_classifies_known_and_unexpected_stops_without_message_parsing() { - assert!(matches!( - EngineRunExit::from(Err(EngineError::Cancelled)), - EngineRunExit::Interrupted(StopReason::Cancelled) - )); - assert!(matches!( - EngineRunExit::from(Err(EngineError::Client(ClientError::ContextWindowExceeded))), - EngineRunExit::Interrupted(StopReason::ContextWindowExceeded) - )); - - let message_only = EngineError::Client(ClientError::Api { - status: Some(400), - code: None, - message: "context_length_exceeded".to_string(), - retry_after: None, - }); - assert!(matches!( - EngineRunExit::from(Err(message_only)), - EngineRunExit::Interrupted(StopReason::Unexpected(EngineError::Client( - ClientError::Api { .. } - ))) - )); -} - // ============================================================================= // Mutable State Tests // ============================================================================= @@ -235,13 +202,11 @@ async fn history_append_failure_stops_before_tool_execution() { }); let mut engine = engine.lock(&history); - let error = engine.run(&mut history, "use the tool").await.unwrap_err(); + let exit = engine.run(&mut history, "use the tool").await; - assert!(matches!( - exit, - EngineRunExit::Interrupted(StopReason::Unexpected(EngineError::HistoryAppend(ref message))) - if message == "simulated ENOSPC" - )); + assert!( + matches!(exit, EngineRunExit::Interrupted(StopReason::Unexpected(EngineError::HistoryAppend(ref message))) if message == "simulated ENOSPC") + ); assert_eq!(tool.call_count(), 0); assert_eq!(history.len(), 1); assert_eq!(history.entries()[0].item.as_text(), Some("use the tool")); @@ -319,7 +284,7 @@ async fn test_mutable_run_updates_history() -> Result<(), EngineError> { let mut history: History = History::new(); // Execute (Mutable::run consumes self, returns EngineRunOutput) - let _out = engine.run(&mut history, "Hi there").await?; + let _out = engine.run(&mut history, "Hi there").await; // History is updated let entries = history.entries(); @@ -368,12 +333,12 @@ async fn test_locked_multi_turn_history_accumulation() { // Turn 1 let result1 = locked_engine.run(&mut history, "Hello!").await; - assert!(result1.is_ok()); + assert!(matches!(result1, EngineRunExit::Finished)); assert_eq!(history.len(), 2); // user + assistant // Turn 2 let result2 = locked_engine.run(&mut history, "Can you help me?").await; - assert!(result2.is_ok()); + assert!(matches!(result2, EngineRunExit::Finished)); assert_eq!(history.len(), 4); // 2 * (user + assistant) // Verify history contents @@ -438,10 +403,7 @@ async fn test_locked_prefix_len_tracking() { assert_eq!(locked_engine.locked_prefix_len(), 2); // 2 items at lock time // Execute turn - locked_engine - .run(&mut history, "New message") - .await - .unwrap(); + locked_engine.run(&mut history, "New message").await; // History grows but locked_prefix_len remains unchanged assert_eq!(history.len(), 4); // 2 + 2 @@ -477,13 +439,16 @@ async fn test_turn_count_increment() -> Result<(), EngineError> { assert_eq!(engine.llm_call_count(), 0); // First run consumes Mutable, returns EngineRunOutput - let mut engine = engine.run(&mut history, "First").await?.engine; + let mut engine = engine.run(&mut history, "First").await.engine; assert_eq!(engine.turn_count(), 1); // Retry not yet implemented β†’ AgentTurn:LlmCall is 1:1. assert_eq!(engine.llm_call_count(), 1); // Subsequent runs on Locked take &mut self - engine.run(&mut history, "Second").await?; + assert!(matches!( + engine.run(&mut history, "Second").await, + EngineRunExit::Finished + )); assert_eq!(engine.turn_count(), 2); assert_eq!(engine.llm_call_count(), 2); @@ -573,7 +538,10 @@ async fn test_lock_unlock_relock_tools_remain_effective() { engine.register_tool(tool_a.definition()); let mut locked = engine.lock(&history); - locked.run(&mut history, "first").await.expect("first run"); + assert!(matches!( + locked.run(&mut history, "first").await, + EngineRunExit::Finished + )); assert_eq!(tool_a.call_count(), 1, "tool_a should be called once"); let mut unlocked = locked.unlock(); @@ -581,10 +549,10 @@ async fn test_lock_unlock_relock_tools_remain_effective() { unlocked.register_tool(tool_b.definition()); let mut relocked = unlocked.lock(&history); - relocked - .run(&mut history, "second") - .await - .expect("second run"); + assert!(matches!( + relocked.run(&mut history, "second").await, + EngineRunExit::Finished + )); assert_eq!(tool_a.call_count(), 1, "tool_a should not be called again"); assert_eq!(tool_b.call_count(), 1, "tool_b should be called once"); @@ -686,54 +654,55 @@ impl Interceptor for ContinueTurnOnce { #[tokio::test] async fn max_turns_is_scoped_to_each_fresh_run() { + let mut history: History = History::new(); let responses = vec![completed_text_events(), completed_text_events()]; let mut engine = Engine::new(MockLlmClient::with_responses(responses)); - let mut history: History = History::new(); engine.set_max_turns(Some(1)); let mut engine = engine.lock(&history); - assert_eq!( - engine.run(&mut history, "first").await.unwrap(), - EngineResult::Finished - ); + assert!(matches!( + engine.run(&mut history, "first").await, + EngineRunExit::Finished + )); assert_eq!(engine.turn_count(), 1); assert_eq!(engine.active_run_turn_count(), None); - assert_eq!( - engine.run(&mut history, "second").await.unwrap(), - EngineResult::Finished - ); + assert!(matches!( + engine.run(&mut history, "second").await, + EngineRunExit::Finished + )); assert_eq!(engine.turn_count(), 2); assert_eq!(engine.active_run_turn_count(), None); } #[tokio::test] async fn yielded_resume_keeps_the_same_unspent_turn_budget() { - let mut engine = Engine::new(MockLlmClient::new(completed_text_events())); let mut history: History = History::new(); + let mut engine = Engine::new(MockLlmClient::new(completed_text_events())); engine.set_max_turns(Some(1)); engine.set_interceptor(YieldOnce { calls: AtomicUsize::new(0), }); let mut engine = engine.lock(&history); - assert_eq!( - engine.run(&mut history, "start").await.unwrap(), - EngineResult::Yielded - ); + assert!(matches!( + engine.run(&mut history, "start").await, + EngineRunExit::Yielded + )); assert_eq!(engine.turn_count(), 0); assert_eq!(engine.active_run_turn_count(), Some(0)); - assert_eq!( - engine.resume(&mut history).await.unwrap(), - EngineResult::Finished - ); + assert!(matches!( + engine.resume(&mut history).await, + EngineRunExit::Finished + )); assert_eq!(engine.turn_count(), 1); assert_eq!(engine.active_run_turn_count(), None); } #[tokio::test] async fn paused_tool_resume_does_not_reset_the_consumed_turn_budget() { + let mut history: History = History::new(); let events = vec![ Event::tool_use_start(0, "call_1", "count_tool"), Event::tool_input_delta(0, "{}"), @@ -744,7 +713,6 @@ async fn paused_tool_resume_does_not_reset_the_consumed_turn_budget() { ]; let tool = CountingTool::new("count_tool"); let mut engine = Engine::new(MockLlmClient::new(events)); - let mut history: History = History::new(); engine.set_max_turns(Some(1)); engine.register_tool(tool.definition()); engine.set_interceptor(PauseToolOnce { @@ -752,18 +720,18 @@ async fn paused_tool_resume_does_not_reset_the_consumed_turn_budget() { }); let mut engine = engine.lock(&history); - assert_eq!( - engine.run(&mut history, "call it").await.unwrap(), - EngineResult::Paused - ); + assert!(matches!( + engine.run(&mut history, "call it").await, + EngineRunExit::Paused + )); assert_eq!(engine.turn_count(), 1); assert_eq!(engine.active_run_turn_count(), Some(1)); assert_eq!(tool.call_count(), 0); - assert_eq!( - engine.resume(&mut history).await.unwrap(), - EngineResult::LimitReached - ); + assert!(matches!( + engine.resume(&mut history).await, + EngineRunExit::Interrupted(StopReason::LimitReached) + )); assert_eq!(engine.turn_count(), 1); assert_eq!(engine.active_run_turn_count(), None); assert_eq!(tool.call_count(), 1, "the consumed turn's tool still runs"); @@ -771,6 +739,7 @@ async fn paused_tool_resume_does_not_reset_the_consumed_turn_budget() { #[tokio::test] async fn fresh_input_abandons_a_paused_run_and_starts_a_new_budget() { + let mut history: History = History::new(); let tool_events = vec![ Event::tool_use_start(0, "call_1", "count_tool"), Event::tool_input_delta(0, "{}"), @@ -782,7 +751,6 @@ async fn fresh_input_abandons_a_paused_run_and_starts_a_new_budget() { let client = MockLlmClient::with_responses(vec![tool_events, completed_text_events()]); let tool = CountingTool::new("count_tool"); let mut engine = Engine::new(client); - let mut history: History = History::new(); engine.set_max_turns(Some(1)); engine.register_tool(tool.definition()); engine.set_interceptor(PauseToolOnce { @@ -790,16 +758,16 @@ async fn fresh_input_abandons_a_paused_run_and_starts_a_new_budget() { }); let mut engine = engine.lock(&history); - assert_eq!( - engine.run(&mut history, "pause").await.unwrap(), - EngineResult::Paused - ); + assert!(matches!( + engine.run(&mut history, "pause").await, + EngineRunExit::Paused + )); assert_eq!(engine.active_run_turn_count(), Some(1)); - assert_eq!( - engine.run(&mut history, "replace").await.unwrap(), - EngineResult::Finished - ); + assert!(matches!( + engine.run(&mut history, "replace").await, + EngineRunExit::Finished + )); assert_eq!(engine.turn_count(), 2); assert_eq!(engine.active_run_turn_count(), None); assert_eq!(tool.call_count(), 1, "pending-tool semantics are unchanged"); @@ -807,18 +775,18 @@ async fn fresh_input_abandons_a_paused_run_and_starts_a_new_budget() { #[tokio::test] async fn interceptor_continuation_consumes_the_logical_run_budget() { - let mut engine = Engine::new(MockLlmClient::new(completed_text_events())); let mut history: History = History::new(); + let mut engine = Engine::new(MockLlmClient::new(completed_text_events())); engine.set_max_turns(Some(1)); engine.set_interceptor(ContinueTurnOnce { calls: AtomicUsize::new(0), }); let mut engine = engine.lock(&history); - assert_eq!( - engine.run(&mut history, "start").await.unwrap(), - EngineResult::LimitReached - ); + assert!(matches!( + engine.run(&mut history, "start").await, + EngineRunExit::Interrupted(StopReason::LimitReached) + )); assert_eq!(engine.turn_count(), 1); assert_eq!(engine.llm_call_count(), 1); assert_eq!(engine.active_run_turn_count(), None); @@ -826,17 +794,17 @@ async fn interceptor_continuation_consumes_the_logical_run_budget() { #[tokio::test] async fn restored_active_run_budget_is_enforced_before_another_llm_call() { - let mut engine = Engine::new(MockLlmClient::new(completed_text_events())); let mut history: History = History::new(); + let mut engine = Engine::new(MockLlmClient::new(completed_text_events())); engine.set_max_turns(Some(1)); engine.set_turn_count(7); engine.set_active_run_turn_count(Some(1)); let mut engine = engine.lock(&history); - assert_eq!( - engine.resume(&mut history).await.unwrap(), - EngineResult::LimitReached - ); + assert!(matches!( + engine.resume(&mut history).await, + EngineRunExit::Interrupted(StopReason::LimitReached) + )); assert_eq!(engine.turn_count(), 7); assert_eq!(engine.llm_call_count(), 0); assert_eq!(engine.active_run_turn_count(), None); diff --git a/crates/agen/tests/parallel_execution_test.rs b/crates/agen/tests/parallel_execution_test.rs index 42134f81..7a68cbf4 100644 --- a/crates/agen/tests/parallel_execution_test.rs +++ b/crates/agen/tests/parallel_execution_test.rs @@ -570,10 +570,7 @@ async fn test_before_tool_call_synthetic_result_committed() { engine.set_interceptor(SyntheticPolicy); - let _result = engine - .run(&mut history, "Test synthetic result") - .await - .unwrap(); + let _result = engine.run(&mut history, "Test synthetic result").await; assert_eq!(blocked_clone.call_count(), 0, "Blocked tool should not run"); assert!(history.items().any(|item| matches!( diff --git a/crates/agen/tests/reasoning_round_trip_test.rs b/crates/agen/tests/reasoning_round_trip_test.rs index 691466c6..21e83f2a 100644 --- a/crates/agen/tests/reasoning_round_trip_test.rs +++ b/crates/agen/tests/reasoning_round_trip_test.rs @@ -66,7 +66,7 @@ async fn anthropic_thinking_round_trips_signature_into_history() { let client = MockLlmClient::new(events); let engine = Engine::new(client); let mut history: History = History::new(); - let _out = engine.run(&mut history, "question?").await.expect("run ok"); + let _out = engine.run(&mut history, "question?").await; let entries = history.entries(); // user / reasoning / assistant_message @@ -110,7 +110,7 @@ async fn openai_reasoning_round_trips_encrypted_and_summary() { let client = MockLlmClient::new(events); let engine = Engine::new(client); let mut history: History = History::new(); - let _out = engine.run(&mut history, "q").await.expect("run ok"); + let _out = engine.run(&mut history, "q").await; let entries = history.entries(); match &entries[1].item { @@ -156,7 +156,7 @@ async fn reasoning_precedes_text_in_assistant_burst() { let client = MockLlmClient::new(events); let engine = Engine::new(client); let mut history: History = History::new(); - let _out = engine.run(&mut history, "q").await.expect("run ok"); + let _out = engine.run(&mut history, "q").await; let entries = history.entries(); // user / reasoning(ε…ˆι ­) / assistant_message @@ -218,7 +218,7 @@ async fn injected_reasoning_survives_into_outgoing_request() { ], ); - let _ = engine.run(&mut history, "follow up").await.expect("run ok"); + let _ = engine.run(&mut history, "follow up").await; let req = captured .lock() diff --git a/crates/session-store/tests/session_test.rs b/crates/session-store/tests/session_test.rs index 484047b5..d8e27fe0 100644 --- a/crates/session-store/tests/session_test.rs +++ b/crates/session-store/tests/session_test.rs @@ -135,7 +135,7 @@ async fn run_and_persist( session_id: session_store::SessionId, segment_id: session_store::SegmentId, input: &str, -) -> (TestWorker, agen::EngineResult) { +) -> (TestWorker, agen::EngineRunExit) { // Mirror Worker's run-entry contract: log the user input as segments // before the worker pushes its flattened user_message; save_delta // skips the resulting user_message item to avoid double-write. diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 2e6a5fbd..38e2195a 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -2941,10 +2941,7 @@ impl Worker { /// 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> - where - St: Clone + 'static, - { + fn apply_interrupt_prep(&mut self) -> Result<(), WorkerError> { let tool_result_summary = self .prompts() .load_full() @@ -2962,17 +2959,24 @@ impl Worker { &tool_result_summary, ); if !closures.is_empty() { - let mut annotate = history_annotator( - self.log_writer_handle(), - Vec::new(), - self.pending_committed_history.clone(), - ); - let (engine, session) = ( - self.engine.as_mut().expect("worker present"), - &mut self.session, - ); - engine.append_history_with(session.history_mut(), closures, &mut annotate)?; - session.note_mutation(); + 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"); @@ -2992,21 +2996,8 @@ impl Worker { })?; let interrupt_entry = HistoryEntry::new(agen::Item::system_message(system_note), interrupt_metadata); - let mut annotate = history_annotator( - self.log_writer_handle(), - vec![interrupt_entry.clone()], - self.pending_committed_history.clone(), - ); - let (engine, session) = ( - self.engine.as_mut().expect("worker present"), - &mut self.session, - ); - engine.append_history_with( - session.history_mut(), - std::iter::once(interrupt_entry.item), - &mut annotate, - )?; - session.note_mutation(); + self.session.history_mut().push_entry(interrupt_entry); + self.session.note_mutation(); Ok(()) }