feat(token-usage): extract request speed / Codex service tier per entry

src/daemon/token_usage_worker.rs

@@ -55,7 +55,7 @@ use crate::error::GitAiError;
 use crate::metrics::{EventAttributes, MetricEvent, PosEncoded, TokenUsageValues};
 use crate::streams::db::{StreamRecord, StreamsDatabase};
 use crate::token_usage::db::{BatchCommit, TokenUsageDatabase, TrackedFile};
-use crate::token_usage::extractor_for_tool;
+use crate::token_usage::{Speed, extractor_for_tool};
 
 /// One raw JSONL line read as bytes: unlike UTF-8-strict `read_line`, a
 /// single invalid byte cannot wedge the cursor forever (the line decodes
@@ -729,6 +729,38 @@ fn reconcile_flagged_sessions(
     Ok(())
 }
 
+/// Speed fallback from the rollout's codex-home `config.toml` (recorded
+/// service tiers win; this covers unmarked entries). Cached by config mtime:
+/// a pass over many rollouts stats the file once per rollout but reads and
+/// parses it only when it changed — never per line.
+fn codex_config_fallback_speed(stream_path: &Path) -> Option<Speed> {
+    use std::sync::{Mutex, OnceLock};
+    use std::time::SystemTime;
+    type Cache = HashMap<PathBuf, (Option<SystemTime>, Option<Speed>)>;
+    static CACHE: OnceLock<Mutex<Cache>> = OnceLock::new();
+
+    let config_path =
+        crate::streams::model_extraction::codex_home_from_transcript_path(stream_path)?
+            .join("config.toml");
+    let modified = std::fs::metadata(&config_path)
+        .ok()
+        .and_then(|meta| meta.modified().ok());
+    let cache = CACHE.get_or_init(Default::default);
+    if let Ok(cache) = cache.lock()
+        && let Some((cached_modified, speed)) = cache.get(&config_path)
+        && *cached_modified == modified
+    {
+        return *speed;
+    }
+    let speed = crate::streams::model_extraction::read_codex_config(&config_path)
+        .as_deref()
+        .and_then(crate::token_usage::codex::config_fallback_speed);
+    if let Ok(mut cache) = cache.lock() {
+        cache.insert(config_path, (modified, speed));
+    }
+    speed
+}
+
 /// Incrementally read one transcript file, persist deduplicated entries, and
 /// emit changed buckets through `sink`. Split out (with injectable repo
 /// resolution and sink) for direct testing without a daemon.
@@ -789,6 +821,12 @@ fn process_file(
         return Ok(());
     }
 
+    // Only lines fed below consume the fallback speed, so quiet-skipped
+    // passes never pay the config lookup.
+    if identity.tool == "codex" {
+        extractor.set_fallback_speed(codex_config_fallback_speed(Path::new(stream_path)));
+    }
+
     let file = std::fs::File::open(stream_path)?;
     let mut reader = BufReader::with_capacity(128 * 1024, file);
     reader.seek(SeekFrom::Start(offset))?;
@@ -1476,6 +1514,63 @@ mod tests {
         );
     }
 
