Add Guardian context cost and request token telemetry (#44164)

## What changed

- Record per-section text bytes, estimated text tokens, image bytes, and image counts for synchronous reviews and asynchronous scoring without logging evidence payloads.
- Emit estimated request tokens through `codex.guardian.context.request_tokens`. Synchronous estimates include assembled history, instructions, tool definitions, and output format, and measure the full logical request before WebSocket delta generation. Asynchronous estimates cover the assembled input.
- Add shared context budgeting helpers, including conservative image token reservations independent of encoded payload size and model-aware input limit calculation.

## Testing

Add coverage for separate text and image accounting, image estimates independent of encoded size, and section estimates that bound delivered messages. Extend asynchronous scorer and app-server tests to verify cost metric emission.

GitOrigin-RevId: aed45ecd9c23706f88caa51f2a4f2c77872d3bdb

codex-rs/app-server/tests/suite/v2/guardian_v2.rs

@@ -179,6 +179,7 @@ struct MockResponsesState {
     allow_guardian_review: Notify,
     classification_completed: Notify,
     truncation_recorded: Notify,
+    sync_cost_recorded: Notify,
     luna_score: f64,
     invalid_classification: bool,
     review_outcome: ReviewOutcome,
@@ -717,6 +718,21 @@ async fn guardian_v2_routes_scoped_tool_approvals(
             "/metrics",
             post(
                 |State(state): State<Arc<MockResponsesState>>, body: String| async move {
+                    if matches!(state.transcript_content, TranscriptContent::MixedEvidence) {
+                        let payload: Value = serde_json::from_str(&body).expect("OTLP JSON");
+                        for metric in payload["resourceMetrics"].as_array().into_iter().flatten()
+                            .flat_map(|resource| resource["scopeMetrics"].as_array().into_iter().flatten())
+                            .flat_map(|scope| scope["metrics"].as_array().into_iter().flatten())
+                            .filter(|metric| metric["name"] == "codex.guardian.context.request_tokens")
+                        {
+                            for point in metric["histogram"]["dataPoints"].as_array().into_iter().flatten() {
+                                let attributes = point["attributes"].as_array().expect("metric attributes");
+                                if attributes.iter().any(|attr| attr["key"] == "target" && attr["value"]["stringValue"] == "sync") {
+                                    state.sync_cost_recorded.notify_one();
+                                }
+                            }
+                        }
+                    }
                     if body.contains("codex.guardian_v2.classification") {
                         state.classification_completed.notify_one();
                     }
@@ -1325,6 +1341,11 @@ async fn guardian_v2_routes_scoped_tool_approvals(
             assert!(text.contains("Planned action JSON:"));
         }
     }
+    if mixed_evidence {
+        timeout(TIMEOUT, responses_state.sync_cost_recorded.notified())
+            .await
+            .expect("sync request cost should be exported");
+    }
     if lifecycle.has_user_answer() {
         let reviews = responses_state
             .guardian_requests

codex-rs/core/src/client.rs

@@ -1640,6 +1640,9 @@ impl ModelClientSession {
             );
             self.client
                 .prepare_response_items_for_request(&mut request.input);
+            if crate::guardian::is_basic_session_source(&self.client.state.session_source) {
+                crate::guardian::observe_guardian_request(session_telemetry, &request);
+            }
             let request_session_telemetry =
                 session_telemetry_for_request(session_telemetry, &request);
             let inference_trace_attempt = inference_trace.start_attempt();
@@ -1830,6 +1833,12 @@ impl ModelClientSession {
                 Err(err) => return Err(provider.map_api_error(err)),
             }
 
+            // Measure the complete logical request, not only the websocket delta.
+            if !warmup
+                && crate::guardian::is_basic_session_source(&self.client.state.session_source)
+            {
+                crate::guardian::observe_guardian_request(session_telemetry, &request);
+            }
             let (incremental_request, previous_response_id_from_untraced_warmup) =
                 self.prepare_websocket_request(&request);
             let inference_trace_attempt = if warmup {

codex-rs/core/src/guardian/mod.rs

@@ -9,6 +9,8 @@ mod decision;
 mod feedback;
 mod metrics;
 mod prompt;
+mod request_budget;
+pub(crate) use request_budget::observe as observe_guardian_request;
 mod review;
 mod review_session;
 mod reviewer_config;

codex-rs/core/src/guardian/prompt.rs

@@ -210,9 +210,21 @@ pub(crate) async fn build_guardian_prompt_items_with_parent_turn(
     if transcript_entries.is_empty() {
         transcript.items.push(placeholder.to_owned());
     }
-    let items = sections
-        .compose(presentation, transcript)?
-        .into_user_inputs()?;
+    let context = sections.compose(presentation, transcript)?;
+    for (section, cost) in context.section_costs() {
+        for (measurement, value) in cost.measurements() {
+            session.services.session_telemetry.histogram(
+                codex_guardian_context::SECTION_COST_METRIC,
+                i64::try_from(value).unwrap_or(i64::MAX),
+                &[
+                    ("target", "sync"),
+                    ("section", section),
+                    ("measurement", measurement),
+                ],
+            );
+        }
+    }
+    let items = context.into_user_inputs()?;
     Ok(GuardianPromptItems {
         items,
         transcript_cursor,

codex-rs/core/src/guardian/request_budget.rs

@@ -0,0 +1,33 @@
+//! Measures the complete synchronous request after image admission and wire
+//! prefix assembly. Includes reused history, tool definitions and output format.
+
+use codex_guardian_context::REQUEST_TOKENS_METRIC;
+use codex_protocol::protocol::TruncationPolicy;
+
+use crate::context_manager::estimate_item_token_count;
+use codex_api::ResponsesApiRequest;
+use codex_otel::SessionTelemetry;
+
+pub(crate) fn observe(telemetry: &SessionTelemetry, request: &ResponsesApiRequest) -> usize {
+    let input = request
+        .input
+        .iter()
+        .map(estimate_item_token_count)
+        .fold(0i64, i64::saturating_add);
+    let instructions = TruncationPolicy::Bytes(request.instructions.len()).token_budget();
+    let metadata = serde_json::to_vec(&(&request.tools, &request.text))
+        .map(|bytes| TruncationPolicy::Bytes(bytes.len()).token_budget())
+        .unwrap_or(usize::MAX);
+    let total = usize::try_from(input)
+        .unwrap_or(usize::MAX)
+        .saturating_add(instructions)
+        .saturating_add(metadata);
+    // The assembled input already includes inherited history and the current
+    // review. Do not report a guessed old/new split after context injection.
+    telemetry.histogram(
+        REQUEST_TOKENS_METRIC,
+        i64::try_from(total).unwrap_or(i64::MAX),
+        &[("target", "sync"), ("component", "total")],
+    );
+    total
+}

codex-rs/ext/guardian-v2/src/async_scorer/extension.rs

@@ -600,6 +600,21 @@ impl GuardianV2Extension {
             drop(history);
             drop(node_repl_images);
             truncations.extend(std::mem::take(&mut transcript.truncations));
+            if let Some(metrics) = metrics.as_deref() {
+                for (section, cost) in transcript.section_costs() {
+                    for (measurement, value) in cost.measurements() {
+                        metrics.histogram(
+                            codex_guardian_context::SECTION_COST_METRIC,
+                            i64::try_from(value).unwrap_or(i64::MAX),
+                            &[
+                                ("target", "async"),
+                                ("section", section),
+                                ("measurement", measurement),
+                            ],
+                        );
+                    }
+                }
+            }
             let classification_input = transcript.into_messages();
             let mut failure_reason = "invalid_output";
             let mut classification_risk = None;

codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs

@@ -349,6 +349,14 @@ fn fast_decision_metric(decision: &str, reason: &str) -> RecordedMetric {
 #[derive(Default)]
 struct RecordingMetrics(Mutex<Vec<RecordedMetric>>);
 
+impl RecordingMetrics {
+    fn classification_samples(&self) -> Vec<RecordedMetric> {
+        self.0.lock().unwrap().iter().filter(|sample| {
+            !matches!(sample, RecordedMetric::Histogram(name, _, _) if name == codex_guardian_context::SECTION_COST_METRIC || name == codex_guardian_context::REQUEST_TOKENS_METRIC)
+        }).cloned().collect()
+    }
+}
+
 impl ExtensionMetrics for RecordingMetrics {
     fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) {
         self.0.lock().unwrap().push(RecordedMetric::Counter(
@@ -1436,12 +1444,22 @@ max_recent_non_user_entries = 8
             .latest_scored_tool_call
             .load(Ordering::Acquire)
             == 0
-            || metrics.0.lock().unwrap().len() < 14
+            || metrics.classification_samples().len() < 14
         {
             tokio::task::yield_now().await;
         }
     })
     .await?;
+    assert!(metrics.0.lock().unwrap().iter().any(|sample| {
+        matches!(sample, RecordedMetric::Histogram(name, value, tags)
+        if name == codex_guardian_context::SECTION_COST_METRIC
+            && *value > 0
+            && tags == &[
+                ("target".to_owned(), "async".to_owned()),
+                ("section".to_owned(), "conversation_transcript".to_owned()),
+                ("measurement".to_owned(), "text_bytes".to_owned()),
+            ])
+    }));
     thread_store.insert(SecurityRiskScore {
         scores: BTreeMap::from([("action_risk".to_owned(), 0.65)]),
         call_id: None,
@@ -1535,7 +1553,7 @@ max_recent_non_user_entries = 8
         Some(ReviewDecision::Approved)
     );
 
-    let samples = initial_metrics.0.lock().unwrap();
+    let samples = initial_metrics.classification_samples();
     let classification_duration_ms = match &samples[9] {
         RecordedMetric::Histogram(name, duration_ms, _)
             if name == CLASSIFICATION_DURATION_METRIC =>
@@ -1545,7 +1563,7 @@ max_recent_non_user_entries = 8
         sample => panic!("expected classification duration metric, got {sample:?}"),
     };
     assert_eq!(
-        *samples,
+        samples,
         [
             ("total", 150),
             ("input", 120),

codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs

@@ -292,6 +292,23 @@ impl LunaSampler {
                 item.set_id(Some(ResponseItemId::new(prefix)));
             }
         }
+        let total_tokens = input
+            .iter()
+            .map(codex_guardian_context::estimate_input_tokens)
+            .fold(0usize, usize::saturating_add);
+        if let Some(metrics) = self.config.metrics.as_deref() {
+            for (component, tokens) in [
+                ("existing_context", 0),
+                ("new_input", total_tokens),
+                ("total", total_tokens),
+            ] {
+                metrics.histogram(
+                    codex_guardian_context::REQUEST_TOKENS_METRIC,
+                    i64::try_from(tokens).unwrap_or(i64::MAX),
+                    &[("target", "async"), ("component", component)],
+                );
+            }
+        }
         let mut request = ResponsesApiRequest {
             model: MODEL.to_owned(),
             instructions: String::new(),

codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs

@@ -276,14 +276,29 @@ async fn sampler_records_token_usage_after_returning_an_early_classification() -
 
     assert_eq!(sampler.sample(sample_request("turn-1")).await?, "low");
     tokio::time::timeout(Duration::from_secs(2), async {
-        while metrics.0.lock().unwrap().len() < 7 {
+        while metrics
+            .0
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|metric| metric.0 == CLASSIFICATION_TOKEN_USAGE_METRIC)
+            .count()
+            < 7
+        {
             tokio::task::yield_now().await;
         }
     })
     .await?;
 
     assert_eq!(
-        *metrics.0.lock().unwrap(),
+        metrics
+            .0
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|metric| metric.0 == CLASSIFICATION_TOKEN_USAGE_METRIC)
+            .cloned()
+            .collect::<Vec<_>>(),
         [
             ("total", 37),
             ("input", 37),
@@ -300,6 +315,27 @@ async fn sampler_records_token_usage_after_returning_an_early_classification() -
         ))
     );
 
+    let request = server
+        .wait_for_request(
+            /*connection_index*/ INITIAL_WEBSOCKET_CONNECTIONS - 1,
+            /*request_index*/ 0,
+        )
+        .await
+        .body_json();
+    let input: Vec<ResponseItem> = serde_json::from_value(request["input"].clone())?;
+    let estimated = input
+        .iter()
+        .map(codex_guardian_context::estimate_input_tokens)
+        .sum::<usize>();
+    assert!(metrics.0.lock().unwrap().contains(&(
+        codex_guardian_context::REQUEST_TOKENS_METRIC.to_owned(),
+        i64::try_from(estimated)?,
+        vec![
+            ("target".to_owned(), "async".to_owned()),
+            ("component".to_owned(), "total".to_owned()),
+        ],
+    )));
+
     Ok(())
 }
 

codex-rs/guardian-context/src/budget.rs

@@ -0,0 +1,200 @@
+//! Payload-free accounting for composed sections and complete Guardian requests.
+//! Text and image bytes stay separate. Hosts provide existing-context estimates;
+//! the shared wire estimate reserves image tokens without counting base64 as text.
+
+use std::io::Write;
+
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::TruncationPolicy;
+
+use crate::ComposedContext;
+use crate::composition::SectionDelivery;
+use crate::composition::SectionOutput;
+
+pub const SECTION_COST_METRIC: &str = "codex.guardian.context.section_cost";
+pub const REQUEST_TOKENS_METRIC: &str = "codex.guardian.context.request_tokens";
+// A conservative reservation matching the existing original-image patch ceiling.
+const IMAGE_TOKEN_RESERVATION: usize = 10_000;
+/// Conservative input ceiling when the model catalog has no authoritative window.
+pub const DEFAULT_MAX_INPUT_TOKENS: usize = 128_000;
+
+/// Applies the same configured-window cap and effective percentage as sync review.
+/// Async callers supply no parent-model override.
+pub fn effective_input_token_limit(
+    model: &codex_protocol::openai_models::ModelInfo,
+    configured_window: Option<i64>,
+) -> usize {
+    let supported = model
+        .resolved_context_window()
+        .unwrap_or(DEFAULT_MAX_INPUT_TOKENS as i64);
+    let supported = if model.used_fallback_model_metadata {
+        supported.min(DEFAULT_MAX_INPUT_TOKENS as i64)
+    } else {
+        supported
+    };
+    let limit = configured_window
+        .unwrap_or(supported)
+        .min(supported)
+        .saturating_mul(model.effective_context_window_percent.clamp(0, 100))
+        / 100;
+    usize::try_from(limit.max(0)).unwrap_or(usize::MAX)
+}
+
+/// Complete input allowance after the host resolves its model and configuration.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct RequestBudget {
+    pub max_input_tokens: usize,
+    pub existing_context_tokens: usize,
+}
+
+/// Size of section evidence, excluding enclosing message serialization overhead.
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+pub struct SectionCost {
+    pub text_bytes: usize,
+    pub image_bytes: usize,
+    pub image_count: usize,
+}
+
+impl SectionCost {
+    pub fn measurements(self) -> [(&'static str, usize); 4] {
+        [
+            ("text_bytes", self.text_bytes),
+            (
+                "estimated_text_tokens",
+                TruncationPolicy::Bytes(self.text_bytes).token_budget(),
+            ),
+            ("image_bytes", self.image_bytes),
+            ("image_count", self.image_count),
+        ]
+    }
+}
+
+impl ComposedContext {
+    /// Conservative cost of composed evidence, including message wrappers.
+    /// Shared user sections each reserve a wrapper, bounding their coalesced cost.
+    pub fn estimated_tokens(&self) -> usize {
+        self.sections
+            .iter()
+            .map(section_tokens)
+            .fold(0usize, usize::saturating_add)
+    }
+
+    /// Stable section names and numeric costs; never exposes evidence in diagnostics.
+    pub fn section_costs(&self) -> impl Iterator<Item = (&'static str, SectionCost)> + '_ {
+        self.sections.iter().map(|section| {
+            let content = match &section.delivery {
+                SectionDelivery::UserContent(content) => content,
+                SectionDelivery::Message(message) => match message.as_ref() {
+                    ResponseItem::Message { content, .. } => content,
+                    item => {
+                        return (
+                            section.id,
+                            SectionCost {
+                                text_bytes: ByteCount::item(item),
+                                ..SectionCost::default()
+                            },
+                        );
+                    }
+                },
+            };
+            let mut cost = SectionCost::default();
+            for item in content {
+                match item {
+                    ContentItem::InputText { text } | ContentItem::OutputText { text } => {
+                        cost.text_bytes = cost.text_bytes.saturating_add(text.len());
+                    }
+                    ContentItem::InputImage { image_url, .. } => {
+                        cost.image_bytes = cost.image_bytes.saturating_add(image_url.len());
+                        cost.image_count = cost.image_count.saturating_add(1);
+                    }
+                    ContentItem::InputAudio { audio_url } => {
+                        // Guardian currently has no audio contributor. Count a future opaque
+                        // payload conservatively until its consumer supplies modality costs.
+                        cost.text_bytes = cost.text_bytes.saturating_add(audio_url.len());
+                    }
+                }
+            }
+            (section.id, cost)
+        })
+    }
+}
+
+/// Conservative estimate including roles, framing, metadata and opaque checkpoints.
+/// Images reserve tokens separately from their wire bytes. This is a budgeting
+/// heuristic, not the model's tokenizer or a measurement of server-side usage.
+pub fn estimate_input_tokens(item: &ResponseItem) -> usize {
+    let content = match item {
+        ResponseItem::Message { content, .. } => content.as_slice(),
+        _ => &[],
+    };
+    adjusted_tokens(ByteCount::item(item), content)
+}
+
+pub(super) fn content_tokens(item: &ContentItem) -> usize {
+    let bytes = ByteCount::measure(|counter| serde_json::to_writer(counter, item));
+    adjusted_tokens(bytes, std::slice::from_ref(item))
+}
+
+pub(super) fn section_tokens(section: &SectionOutput) -> usize {
+    match &section.delivery {
+        SectionDelivery::UserContent(content) if content.is_empty() => 0,
+        SectionDelivery::UserContent(content) => {
+            let separators =
+                TruncationPolicy::Bytes(content.len().saturating_sub(1)).token_budget();
+            content.iter().map(content_tokens).fold(
+                estimate_input_tokens(&crate::composition::user_message(Vec::new()))
+                    .saturating_add(separators),
+                usize::saturating_add,
+            )
+        }
+        SectionDelivery::Message(message) => estimate_input_tokens(message),
+    }
+}
+
+fn adjusted_tokens(mut bytes: usize, content: &[ContentItem]) -> usize {
+    for item in content {
+        if let ContentItem::InputImage { image_url, .. } = item {
+            let payload = ByteCount::measure(|counter| serde_json::to_writer(counter, image_url));
+            if payload == usize::MAX {
+                return usize::MAX;
+            }
+            bytes = bytes
+                .saturating_sub(payload.saturating_sub(2))
+                .saturating_add(TruncationPolicy::Tokens(IMAGE_TOKEN_RESERVATION).byte_budget());
+        }
+    }
+    TruncationPolicy::Bytes(bytes).token_budget()
+}
+
+#[derive(Default)]
+struct ByteCount(usize);
+
+impl ByteCount {
+    fn item(item: &ResponseItem) -> usize {
+        Self::measure(|counter| serde_json::to_writer(counter, item))
+    }
+
+    fn measure(serialize: impl FnOnce(&mut Self) -> serde_json::Result<()>) -> usize {
+        let mut count = Self::default();
+        if serialize(&mut count).is_err() {
+            return usize::MAX;
+        }
+        count.0
+    }
+}
+
+impl Write for ByteCount {
+    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+        self.0 = self.0.saturating_add(bytes.len());
+        Ok(bytes.len())
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+#[path = "budget_tests.rs"]
+mod tests;

codex-rs/guardian-context/src/budget_tests.rs

@@ -0,0 +1,92 @@
+//! Costs distinguish evidence payloads from complete model request estimates.
+
+use super::*;
+use crate::composition::SectionOutput;
+use crate::composition::user_message as message;
+use pretty_assertions::assert_eq;
+
+#[test]
+fn section_costs_keep_multimodal_payloads_separate() {
+    let context = ComposedContext {
+        sections: vec![
+            SectionOutput {
+                id: "transcript",
+                delivery: SectionDelivery::UserContent(vec![
+                    ContentItem::InputText {
+                        text: "évidence".to_owned(),
+                    },
+                    ContentItem::InputImage {
+                        image_url: "data:image/png;base64,AAAA".to_owned(),
+                        detail: None,
+                    },
+                ]),
+            },
+            SectionOutput {
+                id: "trusted",
+                delivery: SectionDelivery::Message(Box::new(message(vec![
+                    ContentItem::InputText {
+                        text: "verified".to_owned(),
+                    },
+                ]))),
+            },
+        ],
+        truncations: Vec::new(),
+    };
+    assert_eq!(
+        context.section_costs().collect::<Vec<_>>(),
+        vec![
+            (
+                "transcript",
+                SectionCost {
+                    text_bytes: "évidence".len(),
+                    image_bytes: "data:image/png;base64,AAAA".len(),
+                    image_count: 1
+                }
+            ),
+            (
+                "trusted",
+                SectionCost {
+                    text_bytes: "verified".len(),
+                    ..SectionCost::default()
+                }
+            ),
+        ]
+    );
+}
+
+#[test]
+fn request_estimate_reserves_images_independently_of_encoded_size() {
+    let image = |payload: &str| {
+        message(vec![ContentItem::InputImage {
+            image_url: format!("data:image/png;base64,{payload}"),
+            detail: None,
+        }])
+    };
+    let short = estimate_input_tokens(&image("AAAA"));
+    assert_eq!(short, estimate_input_tokens(&image(&"A".repeat(200_000))));
+    assert!(short >= IMAGE_TOKEN_RESERVATION);
+}
+
+#[test]
+fn section_estimate_bounds_the_delivered_message() {
+    // Individually rounded costs do not always leave room for JSON separators.
+    for text in ["x", "quoted \"text\"", "évidence"] {
+        for count in 1..=40 {
+            let context = ComposedContext {
+                sections: vec![SectionOutput {
+                    id: "transcript",
+                    delivery: SectionDelivery::UserContent(vec![
+                        ContentItem::InputText {
+                            text: text.to_owned(),
+                        };
+                        count
+                    ]),
+                }],
+                truncations: Vec::new(),
+            };
+            let estimate = context.estimated_tokens();
+            let delivered = context.into_messages();
+            assert!(estimate >= estimate_input_tokens(&delivered[0]));
+        }
+    }
+}

codex-rs/guardian-context/src/composition.rs

@@ -270,7 +270,7 @@ impl ComposedContext {
     }
 }
 
-fn user_message(content: Vec<ContentItem>) -> ResponseItem {
+pub(super) fn user_message(content: Vec<ContentItem>) -> ResponseItem {
     ResponseItem::Message {
         id: None,
         role: "user".to_owned(),

codex-rs/guardian-context/src/lib.rs

@@ -48,7 +48,15 @@ pub use verified_answers::render_verified_answers;
 mod retained_instructions;
 
 mod action;
+mod budget;
 mod composition;
+pub use budget::DEFAULT_MAX_INPUT_TOKENS;
+pub use budget::REQUEST_TOKENS_METRIC;
+pub use budget::RequestBudget;
+pub use budget::SECTION_COST_METRIC;
+pub use budget::SectionCost;
+pub use budget::effective_input_token_limit;
+pub use budget::estimate_input_tokens;
 mod profile;
 pub use composition::CollectedContext;
 pub use composition::ComposedContext;