openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-08 20:40:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Resample Guardian denials with the original request (#59659)

Browse Source 
## What changed

When `features.guardian_denial_reconsideration` is enabled, retry the first terminal denial once using the unchanged inference request. Buffer provisional output so only the selected response enters reviewer history, while allowing tool calls to proceed normally.

Preserve the original denial if the retry fails or times out, and continue to honor cancellation. Account for token usage from both attempts, including failed reviews.

Remove `auto_review.experimental_denial_reconsideration_prompt` and warn when it remains in configuration.

## Testing

Update the Guardian scenario and snapshot to verify identical retry inputs, retained tool evidence, and reuse of the selected assessment in a subsequent review, with asynchronous compaction enabled.

GitOrigin-RevId: cc4e303d40a02acc824c8b4b954852325e64ad52
This commit is contained in:
jif
2026-11-05 13:38:12 +00:00
committed by copyberry
parent 86ac091659
commit 1e9e8bdb00
19 changed files with 569 additions and 587 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/config/config_tests.rs
+1 -37
View File
@@ -8929,9 +8929,6 @@ async fn load_config_uses_auto_review_guardian_policy_config_and_template(
" Configured template: {{ tenant_policy_config }} ".to_string(),
),
experimental_conversation_history_prompt: None,
experimental_denial_reconsideration_prompt: Some(
" Reconsider this review carefully. ".to_owned(),
),
conversation_history_max_output_tokens: None,
}),
..Default::default()
@@ -8953,14 +8950,12 @@ async fn load_config_uses_auto_review_guardian_policy_config_and_template(
config.guardian_policy_config.as_deref(),
config.guardian_extra_policy.as_deref(),
config.guardian_policy_template.as_deref(),
config.guardian_denial_reconsideration_prompt.as_deref(),
),
(
expected_mode,
Some("Use the user-configured guardian policy."),
Some("Use the user-configured additional policy."),
Some("Configured template: {{ tenant_policy_config }}"),
Some("Reconsider this review carefully."),
)
);
@@ -8973,34 +8968,6 @@ async fn load_config_uses_auto_review_guardian_policy_config_and_template(
Ok(())
}
#[tokio::test]
async fn load_config_rejects_oversized_guardian_reconsideration_prompt() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
let cfg = ConfigToml {
auto_review: Some(AutoReviewToml {
experimental_denial_reconsideration_prompt: Some("é".repeat(501)),
..Default::default()
}),
..Default::default()
};
let error = Config::load_from_base_config_with_overrides(
cfg,
ConfigOverrides {
cwd: Some(codex_home.path().to_path_buf()),
..Default::default()
},
codex_home.abs(),
)
.await
.unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
assert_eq!(
error.to_string(),
"auto_review.experimental_denial_reconsideration_prompt exceeds 1,000 UTF-8 bytes"
);
Ok(())
}
#[tokio::test]
async fn sync_elastic_budget_is_opt_in_and_validated() {
let excessive_minimum = codex_guardian_context::DEFAULT_MAX_INPUT_TOKENS + 1;
@@ -9078,7 +9045,6 @@ async fn requirements_guardian_policy_beats_auto_review() -> std::io::Result<()>
extra_policy: Some("Use the user-configured additional policy.".to_string()),
experimental_policy_template: None,
experimental_conversation_history_prompt: None,
experimental_denial_reconsideration_prompt: None,
conversation_history_max_output_tokens: None,
}),
..Default::default()
@@ -9121,7 +9087,6 @@ async fn load_config_ignores_empty_auto_review_guardian_policy_config() -> std::
extra_policy: Some(" ".to_string()),
experimental_policy_template: None,
experimental_conversation_history_prompt: Some(String::new()),
experimental_denial_reconsideration_prompt: Some(" ".to_owned()),
conversation_history_max_output_tokens: None,
}),
..Default::default()
@@ -9142,9 +9107,8 @@ async fn load_config_ignores_empty_auto_review_guardian_policy_config() -> std::
config.guardian_policy_config,
config.guardian_extra_policy,
config.guardian_conversation_history_prompt,
config.guardian_denial_reconsideration_prompt,
),
(None, None, None, None)
(None, None, None)
);
Ok(())
codex-rs/core/src/config/mod.rs
-18
View File
@@ -750,9 +750,6 @@ pub struct Config {
/// Blank config values are treated as unset, like other Guardian policy overrides.
pub guardian_conversation_history_prompt: Option<String>,
/// Optional bounded developer message for the gated denial reconsideration pass.
pub guardian_denial_reconsideration_prompt: Option<String>,
/// Optional per-response history-tool budget. Guardian defaults to 4,000 tokens and
/// preserves stricter parent tool limits.
pub guardian_conversation_history_max_output_tokens: Option<NonZeroUsize>,
@@ -4367,20 +4364,6 @@ impl Config {
.auto_review
.as_ref()
.and_then(|auto_review| auto_review.conversation_history_max_output_tokens);
let guardian_denial_reconsideration_prompt = cfg.auto_review.as_ref().and_then(|auto_review| {
normalize_guardian_policy_config(
auto_review.experimental_denial_reconsideration_prompt.as_deref(),
)
});
if guardian_denial_reconsideration_prompt
.as_ref()
.is_some_and(|prompt| prompt.len() > crate::context::GuardianDenialReconsideration::MAX_BYTES)
{
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"auto_review.experimental_denial_reconsideration_prompt exceeds 1,000 UTF-8 bytes",
));
}
let personality = personality.or(cfg.personality);
let experimental_compact_prompt_path = cfg.experimental_compact_prompt_file.as_ref();
@@ -4752,7 +4735,6 @@ impl Config {
guardian_transcript_mode,
guardian_sync_elastic_budget,
guardian_conversation_history_prompt,
guardian_denial_reconsideration_prompt,
guardian_conversation_history_max_output_tokens,
guardian_review_timeout: cfg
.auto_review
codex-rs/core/src/context/guardian_denial_reconsideration.rs
-44
View File
@@ -1,44 +0,0 @@
//! Supplies a bounded, review-scoped request to reconsider a Guardian denial.
use super::ContextualUserFragment;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ContentItemKind;
use codex_protocol::models::input_text;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct GuardianDenialReconsideration<'a> {
pub(crate) prompt: Option<&'a str>,
}
impl GuardianDenialReconsideration<'_> {
pub(crate) const MAX_BYTES: usize = 1_000;
}
impl ContextualUserFragment for GuardianDenialReconsideration<'_> {
fn content_kind(&self) -> ContentItemKind {
ContentItemKind("guardian.denial_reconsideration".to_owned())
}
fn role(&self) -> &'static str {
"developer"
}
fn markers(&self) -> (&'static str, &'static str) {
Self::type_markers()
}
fn type_markers() -> (&'static str, &'static str) {
("", "")
}
fn body(&self) -> ContentItem {
let prompt = self.prompt.unwrap_or(
"For this review only, reconsider your previous denial of the same planned action. \
Think carefully about the action, the available evidence, and the user's authorization \
under the same security policy. This request grants no new authorization and does not \
change the policy. Return your final assessment using the same JSON schema.",
);
// Config loading rejects oversized overrides; also bound programmatic callers.
input_text(prompt[..prompt.floor_char_boundary(Self::MAX_BYTES)].to_owned())
}
}
codex-rs/core/src/context/mod.rs
-2
View File
@@ -18,7 +18,6 @@ mod guardian_budget_omission;
mod guardian_conversation_history;
mod user_goal;
pub use guardian_budget_omission::GuardianBudgetOmission;
mod guardian_denial_reconsideration;
mod guardian_followup_review_reminder;
mod guardian_node_repl_policy;
mod guardian_policy;
@@ -86,7 +85,6 @@ pub(crate) use guardian_approved_action::GuardianApprovedAction;
pub(crate) use guardian_assistant_context::render_original_assistant_context;
pub use guardian_assistant_context::render_retained_assistant_context;
pub(crate) use guardian_conversation_history::GuardianConversationHistory;
pub(crate) use guardian_denial_reconsideration::GuardianDenialReconsideration;
pub(crate) use guardian_followup_review_reminder::GuardianFollowupReviewReminder;
pub(crate) use guardian_node_repl_policy::GuardianNodeReplPolicy;
pub(crate) use guardian_policy::GuardianPolicy;
codex-rs/core/src/context_manager/history.rs
+2 -2
View File
@@ -1088,8 +1088,8 @@ impl ContextManager {
self.set_token_info_with_boundary(Some(info), covered_item_count);
}
/// Restored or rewritten history keeps the existing last-model-output approximation.
fn reset_last_usage_boundary(&mut self) {
/// Align the context count after history changes or buffered model output is committed.
pub(crate) fn reset_last_usage_boundary(&mut self) {
self.last_model_output_item_count = self
.items
.iter()
codex-rs/core/src/guardian/mod.rs
+2
View File
@@ -19,6 +19,8 @@ pub(crate) use recorded_tool_call::review_tool_call;
mod request_budget;
pub(crate) use request_budget::ExhaustedReviewBudget;
pub(crate) use request_budget::prepare_prompt as prepare_guardian_prompt;
mod response_resampling;
pub(crate) use response_resampling::ResponseResampling;
mod review;
mod review_session;
mod reviewer_config;
codex-rs/core/src/guardian/response_resampling.rs
+123
View File
@@ -0,0 +1,123 @@
//! Buffers provisional Guardian output and resamples a terminal denial once.
//! Completions reach the shared accounting path before output replay or retry.
use std::collections::VecDeque;
use std::future::Future;
use std::sync::Arc;
use codex_async_utils::OrCancelExt;
use codex_guardian_reviewer::DenialResample;
use codex_guardian_reviewer::GuardianReviewResponse;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::ResponseItem;
use futures::StreamExt;
use tokio_util::sync::CancellationToken;
use crate::client_common::ResponseEvent;
use crate::client_common::ResponseStream;
use crate::session::session::Session;
use crate::session::turn::get_last_assistant_message_from_turn;
use crate::tools::ToolRouter;
#[derive(Default)]
pub(crate) struct ResponseResampling {
policy: Option<Arc<DenialResample>>,
buffered_output: Option<Vec<ResponseItem>>,
pub(crate) accepted_output: VecDeque<ResponseItem>,
pub(crate) retry_pending: bool,
pub(crate) update_usage_boundary: bool,
}
impl ResponseResampling {
pub(crate) fn new(session: &Session) -> Self {
let policy = session
.services
.thread_extension_data
.get::<DenialResample>();
Self {
buffered_output: policy.as_ref().map(|_| Vec::new()),
policy,
..Self::default()
}
}
pub(crate) async fn next(
&mut self,
stream: &mut ResponseStream,
) -> Option<CodexResult<ResponseEvent>> {
match self.accepted_output.pop_front() {
Some(item) => Some(Ok(ResponseEvent::OutputItemDone(item))),
None => stream.next().await,
}
}
pub(crate) fn filter(
&mut self,
event: ResponseEvent,
router: &ToolRouter,
) -> Option<ResponseEvent> {
let (Some(policy), Some(output)) = (&self.policy, self.buffered_output.as_mut()) else {
return Some(event);
};
match event {
ResponseEvent::OutputItemDone(item) => {
let dispatch = !matches!(router.build_tool_call(item.clone()), Ok(None));
output.push(item);
if dispatch {
// Tool calls proceed immediately, with their preceding reasoning.
self.accepted_output.extend(std::mem::take(output));
self.buffered_output = None;
}
return None;
}
ResponseEvent::OutputItemAdded(_)
| ResponseEvent::OutputTextDelta(_)
| ResponseEvent::ToolCallInputDelta { .. }
| ResponseEvent::ReasoningSummaryDelta { .. }
| ResponseEvent::ReasoningSummaryDone { .. }
| ResponseEvent::ReasoningContentDelta { .. }
| ResponseEvent::ReasoningSummaryPartAdded { .. } => return None,
ResponseEvent::Completed {
end_turn,
ref token_usage,
..
} => {
let output = std::mem::take(output);
self.buffered_output = None;
if end_turn != Some(false)
&& policy.should_resample(GuardianReviewResponse::new(
get_last_assistant_message_from_turn(output.iter()).unwrap_or_default(),
))
{
self.retry_pending = true;
} else {
self.accepted_output.extend(output);
}
// Account for this completion before replaying any buffered output.
self.update_usage_boundary =
token_usage.is_some() && !self.accepted_output.is_empty();
}
ResponseEvent::Created { .. }
| ResponseEvent::SafetyBuffering(_)
| ResponseEvent::ServerModel(_)
| ResponseEvent::ModelVerifications(_)
| ResponseEvent::TurnModerationMetadata(_)
| ResponseEvent::GuardianAssessment(_)
| ResponseEvent::ServerReasoningIncluded(_)
| ResponseEvent::RateLimits(_)
| ResponseEvent::ModelsEtag(_) => {}
}
Some(event)
}
/// Called after the denied completion has been accounted for, with the unchanged request.
pub(crate) async fn retry(
&mut self,
request: impl Future<Output = CodexResult<ResponseStream>> + Send,
cancellation_token: &CancellationToken,
) -> CodexResult<ResponseStream> {
self.retry_pending = false;
self.buffered_output = Some(Vec::new());
request.or_cancel(cancellation_token).await?
}
}
codex-rs/core/src/guardian/review_session.rs
+11 -62
View File
@@ -67,7 +67,6 @@ use crate::config::ManagedFeatures;
use crate::config::Permissions;
use crate::context::ContextualUserFragment;
use crate::context::GuardianContextMode;
use crate::context::GuardianDenialReconsideration;
use crate::context::GuardianFollowupReviewReminder;
use crate::context::GuardianNodeReplPolicy;
use crate::context_manager::ContextManager;
@@ -78,7 +77,6 @@ use crate::session::turn_context::TurnContext;
use codex_config::types::McpServerConfig;
use codex_features::Feature;
use codex_model_provider_info::ModelProviderInfo;
use codex_protocol::turn_input::TurnInput;
use codex_protocol::turn_input::TurnInputMode;
use codex_protocol::turn_input::TurnInputRequest;
use codex_protocol::turn_input::TurnInputSubmission;
@@ -158,7 +156,6 @@ pub struct GuardianReviewSessionReuseKey {
omit_personality: bool,
permissions: Permissions,
developer_instructions: Option<String>,
denial_reconsideration_prompt: Option<String>,
base_instructions: Option<String>,
user_instructions: Option<Instructions>,
thread_instructions: Option<Instructions>,
@@ -202,9 +199,6 @@ impl GuardianReviewSessionReuseKey {
omit_personality: spawn_config.personality == Some(Personality::None),
permissions: spawn_config.permissions.clone(),
developer_instructions: spawn_config.developer_instructions.clone(),
denial_reconsideration_prompt: spawn_config
.guardian_denial_reconsideration_prompt
.clone(),
base_instructions: spawn_config.base_instructions.clone(),
user_instructions: instructions.user,
thread_instructions: instructions.thread,
@@ -605,22 +599,14 @@ async fn run_review_on_session(
root_turn_id: parent_turn.turn_metadata_state.root_turn_id(),
}
.into_request();
let reconsideration = params
let store = &review_session.session.services.thread_extension_data;
if params
.spawn_config
.features
.enabled(Feature::GuardianDenialReconsideration)
.then(|| {
let mut followup = request.clone();
followup.input = TurnInput::ResponseItem(ContextualUserFragment::into(
GuardianDenialReconsideration {
prompt: params
.spawn_config
.guardian_denial_reconsideration_prompt
.as_deref(),
},
));
followup
});
{
store.insert(codex_guardian_reviewer::DenialResample::default());
}
let child_turn_id = match codex_guardian_reviewer::start_review_turn(
review_session,
request,
@@ -631,6 +617,7 @@ async fn run_review_on_session(
{
Ok(turn_id) => turn_id,
Err(outcome) => {
store.remove::<codex_guardian_reviewer::DenialResample>();
review_session
.session
.services
@@ -666,47 +653,11 @@ async fn run_review_on_session(
)
.await;
execution_attempt.finish(turn_result.turn_completed);
let turn_result = match reconsideration {
Some(request) => {
if let GuardianReviewSessionOutcome::Completed(Ok(Some(response))) =
&turn_result.outcome
&& response.assessment().is_ok_and(|assessment| {
assessment.outcome == codex_protocol::protocol::GuardianAssessmentOutcome::Deny
})
&& let Some(execution) = &params.parent_context.review_execution
{
// Reconsideration's second turn does not yet have an execution observer.
execution.mark_unknown();
}
// The new turn does not own the first turn's retained evidence backup.
// Keep that evidence intact; an over-budget follow-up keeps the denial.
let store = &review_session.session.services.thread_extension_data;
let compaction_was_disabled = store
.get::<crate::compact::AutoCompactionDisabled>()
.is_some();
store.insert(crate::compact::AutoCompactionDisabled);
let result = codex_guardian_reviewer::reconsider_denial(
review_session,
turn_result,
request,
deadline,
params.external_cancel.as_ref(),
&mut analytics_result,
&params.parent_context.turn().session_telemetry,
)
.await;
if !compaction_was_disabled {
store.remove::<crate::compact::AutoCompactionDisabled>();
}
result
}
let turn_result = match store.remove::<codex_guardian_reviewer::DenialResample>() {
Some(resample) => resample.finish(turn_result, &parent_turn.session_telemetry),
None => turn_result,
};
review_session
.session
.services
.thread_extension_data
.remove::<super::PendingReviewRouting>();
store.remove::<super::PendingReviewRouting>();
// Finalizing input consumes this context; failures before admission leave it pending.
let transcript_admitted = review_session
@@ -719,10 +670,8 @@ async fn run_review_on_session(
turn_result.outcome,
GuardianReviewSessionOutcome::Completed(_)
);
if completed_review
&& turn_result.turn_completed
&& let Some(total_token_usage) = review_session.session.total_token_usage().await
{
// Failed reviews can still spend tokens on completed inference requests.
if let Some(total_token_usage) = review_session.session.total_token_usage().await {
analytics_result.token_usage = Some(token_usage_delta(
&token_usage_at_review_start,
&total_token_usage,
codex-rs/core/src/guardian/review_session_tests.rs
+11 -41
View File
@@ -1127,13 +1127,8 @@ async fn parent_checkpoint_recovery_stops_after_one_restart() {
assert!(pool.trunk().await.is_none());
}
#[test_case::test_case(None; "single_review")]
#[test_case::test_case(Some(r#"{"outcome":"allow"}"#); "reconsideration_allows")]
#[test_case::test_case(Some("invalid assessment"); "reconsideration_keeps_denial")]
#[tokio::test]
async fn compaction_and_review_submissions_share_execution_observer(
reconsidered_assessment: Option<&str>,
) {
async fn compaction_and_review_submissions_share_execution_observer() {
let (review_session, events, submissions) = test_review_session().await;
let child_session = Arc::clone(&review_session.session);
let mut params = test_review_params().await;
@@ -1144,13 +1139,6 @@ async fn compaction_and_review_submissions_share_execution_observer(
.unwrap();
params.spawn_config.model_auto_compact_token_limit = Some(0);
params.spawn_config.token_budget = None;
if reconsidered_assessment.is_some() {
params
.spawn_config
.features
.enable(Feature::GuardianDenialReconsideration)
.unwrap();
}
let turn = Arc::get_mut(&mut params.parent_context.turn).expect("unshared turn fixture");
turn.configured_token_budget = None;
turn.use_model_token_budget_defaults = false;
@@ -1169,11 +1157,7 @@ async fn compaction_and_review_submissions_share_execution_observer(
.await
});
let start = std::time::Instant::now();
let mut child_intervals = vec![(10, 30), (40, 70)];
if reconsidered_assessment.is_some() {
child_intervals.push((75, 90));
}
for (index, (left, right)) in child_intervals.into_iter().enumerate() {
for (index, (left, right)) in [(10, 30), (40, 70)].into_iter().enumerate() {
let submission = tokio::time::timeout(Duration::from_secs(10), submissions.recv())
.await
.expect("child submission deadline")
@@ -1195,26 +1179,17 @@ async fn compaction_and_review_submissions_share_execution_observer(
.services
.thread_extension_data
.remove::<crate::turn_timing::review_execution::PendingReviewExecution>();
if index < 2 {
let pending = pending.expect("instrumented child");
assert_eq!(pending.0, execution);
let child_start = start + Duration::from_millis(left);
pending.0.started(child_start);
pending
.0
.completed(child_start, start + Duration::from_millis(right));
} else {
assert!(pending.is_none());
}
let assessment = match (index, reconsidered_assessment) {
(1, Some(_)) => r#"{"outcome":"deny"}"#,
(2, Some(assessment)) => assessment,
_ => "complete",
};
let pending = pending.expect("instrumented child");
assert_eq!(pending.0, execution);
let child_start = start + Duration::from_millis(left);
pending.0.started(child_start);
pending
.0
.completed(child_start, start + Duration::from_millis(right));
events
.send(turn_complete_event(
&submission.id,
Some(assessment),
Some("complete"),
/*time_to_first_token_ms*/ None,
))
.await
@@ -1238,10 +1213,5 @@ async fn compaction_and_review_submissions_share_execution_observer(
.auto_review(start, end, vec![execution])
.durations(start, end);
assert_eq!(totals.auto_review, Duration::from_millis(100));
assert_eq!(
totals.auto_review_overhead,
reconsidered_assessment
.is_none()
.then_some(Duration::from_millis(50))
);
assert_eq!(totals.auto_review_overhead, Some(Duration::from_millis(50)));
}
codex-rs/core/src/session/turn.rs
+65 -27
View File
@@ -2960,7 +2960,7 @@ async fn try_run_sampling_request(
let mut response_id_for_assessment = None;
let mut pending_assessment: Option<codex_api::GuardianAssessment> = None;
let mut response_completed = false;
let mut request_token_usage = None;
let mut request_token_usage: Option<codex_protocol::protocol::TokenUsage> = None;
let mut includes_reasoning = step_context.settings.model_info.use_responses_lite;
// Catalog policies can enable adaptive review independently of the legacy scorer flag.
// Capture evidence whenever Guardian is available; the reviewer resolves the policy.
@@ -2973,21 +2973,25 @@ async fn try_run_sampling_request(
} else {
None
};
let stream = client_session
.stream(
&prompt,
&step_context.settings.model_info,
&step_context.session_telemetry,
effort,
step_context.settings.reasoning_summary,
step_context.settings.service_tier.clone(),
responses_metadata,
&inference_trace,
)
.instrument(trace_span!("stream_request"))
let mut resampling = crate::guardian::ResponseResampling::new(&sess);
let mut stream_request = async |prompt: &Prompt| {
client_session
.stream(
prompt,
&step_context.settings.model_info,
&step_context.session_telemetry,
effort.clone(),
step_context.settings.reasoning_summary,
step_context.settings.service_tier.clone(),
responses_metadata,
&inference_trace,
)
.instrument(trace_span!("stream_request"))
.await
};
let stream = stream_request(&prompt)
.or_cancel(&cancellation_token)
.await?;
drop(prompt);
let mut stream = match stream {
Ok(stream) => stream,
Err(error) => {
@@ -3041,6 +3045,16 @@ async fn try_run_sampling_request(
let mut active_item_is_streaming_to_client = false;
let receiving_span = trace_span!("receiving_stream");
let outcome: CodexResult<SamplingRequestResult> = loop {
if response_completed && resampling.accepted_output.is_empty() {
if resampling.update_usage_boundary {
sess.state.lock().await.history.reset_last_usage_boundary();
}
break Ok(SamplingRequestResult {
needs_follow_up,
mailbox_preempted: *draining_for_mailbox,
last_agent_message,
});
}
let handle_responses = trace_span!(
parent: &receiving_span,
"handle_responses",
@@ -3058,8 +3072,8 @@ async fn try_run_sampling_request(
);
let receive_started = Instant::now();
let event = stream
.next()
let event = resampling
.next(&mut stream)
.instrument(trace_span!(parent: &handle_responses, "receiving"))
.or_cancel(&preempt)
.or_cancel(&cancellation_token)
@@ -3112,7 +3126,11 @@ async fn try_run_sampling_request(
Ok(event) => event,
Err(err) => {
if let CodexErrorDetails::ResponseIncomplete(incomplete) = err.details() {
request_token_usage = incomplete.token_usage.clone();
if let Some(usage) = &incomplete.token_usage {
request_token_usage
.get_or_insert_default()
.add_assign(usage);
}
sess.record_incomplete_response_usage(
&turn_context,
&step_context.settings,
@@ -3134,10 +3152,13 @@ async fn try_run_sampling_request(
}
};
record_turn_ttft_metric(&turn_context, &event).await;
let Some(event) = resampling.filter(event, &step_context.tool_router) else {
continue;
};
sess.services
.session_telemetry
.record_responses(&handle_responses, &event);
record_turn_ttft_metric(&turn_context, &event).await;
match event {
ResponseEvent::Created { response_id } => {
@@ -3490,7 +3511,11 @@ async fn try_run_sampling_request(
.await;
}
response_completed = true;
request_token_usage = token_usage.clone();
if let Some(usage) = &token_usage {
request_token_usage
.get_or_insert_default()
.add_assign(usage);
}
sess.services
.analytics_events_client
.track_code_mode_tool_call(
@@ -3520,9 +3545,13 @@ async fn try_run_sampling_request(
&turn_context,
&step_context.settings,
token_usage.as_ref(),
super::TokenUsageContext::Sampling {
includes_reasoning,
input_item_count,
if resampling.retry_pending {
super::TokenUsageContext::SpendingOnly
} else {
super::TokenUsageContext::Sampling {
includes_reasoning,
input_item_count,
}
},
)
.await;
@@ -3531,14 +3560,23 @@ async fn try_run_sampling_request(
if let Err(err) = budget_result {
break Err(err);
}
if resampling.retry_pending {
response_completed = false;
response_id_for_assessment = None;
pending_assessment = None;
includes_reasoning = step_context.settings.model_info.use_responses_lite;
stream = match resampling
.retry(stream_request(&prompt), &cancellation_token)
.await
{
Ok(stream) => stream,
Err(error) => break Err(error),
};
continue;
}
if let Some(false) = end_turn {
needs_follow_up = true;
}
break Ok(SamplingRequestResult {
needs_follow_up,
mailbox_preempted: *draining_for_mailbox,
last_agent_message,
});
}
ResponseEvent::OutputTextDelta(delta) => {
// In review child threads, suppress assistant text deltas; the