+    #[test]
+    fn codex_config_service_tier_is_the_fallback_for_unmarked_entries() {
+        // A rollout under a codex home whose config.toml requests the fast
+        // tier: unmarked entries resolve to fast (inferred), while a recorded
+        // tier still wins.
+        let (dir, db, _) = setup();
+        let codex_home = dir.path().join(".codex");
+        fs::create_dir_all(codex_home.join("sessions")).unwrap();
+        fs::write(
+            codex_home.join("config.toml"),
+            "model = \"gpt-5.1\"\nservice_tier = \"fast\" # priority lane\n",
+        )
+        .unwrap();
+        let transcript = codex_home.join("sessions").join("rollout-test.jsonl");
+        let usage = |minute: u32, totals: (u64, u64, u64, u64, u64)| {
+            format!(
+                r#"{{"timestamp":"{}","type":"event_msg","payload":{{"type":"token_count","info":{{"total_token_usage":{{"input_tokens":{},"cached_input_tokens":{},"output_tokens":{},"reasoning_output_tokens":{},"total_tokens":{}}}}}}}}}"#,
+                recent_ts(minute, 0),
+                totals.0,
+                totals.1,
+                totals.2,
+                totals.3,
+                totals.4
+            )
+        };
+        fs::write(
+            &transcript,
+            format!(
+                "{}\n{}\n{}\n",
+                usage(1, (100, 40, 50, 10, 150)),
+                r#"{"timestamp":"2026-01-01T00:02:00Z","type":"event_msg","payload":{"type":"thread_settings_applied","thread_settings":{"service_tier":"standard"}}}"#,
+                usage(3, (300, 140, 90, 30, 390)),
+            ),
+        )
+        .unwrap();
+        let identity = SessionIdentity {
+            tool: "codex".to_string(),
+            ..identity()
+        };
+        run_as(&db, &identity, &transcript).unwrap();
+
+        let conn =
+            crate::sqlite::open_with_memory_limits(dir.path().join("token-usage-db")).unwrap();
+        // Both turns land in the same 5-minute bucket, so order by insertion
+        // (file) order rather than the tying bucket_ts.
+        let rows: Vec<(u64, i64, bool)> = conn
+            .prepare("SELECT input_tokens, speed, speed_inferred FROM usage_entries ORDER BY rowid")
+            .unwrap()
+            .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
+            .unwrap()
+            .collect::<Result<_, _>>()
+            .unwrap();
+        // First turn: unmarked -> config fast, inferred. Second: recorded
+        // standard wins, not inferred.
+        assert_eq!(rows, vec![(60, 1, true), (100, 0, false)]);
+    }
+
     #[test]
     fn cross_session_replacement_reconciles_the_previous_owner_without_its_file() {
         let (dir, db, transcript_a) = setup();

src/streams/model_extraction.rs

@@ -205,15 +205,7 @@ fn extract_model_from_codex_jsonl_line(line: &str) -> Option<String> {
 
 fn extract_model_from_codex_config(path: &Path) -> Option<String> {
     let codex_home = codex_home_from_transcript_path(path)?;
-    let config_path = codex_home.join("config.toml");
-    let file = File::open(config_path).ok()?;
-    let mut content = String::new();
-    file.take(MAX_CODEX_CONFIG_BYTES + 1)
-        .read_to_string(&mut content)
-        .ok()?;
-    if content.len() as u64 > MAX_CODEX_CONFIG_BYTES {
-        return None;
-    }
+    let content = read_codex_config(&codex_home.join("config.toml"))?;
     let config: toml::Value = toml::from_str(&content).ok()?;
 
     config
@@ -224,7 +216,21 @@ fn extract_model_from_codex_config(path: &Path) -> Option<String> {
         .or_else(|| toml_string_candidate(config.get("model")))
 }
 
-fn codex_home_from_transcript_path(path: &Path) -> Option<PathBuf> {
+/// Read a codex `config.toml`, size-capped so an absurd file can't balloon
+/// memory. `None` when missing, unreadable, or over the cap.
+pub(crate) fn read_codex_config(config_path: &Path) -> Option<String> {
+    let file = File::open(config_path).ok()?;
+    let mut content = String::new();
+    file.take(MAX_CODEX_CONFIG_BYTES + 1)
+        .read_to_string(&mut content)
+        .ok()?;
+    if content.len() as u64 > MAX_CODEX_CONFIG_BYTES {
+        return None;
+    }
+    Some(content)
+}
+
+pub(crate) fn codex_home_from_transcript_path(path: &Path) -> Option<PathBuf> {
     let configured_home = crate::mdm::utils::codex_home_dir();
     if path.starts_with(&configured_home) {
         return Some(configured_home);

src/token_usage/claude.rs

@@ -8,17 +8,15 @@
 //! - Deduplication and the replacement policy run in the token-usage database
 //!   (entries stream in across runs), via [`should_replace_entry`] and the
 //!   `message_id` fallback; ccusage dedups in memory over whole files.
-//! - Fast-speed entries keep their base model name (no "-fast" suffix or fast
-//!   pricing multiplier); `usage.speed` remains the replacement tie-breaker.
+//! - Fast-speed entries keep their base model name (no "-fast" suffix);
+//!   `usage.speed` remains the replacement tie-breaker.
 //! - Entries whose model is missing or `<synthetic>` are attributed to
 //!   [`UNKNOWN_MODEL`] instead of carrying no model, so tokens aren't lost.
-//! - Long-context tiered pricing is not applied (git-ai's models.dev catalog
-//!   has flat rates); the 1h-ephemeral cache multiplier is (see `cost.rs`).
 
 use serde::Deserialize;
 
 use super::extractor::UsageExtractor;
-use super::types::{TokenCounts, UsageEntry, parse_rfc3339_secs};
+use super::types::{PricingShape, Speed, TokenCounts, UsageEntry, parse_rfc3339_secs};
 
 /// Model recorded for entries with no usable model name.
 pub const UNKNOWN_MODEL: &str = "unknown";
@@ -52,7 +50,7 @@ impl From<&UsageEntry> for ReplacementCandidate {
         Self {
             is_sidechain: entry.is_sidechain,
             token_total: entry.dedupe_token_total(),
-            has_speed: entry.has_speed,
+            has_speed: entry.speed.is_some(),
         }
     }
 }
@@ -103,21 +101,14 @@ struct RawUsage {
     cache_creation_input_tokens: u64,
     #[serde(default)]
     cache_read_input_tokens: u64,
+    /// ccusage `Speed`: unknown values fail the line's parse, matching
+    /// ccusage's strict schema.
     #[serde(default)]
     speed: Option<Speed>,
     #[serde(default)]
     cache_creation: Option<RawCacheCreation>,
 }
 
-/// ccusage `Speed`: unknown values fail the line's parse, matching ccusage's
-/// strict schema.
-#[derive(Deserialize, Clone, Copy)]
-#[serde(rename_all = "lowercase")]
-enum Speed {
-    Standard,
-    Fast,
-}
-
 #[derive(Deserialize, Default, Clone, Copy)]
 struct RawCacheCreation {
     #[serde(default)]
@@ -230,7 +221,12 @@ fn usage_entry(
             .filter(|cost| cost.is_finite() && *cost >= 0.0)
             .map(super::cost::micro_usd),
         is_sidechain,
-        has_speed: usage.speed.is_some(),
+        speed: usage.speed,
+        // The wire flag means "tier not recorded in the transcript" — an
+        // unmarked Claude entry lands in the standard bucket by default,
+        // exactly like an unmarked Codex entry, and must report the same.
+        speed_inferred: usage.speed.is_none(),
+        pricing_shape: PricingShape::Claude,
     }
 }
 
@@ -447,7 +443,9 @@ mod tests {
         assert_eq!(e.cache_write_1h, 0);
         assert_eq!(e.transcript_cost_micro_usd, None);
         assert!(!e.is_sidechain);
-        assert!(!e.has_speed);
+        assert_eq!(e.speed, None);
+        assert!(e.speed_inferred, "no usage.speed marker was recorded");
+        assert_eq!(e.pricing_shape, PricingShape::Claude);
     }
 
     #[test]
@@ -566,7 +564,7 @@ mod tests {
 
         // Tie: speed marker wins.
         let mut with_speed = base.clone();
-        with_speed.has_speed = true;
+        with_speed.speed = Some(Speed::Standard);
         assert!(should_replace_entry(&with_speed, base));
         assert!(!should_replace_entry(base, &with_speed));
         assert!(!should_replace_entry(&base.clone(), base));
@@ -582,9 +580,13 @@ mod tests {
     fn speed_marker_is_extracted() {
         let line = r#"{"timestamp":"2026-01-01T00:00:00Z","message":{"id":"m","model":"x","usage":{"input_tokens":1,"output_tokens":1,"speed":"fast"}},"requestId":"r"}"#;
         let e = &extract(line)[0];
-        assert!(e.has_speed);
+        assert_eq!(e.speed, Some(Speed::Fast));
+        assert!(!e.speed_inferred);
         // Model name keeps its base form (deviation from ccusage's "-fast").
         assert_eq!(e.model, "x");
+
+        let standard = r#"{"timestamp":"2026-01-01T00:00:00Z","message":{"id":"m","model":"x","usage":{"input_tokens":1,"output_tokens":1,"speed":"standard"}},"requestId":"r"}"#;
+        assert_eq!(extract(standard)[0].speed, Some(Speed::Standard));
     }
 
     #[test]