Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: harden interceptor lifecycle contracts

Browse Source
This commit is contained in:
Hare
2026-09-03 16:46:02 +09:00
parent e62c7cf4f5
commit eac4a0c071
11 changed files with 877 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/agen/tests/annotated_history_test.rs
+126
View File
@@ -1,8 +1,15 @@
mod common;
use agen::interceptor::{
AssistantTurnEndContext, Interceptor, InterceptorCallId, InterceptorInvocation,
InterceptorPhase, InterceptorResult, PendingHistoryAppendsContext, PreLlmRequestContext,
PreRequestAction, PromptAction, PromptSubmitContext, RunExitContext, TurnEndAction,
};
use agen::llm_client::event::{Event, ResponseStatus, StatusEvent};
use agen::{Engine, EngineError, History, HistoryEntry, Item, Role};
use async_trait::async_trait;
use common::MockLlmClient;
use std::sync::{Arc, Mutex};
fn completed_text_events(text: &str) -> Vec<Event> {
vec![
@@ -47,6 +54,125 @@ async fn run_preserves_item_annotations_without_projecting_them() {
assert_eq!(history.items_cloned().len(), 2);
}
#[derive(Clone)]
struct AnnotationObservingInterceptor {
observed: Arc<Mutex<Vec<(InterceptorInvocation, Vec<String>)>>>,
}
impl AnnotationObservingInterceptor {
fn record(&self, invocation: &InterceptorInvocation, history: &[HistoryEntry<String>]) {
self.observed.lock().unwrap().push((
invocation.clone(),
history
.iter()
.map(|entry| entry.annotation.clone())
.collect(),
));
}
}
#[async_trait]
impl Interceptor<String> for AnnotationObservingInterceptor {
async fn on_prompt_submit(
&self,
context: PromptSubmitContext<'_, String>,
) -> InterceptorResult<PromptAction> {
self.record(&context.invocation, context.history);
Ok(PromptAction::Continue)
}
async fn pending_history_appends(
&self,
context: PendingHistoryAppendsContext<'_, String>,
) -> InterceptorResult<Vec<Item>> {
self.record(&context.invocation, context.history);
Ok(Vec::new())
}
async fn pre_llm_request(
&self,
context: PreLlmRequestContext<'_, String>,
) -> InterceptorResult<PreRequestAction> {
self.record(&context.invocation, context.history);
Ok(PreRequestAction::Continue)
}
async fn on_assistant_turn_end(
&self,
context: AssistantTurnEndContext<'_, String>,
) -> InterceptorResult<TurnEndAction> {
assert_eq!(context.assistant_entries.len(), 1);
assert_eq!(context.assistant_entries[0].annotation, "2:assistant");
self.record(&context.invocation, context.history);
Ok(TurnEndAction::Finish)
}
async fn on_run_exit(&self, context: RunExitContext<'_, String>) -> InterceptorResult<()> {
self.record(&context.invocation, context.history);
Ok(())
}
}
#[tokio::test]
async fn interceptor_contexts_preserve_annotations_and_typed_lifecycle_identity() {
let client = MockLlmClient::new(completed_text_events("assistant reply"));
let mut engine = Engine::<_, agen::state::Mutable, String>::new_annotated(client);
let observed = Arc::new(Mutex::new(Vec::new()));
engine.set_interceptor(AnnotationObservingInterceptor {
observed: observed.clone(),
});
let mut history = History::<String>::new();
let mut next = 0usize;
let mut annotate = |item: &Item| {
next += 1;
let kind = if item.is_assistant_message() {
"assistant"
} else {
"user"
};
Ok(format!("{next}:{kind}"))
};
let output = engine
.run_with_annotation(&mut history, "hello", &mut annotate)
.await;
assert!(matches!(output.result, agen::EngineRunExit::Finished));
let observed = observed.lock().unwrap();
let phases: Vec<_> = observed
.iter()
.map(|(invocation, _)| invocation.phase)
.collect();
assert_eq!(
phases,
[
InterceptorPhase::PromptSubmit,
InterceptorPhase::PendingHistoryAppends,
InterceptorPhase::PreLlmRequest,
InterceptorPhase::AssistantTurnEnd,
InterceptorPhase::RunExit,
]
);
assert!(
observed
.iter()
.all(|(invocation, _)| invocation.run_id == observed[0].0.run_id)
);
assert_eq!(
observed
.iter()
.map(|(invocation, _)| invocation.counters.invocation.get())
.collect::<Vec<_>>(),
[0, 1, 2, 3, 4]
);
assert_eq!(observed[2].0.call_id, Some(InterceptorCallId::Llm(0)));
assert_eq!(observed[3].0.call_id, Some(InterceptorCallId::Llm(0)));
assert_eq!(observed[1].1, ["1:user"]);
assert_eq!(observed[2].1, ["1:user"]);
assert_eq!(observed[3].1, ["1:user", "2:assistant"]);
assert_eq!(observed[4].1, ["1:user", "2:assistant"]);
}
#[test]
fn append_failure_does_not_make_item_live() {
let client = MockLlmClient::new(vec![]);
crates/agen/tests/engine_state_test.rs
+134 -30
View File
@@ -10,9 +10,11 @@ use std::sync::{Arc, Mutex};
use agen::Item;
use agen::interceptor::{
AssistantTurnEndContext, Interceptor, InterceptorError, InterceptorPoint, InterceptorResult,
PostToolAction, PreLlmRequestContext, PreRequestAction, PreToolAction, PromptAction,
PromptSubmitContext, RunExitContext, ToolCallInfo, ToolResultInfo, TurnEndAction,
AssistantTurnEndContext, Interceptor, InterceptorError, InterceptorErrorCategory,
InterceptorPhase as InterceptorPoint, InterceptorResult, MAX_INTERCEPTOR_DIAGNOSTIC_BYTES,
PendingHistoryAppendsContext, PostToolAction, PreLlmRequestContext, PreRequestAction,
PreToolAction, PromptAction, PromptSubmitContext, RunExitContext, ToolCallInfo, ToolResultInfo,
TurnEndAction,
};
use agen::llm_client::{
ClientError, LlmClient, Request, ResponseStream,
@@ -620,7 +622,7 @@ struct YieldOnce {
impl Interceptor for YieldOnce {
async fn pre_llm_request(
&self,
_context: PreLlmRequestContext<'_>,
_context: PreLlmRequestContext<'_, ()>,
) -> InterceptorResult<PreRequestAction> {
Ok(if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
PreRequestAction::Yield
@@ -636,7 +638,10 @@ struct PauseToolOnce {
#[async_trait]
impl Interceptor for PauseToolOnce {
async fn pre_tool_call(&self, _info: &mut ToolCallInfo) -> InterceptorResult<PreToolAction> {
async fn pre_tool_call(
&self,
_info: &mut ToolCallInfo<'_, ()>,
) -> InterceptorResult<PreToolAction> {
Ok(if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
PreToolAction::Pause
} else {
@@ -653,7 +658,7 @@ struct ContinueTurnOnce {
impl Interceptor for ContinueTurnOnce {
async fn on_assistant_turn_end(
&self,
_context: AssistantTurnEndContext<'_>,
_context: AssistantTurnEndContext<'_, ()>,
) -> InterceptorResult<TurnEndAction> {
Ok(if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
TurnEndAction::ContinueWithMessages(vec![Item::system_message("continue")])
@@ -680,7 +685,10 @@ impl FailingLifecycleInterceptor {
fn record<T>(&self, point: InterceptorPoint, action: T) -> InterceptorResult<T> {
self.calls.lock().unwrap().push(point);
if self.failure == point {
Err(InterceptorError::new(format!("{point} rejected")))
Err(InterceptorError::new(
InterceptorErrorCategory::Policy,
format!("{point} rejected"),
))
} else {
Ok(action)
}
@@ -695,47 +703,56 @@ impl FailingLifecycleInterceptor {
impl Interceptor for FailingLifecycleInterceptor {
async fn on_prompt_submit(
&self,
_context: PromptSubmitContext<'_>,
_context: PromptSubmitContext<'_, ()>,
) -> InterceptorResult<PromptAction> {
tokio::task::yield_now().await;
self.record(InterceptorPoint::PromptSubmit, PromptAction::Continue)
}
async fn pending_history_appends(&self) -> InterceptorResult<Vec<Item>> {
async fn pending_history_appends(
&self,
_context: PendingHistoryAppendsContext<'_, ()>,
) -> InterceptorResult<Vec<Item>> {
tokio::task::yield_now().await;
self.record(InterceptorPoint::PendingHistoryAppends, Vec::new())
}
async fn pre_llm_request(
&self,
_context: PreLlmRequestContext<'_>,
_context: PreLlmRequestContext<'_, ()>,
) -> InterceptorResult<PreRequestAction> {
tokio::task::yield_now().await;
self.record(InterceptorPoint::PreLlmRequest, PreRequestAction::Continue)
}
async fn pre_tool_call(&self, _info: &mut ToolCallInfo) -> InterceptorResult<PreToolAction> {
async fn pre_tool_call(
&self,
_info: &mut ToolCallInfo<'_, ()>,
) -> InterceptorResult<PreToolAction> {
tokio::task::yield_now().await;
self.record(InterceptorPoint::PreToolCall, PreToolAction::Continue)
}
async fn post_tool_call(&self, _info: &ToolResultInfo) -> InterceptorResult<PostToolAction> {
async fn post_tool_call(
&self,
_info: &ToolResultInfo<'_, ()>,
) -> InterceptorResult<PostToolAction> {
tokio::task::yield_now().await;
self.record(InterceptorPoint::PostToolCall, PostToolAction::Continue)
}
async fn on_assistant_turn_end(
&self,
context: AssistantTurnEndContext<'_>,
context: AssistantTurnEndContext<'_, ()>,
) -> InterceptorResult<TurnEndAction> {
tokio::task::yield_now().await;
assert!(context.history.ends_with(context.assistant_items));
assert!(context.history.ends_with(context.assistant_entries));
if !context.tool_calls.is_empty() {
assert_eq!(
context
.assistant_items
.assistant_entries
.iter()
.filter(|item| matches!(item, Item::ToolCall { .. }))
.filter(|entry| matches!(&entry.item, Item::ToolCall { .. }))
.count(),
context.tool_calls.len()
);
@@ -743,7 +760,7 @@ impl Interceptor for FailingLifecycleInterceptor {
self.record(InterceptorPoint::AssistantTurnEnd, TurnEndAction::Finish)
}
async fn on_run_exit(&self, _context: RunExitContext<'_>) -> InterceptorResult<()> {
async fn on_run_exit(&self, _context: RunExitContext<'_, ()>) -> InterceptorResult<()> {
tokio::task::yield_now().await;
self.record(InterceptorPoint::RunExit, ())
}
@@ -795,7 +812,7 @@ fn expected_interceptor_calls(failure: InterceptorPoint) -> Vec<InterceptorPoint
}
#[tokio::test]
async fn interceptor_failures_are_typed_unexpected_and_each_lifecycle_point_runs_once() {
async fn interceptor_failures_are_typed_and_terminal_observer_preserves_original_exit() {
use InterceptorPoint as Point;
for failure_point in [
@@ -828,15 +845,23 @@ async fn interceptor_failures_are_typed_unexpected_and_each_lifecycle_point_runs
let mut engine = engine.lock(&history);
let exit = engine.run(&mut history, "test").await;
let EngineRunExit::Interrupted(RunInterruptionReason::Unexpected(
EngineError::Interceptor(failure),
)) = exit
else {
panic!("expected typed interceptor interruption at {failure_point}, got {exit:?}");
let failure = if failure_point == Point::RunExit {
assert!(matches!(exit, EngineRunExit::Finished));
engine
.last_run_exit_observer_failure()
.expect("terminal observer diagnostic should be retained")
} else {
let EngineRunExit::Interrupted(RunInterruptionReason::Unexpected(
EngineError::Interceptor(failure),
)) = &exit
else {
panic!("expected typed interceptor interruption at {failure_point}, got {exit:?}");
};
failure
};
assert_eq!(failure.point(), failure_point);
assert_eq!(failure.phase(), failure_point);
assert_eq!(
failure.error().message(),
failure.error().diagnostic(),
format!("{failure_point} rejected")
);
assert_eq!(
@@ -854,6 +879,85 @@ async fn interceptor_failures_are_typed_unexpected_and_each_lifecycle_point_runs
}
}
#[test]
fn interceptor_error_keeps_typed_category_and_bounded_utf8_diagnostic() {
let error = InterceptorError::new(
InterceptorErrorCategory::Dependency,
"界".repeat(MAX_INTERCEPTOR_DIAGNOSTIC_BYTES),
);
assert_eq!(error.category(), InterceptorErrorCategory::Dependency);
assert!(error.diagnostic().len() <= MAX_INTERCEPTOR_DIAGNOSTIC_BYTES);
assert!(
error
.diagnostic()
.is_char_boundary(error.diagnostic().len())
);
}
struct FailingRunExitObserver {
pause: bool,
}
#[async_trait]
impl Interceptor for FailingRunExitObserver {
async fn on_assistant_turn_end(
&self,
_context: AssistantTurnEndContext<'_, ()>,
) -> InterceptorResult<TurnEndAction> {
Ok(if self.pause {
TurnEndAction::Pause
} else {
TurnEndAction::Finish
})
}
async fn on_run_exit(&self, _context: RunExitContext<'_, ()>) -> InterceptorResult<()> {
Err(InterceptorError::new(
InterceptorErrorCategory::Dependency,
"terminal audit unavailable",
))
}
}
#[tokio::test]
async fn terminal_observer_failure_preserves_paused_and_interrupted_exits() {
let mut paused_engine = Engine::new(MockLlmClient::new(completed_text_events()));
paused_engine.set_interceptor(FailingRunExitObserver { pause: true });
let mut paused_history = History::new();
let mut paused_engine = paused_engine.lock(&paused_history);
assert!(matches!(
paused_engine.run(&mut paused_history, "pause").await,
EngineRunExit::Paused
));
assert_eq!(
paused_engine
.last_run_exit_observer_failure()
.expect("paused observer diagnostic")
.error()
.category(),
InterceptorErrorCategory::Dependency
);
let mut interrupted_engine = Engine::new(MockLlmClient::new(completed_text_events()));
interrupted_engine.set_max_turns(Some(0));
interrupted_engine.set_interceptor(FailingRunExitObserver { pause: false });
let mut interrupted_history = History::new();
let mut interrupted_engine = interrupted_engine.lock(&interrupted_history);
assert!(matches!(
interrupted_engine
.run(&mut interrupted_history, "limit")
.await,
EngineRunExit::Interrupted(RunInterruptionReason::LimitReached)
));
assert_eq!(
interrupted_engine
.last_run_exit_observer_failure()
.expect("interrupted observer diagnostic")
.phase(),
InterceptorPoint::RunExit
);
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TerminalMode {
Finish,
@@ -886,7 +990,7 @@ impl RecordingTerminalInterceptor {
impl Interceptor for RecordingTerminalInterceptor {
async fn pre_llm_request(
&self,
_context: PreLlmRequestContext<'_>,
_context: PreLlmRequestContext<'_, ()>,
) -> InterceptorResult<PreRequestAction> {
Ok(if self.mode == TerminalMode::Yield {
PreRequestAction::Yield
@@ -897,11 +1001,11 @@ impl Interceptor for RecordingTerminalInterceptor {
async fn on_assistant_turn_end(
&self,
context: AssistantTurnEndContext<'_>,
context: AssistantTurnEndContext<'_, ()>,
) -> InterceptorResult<TurnEndAction> {
assert!(!context.assistant_items.is_empty());
assert!(!context.assistant_entries.is_empty());
assert!(
context.history.ends_with(context.assistant_items),
context.history.ends_with(context.assistant_entries),
"assistant-turn callback must observe committed terminal items"
);
let turn = self.assistant_turns.fetch_add(1, Ordering::SeqCst);
@@ -912,7 +1016,7 @@ impl Interceptor for RecordingTerminalInterceptor {
})
}
async fn on_run_exit(&self, context: RunExitContext<'_>) -> InterceptorResult<()> {
async fn on_run_exit(&self, context: RunExitContext<'_, ()>) -> InterceptorResult<()> {
let kind = match context.exit {
EngineRunExit::Finished => "finished",
EngineRunExit::Paused => "paused",
crates/agen/tests/parallel_execution_test.rs
+109 -8
View File
@@ -7,14 +7,17 @@ use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use agen::interceptor::{
Interceptor, InterceptorResult, PostToolAction, PreToolAction, ToolCallInfo, ToolResultInfo,
Interceptor, InterceptorErrorCategory, InterceptorPhase, InterceptorResult, PostToolAction,
PreToolAction, ToolCallInfo, ToolResultInfo,
};
use agen::llm_client::event::{Event, ResponseStatus, StatusEvent};
use agen::tool::{
Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput, ToolResult,
ToolResultDisposition,
};
use agen::{Engine, History, Item, ToolExecutionPolicy};
use agen::{
Engine, EngineError, EngineRunExit, History, Item, RunInterruptionReason, ToolExecutionPolicy,
};
use async_trait::async_trait;
mod common;
@@ -907,7 +910,10 @@ async fn test_tool_execution_context_for_skipped_and_synthetic_paths() {
#[async_trait]
impl Interceptor for ContextPolicy {
async fn pre_tool_call(&self, info: &mut ToolCallInfo) -> InterceptorResult<PreToolAction> {
async fn pre_tool_call(
&self,
info: &mut ToolCallInfo<'_, ()>,
) -> InterceptorResult<PreToolAction> {
self.pre_contexts.lock().unwrap().push(info.context.clone());
Ok(match info.call.name.as_str() {
"skip_tool" => PreToolAction::Skip,
@@ -919,7 +925,10 @@ async fn test_tool_execution_context_for_skipped_and_synthetic_paths() {
})
}
async fn post_tool_call(&self, info: &ToolResultInfo) -> InterceptorResult<PostToolAction> {
async fn post_tool_call(
&self,
info: &ToolResultInfo<'_, ()>,
) -> InterceptorResult<PostToolAction> {
self.post_contexts
.lock()
.unwrap()
@@ -996,7 +1005,10 @@ async fn test_before_tool_call_skip() {
#[async_trait]
impl Interceptor for BlockingPolicy {
async fn pre_tool_call(&self, info: &mut ToolCallInfo) -> InterceptorResult<PreToolAction> {
async fn pre_tool_call(
&self,
info: &mut ToolCallInfo<'_, ()>,
) -> InterceptorResult<PreToolAction> {
Ok(if info.call.name == "blocked_tool" {
PreToolAction::Skip
} else {
@@ -1083,7 +1095,19 @@ async fn test_post_tool_call_observes_committed_result() {
#[async_trait]
impl Interceptor for ObservingPolicy {
async fn post_tool_call(&self, info: &ToolResultInfo) -> InterceptorResult<PostToolAction> {
async fn post_tool_call(
&self,
info: &ToolResultInfo<'_, ()>,
) -> InterceptorResult<PostToolAction> {
assert_eq!(info.invocation.phase, InterceptorPhase::PostToolCall);
assert_eq!(
info.invocation.call_id,
Some(agen::InterceptorCallId::Tool(info.call.id.clone()))
);
assert!(matches!(
info.history.last().map(|entry| &entry.item),
Some(Item::ToolResult { call_id, .. }) if call_id == &info.call.id
));
*self.observed_content.lock().unwrap() = Some(info.result.summary.clone());
Ok(PostToolAction::Continue)
}
@@ -1144,7 +1168,10 @@ async fn test_before_tool_call_synthetic_result_committed() {
#[async_trait]
impl Interceptor for SyntheticPolicy {
async fn pre_tool_call(&self, info: &mut ToolCallInfo) -> InterceptorResult<PreToolAction> {
async fn pre_tool_call(
&self,
info: &mut ToolCallInfo<'_, ()>,
) -> InterceptorResult<PreToolAction> {
Ok(PreToolAction::SyntheticResult(ToolResult::error(
info.call.id.clone(),
"permission denied",
@@ -1168,6 +1195,80 @@ async fn test_before_tool_call_synthetic_result_committed() {
)));
}
#[derive(Clone, Copy)]
enum InvalidIdentityMode {
ContinuedCall,
SyntheticResult,
}
struct InvalidIdentityPolicy(InvalidIdentityMode);
#[async_trait]
impl Interceptor for InvalidIdentityPolicy {
async fn pre_tool_call(
&self,
info: &mut ToolCallInfo<'_, ()>,
) -> InterceptorResult<PreToolAction> {
assert_eq!(info.invocation.phase, InterceptorPhase::PreToolCall);
assert_eq!(
info.invocation.call_id,
Some(agen::InterceptorCallId::Tool("call_1".to_string()))
);
assert!(matches!(
info.history.last().map(|entry| &entry.item),
Some(Item::ToolCall { call_id, .. }) if call_id == "call_1"
));
Ok(match self.0 {
InvalidIdentityMode::ContinuedCall => {
info.call.id = "different-call".to_string();
PreToolAction::Continue
}
InvalidIdentityMode::SyntheticResult => PreToolAction::SyntheticResult(
ToolResult::error("different-call", "invalid synthetic result"),
),
})
}
}
#[tokio::test]
async fn interceptor_cannot_change_tool_call_identity() {
for mode in [
InvalidIdentityMode::ContinuedCall,
InvalidIdentityMode::SyntheticResult,
] {
let client = MockLlmClient::new(vec![
Event::tool_use_start(0, "call_1", "echo"),
Event::tool_input_delta(0, r#"{}"#),
Event::tool_use_stop(0),
Event::Status(StatusEvent {
status: ResponseStatus::Completed,
}),
]);
let mut engine = Engine::new(client);
engine.register_tool(SlowTool::new("echo", 1).definition());
engine.set_interceptor(InvalidIdentityPolicy(mode));
let mut history = History::new();
let result = engine.run(&mut history, "identity").await;
let EngineRunExit::Interrupted(RunInterruptionReason::Unexpected(
EngineError::Interceptor(failure),
)) = result.result
else {
panic!("invalid tool identity must interrupt with a typed failure");
};
assert_eq!(failure.phase(), InterceptorPhase::PreToolCall);
assert_eq!(
failure.error().category(),
InterceptorErrorCategory::ContractViolation
);
assert!(
!history
.items()
.any(|item| matches!(item, Item::ToolResult { .. }))
);
}
}
#[tokio::test]
async fn post_tool_abort_commits_confirmed_result_before_stopping_run() {
let client = MockLlmClient::new(vec![
@@ -1190,7 +1291,7 @@ async fn post_tool_abort_commits_confirmed_result_before_stopping_run() {
impl Interceptor for AbortAfterResult {
async fn post_tool_call(
&self,
_info: &ToolResultInfo,
_info: &ToolResultInfo<'_, ()>,
) -> InterceptorResult<PostToolAction> {
self.lifecycle.lock().unwrap().push("post_tool_call");
Ok(PostToolAction::Abort("policy stopped the run".to_string()))