chore: merge develop preserving documentation structure and console UX
Integrate develop at 523bade3 while retaining concise agent guidance, rustdoc-owned design rationale, and current console history/rewind behavior. Adapt incoming tests to public entry-ID rewind and synchronize OpenAPI/TypeScript artifacts.
Validation: default-member cargo test --all-features (4936 passed, 23 ignored), root cargo check, all-feature rustdoc with broken links denied, format/diff/name guards, OpenAPI and 11 TypeScript freshness checks. Web: 613 unit, 327 component, 7 Ticket browser and 3 Console browser tests; check/build passed. Merged-state Ticket visual review at 1440/768/390px in light/dark; no matching pre-merge image baseline. E2E/live-provider checks not run.
This commit is contained in:
@@ -37,6 +37,7 @@ chrono = { version = "0.4", default-features = false, features = ["serde", "cloc
|
||||
agen-macros = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
tokio = { workspace = true, features = ["test-util"] }
|
||||
clap = { version = "4.5", features = ["derive", "env"] }
|
||||
tempfile = { workspace = true }
|
||||
dotenv = "0.15"
|
||||
|
||||
+171
-127
@@ -88,6 +88,63 @@ pub enum EngineError {
|
||||
ToolAttemptFence(String),
|
||||
}
|
||||
|
||||
/// Policy completion is not permission to start work when an interruption was
|
||||
/// queued during its final poll. Cancel wins when both requests are observed.
|
||||
fn check_policy_interruption(
|
||||
cancel_rx: &mut mpsc::Receiver<()>,
|
||||
pause_rx: &mut mpsc::Receiver<()>,
|
||||
) -> Result<(), EngineError> {
|
||||
if cancel_rx.try_recv().is_ok() {
|
||||
return Err(EngineError::Cancelled);
|
||||
}
|
||||
if pause_rx.try_recv().is_ok() {
|
||||
return Err(EngineError::PauseRequested);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Poll one policy invocation, never a second copy of its side effects. Dropping
|
||||
/// a pending policy stops admission, not any already-owned tool execution; the
|
||||
/// caller must drain that execution through its original attempt fence.
|
||||
async fn await_tool_policy<T>(
|
||||
policy: impl Future<Output = Result<T, InterceptorError>>,
|
||||
phase: InterceptorPhase,
|
||||
cancel_rx: &mut mpsc::Receiver<()>,
|
||||
pause_rx: &mut mpsc::Receiver<()>,
|
||||
) -> Result<T, EngineError> {
|
||||
let result = tokio::select! {
|
||||
biased;
|
||||
Some(()) = cancel_rx.recv() => return Err(EngineError::Cancelled),
|
||||
Some(()) = pause_rx.recv() => return Err(EngineError::PauseRequested),
|
||||
result = policy => result,
|
||||
};
|
||||
check_policy_interruption(cancel_rx, pause_rx)?;
|
||||
result.map_err(|error| InterceptorFailure::new(phase, error).into())
|
||||
}
|
||||
|
||||
/// A consumed Pause still bounds post-policy processing while owned tools drain.
|
||||
/// Terminal history and publication happen before this wait, so expiring it
|
||||
/// cannot discard the result or reopen the execution attempt.
|
||||
#[derive(Clone, Copy)]
|
||||
enum PostToolPolicyWait {
|
||||
Skip,
|
||||
Interruptible,
|
||||
UntilPauseSafeBoundary(TokioInstant),
|
||||
}
|
||||
|
||||
impl PostToolPolicyWait {
|
||||
fn for_pause_deadline(deadline: Option<TokioInstant>) -> Self {
|
||||
deadline.map_or(Self::Interruptible, Self::UntilPauseSafeBoundary)
|
||||
}
|
||||
|
||||
fn pause_deadline(self) -> Option<TokioInstant> {
|
||||
match self {
|
||||
Self::UntilPauseSafeBoundary(deadline) => Some(deadline),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Tool registration error
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum ToolRegistryError {
|
||||
@@ -770,14 +827,6 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
}
|
||||
}
|
||||
|
||||
fn try_paused(&mut self) -> bool {
|
||||
use tokio::sync::mpsc::error::TryRecvError;
|
||||
match self.pause_rx.try_recv() {
|
||||
Ok(()) => true,
|
||||
Err(TryRecvError::Empty | TryRecvError::Disconnected) => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Register a text block observer with scoped callbacks.
|
||||
///
|
||||
/// The setup closure is called once per text block. Inside it, register
|
||||
@@ -975,10 +1024,10 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
|
||||
/// Register a callback invoked once per completed tool execution.
|
||||
///
|
||||
/// Fired after `post_tool_call` interceptors and any `content`
|
||||
/// truncation from `tool_output_limits`, so the callback observes
|
||||
/// exactly what is persisted to history. Intended for callers that need
|
||||
/// to forward tool results to clients.
|
||||
/// Fired after `content` truncation from `tool_output_limits` and terminal
|
||||
/// history commit, before `post_tool_call` policy. The callback observes
|
||||
/// exactly what is persisted even if that policy is interrupted. Intended
|
||||
/// for callers that need to forward tool results to clients.
|
||||
pub fn on_tool_result(&mut self, callback: impl Fn(&ToolResult) + Send + Sync + 'static) {
|
||||
self.tool_result_cbs.push(Box::new(callback));
|
||||
}
|
||||
@@ -1552,43 +1601,24 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
tool,
|
||||
context,
|
||||
};
|
||||
let action = tokio::select! {
|
||||
result = self.interceptor.pre_tool_call(&mut info) => {
|
||||
result.map_err(|error| {
|
||||
EngineError::from(InterceptorFailure::new(
|
||||
InterceptorPhase::PreToolCall,
|
||||
error,
|
||||
))
|
||||
})?
|
||||
}
|
||||
pause = self.pause_rx.recv() => {
|
||||
if pause.is_some() {
|
||||
info!(call_id = %expected_tool_use_id, "Paused while awaiting pre-tool policy");
|
||||
}
|
||||
self.append_early_tool_call(
|
||||
history,
|
||||
annotate,
|
||||
&tool_call,
|
||||
call_index,
|
||||
None,
|
||||
)?;
|
||||
let action = await_tool_policy(
|
||||
self.interceptor.pre_tool_call(&mut info),
|
||||
InterceptorPhase::PreToolCall,
|
||||
&mut self.cancel_rx,
|
||||
&mut self.pause_rx,
|
||||
)
|
||||
.await;
|
||||
let action = match action {
|
||||
Ok(action) => action,
|
||||
Err(error @ (EngineError::PauseRequested | EngineError::Cancelled)) => {
|
||||
self.append_early_tool_call(history, annotate, &tool_call, call_index, None)?;
|
||||
batch.dispatched_any = true;
|
||||
return Ok(EarlyToolAdmission::Pause);
|
||||
}
|
||||
cancel = self.cancel_rx.recv() => {
|
||||
if cancel.is_some() {
|
||||
info!(call_id = %expected_tool_use_id, "Cancelled while awaiting pre-tool policy");
|
||||
}
|
||||
self.append_early_tool_call(
|
||||
history,
|
||||
annotate,
|
||||
&tool_call,
|
||||
call_index,
|
||||
None,
|
||||
)?;
|
||||
batch.dispatched_any = true;
|
||||
return Err(EngineError::Cancelled);
|
||||
return match error {
|
||||
EngineError::PauseRequested => Ok(EarlyToolAdmission::Pause),
|
||||
_ => Err(error),
|
||||
};
|
||||
}
|
||||
Err(error) => return Err(error),
|
||||
};
|
||||
if info.call.id != expected_tool_use_id {
|
||||
return Err(InterceptorFailure::new(
|
||||
@@ -1737,7 +1767,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
result,
|
||||
None,
|
||||
batch.turn_id,
|
||||
true,
|
||||
PostToolPolicyWait::Interruptible,
|
||||
&batch.call_info_map,
|
||||
&mut batch.attempt_fence,
|
||||
&mut batch.terminal_call_ids,
|
||||
@@ -1764,7 +1794,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
result,
|
||||
None,
|
||||
batch.turn_id,
|
||||
false,
|
||||
PostToolPolicyWait::Skip,
|
||||
&batch.call_info_map,
|
||||
&mut batch.attempt_fence,
|
||||
&mut batch.terminal_call_ids,
|
||||
@@ -1812,7 +1842,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
result,
|
||||
Some(&attempt_id),
|
||||
batch.turn_id,
|
||||
false,
|
||||
PostToolPolicyWait::Skip,
|
||||
&batch.call_info_map,
|
||||
&mut batch.attempt_fence,
|
||||
&mut batch.terminal_call_ids,
|
||||
@@ -1835,7 +1865,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
ToolResult::outcome_unknown(call_id),
|
||||
Some(attempt_id),
|
||||
batch.turn_id,
|
||||
false,
|
||||
PostToolPolicyWait::Skip,
|
||||
&batch.call_info_map,
|
||||
&mut batch.attempt_fence,
|
||||
&mut batch.terminal_call_ids,
|
||||
@@ -1858,28 +1888,6 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
force_cancel: bool,
|
||||
initial_pause_requested: bool,
|
||||
) -> Result<ToolExecutionResult, EngineError> {
|
||||
while let Some(result) = batch.synthetic_results.pop_front() {
|
||||
if let Err(error) = self
|
||||
.finalize_and_commit_tool_result(
|
||||
history,
|
||||
annotate,
|
||||
result,
|
||||
None,
|
||||
batch.turn_id,
|
||||
true,
|
||||
&batch.call_info_map,
|
||||
&mut batch.attempt_fence,
|
||||
&mut batch.terminal_call_ids,
|
||||
)
|
||||
.await
|
||||
{
|
||||
let cleanup_error = self
|
||||
.cancel_and_terminalize_early_batch(history, annotate, &mut batch)
|
||||
.await;
|
||||
return Err(cleanup_error.unwrap_or(error));
|
||||
}
|
||||
}
|
||||
|
||||
if force_cancel {
|
||||
if let Some(error) = self
|
||||
.cancel_and_terminalize_early_batch(history, annotate, &mut batch)
|
||||
@@ -1893,6 +1901,28 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
let mut pause_requested = initial_pause_requested;
|
||||
let mut pause_deadline = initial_pause_requested
|
||||
.then(|| TokioInstant::now() + self.tool_execution_policy.pause_safe_boundary_timeout);
|
||||
while let Some(result) = batch.synthetic_results.pop_front() {
|
||||
if let Err(error) = self
|
||||
.finalize_and_commit_tool_result(
|
||||
history,
|
||||
annotate,
|
||||
result,
|
||||
None,
|
||||
batch.turn_id,
|
||||
PostToolPolicyWait::for_pause_deadline(pause_deadline),
|
||||
&batch.call_info_map,
|
||||
&mut batch.attempt_fence,
|
||||
&mut batch.terminal_call_ids,
|
||||
)
|
||||
.await
|
||||
{
|
||||
let cleanup_error = self
|
||||
.cancel_and_terminalize_early_batch(history, annotate, &mut batch)
|
||||
.await;
|
||||
return Err(cleanup_error.unwrap_or(error));
|
||||
}
|
||||
}
|
||||
|
||||
while !batch.futures.is_empty() {
|
||||
tokio::select! {
|
||||
biased;
|
||||
@@ -1904,7 +1934,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
result,
|
||||
Some(&attempt_id),
|
||||
batch.turn_id,
|
||||
true,
|
||||
PostToolPolicyWait::for_pause_deadline(pause_deadline),
|
||||
&batch.call_info_map,
|
||||
&mut batch.attempt_fence,
|
||||
&mut batch.terminal_call_ids,
|
||||
@@ -1966,8 +1996,10 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
|
||||
// A pause observed before provider ownership starts leaves every call
|
||||
// NotStarted and therefore eligible for an explicit later retry.
|
||||
if self.try_paused() {
|
||||
return Ok(ToolExecutionResult::Paused);
|
||||
match check_policy_interruption(&mut self.cancel_rx, &mut self.pause_rx) {
|
||||
Ok(()) => {}
|
||||
Err(EngineError::PauseRequested) => return Ok(ToolExecutionResult::Paused),
|
||||
Err(error) => return Err(error),
|
||||
}
|
||||
|
||||
// Map from tool call ID to (ToolCall, Meta, Tool, Context)
|
||||
@@ -2006,16 +2038,18 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
context,
|
||||
};
|
||||
|
||||
let pre_tool_action =
|
||||
self.interceptor
|
||||
.pre_tool_call(&mut info)
|
||||
.await
|
||||
.map_err(|error| {
|
||||
EngineError::from(InterceptorFailure::new(
|
||||
InterceptorPhase::PreToolCall,
|
||||
error,
|
||||
))
|
||||
})?;
|
||||
let pre_tool_action = match await_tool_policy(
|
||||
self.interceptor.pre_tool_call(&mut info),
|
||||
InterceptorPhase::PreToolCall,
|
||||
&mut self.cancel_rx,
|
||||
&mut self.pause_rx,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(action) => action,
|
||||
Err(EngineError::PauseRequested) => return Ok(ToolExecutionResult::Paused),
|
||||
Err(error) => return Err(error),
|
||||
};
|
||||
if info.call.id != expected_tool_use_id {
|
||||
return Err(InterceptorFailure::new(
|
||||
InterceptorPhase::PreToolCall,
|
||||
@@ -2082,6 +2116,14 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
}
|
||||
}
|
||||
|
||||
// Recheck after all policies and immediately before provider ownership.
|
||||
// No tool task exists yet: these calls remain proven NotStarted.
|
||||
match check_policy_interruption(&mut self.cancel_rx, &mut self.pause_rx) {
|
||||
Ok(()) => {}
|
||||
Err(EngineError::PauseRequested) => return Ok(ToolExecutionResult::Paused),
|
||||
Err(error) => return Err(error),
|
||||
}
|
||||
|
||||
// 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.
|
||||
@@ -2158,7 +2200,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
result,
|
||||
None,
|
||||
turn_id,
|
||||
true,
|
||||
PostToolPolicyWait::Interruptible,
|
||||
&call_info_map,
|
||||
&mut attempt_fence,
|
||||
&mut terminal_call_ids,
|
||||
@@ -2190,7 +2232,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
result,
|
||||
Some(&attempt_id),
|
||||
turn_id,
|
||||
true,
|
||||
PostToolPolicyWait::for_pause_deadline(pause_deadline),
|
||||
&call_info_map,
|
||||
&mut attempt_fence,
|
||||
&mut terminal_call_ids,
|
||||
@@ -2271,7 +2313,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
result,
|
||||
Some(&attempt_id),
|
||||
turn_id,
|
||||
false,
|
||||
PostToolPolicyWait::Skip,
|
||||
&call_info_map,
|
||||
&mut attempt_fence,
|
||||
&mut terminal_call_ids,
|
||||
@@ -2299,7 +2341,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
ToolResult::outcome_unknown(call_id),
|
||||
Some(attempt_id),
|
||||
turn_id,
|
||||
false,
|
||||
PostToolPolicyWait::Skip,
|
||||
&call_info_map,
|
||||
&mut attempt_fence,
|
||||
&mut terminal_call_ids,
|
||||
@@ -2340,8 +2382,8 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
})
|
||||
}
|
||||
|
||||
/// Apply post-execution policy, bound the model-visible payload, durably
|
||||
/// append one terminal ToolResult, and only then publish it to observers.
|
||||
/// Bound the model-visible payload, durably append one terminal ToolResult,
|
||||
/// publish it, then await interruptible post-execution policy.
|
||||
async fn finalize_and_commit_tool_result(
|
||||
&mut self,
|
||||
history: &mut History<A>,
|
||||
@@ -2349,7 +2391,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
mut tool_result: ToolResult,
|
||||
execution_attempt_id: Option<&str>,
|
||||
turn_id: usize,
|
||||
invoke_post_hook: bool,
|
||||
post_policy_wait: PostToolPolicyWait,
|
||||
call_info_map: &HashMap<
|
||||
String,
|
||||
(
|
||||
@@ -2434,7 +2476,9 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
);
|
||||
self.emit_tool_result(&tool_result);
|
||||
|
||||
if invoke_post_hook && let Some((tool_call, meta, tool, context)) = call_info {
|
||||
if !matches!(post_policy_wait, PostToolPolicyWait::Skip)
|
||||
&& let Some((tool_call, meta, tool, context)) = call_info
|
||||
{
|
||||
let invocation = self.interceptor_invocation(
|
||||
InterceptorPhase::PostToolCall,
|
||||
Some(turn_id),
|
||||
@@ -2450,27 +2494,20 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
tool: tool.clone(),
|
||||
context: context.clone(),
|
||||
};
|
||||
let pause_deadline = post_policy_wait.pause_deadline();
|
||||
let post_tool_action = tokio::select! {
|
||||
result = self.interceptor.post_tool_call(&info) => {
|
||||
result.map_err(|error| {
|
||||
EngineError::from(InterceptorFailure::new(
|
||||
InterceptorPhase::PostToolCall,
|
||||
error,
|
||||
))
|
||||
})?
|
||||
}
|
||||
pause = self.pause_rx.recv() => {
|
||||
if pause.is_some() {
|
||||
info!(call_id = %tool_call.id, "Paused while awaiting post-tool policy");
|
||||
}
|
||||
biased;
|
||||
// Poll the existing safe-boundary deadline even when the Pause
|
||||
// that established it was consumed by an outer policy gate.
|
||||
_ = tokio::time::sleep_until(pause_deadline.unwrap_or_else(TokioInstant::now)), if pause_deadline.is_some() => {
|
||||
return Err(EngineError::PauseRequested);
|
||||
}
|
||||
cancel = self.cancel_rx.recv() => {
|
||||
if cancel.is_some() {
|
||||
info!(call_id = %tool_call.id, "Cancelled while awaiting post-tool policy");
|
||||
}
|
||||
return Err(EngineError::Cancelled);
|
||||
}
|
||||
result = await_tool_policy(
|
||||
self.interceptor.post_tool_call(&info),
|
||||
InterceptorPhase::PostToolCall,
|
||||
&mut self.cancel_rx,
|
||||
&mut self.pause_rx,
|
||||
) => result?,
|
||||
};
|
||||
if let PostToolAction::Abort(reason) = post_tool_action {
|
||||
return Err(EngineError::Aborted(reason));
|
||||
@@ -2508,7 +2545,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
ToolResult::outcome_unknown(&call_id),
|
||||
None,
|
||||
self.turn_count,
|
||||
false,
|
||||
PostToolPolicyWait::Skip,
|
||||
&call_info_map,
|
||||
&mut attempt_fence,
|
||||
&mut terminal_call_ids,
|
||||
@@ -2970,22 +3007,29 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
)),
|
||||
0,
|
||||
);
|
||||
let assistant_turn_action = self
|
||||
.interceptor
|
||||
.on_assistant_turn_end(AssistantTurnEndContext {
|
||||
invocation: assistant_invocation,
|
||||
assistant_entries: &history.entries()[assistant_start..],
|
||||
history: history.entries(),
|
||||
tool_calls: &tool_calls,
|
||||
})
|
||||
.await;
|
||||
let assistant_turn_action = await_tool_policy(
|
||||
self.interceptor
|
||||
.on_assistant_turn_end(AssistantTurnEndContext {
|
||||
invocation: assistant_invocation,
|
||||
assistant_entries: &history.entries()[assistant_start..],
|
||||
history: history.entries(),
|
||||
tool_calls: &tool_calls,
|
||||
}),
|
||||
InterceptorPhase::AssistantTurnEnd,
|
||||
&mut self.cancel_rx,
|
||||
&mut self.pause_rx,
|
||||
)
|
||||
.await;
|
||||
let assistant_turn_action = match assistant_turn_action {
|
||||
Ok(action) => action,
|
||||
Err(EngineError::PauseRequested) => {
|
||||
if let Some(batch) = early_tools.take() {
|
||||
self.drain_early_tool_batch(history, annotate, batch, false, true)
|
||||
.await?;
|
||||
}
|
||||
return Ok(EngineResult::Paused);
|
||||
}
|
||||
Err(error) => {
|
||||
let error = EngineError::from(InterceptorFailure::new(
|
||||
InterceptorPhase::AssistantTurnEnd,
|
||||
error,
|
||||
));
|
||||
if let Some(batch) = early_tools.take()
|
||||
&& let Err(cleanup_error) = self
|
||||
.drain_early_tool_batch(history, annotate, batch, true, false)
|
||||
@@ -3551,7 +3595,7 @@ impl<C: LlmClient, S: EngineState, A: Send + Sync> Engine<C, S, A> {
|
||||
result,
|
||||
Some(&attempt_id),
|
||||
batch.turn_id,
|
||||
true,
|
||||
PostToolPolicyWait::Interruptible,
|
||||
&batch.call_info_map,
|
||||
&mut batch.attempt_fence,
|
||||
&mut batch.terminal_call_ids,
|
||||
|
||||
@@ -4,6 +4,13 @@
|
||||
//! keeping host-specific policy out of the Engine. Additional model input must go
|
||||
//! through history appends: inserting new facts only into an in-flight request
|
||||
//! would make its input impossible to reconstruct from stored history.
|
||||
//!
|
||||
//! Tool pre/post policy and assistant-turn-end waits are interruptible by Cancel or Pause;
|
||||
//! Cancel wins when both are observed. Pending policy futures may be dropped, not replayed, so
|
||||
//! hosts must not depend on their completion for cleanup of already-owned effects. The Engine
|
||||
//! drains owned tool attempts through their original fences. Terminal tool history and result
|
||||
//! publication precede post-tool policy; interruption cannot erase that result. Pause draining
|
||||
//! bounds post-tool waits by the existing safe-boundary deadline.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
|
||||
@@ -106,15 +106,23 @@ pub async fn load(path: &Path) -> Result<AuthSnapshot, CodexAuthError> {
|
||||
/// Reload the existing file, update `tokens.{id_token,access_token,refresh_token}`
|
||||
/// and `last_refresh`, and write it back. Equivalent to Codex CLI's `persist_tokens`.
|
||||
///
|
||||
/// Reload and merge before writing to preserve fields if a concurrent
|
||||
/// Codex CLI / another process has already refreshed.
|
||||
/// Reload and merge before writing to preserve fields updated by a concurrent
|
||||
/// Codex CLI / another process. Account and refresh token must still match the
|
||||
/// request's input: merge unrelated fields, but return the current snapshot
|
||||
/// without writing if another login or token rotation is observed.
|
||||
pub async fn persist_refreshed(
|
||||
path: &Path,
|
||||
refresh_input: &AuthSnapshot,
|
||||
new_id_token: Option<String>,
|
||||
new_access_token: Option<String>,
|
||||
new_refresh_token: Option<String>,
|
||||
) -> Result<AuthSnapshot, CodexAuthError> {
|
||||
let mut current = load(path).await?;
|
||||
if current.account_id != refresh_input.account_id
|
||||
|| current.refresh_token != refresh_input.refresh_token
|
||||
{
|
||||
return Ok(current);
|
||||
}
|
||||
let raw = &mut current.raw;
|
||||
let tokens = raw
|
||||
.get_mut("tokens")
|
||||
@@ -243,10 +251,16 @@ mod tests {
|
||||
"agent_identity":{"workspace_id":"w","agent_runtime_id":"r","agent_private_key":"k","registered_at":"x"}
|
||||
}"#,
|
||||
);
|
||||
let updated =
|
||||
persist_refreshed(&path, None, Some("new-acc".into()), Some("new-ref".into()))
|
||||
.await
|
||||
.unwrap();
|
||||
let input = load(&path).await.unwrap();
|
||||
let updated = persist_refreshed(
|
||||
&path,
|
||||
&input,
|
||||
None,
|
||||
Some("new-acc".into()),
|
||||
Some("new-ref".into()),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(updated.access_token, "new-acc");
|
||||
assert_eq!(updated.refresh_token, "new-ref");
|
||||
// The unknown field agent_identity is preserved
|
||||
@@ -266,7 +280,8 @@ mod tests {
|
||||
);
|
||||
// Change the existing file to 644, then persist → check that it returns to 600
|
||||
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
|
||||
persist_refreshed(&path, None, Some("a2".into()), None)
|
||||
let input = load(&path).await.unwrap();
|
||||
persist_refreshed(&path, &input, None, Some("a2".into()), None)
|
||||
.await
|
||||
.unwrap();
|
||||
let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
|
||||
|
||||
@@ -17,6 +17,8 @@ mod auth_json;
|
||||
mod error;
|
||||
mod jwt;
|
||||
mod refresh;
|
||||
#[cfg(test)]
|
||||
mod refresh_tests;
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
@@ -40,6 +42,9 @@ pub struct CodexAuthProvider {
|
||||
auth_path: PathBuf,
|
||||
refresh_endpoint: String,
|
||||
http: reqwest::Client,
|
||||
refresh_limits: refresh::RefreshLimits,
|
||||
#[cfg(test)]
|
||||
before_guarded_reload: Option<Arc<tokio::sync::Barrier>>,
|
||||
state: Arc<Mutex<State>>,
|
||||
}
|
||||
|
||||
@@ -76,6 +81,9 @@ impl CodexAuthProvider {
|
||||
auth_path: codex_home.join("auth.json"),
|
||||
refresh_endpoint: refresh::REFRESH_URL.to_string(),
|
||||
http: reqwest::Client::new(),
|
||||
refresh_limits: refresh::RefreshLimits::default(),
|
||||
#[cfg(test)]
|
||||
before_guarded_reload: None,
|
||||
state: Arc::new(Mutex::new(State::default())),
|
||||
}
|
||||
}
|
||||
@@ -106,8 +114,13 @@ impl CodexAuthProvider {
|
||||
return Ok(snap);
|
||||
}
|
||||
|
||||
// 3. Reload immediately before refreshing and check that account_id matches (guarded reload).
|
||||
// If it differs, another process has already updated it → skip refreshing
|
||||
// 3. Reload immediately before refreshing (guarded reload).
|
||||
// Skip refresh if another process supplied fresh tokens or changed account_id.
|
||||
#[cfg(test)]
|
||||
if let Some(barrier) = &self.before_guarded_reload {
|
||||
barrier.wait().await;
|
||||
barrier.wait().await;
|
||||
}
|
||||
let pre_refresh = auth_json::load(&self.auth_path).await?;
|
||||
if !is_stale(&pre_refresh) {
|
||||
state.cached = Some(pre_refresh.clone());
|
||||
@@ -124,12 +137,14 @@ impl CodexAuthProvider {
|
||||
&self.http,
|
||||
&self.refresh_endpoint,
|
||||
&pre_refresh.refresh_token,
|
||||
&self.refresh_limits,
|
||||
)
|
||||
.await?;
|
||||
|
||||
// 5. Write back (persist_refreshed reloads and merges before writing)
|
||||
let new_snap = auth_json::persist_refreshed(
|
||||
&self.auth_path,
|
||||
&pre_refresh,
|
||||
refreshed.id_token,
|
||||
refreshed.access_token,
|
||||
refreshed.refresh_token,
|
||||
@@ -210,13 +225,18 @@ mod tests {
|
||||
use wiremock::matchers::{method, path as url_path};
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
fn make_jwt(payload: serde_json::Value) -> String {
|
||||
pub(super) fn make_jwt(payload: serde_json::Value) -> String {
|
||||
let p = base64::engine::general_purpose::URL_SAFE_NO_PAD
|
||||
.encode(serde_json::to_vec(&payload).unwrap());
|
||||
format!("h.{p}.s")
|
||||
}
|
||||
|
||||
fn write_auth(dir: &std::path::Path, exp: i64, fedramp: bool, refresh: &str) -> PathBuf {
|
||||
pub(super) fn write_auth(
|
||||
dir: &std::path::Path,
|
||||
exp: i64,
|
||||
fedramp: bool,
|
||||
refresh: &str,
|
||||
) -> PathBuf {
|
||||
let id_token = make_jwt(json!({
|
||||
"https://api.openai.com/auth": {
|
||||
"chatgpt_account_id": "acc-xyz",
|
||||
|
||||
@@ -10,7 +10,26 @@ use super::error::{CodexAuthError, PermanentReason};
|
||||
|
||||
pub const CLIENT_ID: &str = "app_EMoamEEZ73f0CkXaXp7hrann";
|
||||
pub const REFRESH_URL: &str = "https://auth.openai.com/oauth/token";
|
||||
pub const DEFAULT_REFRESH_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
/// Limits cover send, both response bodies, and decode, not auth file I/O.
|
||||
/// OAuth responses contain only a few tokens; 64 KiB leaves ample JWT headroom.
|
||||
#[derive(Clone)]
|
||||
pub(super) struct RefreshLimits {
|
||||
pub timeout: Duration,
|
||||
pub max_body_bytes: usize,
|
||||
#[cfg(test)]
|
||||
pub body_started: Option<std::sync::Arc<tokio::sync::Notify>>,
|
||||
}
|
||||
|
||||
impl Default for RefreshLimits {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
timeout: Duration::from_secs(30),
|
||||
max_body_bytes: 64 * 1024,
|
||||
#[cfg(test)]
|
||||
body_started: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct RefreshRequest<'a> {
|
||||
@@ -37,54 +56,94 @@ pub async fn request_refresh(
|
||||
client: &reqwest::Client,
|
||||
endpoint: &str,
|
||||
refresh_token: &str,
|
||||
limits: &RefreshLimits,
|
||||
) -> Result<RefreshResponse, CodexAuthError> {
|
||||
let deadline = tokio::time::Instant::now() + limits.timeout;
|
||||
let result = tokio::time::timeout_at(
|
||||
deadline,
|
||||
request_refresh_bounded(client, endpoint, refresh_token, limits),
|
||||
)
|
||||
.await;
|
||||
// JSON decode is synchronous. The byte bound limits its work, and this
|
||||
// post-check prevents a decode that crosses the deadline from being persisted.
|
||||
if tokio::time::Instant::now() >= deadline {
|
||||
return Err(CodexAuthError::RefreshTransient(
|
||||
"oauth_token_refresh timed out".into(),
|
||||
));
|
||||
}
|
||||
result.map_err(|_| CodexAuthError::RefreshTransient("oauth_token_refresh timed out".into()))?
|
||||
}
|
||||
|
||||
async fn request_refresh_bounded(
|
||||
client: &reqwest::Client,
|
||||
endpoint: &str,
|
||||
refresh_token: &str,
|
||||
limits: &RefreshLimits,
|
||||
) -> Result<RefreshResponse, CodexAuthError> {
|
||||
let body = RefreshRequest {
|
||||
client_id: CLIENT_ID,
|
||||
grant_type: "refresh_token",
|
||||
refresh_token,
|
||||
};
|
||||
let response = response_with_timeout(
|
||||
client
|
||||
.post(endpoint)
|
||||
.header("Content-Type", "application/json")
|
||||
.json(&body)
|
||||
.send(),
|
||||
DEFAULT_REFRESH_TIMEOUT,
|
||||
)
|
||||
.await?;
|
||||
let mut response = client
|
||||
.post(endpoint)
|
||||
.header("Content-Type", "application/json")
|
||||
.json(&body)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|_| CodexAuthError::RefreshTransient("send failed".into()))?;
|
||||
|
||||
let status = response.status();
|
||||
if status.is_success() {
|
||||
response
|
||||
.json::<RefreshResponse>()
|
||||
.await
|
||||
.map_err(|e| CodexAuthError::RefreshTransient(format!("parse response: {e}")))
|
||||
} else {
|
||||
let body = response.text().await.unwrap_or_default();
|
||||
if status == reqwest::StatusCode::UNAUTHORIZED {
|
||||
let (reason, message) = classify_permanent(&body);
|
||||
Err(CodexAuthError::RefreshPermanent { reason, message })
|
||||
} else {
|
||||
Err(CodexAuthError::RefreshTransient(format!(
|
||||
"{status}: {body}"
|
||||
)))
|
||||
if response
|
||||
.content_length()
|
||||
.is_some_and(|len| len > limits.max_body_bytes as u64)
|
||||
{
|
||||
return Err(body_too_large());
|
||||
}
|
||||
#[cfg(test)]
|
||||
if let Some(started) = &limits.body_started {
|
||||
started.notify_one();
|
||||
}
|
||||
let mut bytes = Vec::new();
|
||||
while let Some(chunk) = response
|
||||
.chunk()
|
||||
.await
|
||||
.map_err(|_| CodexAuthError::RefreshTransient("read response body failed".into()))?
|
||||
{
|
||||
// Content-Length can be absent or untrustworthy. Bound actual bytes too,
|
||||
// before copying a chunk into the accumulated body.
|
||||
if chunk.len() > limits.max_body_bytes.saturating_sub(bytes.len()) {
|
||||
return Err(body_too_large());
|
||||
}
|
||||
bytes.extend_from_slice(&chunk);
|
||||
}
|
||||
if status.is_success() {
|
||||
let tokens: RefreshResponse = serde_json::from_slice(&bytes).map_err(|_| {
|
||||
// Do not expose untrusted body values (which can contain secrets).
|
||||
CodexAuthError::RefreshTransient("decode response failed".into())
|
||||
})?;
|
||||
if tokens.access_token.as_deref().is_none_or(str::is_empty)
|
||||
|| tokens.id_token.as_deref().is_some_and(str::is_empty)
|
||||
|| tokens.refresh_token.as_deref().is_some_and(str::is_empty)
|
||||
{
|
||||
return Err(CodexAuthError::RefreshTransient(
|
||||
"incomplete token response".into(),
|
||||
));
|
||||
}
|
||||
Ok(tokens)
|
||||
} else if status == reqwest::StatusCode::UNAUTHORIZED {
|
||||
let (reason, message) = classify_permanent(&String::from_utf8_lossy(&bytes));
|
||||
Err(CodexAuthError::RefreshPermanent { reason, message })
|
||||
} else {
|
||||
// Error bodies are consumed within the same limits but never echoed.
|
||||
Err(CodexAuthError::RefreshTransient(format!(
|
||||
"refresh endpoint returned {status}"
|
||||
)))
|
||||
}
|
||||
}
|
||||
|
||||
async fn response_with_timeout(
|
||||
future: impl std::future::Future<Output = Result<reqwest::Response, reqwest::Error>>,
|
||||
timeout: Duration,
|
||||
) -> Result<reqwest::Response, CodexAuthError> {
|
||||
tokio::time::timeout(timeout, future)
|
||||
.await
|
||||
.map_err(|_| {
|
||||
CodexAuthError::RefreshTransient(format!(
|
||||
"oauth_token_refresh timed out after {}s",
|
||||
timeout.as_secs()
|
||||
))
|
||||
})?
|
||||
.map_err(|e| CodexAuthError::RefreshTransient(format!("send: {e}")))
|
||||
fn body_too_large() -> CodexAuthError {
|
||||
CodexAuthError::RefreshTransient("response body exceeds size limit".into())
|
||||
}
|
||||
|
||||
fn classify_permanent(body: &str) -> (PermanentReason, String) {
|
||||
@@ -99,7 +158,7 @@ fn classify_permanent(body: &str) -> (PermanentReason, String) {
|
||||
PermanentReason::Expired => "Your refresh token has expired".to_string(),
|
||||
PermanentReason::Reused => "Your refresh token was already used".to_string(),
|
||||
PermanentReason::Revoked => "Your refresh token was revoked".to_string(),
|
||||
PermanentReason::Other => format!("Unknown 401 from refresh endpoint: {body}"),
|
||||
PermanentReason::Other => "Unknown 401 from refresh endpoint".to_string(),
|
||||
};
|
||||
(reason, message)
|
||||
}
|
||||
@@ -126,23 +185,6 @@ fn extract_error_code(body: &str) -> Option<String> {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn refresh_response_timeout_is_transient() {
|
||||
let err = match response_with_timeout(
|
||||
std::future::pending::<Result<reqwest::Response, reqwest::Error>>(),
|
||||
Duration::from_millis(5),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(_) => panic!("expected refresh timeout"),
|
||||
Err(err) => err,
|
||||
};
|
||||
|
||||
assert!(
|
||||
matches!(err, CodexAuthError::RefreshTransient(message) if message.contains("timed out"))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_expired() {
|
||||
let body = r#"{"error":{"code":"refresh_token_expired"}}"#;
|
||||
|
||||
@@ -0,0 +1,431 @@
|
||||
//! Real AuthProvider calls against owned auth files and a loopback HTTP peer.
|
||||
//! Gates place callers at response-body / reload boundaries; paused Tokio time
|
||||
//! advances the injected HTTP deadline only after headers have been received.
|
||||
use super::tests::{make_jwt, write_auth};
|
||||
use super::*;
|
||||
use serde_json::{Value, json};
|
||||
use std::time::Duration as StdDuration;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::{Notify, mpsc};
|
||||
use tokio::task::{JoinHandle, JoinSet};
|
||||
|
||||
struct Reply {
|
||||
status: u16,
|
||||
body: Vec<u8>,
|
||||
chunked: bool,
|
||||
gate: Option<Arc<Notify>>,
|
||||
truncate: bool,
|
||||
}
|
||||
|
||||
impl Reply {
|
||||
fn body(status: u16, body: impl Into<Vec<u8>>) -> Self {
|
||||
Self {
|
||||
status,
|
||||
body: body.into(),
|
||||
chunked: false,
|
||||
gate: None,
|
||||
truncate: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn gated(status: u16, body: impl Into<Vec<u8>>, gate: &Arc<Notify>) -> Self {
|
||||
Self {
|
||||
gate: Some(gate.clone()),
|
||||
..Self::body(status, body)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct Stub {
|
||||
endpoint: String,
|
||||
requests: mpsc::UnboundedReceiver<Value>,
|
||||
task: JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl Drop for Stub {
|
||||
fn drop(&mut self) {
|
||||
// Dropping the server's JoinSet also aborts stalled connection handlers.
|
||||
self.task.abort();
|
||||
}
|
||||
}
|
||||
|
||||
impl Stub {
|
||||
async fn start(replies: Vec<Reply>) -> Self {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let endpoint = format!("http://{}/oauth/token", listener.local_addr().unwrap());
|
||||
let (tx, requests) = mpsc::unbounded_channel();
|
||||
let task = tokio::spawn(async move {
|
||||
let mut handlers = JoinSet::new();
|
||||
for reply in replies {
|
||||
let (mut stream, _) = listener.accept().await.unwrap();
|
||||
let tx = tx.clone();
|
||||
handlers.spawn(async move {
|
||||
let mut request = Vec::new();
|
||||
let header_end = loop {
|
||||
let byte = stream.read_u8().await.unwrap();
|
||||
request.push(byte);
|
||||
if request.ends_with(b"\r\n\r\n") { break request.len(); }
|
||||
assert!(request.len() < 16 * 1024, "fixture request headers too large");
|
||||
};
|
||||
let headers = std::str::from_utf8(&request).unwrap();
|
||||
assert!(headers.starts_with("POST /oauth/token HTTP/1.1\r\n"));
|
||||
let len: usize = headers.lines().find_map(|line| {
|
||||
let (name, value) = line.split_once(':')?;
|
||||
name.eq_ignore_ascii_case("content-length").then(|| value.trim().parse().unwrap())
|
||||
}).unwrap();
|
||||
request.resize(header_end + len, 0);
|
||||
stream.read_exact(&mut request[header_end..]).await.unwrap();
|
||||
tx.send(serde_json::from_slice(&request[header_end..]).unwrap()).unwrap();
|
||||
let framing = if reply.chunked {
|
||||
"Transfer-Encoding: chunked".to_string()
|
||||
} else {
|
||||
format!("Content-Length: {}", reply.body.len())
|
||||
};
|
||||
stream.write_all(format!(
|
||||
"HTTP/1.1 {} Fixture\r\nContent-Type: application/json\r\n{framing}\r\nConnection: close\r\n\r\n", reply.status
|
||||
).as_bytes()).await.unwrap();
|
||||
if let Some(gate) = reply.gate { gate.notified().await; }
|
||||
let body = if reply.truncate { &reply.body[..reply.body.len() / 2] } else { &reply.body[..] };
|
||||
if reply.chunked {
|
||||
for chunk in body.chunks(512) {
|
||||
let _ = stream.write_all(format!("{:x}\r\n", chunk.len()).as_bytes()).await;
|
||||
let _ = stream.write_all(chunk).await;
|
||||
let _ = stream.write_all(b"\r\n").await;
|
||||
}
|
||||
let _ = stream.write_all(b"0\r\n\r\n").await;
|
||||
} else {
|
||||
let _ = stream.write_all(body).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
while let Some(result) = handlers.join_next().await {
|
||||
result.unwrap();
|
||||
}
|
||||
});
|
||||
Self {
|
||||
endpoint,
|
||||
requests,
|
||||
task,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn rotated_body() -> Vec<u8> {
|
||||
serde_json::to_vec(&json!({
|
||||
"access_token": make_jwt(json!({"exp": Utc::now().timestamp() + 3600})),
|
||||
"refresh_token": "fixture-rotated-secret",
|
||||
}))
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn provider(dir: &std::path::Path, stub: &Stub, started: &Arc<Notify>) -> Arc<CodexAuthProvider> {
|
||||
let mut provider = CodexAuthProvider::new(dir.to_path_buf())
|
||||
.with_refresh_endpoint(&stub.endpoint)
|
||||
.with_http_client(reqwest::Client::builder().no_proxy().build().unwrap());
|
||||
provider.refresh_limits = refresh::RefreshLimits {
|
||||
timeout: StdDuration::from_secs(10),
|
||||
max_body_bytes: 1024,
|
||||
body_started: Some(started.clone()),
|
||||
};
|
||||
Arc::new(provider)
|
||||
}
|
||||
|
||||
async fn bounded<F: std::future::Future>(future: F) -> F::Output {
|
||||
tokio::time::timeout(StdDuration::from_secs(5), future)
|
||||
.await
|
||||
.expect("fixture/caller did not progress")
|
||||
}
|
||||
|
||||
fn assert_rotated(path: &std::path::Path, headers: &[(HeaderName, HeaderValue)]) {
|
||||
let value: Value = serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap();
|
||||
assert_eq!(value["tokens"]["refresh_token"], "fixture-rotated-secret");
|
||||
assert_eq!(value["fixture_extra"], "preserve me");
|
||||
let auth = headers
|
||||
.iter()
|
||||
.find(|(name, _)| name == "authorization")
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
auth.1.to_str().unwrap(),
|
||||
format!(
|
||||
"Bearer {}",
|
||||
value["tokens"]["access_token"].as_str().unwrap()
|
||||
)
|
||||
);
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
assert_eq!(
|
||||
std::fs::metadata(path).unwrap().permissions().mode() & 0o777,
|
||||
0o600
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn stale_auth(dir: &std::path::Path) -> PathBuf {
|
||||
let path = write_auth(
|
||||
dir,
|
||||
Utc::now().timestamp() - 60,
|
||||
false,
|
||||
"fixture-old-secret",
|
||||
);
|
||||
let mut value: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
|
||||
value["fixture_extra"] = json!("preserve me");
|
||||
std::fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
|
||||
path
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stalled_success_and_error_bodies_timeout_without_persist_and_release_waiting_caller() {
|
||||
for status in [200, 401, 503] {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = stale_auth(dir.path());
|
||||
let original = std::fs::read(&path).unwrap();
|
||||
let gate = Arc::new(Notify::new());
|
||||
let started = Arc::new(Notify::new());
|
||||
let mut stub = Stub::start(vec![
|
||||
Reply::gated(status, b"unfinished".to_vec(), &gate),
|
||||
Reply::body(200, rotated_body()),
|
||||
])
|
||||
.await;
|
||||
let provider = provider(dir.path(), &stub, &started);
|
||||
let first = tokio::spawn({
|
||||
let provider = provider.clone();
|
||||
async move { provider.headers().await }
|
||||
});
|
||||
bounded(started.notified()).await; // provider has received headers and is consuming the body
|
||||
let request = bounded(stub.requests.recv()).await.unwrap();
|
||||
assert_eq!(request["refresh_token"], "fixture-old-secret");
|
||||
let second = provider.headers();
|
||||
tokio::pin!(second);
|
||||
assert!(futures::poll!(&mut second).is_pending());
|
||||
assert!(provider.state.try_lock().is_err());
|
||||
assert!(
|
||||
stub.requests.try_recv().is_err(),
|
||||
"waiter must not refresh concurrently"
|
||||
);
|
||||
tokio::time::pause();
|
||||
tokio::time::advance(StdDuration::from_secs(11)).await;
|
||||
let error = first.await.unwrap().unwrap_err();
|
||||
tokio::time::resume();
|
||||
assert!(error.to_string().contains("timed out"), "{error}");
|
||||
assert_eq!(std::fs::read(&path).unwrap(), original);
|
||||
let headers = bounded(second).await.unwrap();
|
||||
assert_rotated(&path, &headers);
|
||||
let retry = bounded(stub.requests.recv()).await.unwrap();
|
||||
assert_eq!(retry["refresh_token"], "fixture-old-secret");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn oversized_malformed_and_incomplete_bodies_do_not_persist_and_allow_later_refresh() {
|
||||
let mut cases = vec![
|
||||
(Reply::body(200, vec![b'x'; 1025]), "size limit"),
|
||||
(
|
||||
Reply {
|
||||
chunked: true,
|
||||
..Reply::body(200, vec![b'x'; 1025])
|
||||
},
|
||||
"size limit",
|
||||
),
|
||||
(
|
||||
Reply {
|
||||
chunked: true,
|
||||
..Reply::body(401, vec![b'x'; 1025])
|
||||
},
|
||||
"size limit",
|
||||
),
|
||||
(Reply::body(503, vec![b'x'; 1025]), "size limit"),
|
||||
(
|
||||
Reply::body(
|
||||
200,
|
||||
br#"{"access_token": {"secret":"fixture-body-secret"}}"#.to_vec(),
|
||||
),
|
||||
"decode",
|
||||
),
|
||||
(Reply::body(200, b"{".to_vec()), "decode"),
|
||||
(Reply::body(200, b"{}".to_vec()), "incomplete"),
|
||||
(
|
||||
Reply::body(
|
||||
200,
|
||||
br#"{"access_token":"","refresh_token":"fixture-body-secret"}"#.to_vec(),
|
||||
),
|
||||
"incomplete",
|
||||
),
|
||||
(
|
||||
Reply {
|
||||
truncate: true,
|
||||
..Reply::body(200, rotated_body())
|
||||
},
|
||||
"read response body",
|
||||
),
|
||||
(
|
||||
Reply {
|
||||
truncate: true,
|
||||
..Reply::body(401, b"fixture-body-secret".to_vec())
|
||||
},
|
||||
"read response body",
|
||||
),
|
||||
(
|
||||
Reply::body(401, b"fixture-body-secret".to_vec()),
|
||||
"codex login",
|
||||
),
|
||||
(Reply::body(503, b"fixture-body-secret".to_vec()), "503"),
|
||||
];
|
||||
for (reply, expected) in cases.drain(..) {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = stale_auth(dir.path());
|
||||
let original = std::fs::read(&path).unwrap();
|
||||
let started = Arc::new(Notify::new());
|
||||
let mut stub = Stub::start(vec![reply, Reply::body(200, rotated_body())]).await;
|
||||
let provider = provider(dir.path(), &stub, &started);
|
||||
let error = bounded(provider.headers()).await.unwrap_err();
|
||||
let message = error.to_string();
|
||||
assert!(message.contains(expected), "expected {expected}: {message}");
|
||||
assert!(!message.contains("fixture-body-secret"));
|
||||
assert!(!message.contains("fixture-old-secret"));
|
||||
assert_eq!(std::fs::read(&path).unwrap(), original);
|
||||
let headers = bounded(provider.headers()).await.unwrap();
|
||||
assert_rotated(&path, &headers);
|
||||
for _ in 0..2 {
|
||||
assert_eq!(
|
||||
bounded(stub.requests.recv()).await.unwrap()["refresh_token"],
|
||||
"fixture-old-secret"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancelled_body_wait_leaves_auth_unchanged_and_releases_waiting_caller() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = stale_auth(dir.path());
|
||||
let original = std::fs::read(&path).unwrap();
|
||||
let started = Arc::new(Notify::new());
|
||||
let gate = Arc::new(Notify::new());
|
||||
let mut stub = Stub::start(vec![
|
||||
Reply::gated(200, rotated_body(), &gate),
|
||||
Reply::body(200, rotated_body()),
|
||||
])
|
||||
.await;
|
||||
let provider = provider(dir.path(), &stub, &started);
|
||||
let first = tokio::spawn({
|
||||
let provider = provider.clone();
|
||||
async move { provider.headers().await }
|
||||
});
|
||||
bounded(started.notified()).await;
|
||||
bounded(stub.requests.recv()).await.unwrap();
|
||||
let second = provider.headers();
|
||||
tokio::pin!(second);
|
||||
assert!(futures::poll!(&mut second).is_pending());
|
||||
assert!(stub.requests.try_recv().is_err());
|
||||
first.abort();
|
||||
assert!(first.await.unwrap_err().is_cancelled());
|
||||
assert_eq!(std::fs::read(&path).unwrap(), original);
|
||||
assert_rotated(&path, &bounded(second).await.unwrap());
|
||||
assert_eq!(
|
||||
bounded(stub.requests.recv()).await.unwrap()["refresh_token"],
|
||||
"fixture-old-secret"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn shared_callers_refresh_once_and_use_persisted_rotated_tokens() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = stale_auth(dir.path());
|
||||
let started = Arc::new(Notify::new());
|
||||
let gate = Arc::new(Notify::new());
|
||||
let mut stub = Stub::start(vec![Reply::gated(200, rotated_body(), &gate)]).await;
|
||||
let provider = provider(dir.path(), &stub, &started);
|
||||
let first = tokio::spawn({
|
||||
let provider = provider.clone();
|
||||
async move { provider.headers().await }
|
||||
});
|
||||
bounded(started.notified()).await;
|
||||
let second = provider.headers();
|
||||
tokio::pin!(second);
|
||||
assert!(futures::poll!(&mut second).is_pending());
|
||||
assert_eq!(
|
||||
bounded(stub.requests.recv()).await.unwrap()["refresh_token"],
|
||||
"fixture-old-secret"
|
||||
);
|
||||
let mut external: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
|
||||
external["concurrent_extra"] = json!("merge me");
|
||||
std::fs::write(&path, serde_json::to_vec(&external).unwrap()).unwrap();
|
||||
gate.notify_one();
|
||||
let first_headers = bounded(first).await.unwrap().unwrap();
|
||||
let second_headers = bounded(second).await.unwrap();
|
||||
assert_eq!(first_headers, second_headers);
|
||||
assert_rotated(&path, &second_headers);
|
||||
let persisted: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
|
||||
assert_eq!(persisted["concurrent_extra"], "merge me");
|
||||
assert!(
|
||||
stub.requests.try_recv().is_err(),
|
||||
"single-flight must make only one request"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn account_switch_before_guarded_reload_skips_old_account_refresh() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = stale_auth(dir.path());
|
||||
let barrier = Arc::new(tokio::sync::Barrier::new(2));
|
||||
let mut provider = CodexAuthProvider::new(dir.path().to_path_buf());
|
||||
let mut stub = Stub::start(vec![]).await;
|
||||
provider.refresh_endpoint = stub.endpoint.clone();
|
||||
provider.before_guarded_reload = Some(barrier.clone());
|
||||
let task = tokio::spawn(async move { provider.headers().await });
|
||||
bounded(barrier.wait()).await;
|
||||
let mut switched: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
|
||||
switched["tokens"]["account_id"] = json!("switched-account");
|
||||
switched["tokens"]["refresh_token"] = json!("fixture-switched-secret");
|
||||
std::fs::write(&path, serde_json::to_vec(&switched).unwrap()).unwrap();
|
||||
let original = std::fs::read(&path).unwrap();
|
||||
barrier.wait().await;
|
||||
let headers = bounded(task).await.unwrap().unwrap();
|
||||
assert!(
|
||||
headers
|
||||
.iter()
|
||||
.any(|(name, value)| name == "chatgpt-account-id" && value == "switched-account")
|
||||
);
|
||||
assert_eq!(std::fs::read(&path).unwrap(), original);
|
||||
assert!(stub.requests.try_recv().is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn account_or_token_switch_during_refresh_is_not_overwritten_by_old_response() {
|
||||
for account_changed in [false, true] {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = stale_auth(dir.path());
|
||||
let started = Arc::new(Notify::new());
|
||||
let gate = Arc::new(Notify::new());
|
||||
let mut stub = Stub::start(vec![Reply::gated(200, rotated_body(), &gate)]).await;
|
||||
let provider = provider(dir.path(), &stub, &started);
|
||||
let task = tokio::spawn({
|
||||
let provider = provider.clone();
|
||||
async move { provider.headers().await }
|
||||
});
|
||||
bounded(started.notified()).await;
|
||||
bounded(stub.requests.recv()).await.unwrap();
|
||||
let mut switched: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
|
||||
if account_changed {
|
||||
switched["tokens"]["account_id"] = json!("switched-account");
|
||||
} else {
|
||||
switched["tokens"]["refresh_token"] = json!("fixture-switched-secret");
|
||||
}
|
||||
switched["tokens"]["access_token"] =
|
||||
json!(make_jwt(json!({"exp": Utc::now().timestamp() + 7200})));
|
||||
std::fs::write(&path, serde_json::to_vec(&switched).unwrap()).unwrap();
|
||||
let original = std::fs::read(&path).unwrap();
|
||||
gate.notify_one();
|
||||
let headers = bounded(task).await.unwrap().unwrap();
|
||||
assert_eq!(std::fs::read(&path).unwrap(), original);
|
||||
let expected = format!(
|
||||
"Bearer {}",
|
||||
switched["tokens"]["access_token"].as_str().unwrap()
|
||||
);
|
||||
assert!(headers.iter().any(|(name, value)| name == "authorization" && value.to_str().unwrap() == expected));
|
||||
assert_eq!(bounded(provider.headers()).await.unwrap(), headers);
|
||||
assert!(stub.requests.try_recv().is_err());
|
||||
}
|
||||
}
|
||||
@@ -2873,3 +2873,737 @@ async fn post_tool_stop_terminalizes_started_parallel_siblings_before_returning(
|
||||
assert!(matches!(next, EngineRunExit::Finished));
|
||||
}
|
||||
}
|
||||
|
||||
// The same policy barrier is exercised through the real response-complete,
|
||||
// provider fallback, interceptor fallback, and early-dispatch engine paths.
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
enum PolicyRoute {
|
||||
AfterResponse,
|
||||
ProviderFallback,
|
||||
InterceptorFallback,
|
||||
Early,
|
||||
}
|
||||
|
||||
impl PolicyRoute {
|
||||
fn completion_support(self) -> ToolCallCompletionSupport {
|
||||
match self {
|
||||
Self::ProviderFallback => ToolCallCompletionSupport::ResponseComplete,
|
||||
_ => ToolCallCompletionSupport::PerBlock,
|
||||
}
|
||||
}
|
||||
|
||||
fn configure(self, engine: &mut Engine<ControlledStreamClient>) {
|
||||
engine.set_max_turns(Some(1));
|
||||
if matches!(self, Self::AfterResponse) {
|
||||
engine.set_tool_call_dispatch_mode(ToolCallDispatchMode::AfterResponse);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct NotifyOnDrop(Arc<tokio::sync::Notify>);
|
||||
|
||||
impl Drop for NotifyOnDrop {
|
||||
fn drop(&mut self) {
|
||||
self.0.notify_one();
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct PolicyBarrier {
|
||||
phase: InterceptorPhase,
|
||||
early_compatible: bool,
|
||||
entered: Arc<tokio::sync::Notify>,
|
||||
release: Arc<tokio::sync::Notify>,
|
||||
invocations: Arc<AtomicUsize>,
|
||||
continues: Arc<AtomicUsize>,
|
||||
retired: Arc<tokio::sync::Notify>,
|
||||
// Enqueue from the policy's final poll, after select has polled signals.
|
||||
// This deterministically exercises the policy-completion/ownership race.
|
||||
signal_on_continue: Option<tokio::sync::mpsc::Sender<()>>,
|
||||
}
|
||||
|
||||
impl PolicyBarrier {
|
||||
fn new(phase: InterceptorPhase, route: PolicyRoute) -> Self {
|
||||
Self {
|
||||
phase,
|
||||
early_compatible: !matches!(route, PolicyRoute::InterceptorFallback),
|
||||
entered: Arc::new(tokio::sync::Notify::new()),
|
||||
release: Arc::new(tokio::sync::Notify::new()),
|
||||
invocations: Arc::new(AtomicUsize::new(0)),
|
||||
continues: Arc::new(AtomicUsize::new(0)),
|
||||
retired: Arc::new(tokio::sync::Notify::new()),
|
||||
signal_on_continue: None,
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait(&self) {
|
||||
let _retired = NotifyOnDrop(self.retired.clone());
|
||||
self.invocations.fetch_add(1, Ordering::SeqCst);
|
||||
self.entered.notify_one();
|
||||
self.release.notified().await;
|
||||
if let Some(signal) = &self.signal_on_continue {
|
||||
signal.try_send(()).unwrap();
|
||||
}
|
||||
self.continues.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Interceptor for PolicyBarrier {
|
||||
fn supports_early_tool_dispatch(&self) -> bool {
|
||||
self.early_compatible
|
||||
}
|
||||
|
||||
async fn pre_tool_call(
|
||||
&self,
|
||||
_info: &mut ToolCallInfo<'_>,
|
||||
) -> InterceptorResult<PreToolAction> {
|
||||
if self.phase == InterceptorPhase::PreToolCall {
|
||||
self.wait().await;
|
||||
}
|
||||
Ok(PreToolAction::Continue)
|
||||
}
|
||||
|
||||
async fn on_assistant_turn_end(
|
||||
&self,
|
||||
_context: AssistantTurnEndContext<'_>,
|
||||
) -> InterceptorResult<TurnEndAction> {
|
||||
if self.phase == InterceptorPhase::AssistantTurnEnd {
|
||||
self.wait().await;
|
||||
}
|
||||
Ok(TurnEndAction::Finish)
|
||||
}
|
||||
}
|
||||
|
||||
fn send_policy_response(tx: &tokio::sync::mpsc::UnboundedSender<Result<Event, ClientError>>) {
|
||||
for event in [
|
||||
Event::tool_use_start(0, "policy_call", "policy_tool"),
|
||||
Event::tool_input_delta(0, "{}"),
|
||||
Event::tool_use_stop(0),
|
||||
Event::usage(10, 3),
|
||||
Event::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
}),
|
||||
] {
|
||||
tx.send(Ok(event)).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
fn assert_policy_interruption(exit: &EngineRunExit, pause: bool) {
|
||||
if pause {
|
||||
assert!(matches!(exit, EngineRunExit::Paused), "{exit:?}");
|
||||
} else {
|
||||
assert!(
|
||||
matches!(
|
||||
exit,
|
||||
EngineRunExit::Interrupted(RunInterruptionReason::Cancelled)
|
||||
),
|
||||
"{exit:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn call_and_result_count(history: &History) -> (usize, usize) {
|
||||
(
|
||||
history
|
||||
.items()
|
||||
.filter(
|
||||
|item| matches!(item, Item::ToolCall { call_id, .. } if call_id == "policy_call"),
|
||||
)
|
||||
.count(),
|
||||
history
|
||||
.items()
|
||||
.filter(
|
||||
|item| matches!(item, Item::ToolResult { call_id, .. } if call_id == "policy_call"),
|
||||
)
|
||||
.count(),
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn response_complete_pending_policies_interrupt_before_release_and_preserve_not_started() {
|
||||
for route in [
|
||||
PolicyRoute::AfterResponse,
|
||||
PolicyRoute::ProviderFallback,
|
||||
PolicyRoute::InterceptorFallback,
|
||||
] {
|
||||
for phase in [
|
||||
InterceptorPhase::PreToolCall,
|
||||
InterceptorPhase::AssistantTurnEnd,
|
||||
] {
|
||||
for pause in [false, true] {
|
||||
let (client, tx) = ControlledStreamClient::new(route.completion_support());
|
||||
let mut engine = Engine::new(client);
|
||||
route.configure(&mut engine);
|
||||
let policy = PolicyBarrier::new(phase, route);
|
||||
let probe = policy.clone();
|
||||
engine.set_interceptor(policy);
|
||||
let tool = SlowTool::new("policy_tool", 0);
|
||||
engine.register_tool(tool.definition());
|
||||
let usage = Arc::new(AtomicUsize::new(0));
|
||||
let usage_probe = usage.clone();
|
||||
engine.on_usage(move |_| {
|
||||
usage_probe.fetch_add(1, Ordering::SeqCst);
|
||||
});
|
||||
let signal = if pause {
|
||||
engine.pause_sender()
|
||||
} else {
|
||||
engine.cancel_sender()
|
||||
};
|
||||
let run = tokio::spawn(async move {
|
||||
let mut history = History::new();
|
||||
let output = engine.run(&mut history, "run").await;
|
||||
(output, history)
|
||||
});
|
||||
send_policy_response(&tx);
|
||||
drop(tx);
|
||||
tokio::time::timeout(Duration::from_secs(2), probe.entered.notified())
|
||||
.await
|
||||
.expect("policy entered");
|
||||
signal.send(()).await.unwrap();
|
||||
let (output, history) = tokio::time::timeout(Duration::from_secs(2), run).await
|
||||
.unwrap_or_else(|_| panic!("{route:?}/{phase:?}/pause={pause}: interruption waited for policy release"))
|
||||
.unwrap();
|
||||
assert_policy_interruption(&output.result, pause);
|
||||
assert_eq!(tool.call_count(), 0);
|
||||
assert_eq!(call_and_result_count(&history), (1, 0));
|
||||
assert!(history.items().any(|item| matches!(
|
||||
item,
|
||||
Item::ToolCall {
|
||||
execution_id: None,
|
||||
status: None | Some(agen::llm_client::ItemStatus::Completed),
|
||||
..
|
||||
}
|
||||
)));
|
||||
assert_eq!(usage.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(probe.invocations.load(Ordering::SeqCst), 1);
|
||||
probe.release.notify_one(); // late permission cannot create a task
|
||||
assert_eq!(probe.continues.load(Ordering::SeqCst), 0);
|
||||
assert_eq!(tool.call_count(), 0);
|
||||
|
||||
// Recreate both history and engine as a restore boundary. Only
|
||||
// proven NotStarted is allowed to run, with normal pre-tool policy.
|
||||
let mut restored_history: History =
|
||||
serde_json::from_str(&serde_json::to_string(&history).unwrap()).unwrap();
|
||||
let mut restored =
|
||||
Engine::new(MockLlmClient::new(vec![Event::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
})]));
|
||||
restored.set_max_turns(Some(1));
|
||||
restored.register_tool(tool.definition());
|
||||
let _ = restored.resume(&mut restored_history).await;
|
||||
assert_eq!(tool.call_count(), 1);
|
||||
assert_eq!(call_and_result_count(&restored_history), (1, 1));
|
||||
assert_eq!(usage.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn policy_final_poll_signal_prevents_provider_ownership_even_with_continue() {
|
||||
for route in [
|
||||
PolicyRoute::AfterResponse,
|
||||
PolicyRoute::ProviderFallback,
|
||||
PolicyRoute::InterceptorFallback,
|
||||
PolicyRoute::Early,
|
||||
] {
|
||||
for pause in [false, true] {
|
||||
let (client, tx) = ControlledStreamClient::new(route.completion_support());
|
||||
let mut engine = Engine::new(client);
|
||||
route.configure(&mut engine);
|
||||
let mut policy = PolicyBarrier::new(InterceptorPhase::PreToolCall, route);
|
||||
policy.signal_on_continue = Some(if pause {
|
||||
engine.pause_sender()
|
||||
} else {
|
||||
engine.cancel_sender()
|
||||
});
|
||||
let probe = policy.clone();
|
||||
engine.set_interceptor(policy);
|
||||
let tool = SlowTool::new("policy_tool", 0);
|
||||
engine.register_tool(tool.definition());
|
||||
let run = tokio::spawn(async move {
|
||||
let mut history = History::new();
|
||||
let output = engine.run(&mut history, "run").await;
|
||||
(output, history)
|
||||
});
|
||||
send_policy_response(&tx);
|
||||
drop(tx);
|
||||
tokio::time::timeout(Duration::from_secs(2), probe.entered.notified())
|
||||
.await
|
||||
.expect("policy entered");
|
||||
probe.release.notify_one();
|
||||
let (output, history) = tokio::time::timeout(Duration::from_secs(2), run)
|
||||
.await
|
||||
.expect("final-poll signal processed")
|
||||
.unwrap();
|
||||
assert_policy_interruption(&output.result, pause);
|
||||
assert_eq!(probe.continues.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(probe.invocations.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(
|
||||
tool.call_count(),
|
||||
0,
|
||||
"{route:?}/pause={pause}: Continue bypassed queued interruption"
|
||||
);
|
||||
assert_eq!(call_and_result_count(&history), (1, 0));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Provider owns the terminal channel. Cancel acknowledges receipt but does not
|
||||
// manufacture a result, so tests can release a real terminal output or deliver
|
||||
// it after the engine's confirmation deadline (without a live side effect).
|
||||
#[derive(Clone)]
|
||||
struct PolicyOwnedTool {
|
||||
starts: Arc<AtomicUsize>,
|
||||
started: Arc<tokio::sync::Notify>,
|
||||
cancel_requested: Arc<tokio::sync::Notify>,
|
||||
terminal: Arc<tokio::sync::Mutex<Option<tokio::sync::oneshot::Receiver<ToolOutput>>>>,
|
||||
retired: Arc<tokio::sync::Notify>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for PolicyOwnedTool {
|
||||
async fn execute(&self, _: &str, _: ToolExecutionContext) -> Result<ToolOutput, ToolError> {
|
||||
let _retired = NotifyOnDrop(self.retired.clone());
|
||||
self.starts.fetch_add(1, Ordering::SeqCst);
|
||||
self.started.notify_one();
|
||||
Ok(self.terminal.lock().await.take().unwrap().await.unwrap())
|
||||
}
|
||||
async fn cancel(&self, _: &str) -> Result<(), ToolError> {
|
||||
self.cancel_requested.notify_one();
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn pending_turn_end_interrupts_drain_owned_early_attempt_and_fence_late_result_on_restore() {
|
||||
for pause in [false, true] {
|
||||
for late in [false, true] {
|
||||
let (client, tx) = ControlledStreamClient::new(ToolCallCompletionSupport::PerBlock);
|
||||
let mut engine = Engine::new(client);
|
||||
PolicyRoute::Early.configure(&mut engine);
|
||||
engine.set_tool_execution_policy(ToolExecutionPolicy {
|
||||
pause_safe_boundary_timeout: Duration::ZERO,
|
||||
cancellation_request_timeout: Duration::from_secs(1),
|
||||
terminal_confirmation_timeout: if late {
|
||||
Duration::ZERO
|
||||
} else {
|
||||
Duration::from_secs(2)
|
||||
},
|
||||
});
|
||||
let policy = PolicyBarrier::new(InterceptorPhase::AssistantTurnEnd, PolicyRoute::Early);
|
||||
let probe = policy.clone();
|
||||
engine.set_interceptor(policy);
|
||||
let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
|
||||
let tool = PolicyOwnedTool {
|
||||
starts: Arc::new(AtomicUsize::new(0)),
|
||||
started: Arc::new(tokio::sync::Notify::new()),
|
||||
cancel_requested: Arc::new(tokio::sync::Notify::new()),
|
||||
terminal: Arc::new(tokio::sync::Mutex::new(Some(terminal_rx))),
|
||||
retired: Arc::new(tokio::sync::Notify::new()),
|
||||
};
|
||||
let definition: ToolDefinition = {
|
||||
let tool = tool.clone();
|
||||
Arc::new(move || {
|
||||
(
|
||||
ToolMeta::new("policy_tool")
|
||||
.input_schema(serde_json::json!({"type":"object"})),
|
||||
Arc::new(tool.clone()) as Arc<dyn Tool>,
|
||||
)
|
||||
})
|
||||
};
|
||||
engine.register_tool(definition.clone());
|
||||
let committed = Arc::new(AtomicUsize::new(0));
|
||||
let commit_probe = committed.clone();
|
||||
engine.on_history_append(move |item| {
|
||||
if matches!(item, Item::ToolResult { .. }) {
|
||||
commit_probe.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
Ok(())
|
||||
});
|
||||
let published = Arc::new(AtomicUsize::new(0));
|
||||
let publish_probe = published.clone();
|
||||
let commit_probe = committed.clone();
|
||||
engine.on_tool_result(move |_| {
|
||||
assert_eq!(
|
||||
commit_probe.load(Ordering::SeqCst),
|
||||
1,
|
||||
"commit before publication"
|
||||
);
|
||||
publish_probe.fetch_add(1, Ordering::SeqCst);
|
||||
});
|
||||
let usage = Arc::new(AtomicUsize::new(0));
|
||||
let usage_probe = usage.clone();
|
||||
engine.on_usage(move |_| {
|
||||
usage_probe.fetch_add(1, Ordering::SeqCst);
|
||||
});
|
||||
let signal = if pause {
|
||||
engine.pause_sender()
|
||||
} else {
|
||||
engine.cancel_sender()
|
||||
};
|
||||
let run = tokio::spawn(async move {
|
||||
let mut history = History::new();
|
||||
let output = engine.run(&mut history, "run").await;
|
||||
(output, history)
|
||||
});
|
||||
// Wait for ownership before completing the assistant response.
|
||||
for event in [
|
||||
Event::tool_use_start(0, "policy_call", "policy_tool"),
|
||||
Event::tool_input_delta(0, "{}"),
|
||||
Event::tool_use_stop(0),
|
||||
] {
|
||||
tx.send(Ok(event)).unwrap();
|
||||
}
|
||||
tokio::time::timeout(Duration::from_secs(2), tool.started.notified())
|
||||
.await
|
||||
.expect("early provider started");
|
||||
tx.send(Ok(Event::usage(10, 3))).unwrap();
|
||||
tx.send(Ok(Event::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
})))
|
||||
.unwrap();
|
||||
drop(tx);
|
||||
tokio::time::timeout(Duration::from_secs(2), probe.entered.notified())
|
||||
.await
|
||||
.expect("turn-end policy pending");
|
||||
signal.send(()).await.unwrap();
|
||||
tokio::time::timeout(Duration::from_secs(2), tool.cancel_requested.notified())
|
||||
.await
|
||||
.expect("owned provider cancellation progresses before policy release");
|
||||
let terminal_tx = if late {
|
||||
Some(terminal_tx)
|
||||
} else {
|
||||
terminal_tx
|
||||
.send(ToolOutput::from("confirmed after interruption".to_string()))
|
||||
.unwrap();
|
||||
None
|
||||
};
|
||||
let (output, history) = tokio::time::timeout(Duration::from_secs(2), run)
|
||||
.await
|
||||
.expect("terminalization does not await turn-end policy")
|
||||
.unwrap();
|
||||
assert_policy_interruption(&output.result, pause);
|
||||
assert_eq!(call_and_result_count(&history), (1, 1));
|
||||
assert_eq!(committed.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(published.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(usage.load(Ordering::SeqCst), 1);
|
||||
assert!(history.items().any(|item| matches!(
|
||||
item,
|
||||
Item::ToolCall {
|
||||
execution_id: Some(_),
|
||||
status: Some(agen::llm_client::ItemStatus::InProgress),
|
||||
..
|
||||
}
|
||||
)));
|
||||
assert!(history.items().any(|item| matches!(item, Item::ToolResult { disposition, .. } if *disposition == if late { ToolResultDisposition::OutcomeUnknown } else { ToolResultDisposition::Success })));
|
||||
if let Some(terminal_tx) = terminal_tx {
|
||||
tokio::time::timeout(Duration::from_secs(2), tool.retired.notified())
|
||||
.await
|
||||
.expect("forced-close retired the owned receiver");
|
||||
// Force-close retires the receiver; a provider's late completion
|
||||
// cannot reenter history or replace OutcomeUnknown.
|
||||
assert!(
|
||||
terminal_tx
|
||||
.send(ToolOutput::from("late completion".to_string()))
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
probe.release.notify_one();
|
||||
assert_eq!(probe.continues.load(Ordering::SeqCst), 0);
|
||||
let original_items = history.items().cloned().collect::<Vec<_>>();
|
||||
let mut resumed_history = history.clone();
|
||||
let mut resumed_engine = output.engine;
|
||||
let _ = resumed_engine.resume(&mut resumed_history).await;
|
||||
assert_eq!(
|
||||
tool.starts.load(Ordering::SeqCst),
|
||||
1,
|
||||
"same-engine resume cannot rerun a terminal attempt"
|
||||
);
|
||||
assert_eq!(call_and_result_count(&resumed_history), (1, 1));
|
||||
let mut restored_history: History =
|
||||
serde_json::from_str(&serde_json::to_string(&history).unwrap()).unwrap();
|
||||
let restored_client = MockLlmClient::new(vec![Event::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
})]);
|
||||
let request_probe = restored_client.clone();
|
||||
let mut restored = Engine::new(restored_client);
|
||||
restored.register_tool(definition);
|
||||
restored.set_max_turns(Some(1));
|
||||
let _ = restored.resume(&mut restored_history).await;
|
||||
let requests = request_probe.requests();
|
||||
assert_eq!(requests.len(), 1);
|
||||
let replay = requests[0]
|
||||
.items
|
||||
.iter()
|
||||
.filter_map(|item| match item {
|
||||
Item::ToolCall { call_id, .. } => Some(format!("call:{call_id}")),
|
||||
Item::ToolResult { call_id, .. } => Some(format!("result:{call_id}")),
|
||||
_ => None,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(replay, ["call:policy_call", "result:policy_call"]);
|
||||
assert_eq!(tool.starts.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(call_and_result_count(&restored_history), (1, 1));
|
||||
assert_eq!(
|
||||
restored_history
|
||||
.items()
|
||||
.take(original_items.len())
|
||||
.cloned()
|
||||
.collect::<Vec<_>>(),
|
||||
original_items,
|
||||
"restore must not rewrite original history"
|
||||
);
|
||||
assert_eq!(published.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn pending_turn_end_pause_preserves_natural_safe_boundary_without_cancelling_started_tool() {
|
||||
let (client, tx) = ControlledStreamClient::new(ToolCallCompletionSupport::PerBlock);
|
||||
let mut engine = Engine::new(client);
|
||||
PolicyRoute::Early.configure(&mut engine);
|
||||
engine.set_tool_execution_policy(ToolExecutionPolicy {
|
||||
pause_safe_boundary_timeout: Duration::from_secs(2),
|
||||
..ToolExecutionPolicy::default()
|
||||
});
|
||||
let policy = PolicyBarrier::new(InterceptorPhase::AssistantTurnEnd, PolicyRoute::Early);
|
||||
let probe = policy.clone();
|
||||
engine.set_interceptor(policy);
|
||||
let tool = SafePauseTool::new();
|
||||
// Use this tool's actual public name in the stream fixture.
|
||||
engine.register_tool(tool.definition());
|
||||
let pause = engine.pause_sender();
|
||||
let run = tokio::spawn(async move {
|
||||
let mut history = History::new();
|
||||
let output = engine.run(&mut history, "run").await;
|
||||
(output, history)
|
||||
});
|
||||
for event in [
|
||||
Event::tool_use_start(0, "policy_call", "safe_pause"),
|
||||
Event::tool_input_delta(0, "{}"),
|
||||
Event::tool_use_stop(0),
|
||||
Event::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
}),
|
||||
] {
|
||||
tx.send(Ok(event)).unwrap();
|
||||
}
|
||||
drop(tx);
|
||||
tokio::time::timeout(Duration::from_secs(2), probe.entered.notified())
|
||||
.await
|
||||
.expect("turn-end entered");
|
||||
pause.send(()).await.unwrap();
|
||||
tokio::time::timeout(Duration::from_secs(2), probe.retired.notified())
|
||||
.await
|
||||
.expect("pending policy interrupted before tool release");
|
||||
tool.release.notify_one();
|
||||
let (output, history) = tokio::time::timeout(Duration::from_secs(2), run)
|
||||
.await
|
||||
.expect("safe boundary collected")
|
||||
.unwrap();
|
||||
assert_policy_interruption(&output.result, true);
|
||||
assert_eq!(tool.calls.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(tool.cancellations.load(Ordering::SeqCst), 0);
|
||||
assert_eq!(call_and_result_count(&history), (1, 1));
|
||||
probe.release.notify_one();
|
||||
assert_eq!(probe.continues.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn turn_end_final_poll_signal_wins_over_finish_without_tools() {
|
||||
for pause in [false, true] {
|
||||
let (client, tx) = ControlledStreamClient::new(ToolCallCompletionSupport::PerBlock);
|
||||
let mut engine = Engine::new(client);
|
||||
let mut policy = PolicyBarrier::new(InterceptorPhase::AssistantTurnEnd, PolicyRoute::Early);
|
||||
policy.signal_on_continue = Some(if pause {
|
||||
engine.pause_sender()
|
||||
} else {
|
||||
engine.cancel_sender()
|
||||
});
|
||||
let probe = policy.clone();
|
||||
engine.set_interceptor(policy);
|
||||
let run = tokio::spawn(async move {
|
||||
let mut history = History::new();
|
||||
engine.run(&mut history, "run").await
|
||||
});
|
||||
tx.send(Ok(Event::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
})))
|
||||
.unwrap();
|
||||
drop(tx);
|
||||
tokio::time::timeout(Duration::from_secs(2), probe.entered.notified())
|
||||
.await
|
||||
.expect("turn-end entered");
|
||||
probe.release.notify_one();
|
||||
let output = tokio::time::timeout(Duration::from_secs(2), run)
|
||||
.await
|
||||
.expect("final-poll signal processed")
|
||||
.unwrap();
|
||||
assert_policy_interruption(&output.result, pause);
|
||||
assert_eq!(probe.continues.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn turn_end_pause_deadline_reclaims_started_sibling_while_post_policy_is_pending() {
|
||||
struct TurnAndPostBarrier {
|
||||
turn: PolicyBarrier,
|
||||
post: PolicyBarrier,
|
||||
}
|
||||
#[async_trait]
|
||||
impl Interceptor for TurnAndPostBarrier {
|
||||
fn supports_early_tool_dispatch(&self) -> bool {
|
||||
true
|
||||
}
|
||||
async fn on_assistant_turn_end(
|
||||
&self,
|
||||
_: AssistantTurnEndContext<'_>,
|
||||
) -> InterceptorResult<TurnEndAction> {
|
||||
self.turn.wait().await;
|
||||
Ok(TurnEndAction::Finish)
|
||||
}
|
||||
async fn post_tool_call(
|
||||
&self,
|
||||
_: &ToolResultInfo<'_>,
|
||||
) -> InterceptorResult<PostToolAction> {
|
||||
self.post.wait().await;
|
||||
Ok(PostToolAction::Continue)
|
||||
}
|
||||
}
|
||||
|
||||
let (client, tx) = ControlledStreamClient::new(ToolCallCompletionSupport::PerBlock);
|
||||
let mut engine = Engine::new(client);
|
||||
engine.set_tool_execution_policy(ToolExecutionPolicy {
|
||||
pause_safe_boundary_timeout: Duration::from_millis(500),
|
||||
cancellation_request_timeout: Duration::from_secs(1),
|
||||
terminal_confirmation_timeout: Duration::from_secs(2),
|
||||
});
|
||||
let turn = PolicyBarrier::new(InterceptorPhase::AssistantTurnEnd, PolicyRoute::Early);
|
||||
let post = PolicyBarrier::new(InterceptorPhase::PostToolCall, PolicyRoute::Early);
|
||||
engine.set_interceptor(TurnAndPostBarrier {
|
||||
turn: turn.clone(),
|
||||
post: post.clone(),
|
||||
});
|
||||
let completed = BarrierTool::new("completed_before_pause");
|
||||
engine.register_tool(completed.definition());
|
||||
let (terminal_tx, terminal_rx) = tokio::sync::oneshot::channel();
|
||||
let sibling = PolicyOwnedTool {
|
||||
starts: Arc::new(AtomicUsize::new(0)),
|
||||
started: Arc::new(tokio::sync::Notify::new()),
|
||||
cancel_requested: Arc::new(tokio::sync::Notify::new()),
|
||||
terminal: Arc::new(tokio::sync::Mutex::new(Some(terminal_rx))),
|
||||
retired: Arc::new(tokio::sync::Notify::new()),
|
||||
};
|
||||
let definition: ToolDefinition = {
|
||||
let sibling = sibling.clone();
|
||||
Arc::new(move || {
|
||||
(
|
||||
ToolMeta::new("pending_sibling").input_schema(serde_json::json!({"type":"object"})),
|
||||
Arc::new(sibling.clone()) as Arc<dyn Tool>,
|
||||
)
|
||||
})
|
||||
};
|
||||
engine.register_tool(definition.clone());
|
||||
let committed = Arc::new(AtomicUsize::new(0));
|
||||
let commit_probe = committed.clone();
|
||||
engine.on_history_append(move |item| {
|
||||
if matches!(item, Item::ToolResult { .. }) {
|
||||
commit_probe.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
Ok(())
|
||||
});
|
||||
let published = Arc::new(AtomicUsize::new(0));
|
||||
let publish_probe = published.clone();
|
||||
let commit_probe = committed.clone();
|
||||
engine.on_tool_result(move |_| {
|
||||
let n = publish_probe.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
assert_eq!(commit_probe.load(Ordering::SeqCst), n);
|
||||
});
|
||||
let pause = engine.pause_sender();
|
||||
let run = tokio::spawn(async move {
|
||||
let mut history = History::new();
|
||||
let output = engine.run(&mut history, "run").await;
|
||||
(output, history)
|
||||
});
|
||||
for (index, id, name) in [
|
||||
(0, "policy_call", "completed_before_pause"),
|
||||
(1, "sibling_call", "pending_sibling"),
|
||||
] {
|
||||
for event in [
|
||||
Event::tool_use_start(index, id, name),
|
||||
Event::tool_input_delta(index, "{}"),
|
||||
Event::tool_use_stop(index),
|
||||
] {
|
||||
tx.send(Ok(event)).unwrap();
|
||||
}
|
||||
}
|
||||
tokio::time::timeout(Duration::from_secs(2), completed.started.notified())
|
||||
.await
|
||||
.expect("first provider owned");
|
||||
tokio::time::timeout(Duration::from_secs(2), sibling.started.notified())
|
||||
.await
|
||||
.expect("sibling provider owned");
|
||||
tx.send(Ok(Event::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
})))
|
||||
.unwrap();
|
||||
drop(tx);
|
||||
tokio::time::timeout(Duration::from_secs(2), turn.entered.notified())
|
||||
.await
|
||||
.expect("turn-end pending");
|
||||
pause.send(()).await.unwrap(); // only one interruption signal
|
||||
tokio::time::timeout(Duration::from_secs(2), turn.retired.notified())
|
||||
.await
|
||||
.expect("turn-end interrupted");
|
||||
completed.release.notify_one();
|
||||
tokio::time::timeout(Duration::from_secs(2), post.entered.notified())
|
||||
.await
|
||||
.expect("completed result's post policy pending");
|
||||
assert_eq!(committed.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(published.load(Ordering::SeqCst), 1);
|
||||
tokio::time::timeout(Duration::from_secs(2), sibling.cancel_requested.notified()).await.expect("original Pause deadline must reach sibling cancellation without releasing either policy");
|
||||
terminal_tx
|
||||
.send(ToolOutput::from("confirmed sibling terminal".to_string()))
|
||||
.unwrap();
|
||||
let (output, history) = tokio::time::timeout(Duration::from_secs(2), run)
|
||||
.await
|
||||
.expect("all owned tools reclaimed")
|
||||
.unwrap();
|
||||
assert_policy_interruption(&output.result, true);
|
||||
assert_eq!(committed.load(Ordering::SeqCst), 2);
|
||||
assert_eq!(published.load(Ordering::SeqCst), 2);
|
||||
assert_eq!(
|
||||
history
|
||||
.items()
|
||||
.filter(|item| matches!(item, Item::ToolResult { .. }))
|
||||
.count(),
|
||||
2
|
||||
);
|
||||
assert_eq!(completed.starts.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(sibling.starts.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(post.invocations.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(post.continues.load(Ordering::SeqCst), 0);
|
||||
turn.release.notify_one();
|
||||
post.release.notify_one();
|
||||
let mut restored_history: History =
|
||||
serde_json::from_str(&serde_json::to_string(&history).unwrap()).unwrap();
|
||||
let mut restored = Engine::new(MockLlmClient::new(vec![Event::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
})]));
|
||||
restored.register_tool(completed.definition());
|
||||
restored.register_tool(definition);
|
||||
restored.set_max_turns(Some(1));
|
||||
let _ = restored.resume(&mut restored_history).await;
|
||||
assert_eq!(completed.starts.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(sibling.starts.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(
|
||||
restored_history
|
||||
.items()
|
||||
.filter(|item| matches!(item, Item::ToolResult { .. }))
|
||||
.count(),
|
||||
2
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user