fix(engine): keep tool truncation notices in tool results
Stop emitting a user-facing Engine warning for ordinary output limiting while preserving structured size diagnostics and the model-visible truncation marker. Keep the Worker warning bridge for genuine warnings. Regression coverage proves bounded ASCII/UTF-8 results, per-tool limits, attachments, dispositions and commit/model agreement, plus absence of live/buffered truncation alerts and preservation of other Engine alerts. Validation: cargo check; cargo test -p agen; cargo test -p worker; targeted callback test with codex; cargo fmt --all -- --check; git diff --check HEAD.
This commit is contained in:
@@ -937,7 +937,8 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
///
|
||||
/// 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<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
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.
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Mutex<Vec<u8>>>);
|
||||
|
||||
impl std::io::Write for DiagnosticLog {
|
||||
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
|
||||
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::<ToolResult>::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<AtomicUsize>,
|
||||
|
||||
@@ -1303,10 +1303,7 @@ pub(crate) fn wire_event_bridges_on_engine<C, St>(
|
||||
});
|
||||
});
|
||||
|
||||
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<C, St>(
|
||||
// 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<C: LlmClient, A: Send + Sync>(
|
||||
engine: &mut agen::Engine<C, agen::state::Mutable, A>,
|
||||
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<Mutex<VecDeque<Vec<LlmEvent>>>>,
|
||||
requests: Arc<Mutex<Vec<Request>>>,
|
||||
}
|
||||
|
||||
impl ScriptedClient {
|
||||
fn new(responses: Vec<Vec<LlmEvent>>) -> 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<dyn LlmClient> {
|
||||
Box::new(self.clone())
|
||||
}
|
||||
|
||||
fn tool_call_completion_support(&self) -> ToolCallCompletionSupport {
|
||||
ToolCallCompletionSupport::ResponseComplete
|
||||
}
|
||||
|
||||
async fn stream(
|
||||
&self,
|
||||
request: Request,
|
||||
) -> Result<
|
||||
Pin<Box<dyn futures::Stream<Item = Result<LlmEvent, ClientError>> + 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<ToolOutput, ToolError> {
|
||||
Ok(ToolOutput {
|
||||
summary: "Fixed large output".into(),
|
||||
content: Some("useful output line\n".repeat(512)),
|
||||
attachments: Vec::new(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn finished_response() -> Vec<LlmEvent> {
|
||||
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")
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user