diff --git a/crates/agen/src/engine.rs b/crates/agen/src/engine.rs index d35c1778b..8638332d4 100644 --- a/crates/agen/src/engine.rs +++ b/crates/agen/src/engine.rs @@ -937,7 +937,8 @@ impl Engine { /// /// The callback is invoked with a short human-readable message /// whenever the Engine encounters a condition that should be - /// surfaced to a human (e.g. tool output byte-cap truncation). + /// surfaced to a human (e.g. a provider-rejected image). Tool output + /// byte-cap truncation is reported in the tool result, not this channel. /// This channel is separate from `tracing::warn!`, which remains /// in place for developer logs. pub fn on_warning(&mut self, callback: impl Fn(&str) + Send + Sync + 'static) { @@ -2379,13 +2380,9 @@ impl Engine { limit_bytes = limit, "Tool output exceeded byte limit and was truncated" ); - self.emit_warning(&format!( - "tool `{}` output truncated from {} to {} bytes (limit {})", - tool_call.name, - before, - content.len(), - limit - )); + // Truncation is a tool-result constraint, not an actionable + // engine warning. The content marker tells the model to refine + // its query; keep diagnostics in tracing, not user alerts. } } diff --git a/crates/agen/tests/callback_test.rs b/crates/agen/tests/callback_test.rs index 1c3535f8e..162cbfcec 100644 --- a/crates/agen/tests/callback_test.rs +++ b/crates/agen/tests/callback_test.rs @@ -16,6 +16,182 @@ use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; use async_trait::async_trait; use common::MockLlmClient; +/// A per-run tracing sink: no global subscriber or output-body logging. +#[derive(Clone)] +struct DiagnosticLog(Arc>>); + +impl std::io::Write for DiagnosticLog { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[tokio::test] +async fn output_limits_keep_partial_results_without_warning_callbacks() { + use agen::tool::{Attachment, ImageAttachment, ToolOutputLimits, ToolResult}; + use agen::{History, Item, ToolResultDisposition}; + use tracing::instrument::WithSubscriber; + + let limit = 256; + let cases = [ + ("below", "small result".to_string(), limit), + ("exact", "x".repeat(limit), limit), + ("Grep", "matched line\n".repeat(200), limit), + ("Read", "あいうえお".repeat(200), 120), + ]; + let mut events = Vec::new(); + for (index, name) in cases.iter().map(|c| c.0).chain(["erroring"]).enumerate() { + events.extend([ + Event::tool_use_start(index, name, name), + Event::tool_input_delta(index, "{}"), + Event::tool_use_stop(index), + ]); + } + events.push(Event::Status(ClientStatusEvent { + status: ResponseStatus::Completed, + })); + let client = MockLlmClient::with_responses(vec![ + events, + vec![Event::Status(ClientStatusEvent { + status: ResponseStatus::Completed, + })], + ]); + let probe = client.clone(); + let mut engine = Engine::new(client); + engine.set_tool_call_dispatch_mode(agen::ToolCallDispatchMode::AfterResponse); + engine.set_tool_output_limits(Some(ToolOutputLimits { + default_max_bytes: limit, + per_tool: [("Read".to_string(), 120)].into(), + })); + let attachments = vec![Attachment::Image(ImageAttachment::new( + "image/png", + b"\x89PNG\r\n\x1a\n\0\0\0\rIHDR\0\0\0\x01\0\0\0\x01".to_vec(), + ))]; + for (name, content, _) in &cases { + engine.register_tool(fixed_tool( + name, + ToolOutput { + summary: format!("result from {name}"), + content: Some(content.clone()), + attachments: attachments.clone(), + }, + )); + } + engine.register_tool(erroring_tool("erroring", "execution failed")); + + let warnings = Arc::new(Mutex::new(Vec::new())); + let sink = warnings.clone(); + engine.on_warning(move |warning| sink.lock().unwrap().push(warning.to_owned())); + let published = Arc::new(Mutex::new(Vec::::new())); + let sink = published.clone(); + engine.on_tool_result(move |result| sink.lock().unwrap().push(result.clone())); + let committed = Arc::new(Mutex::new(Vec::new())); + let sink = committed.clone(); + engine.on_history_append(move |item| { + if matches!(item, Item::ToolResult { .. }) { + sink.lock().unwrap().push(item.clone()); + } + Ok(()) + }); + + let log = DiagnosticLog(Arc::new(Mutex::new(Vec::new()))); + let writer = log.clone(); + let subscriber = tracing_subscriber::fmt() + .without_time() + .with_ansi(false) + .with_max_level(tracing::Level::WARN) + .with_writer(move || writer.clone()) + .finish(); + let mut history = History::new(); + let run = engine + .run(&mut history, "run tools") + .with_subscriber(subscriber) + .await; + assert!(matches!(run.result, agen::EngineRunExit::Finished)); + assert!(warnings.lock().unwrap().is_empty()); + + let requests = probe.requests(); + assert_eq!(requests.len(), 2); + let saved: Vec<_> = history + .iter() + .filter_map(|entry| matches!(entry.item, Item::ToolResult { .. }).then_some(&entry.item)) + .collect(); + assert_eq!(saved.len(), cases.len() + 1); + let committed = committed.lock().unwrap(); + let published = published.lock().unwrap(); + let logs = String::from_utf8(log.0.lock().unwrap().clone()).unwrap(); + for item in saved { + let Item::ToolResult { + call_id, + summary, + content, + disposition, + is_error, + attachments: saved_attachments, + .. + } = item + else { + unreachable!() + }; + assert!(committed.iter().any(|candidate| candidate == item)); + assert!(requests[1].items.iter().any(|candidate| candidate == item)); + let result = published + .iter() + .find(|result| result.tool_use_id == *call_id) + .unwrap(); + assert_eq!(&result.summary, summary); + assert_eq!(&result.content, content); + assert_eq!(&result.disposition, disposition); + assert_eq!(&result.is_error, is_error); + assert_eq!(&result.attachments, saved_attachments); + if call_id == "erroring" { + assert_eq!(*disposition, ToolResultDisposition::Error); + assert!(*is_error); + assert!(content.is_none()); + continue; + } + assert_eq!(*disposition, ToolResultDisposition::Success); + assert!(!*is_error); + assert_eq!(*summary, format!("result from {call_id}")); + assert_eq!(saved_attachments, &attachments); + let (_, original, cap) = cases.iter().find(|c| c.0 == call_id).unwrap(); + let content = content.as_ref().unwrap(); + assert!(content.len() <= *cap); + if original.len() <= *cap { + assert_eq!(content, original); + } else { + let (body, marker) = content.split_once("\n\n[truncated: ").unwrap(); + assert!(original.starts_with(body)); + assert_eq!( + marker, + format!( + "{} bytes dropped, refine your query]", + original.len() - body.len() + ) + ); + let diagnostic = logs + .lines() + .find(|line| line.contains(&format!("tool={call_id}"))) + .unwrap(); + assert!(diagnostic.contains(&format!("before_bytes={}", original.len()))); + assert!(diagnostic.contains(&format!("after_bytes={}", content.len()))); + assert!(diagnostic.contains(&format!("limit_bytes={cap}"))); + assert!(!diagnostic.contains(body)); + } + } + assert_eq!( + logs.lines() + .filter(|line| line.contains("Tool output exceeded byte limit")) + .count(), + 2 + ); +} + #[derive(Clone)] struct FailOnceClient { calls: Arc, diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 38e36e541..9b8242604 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1303,10 +1303,7 @@ pub(crate) fn wire_event_bridges_on_engine( }); }); - let alerter_for_worker = alerter.clone(); - worker.on_warning(move |message| { - alerter_for_worker.alert(AlertLevel::Warn, AlertSource::Engine, message.to_owned()); - }); + wire_engine_warnings(worker, alerter); // History-append broadcasts (previously `Event::SystemMessage`) // have been removed: every persistent history item is now committed @@ -1315,6 +1312,17 @@ pub(crate) fn wire_event_bridges_on_engine( // per-item commit channel is wired at the top of this function. } +/// Forward genuine Engine warnings to both live clients and the alert snapshot. +fn wire_engine_warnings( + engine: &mut agen::Engine, + alerter: &Alerter, +) { + let alerter = alerter.clone(); + engine.on_warning(move |message| { + alerter.alert(AlertLevel::Warn, AlertSource::Engine, message.to_owned()); + }); +} + fn validate_memory_lifecycle_targets( memory: &manifest::MemoryFeatureProfileConfig, subjektiv: &manifest::SubjektivFeatureProfileConfig, @@ -3906,6 +3914,198 @@ mod tests { use tempfile::TempDir; use tokio::net::UnixListener; + mod engine_alerts { + use super::*; + use agen::llm_client::event::Event as LlmEvent; + use agen::llm_client::{ClientError, Item, Request, ToolCallCompletionSupport}; + use agen::tool::{ + Tool, ToolError, ToolExecutionContext, ToolMeta, ToolOutput, ToolOutputLimits, + }; + use agen::{Engine, EngineRunExit, History, ToolCallDispatchMode}; + use std::collections::VecDeque; + use std::pin::Pin; + use std::sync::Mutex; + + #[derive(Clone)] + struct ScriptedClient { + responses: Arc>>>, + requests: Arc>>, + } + + impl ScriptedClient { + fn new(responses: Vec>) -> Self { + Self { + responses: Arc::new(Mutex::new(responses.into())), + requests: Arc::new(Mutex::new(Vec::new())), + } + } + } + + #[async_trait::async_trait] + impl LlmClient for ScriptedClient { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + fn tool_call_completion_support(&self) -> ToolCallCompletionSupport { + ToolCallCompletionSupport::ResponseComplete + } + + async fn stream( + &self, + request: Request, + ) -> Result< + Pin> + Send>>, + ClientError, + > { + self.requests.lock().unwrap().push(request); + let events = self + .responses + .lock() + .unwrap() + .pop_front() + .expect("unexpected LLM request"); + Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok)))) + } + } + + struct LargeOutputTool; + + #[async_trait::async_trait] + impl Tool for LargeOutputTool { + async fn execute( + &self, + _input_json: &str, + _ctx: ToolExecutionContext, + ) -> Result { + Ok(ToolOutput { + summary: "Fixed large output".into(), + content: Some("useful output line\n".repeat(512)), + attachments: Vec::new(), + }) + } + } + + fn finished_response() -> Vec { + vec![ + LlmEvent::text_block_start(0), + LlmEvent::text_delta(0, "Done"), + LlmEvent::text_block_stop(0, None), + ] + } + + #[tokio::test] + async fn truncated_tool_output_preserves_model_result_without_live_or_buffered_alerts() { + let client = ScriptedClient::new(vec![ + vec![ + LlmEvent::tool_use_start(0, "large-call", "large_output"), + LlmEvent::tool_input_delta(0, "{}"), + LlmEvent::tool_use_stop(0), + ], + finished_response(), + ]); + let requests = client.requests.clone(); + let mut engine = Engine::new(client); + // Isolate output limiting from the independent early-dispatch warning. + engine.set_tool_call_dispatch_mode(ToolCallDispatchMode::AfterResponse); + engine.set_tool_output_limits(Some(ToolOutputLimits { + default_max_bytes: 256, + per_tool: Default::default(), + })); + engine.register_tool(Arc::new(|| { + ( + ToolMeta::new("large_output") + .input_schema(serde_json::json!({"type": "object"})), + Arc::new(LargeOutputTool), + ) + })); + let (tx, _) = broadcast::channel(8); + let alerter = Alerter::new(tx); + let (_, mut live) = alerter.subscribe_with_snapshot(); + wire_engine_warnings(&mut engine, &alerter); + + let output = engine.run(&mut History::new(), "Run the tool").await; + assert!( + matches!(output.result, EngineRunExit::Finished), + "{:?}", + output.result + ); + let requests = requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + let result = requests[1].items.iter().find(|item| { + matches!(item, Item::ToolResult { call_id, .. } if call_id == "large-call") + }).expect("the next model request must include the bounded tool result"); + match result { + Item::ToolResult { + summary, + content: Some(content), + is_error, + disposition, + .. + } => { + assert_eq!(summary, "Fixed large output"); + assert!(!is_error); + assert_eq!(*disposition, agen::ToolResultDisposition::Success); + assert!(content.len() <= 256, "{content}"); + assert!(content.starts_with("useful output line\n"), "{content}"); + assert!(content.contains("[truncated:"), "{content}"); + assert!(content.contains("refine your query]"), "{content}"); + } + other => panic!("unexpected model-visible result: {other:?}"), + } + assert!(matches!( + live.try_recv(), + Err(broadcast::error::TryRecvError::Empty) + )); + let (snapshot, _) = alerter.subscribe_with_snapshot(); + assert!(snapshot.is_empty(), "unexpected alerts: {snapshot:?}"); + } + + #[tokio::test] + async fn unsupported_early_dispatch_warning_reaches_live_and_buffered_engine_alerts() { + let mut engine = Engine::new(ScriptedClient::new(vec![finished_response()])); + engine.set_tool_call_dispatch_mode(ToolCallDispatchMode::OnToolCallComplete); + let (tx, _) = broadcast::channel(8); + let alerter = Alerter::new(tx); + let (_, mut live) = alerter.subscribe_with_snapshot(); + wire_engine_warnings(&mut engine, &alerter); + + let output = engine.run(&mut History::new(), "Finish").await; + assert!( + matches!(output.result, EngineRunExit::Finished), + "{:?}", + output.result + ); + let alert = match live.try_recv() { + Ok(Event::Alert(alert)) => alert, + other => panic!("expected live Engine warning: {other:?}"), + }; + assert_eq!(alert.level, AlertLevel::Warn); + assert_eq!(alert.source, AlertSource::Engine); + assert!( + alert.message.contains( + "provider does not expose a trustworthy per-call completion boundary" + ), + "{}", + alert.message + ); + assert!(matches!( + live.try_recv(), + Err(broadcast::error::TryRecvError::Empty) + )); + let (snapshot, mut late_live) = alerter.subscribe_with_snapshot(); + assert_eq!(snapshot.len(), 1, "unexpected alerts: {snapshot:?}"); + assert_eq!(snapshot[0].level, alert.level); + assert_eq!(snapshot[0].source, alert.source); + assert_eq!(snapshot[0].message, alert.message); + assert_eq!(snapshot[0].timestamp_ms, alert.timestamp_ms); + assert!(matches!( + late_live.try_recv(), + Err(broadcast::error::TryRecvError::Empty) + )); + } + } + #[test] fn runtime_session_head_is_materialized_after_feature_installation_before_exposure() { let production = include_str!("controller.rs") diff --git a/crates/worker/src/ipc/alerter.rs b/crates/worker/src/ipc/alerter.rs index 1ab5f55b1..b36ee2b86 100644 --- a/crates/worker/src/ipc/alerter.rs +++ b/crates/worker/src/ipc/alerter.rs @@ -2,7 +2,7 @@ //! //! Separate from `tracing` (which is for developer logs). Alerts //! are short human-readable messages the Worker layer wants a client to -//! see — for example "compaction failed", "tool output truncated". +//! see — for example "compaction failed", "tool-call early dispatch deferred". //! //! Each alert is broadcast on the shared `Event` channel and //! also appended to an in-memory buffer so that clients connecting