diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index ded51c95b2..4cfe1a909f 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -239,6 +239,17 @@ dependencies = [ "sha2 0.10.9", ] +[[package]] +name = "ahash" +version = "0.7.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "891477e0c6a8957309ee5c45a6368af3ae14bb510732d2684ffa19af310920f9" +dependencies = [ + "getrandom 0.2.17", + "once_cell", + "version_check", +] + [[package]] name = "ahash" version = "0.8.12" @@ -246,6 +257,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" dependencies = [ "cfg-if", + "const-random", "getrandom 0.3.4", "once_cell", "version_check", @@ -469,6 +481,169 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" +[[package]] +name = "arrow" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "378530e55cd479eda3c14eb345310799717e6f76d0c332041e8487022166b471" +dependencies = [ + "arrow-arith", + "arrow-array", + "arrow-buffer", + "arrow-cast", + "arrow-data", + "arrow-ord", + "arrow-row", + "arrow-schema", + "arrow-select", + "arrow-string", +] + +[[package]] +name = "arrow-arith" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a0ab212d2c1886e802f51c5212d78ebbcbb0bec980fff9dadc1eb8d45cd0b738" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "chrono", + "num-traits", +] + +[[package]] +name = "arrow-array" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfd33d3e92f207444098c75b42de99d329562be0cf686b307b097cc52b4e999e" +dependencies = [ + "ahash 0.8.12", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "chrono", + "half", + "hashbrown 0.17.1", + "num-complex", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-buffer" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c6cd424c2693bcdbc150d843dc9d4d137dd2de4782ce6df491ad11a3a0416c0" +dependencies = [ + "bytes", + "half", + "num-bigint", + "num-traits", +] + +[[package]] +name = "arrow-cast" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c5aefb56a2c02e9e2b30746241058b85f8983f0fcff2ba0c6d09006e1cded7f" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-ord", + "arrow-schema", + "arrow-select", + "atoi", + "base64 0.22.1", + "chrono", + "comfy-table", + "half", + "lexical-core", + "num-traits", + "ryu", +] + +[[package]] +name = "arrow-data" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c88210023a2bfee1896af366309a3028fc3bcbd6515fa29a7990ee1baa08ee0" +dependencies = [ + "arrow-buffer", + "arrow-schema", + "half", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-ord" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1bffd8fd2579286a5d63bac898159873e5094a79009940bcb42bbfce4f19f1d0" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "arrow-select", +] + +[[package]] +name = "arrow-row" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bab5994731204603c73ba69267616c50f80780774c6bb0476f1f830625115e0c" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "half", +] + +[[package]] +name = "arrow-schema" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f633dbfdf39c039ada1bf9e34c694816eb71fbb7dc78f613993b7245e078a1ed" +dependencies = [ + "bitflags 2.10.0", +] + +[[package]] +name = "arrow-select" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8cd065c54172ac787cf3f2f8d4107e0d3fdc26edba76fdf4f4cc170258942222" +dependencies = [ + "ahash 0.8.12", + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "num-traits", +] + +[[package]] +name = "arrow-string" +version = "58.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29dd7cda3ab9692f43a2e4acc444d760cc17b12bb6d8232ddf64e9bab7c06b42" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "arrow-select", + "memchr", + "num-traits", + "regex", + "regex-syntax 0.8.8", +] + [[package]] name = "ascii" version = "1.1.0" @@ -1318,6 +1493,18 @@ dependencies = [ "serde_core", ] +[[package]] +name = "bitvec" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1bc2832c24239b0141d5674bb9174f9d68a8b5b3f2753311927c172ca46f7e9c" +dependencies = [ + "funty", + "radium", + "tap", + "wyz", +] + [[package]] name = "blake2" version = "0.10.6" @@ -1396,9 +1583,23 @@ version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d1da5ab77c1437701eeff7c88d968729e7766172279eab0676857b3d63af7a6f" dependencies = [ + "borsh-derive", "cfg_aliases 0.2.1", ] +[[package]] +name = "borsh-derive" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfcfdc083699101d5a7965e49925975f2f55060f94f9a05e7187be95d530ca59" +dependencies = [ + "once_cell", + "proc-macro-crate", + "proc-macro2", + "quote", + "syn 2.0.114", +] + [[package]] name = "bstr" version = "1.12.1" @@ -1416,6 +1617,28 @@ version = "3.19.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510" +[[package]] +name = "bytecheck" +version = "0.6.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23cdc57ce23ac53c931e88a43d06d070a6fd142f2617be5855eb75efc9beb1c2" +dependencies = [ + "bytecheck_derive", + "ptr_meta", + "simdutf8", +] + +[[package]] +name = "bytecheck_derive" +version = "0.6.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3db406d29fbcd95542e92559bed4d8ad92636d1ca8b3b72ede10b4bcc010e659" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "bytemuck" version = "1.25.0" @@ -1494,7 +1717,7 @@ version = "0.56.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "801927ee168e17809ab8901d9f01f700cd7d8d6a6527997fee44e4b0327a253c" dependencies = [ - "ahash", + "ahash 0.8.12", "cached_proc_macro", "cached_proc_macro_types", "hashbrown 0.15.5", @@ -1537,6 +1760,12 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df8670b8c7b9dae1793364eafadf7239c40d669904660c5960d74cfd80b46a53" +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "castaway" version = "0.2.4" @@ -1852,6 +2081,18 @@ dependencies = [ "tracing", ] +[[package]] +name = "codex-analytics-materializer" +version = "0.0.0" +dependencies = [ + "anyhow", + "clap", + "codex-analytics", + "duckdb", + "pretty_assertions", + "serde_json", +] + [[package]] name = "codex-ansi-escape" version = "0.0.0" @@ -3263,6 +3504,7 @@ version = "0.0.0" dependencies = [ "anyhow", "chrono", + "codex-analytics", "codex-backend-client", "codex-config", "codex-core", @@ -4296,6 +4538,17 @@ dependencies = [ "memchr", ] +[[package]] +name = "comfy-table" +version = "7.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a65ebfec4fb190b6f90e944a817d60499ee0744e582530e2c9900a22e591d9a" +dependencies = [ + "crossterm", + "unicode-segmentation", + "unicode-width 0.2.1", +] + [[package]] name = "compact_str" version = "0.8.1" @@ -4355,6 +4608,26 @@ version = "0.9.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" +[[package]] +name = "const-random" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359" +dependencies = [ + "const-random-macro", +] + +[[package]] +name = "const-random-macro" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e" +dependencies = [ + "getrandom 0.2.17", + "once_cell", + "tiny-keccak", +] + [[package]] name = "const_format" version = "0.2.35" @@ -5432,6 +5705,24 @@ version = "0.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f678cf4a922c215c63e0de95eb1ff08a958a81d47e485cf9da1e27bf6305cfa5" +[[package]] +name = "duckdb" +version = "1.10503.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0cb42b21e3be42ef14acda15ce8dc99c125ce5376b87c8016a432cf14ae65de1" +dependencies = [ + "arrow", + "cast", + "comfy-table", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink 0.10.0", + "libduckdb-sys", + "num-integer", + "rust_decimal", + "strum 0.27.2", +] + [[package]] name = "dunce" version = "1.0.5" @@ -5718,6 +6009,18 @@ dependencies = [ "once_cell", ] +[[package]] +name = "fallible-iterator" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + [[package]] name = "faster-hex" version = "0.10.0" @@ -5874,6 +6177,7 @@ dependencies = [ "crc32fast", "libz-sys", "miniz_oxide", + "zlib-rs 0.5.5", ] [[package]] @@ -6018,6 +6322,12 @@ dependencies = [ "winapi", ] +[[package]] +name = "funty" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" + [[package]] name = "futures" version = "0.3.31" @@ -6542,7 +6852,7 @@ dependencies = [ "prodash", "thiserror 2.0.18", "walkdir", - "zlib-rs", + "zlib-rs 0.6.3", ] [[package]] @@ -7230,6 +7540,7 @@ checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" dependencies = [ "cfg-if", "crunchy", + "num-traits", "zerocopy", ] @@ -7247,6 +7558,9 @@ name = "hashbrown" version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" +dependencies = [ + "ahash 0.7.8", +] [[package]] name = "hashbrown" @@ -7254,7 +7568,7 @@ version = "0.14.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" dependencies = [ - "ahash", + "ahash 0.8.12", "allocator-api2", ] @@ -7280,6 +7594,21 @@ dependencies = [ "foldhash 0.2.0", ] +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + +[[package]] +name = "hashlink" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" +dependencies = [ + "hashbrown 0.15.5", +] + [[package]] name = "hashlink" version = "0.11.0" @@ -8504,6 +8833,63 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" +[[package]] +name = "lexical-core" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d8d125a277f807e55a77304455eb7b1cb52f2b18c143b60e766c120bd64a594" +dependencies = [ + "lexical-parse-float", + "lexical-parse-integer", + "lexical-util", + "lexical-write-float", + "lexical-write-integer", +] + +[[package]] +name = "lexical-parse-float" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52a9f232fbd6f550bc0137dcb5f99ab674071ac2d690ac69704593cb4abbea56" +dependencies = [ + "lexical-parse-integer", + "lexical-util", +] + +[[package]] +name = "lexical-parse-integer" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a7a039f8fb9c19c996cd7b2fcce303c1b2874fe1aca544edc85c4a5f8489b34" +dependencies = [ + "lexical-util", +] + +[[package]] +name = "lexical-util" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2604dd126bb14f13fb5d1bd6a66155079cb9fa655b37f875b3a742c705dbed17" + +[[package]] +name = "lexical-write-float" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50c438c87c013188d415fbabbb1dceb44249ab81664efbd31b14ae55dabb6361" +dependencies = [ + "lexical-util", + "lexical-write-integer", +] + +[[package]] +name = "lexical-write-integer" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "409851a618475d2d5796377cad353802345cba92c867d9fbcde9cf4eac4e14df" +dependencies = [ + "lexical-util", +] + [[package]] name = "libc" version = "0.2.182" @@ -8519,6 +8905,23 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "libduckdb-sys" +version = "1.10503.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9a4389c243e7afa0fbc8d8471031335ddc8a2dac4534f3290c9c2ba3edabab5" +dependencies = [ + "cc", + "flate2", + "pkg-config", + "reqwest 0.12.28", + "serde", + "serde_json", + "tar", + "vcpkg", + "zip 6.0.0", +] + [[package]] name = "libloading" version = "0.8.9" @@ -9253,6 +9656,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" dependencies = [ "autocfg", + "libm", ] [[package]] @@ -10333,7 +10737,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "22505a5c94da8e3b7c2996394d1c933236c4d743e81a410bcca4e6989fc066a4" dependencies = [ "bytes", - "heck 0.4.1", + "heck 0.5.0", "itertools 0.11.0", "log", "multimap", @@ -10353,7 +10757,7 @@ version = "0.14.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" dependencies = [ - "heck 0.4.1", + "heck 0.5.0", "itertools 0.14.0", "log", "multimap", @@ -10425,6 +10829,26 @@ version = "2.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "33cb294fe86a74cbcf50d4445b37da762029549ebeea341421c7c70370f86cac" +[[package]] +name = "ptr_meta" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0738ccf7ea06b608c10564b31debd4f5bc5e197fc8bfe088f68ae5ce81e7a4f1" +dependencies = [ + "ptr_meta_derive", +] + +[[package]] +name = "ptr_meta_derive" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16b845dbfca988fa33db069c0e230574d15a3088f147a87b64c7589eb662c9ac" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "publicsuffix" version = "2.3.0" @@ -10556,6 +10980,12 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "radium" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09" + [[package]] name = "radix_trie" version = "0.2.1" @@ -10582,7 +11012,7 @@ version = "0.3.0-alpha.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b93751ab27c9d151e84c1100057eab3f2a6a1378bc31b62abd416ecb1847658" dependencies = [ - "ahash", + "ahash 0.8.12", "asynk-strim", "bytes", "futures", @@ -10605,7 +11035,7 @@ version = "0.3.0-alpha.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e340fef2799277e204260b17af01bc23604712092eacd6defe40167f304baed8" dependencies = [ - "ahash", + "ahash 0.8.12", "hickory-resolver", "rama-core", "rama-net", @@ -10626,7 +11056,7 @@ version = "0.3.0-alpha.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "453d60af031e23af2d48995e41b17023f6150044738680508b63671f8d7417dd" dependencies = [ - "ahash", + "ahash 0.8.12", "base64 0.22.1", "bitflags 2.10.0", "chrono", @@ -10681,7 +11111,7 @@ version = "0.3.0-alpha.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3822be6703e010afec0bcfeb5dbb6e5a3b23ca5689d9b1215b66ce6446653b77" dependencies = [ - "ahash", + "ahash 0.8.12", "atomic-waker", "futures-channel", "httparse", @@ -10706,7 +11136,7 @@ version = "0.3.0-alpha.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d74fe0cd9bd4440827dc6dc0f504cf66065396532e798891dee2c1b740b2285" dependencies = [ - "ahash", + "ahash 0.8.12", "base64 0.22.1", "chrono", "const_format", @@ -10728,7 +11158,7 @@ version = "0.3.0-alpha.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6dae655a72da5f2b97cfacb67960d8b28c5025e62707b4c8c5f0c5c9843a444" dependencies = [ - "ahash", + "ahash 0.8.12", "bytes", "const_format", "fnv", @@ -10770,7 +11200,7 @@ version = "0.3.0-alpha.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b28ee9e1e5d39264414b71f5c33e7fbb66b382c3fac456fe0daad39cf5509933" dependencies = [ - "ahash", + "ahash 0.8.12", "const_format", "flume", "hex", @@ -11135,6 +11565,15 @@ version = "0.8.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7a2d987857b319362043e95f5353c0535c1f58eec5336fdfcf626430af7def58" +[[package]] +name = "rend" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71fe3824f5629716b1589be05dacd749f6aa084c87e00e016714a8cdfccc997c" +dependencies = [ + "bytecheck", +] + [[package]] name = "reqwest" version = "0.12.28" @@ -11266,6 +11705,35 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rkyv" +version = "0.7.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2297bf9c81a3f0dc96bc9521370b88f054168c29826a75e89c55ff196e7ed6a1" +dependencies = [ + "bitvec", + "bytecheck", + "bytes", + "hashbrown 0.12.3", + "ptr_meta", + "rend", + "rkyv_derive", + "seahash", + "tinyvec", + "uuid", +] + +[[package]] +name = "rkyv_derive" +version = "0.7.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84d7b42d4b8d06048d3ac8db0eb31bcb942cbeb709f0b5f2b2ebde398d3038f5" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "rmcp" version = "1.7.0" @@ -11369,6 +11837,23 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "rust_decimal" +version = "1.42.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c5108e3d4d903e21aac27f12ba5377b6b34f9f44b325e4894c7924169d06995" +dependencies = [ + "arrayvec", + "borsh", + "bytes", + "num-traits", + "rand 0.8.5", + "rkyv", + "serde", + "serde_json", + "wasm-bindgen", +] + [[package]] name = "rustc-demangle" version = "0.1.27" @@ -11487,7 +11972,7 @@ dependencies = [ "security-framework 3.5.1", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -11717,6 +12202,12 @@ version = "3.0.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "490dcfcbfef26be6800d11870ff2df8774fa6e86d047e3e8c8a76b25655e41ca" +[[package]] +name = "seahash" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b" + [[package]] name = "sec1" version = "0.7.3" @@ -12428,7 +12919,7 @@ dependencies = [ "futures-io", "futures-util", "hashbrown 0.16.1", - "hashlink", + "hashlink 0.11.0", "indexmap 2.13.0", "log", "memchr", @@ -12947,6 +13438,12 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" +[[package]] +name = "tap" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369" + [[package]] name = "tar" version = "0.4.45" @@ -12955,6 +13452,7 @@ checksum = "22692a6476a21fa75fdfc11d452fda482af402c008cdbaf3476414e122040973" dependencies = [ "filetime", "libc", + "xattr", ] [[package]] @@ -14201,6 +14699,7 @@ dependencies = [ "cfg-if", "once_cell", "rustversion", + "serde", "wasm-bindgen-macro", "wasm-bindgen-shared", ] @@ -15247,6 +15746,15 @@ version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9edde0db4769d2dc68579893f2306b26c6ecfbe0ef499b013d731b7b9247e0b9" +[[package]] +name = "wyz" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05f360fc0b24296329c78fda852a1e9ae82de9cf7b27dae4b7f62f118f77b9ed" +dependencies = [ + "tap", +] + [[package]] name = "x11rb" version = "0.13.2" @@ -15295,6 +15803,16 @@ dependencies = [ "time", ] +[[package]] +name = "xattr" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" +dependencies = [ + "libc", + "rustix 1.1.4", +] + [[package]] name = "xdg-home" version = "1.3.0" @@ -15575,6 +16093,26 @@ dependencies = [ "zstd 0.13.3", ] +[[package]] +name = "zip" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb2a05c7c36fde6c09b08576c9f7fb4cda705990f73b58fe011abf7dfb24168b" +dependencies = [ + "arbitrary", + "crc32fast", + "flate2", + "indexmap 2.13.0", + "memchr", + "zopfli", +] + +[[package]] +name = "zlib-rs" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40990edd51aae2c2b6907af74ffb635029d5788228222c4bb811e9351c0caad3" + [[package]] name = "zlib-rs" version = "0.6.3" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 9d870e29be..94cbff5612 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -2,6 +2,7 @@ members = [ "aws-auth", "analytics", + "analytics-materializer", "agent-graph-store", "agent-identity", "backend-client", @@ -285,6 +286,7 @@ dirs = "6" divan = "0.1.21" dns-lookup = "3.0.1" dotenvy = "0.15.7" +duckdb = { version = "1.10503.1", default-features = false, features = ["bundled", "json"] } dunce = "1.0.4" ed25519-dalek = { version = "2.2.0", features = ["pkcs8"] } encoding_rs = "0.8.35" diff --git a/codex-rs/analytics-materializer/BUILD.bazel b/codex-rs/analytics-materializer/BUILD.bazel new file mode 100644 index 0000000000..efca4df92e --- /dev/null +++ b/codex-rs/analytics-materializer/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "analytics-materializer", + crate_name = "codex_analytics_materializer", +) diff --git a/codex-rs/analytics-materializer/Cargo.toml b/codex-rs/analytics-materializer/Cargo.toml new file mode 100644 index 0000000000..fd23d5a70a --- /dev/null +++ b/codex-rs/analytics-materializer/Cargo.toml @@ -0,0 +1,27 @@ +[package] +edition.workspace = true +license.workspace = true +name = "codex-analytics-materializer" +version.workspace = true + +[lib] +doctest = false +name = "codex_analytics_materializer" +path = "src/lib.rs" + +[[bin]] +name = "codex-analytics-materializer" +path = "src/main.rs" + +[lints] +workspace = true + +[dependencies] +anyhow = { workspace = true } +clap = { workspace = true, features = ["derive"] } +codex-analytics = { workspace = true } +duckdb = { workspace = true } +serde_json = { workspace = true } + +[dev-dependencies] +pretty_assertions = { workspace = true } diff --git a/codex-rs/analytics-materializer/src/lib.rs b/codex-rs/analytics-materializer/src/lib.rs new file mode 100644 index 0000000000..b8ea8184d5 --- /dev/null +++ b/codex-rs/analytics-materializer/src/lib.rs @@ -0,0 +1,488 @@ +#![recursion_limit = "256"] + +use anyhow::Context; +use anyhow::Result; +use anyhow::bail; +use codex_analytics::LOCAL_ANALYTICS_SCHEMA_VERSION; +use codex_analytics::LocalAnalyticsRecord; +use codex_analytics::LocalAnalyticsRecordType; +use duckdb::Connection; +use duckdb::params_from_iter; +use duckdb::types::Value as DuckValue; +use std::fs; +use std::fs::File; +use std::io::BufRead; +use std::io::BufReader; +use std::path::Path; +use std::path::PathBuf; + +const MATERIALIZE_SQL: &str = r#" +CREATE TEMP VIEW codex_events AS +SELECT + *, + json_extract_string(payload, '$.event_type') AS event_type +FROM local_records +WHERE record_type = 'codex_analytics_event'; + +CREATE TABLE viewer_threads_v1 AS +WITH RECURSIVE thread_base AS ( + SELECT + json_extract_string(payload, '$.event_params.session_id') AS session_id, + json_extract_string(payload, '$.event_params.thread_id') AS thread_id, + json_extract_string(payload, '$.event_params.parent_thread_id') AS parent_thread_id, + json_extract_string(payload, '$.event_params.forked_from_thread_id') AS forked_from_thread_id, + json_extract_string(payload, '$.event_params.thread_source') AS thread_source, + json_extract_string(payload, '$.event_params.subagent_source') AS subagent_source, + json_extract_string(payload, '$.event_params.initialization_mode') AS initialization_mode, + TRY_CAST(json_extract_string(payload, '$.event_params.created_at') AS BIGINT) AS thread_created_at_epoch_seconds, + json_extract_string(payload, '$.event_params.model') AS model, + json_extract_string(payload, '$.event_params.app_server_client.product_client_id') AS product_client_id, + json_extract_string(payload, '$.event_params.app_server_client.client_name') AS client_name, + json_extract_string(payload, '$.event_params.app_server_client.client_version') AS client_version, + json_extract_string(payload, '$.event_params.app_server_client.rpc_transport') AS rpc_transport, + TRY_CAST(json_extract_string(payload, '$.event_params.app_server_client.experimental_api_enabled') AS BOOLEAN) AS experimental_api_enabled, + json_extract_string(payload, '$.event_params.runtime.codex_rs_version') AS codex_rs_version, + json_extract_string(payload, '$.event_params.runtime.runtime_os') AS runtime_os, + json_extract_string(payload, '$.event_params.runtime.runtime_os_version') AS runtime_os_version, + json_extract_string(payload, '$.event_params.runtime.runtime_arch') AS runtime_arch, + TRY_CAST(json_extract_string(payload, '$.event_params.ephemeral') AS BOOLEAN) AS ephemeral + FROM codex_events + WHERE event_type = 'codex_thread_initialized' + QUALIFY row_number() OVER ( + PARTITION BY + json_extract_string(payload, '$.event_params.session_id'), + json_extract_string(payload, '$.event_params.thread_id') + ORDER BY sink_line_number DESC + ) = 1 +), +thread_ancestors AS ( + SELECT session_id, thread_id, thread_id AS root_thread_id, parent_thread_id AS next_parent_thread_id, 0 AS depth + FROM thread_base + UNION ALL + SELECT ancestors.session_id, ancestors.thread_id, parent.thread_id, parent.parent_thread_id, ancestors.depth + 1 + FROM thread_ancestors ancestors + JOIN thread_base parent + ON parent.session_id = ancestors.session_id + AND parent.thread_id = ancestors.next_parent_thread_id + WHERE ancestors.depth < 100 +), +thread_roots AS ( + SELECT session_id, thread_id, root_thread_id + FROM thread_ancestors + QUALIFY row_number() OVER (PARTITION BY session_id, thread_id ORDER BY depth DESC) = 1 +) +SELECT + base.session_id, + base.thread_id, + roots.root_thread_id, + base.parent_thread_id, + base.forked_from_thread_id, + base.thread_source, + base.subagent_source, + base.initialization_mode, + base.thread_created_at_epoch_seconds, + base.model, + base.product_client_id, + base.client_name, + base.client_version, + base.rpc_transport, + base.experimental_api_enabled, + base.codex_rs_version, + base.runtime_os, + base.runtime_os_version, + base.runtime_arch, + base.ephemeral, + base.parent_thread_id IS NULL AS is_root +FROM thread_base base +JOIN thread_roots roots USING (session_id, thread_id); + +CREATE TABLE viewer_responses_calls_v1 AS +SELECT + session_id, + thread_id, + turn_id, + json_extract_string(payload, '$.responses_call_id') AS responses_call_id, + row_number() OVER ( + PARTITION BY session_id, thread_id, turn_id + ORDER BY TRY_CAST(json_extract_string(payload, '$.request_started_at_epoch_millis') AS BIGINT), json_extract_string(payload, '$.responses_call_id') + ) AS call_ordinal, + json_extract_string(payload, '$.transport') AS transport, + json_extract_string(payload, '$.status') AS status, + TRY_CAST(json_extract_string(payload, '$.request_started_at_epoch_millis') AS BIGINT) AS request_started_at_epoch_millis, + TRY_CAST(json_extract_string(payload, '$.completed_at_epoch_millis') AS BIGINT) AS completed_at_epoch_millis, + json_extract_string(payload, '$.response_id') AS response_id, + json_extract_string(payload, '$.upstream_request_id') AS upstream_request_id, + CAST(json_extract(payload, '$.request_json') AS VARCHAR) AS request_json, + NULLIF(CAST(json_extract(payload, '$.response_json') AS VARCHAR), 'null') AS response_json, + NULLIF(CAST(json_extract(payload, '$.token_usage_json') AS VARCHAR), 'null') AS token_usage_json, + NULLIF(CAST(json_extract(payload, '$.error_json') AS VARCHAR), 'null') AS error_json +FROM local_records +WHERE record_type = 'responses_api_call'; + +CREATE TEMP VIEW viewer_turn_base AS +SELECT + json_extract_string(payload, '$.event_params.session_id') AS session_id, + json_extract_string(payload, '$.event_params.thread_id') AS thread_id, + json_extract_string(payload, '$.event_params.turn_id') AS turn_id, + json_extract_string(payload, '$.event_params.parent_thread_id') AS parent_thread_id, + json_extract_string(payload, '$.event_params.thread_source') AS thread_source, + json_extract_string(payload, '$.event_params.subagent_source') AS subagent_source, + json_extract_string(payload, '$.event_params.app_server_client.product_client_id') AS product_client_id, + json_extract_string(payload, '$.event_params.app_server_client.client_name') AS client_name, + json_extract_string(payload, '$.event_params.app_server_client.client_version') AS client_version, + json_extract_string(payload, '$.event_params.app_server_client.rpc_transport') AS rpc_transport, + TRY_CAST(json_extract_string(payload, '$.event_params.app_server_client.experimental_api_enabled') AS BOOLEAN) AS experimental_api_enabled, + json_extract_string(payload, '$.event_params.runtime.codex_rs_version') AS codex_rs_version, + json_extract_string(payload, '$.event_params.runtime.runtime_os') AS runtime_os, + json_extract_string(payload, '$.event_params.runtime.runtime_os_version') AS runtime_os_version, + json_extract_string(payload, '$.event_params.runtime.runtime_arch') AS runtime_arch, + json_extract_string(payload, '$.event_params.model_provider') AS model_provider, + json_extract_string(payload, '$.event_params.service_tier') AS service_tier, + json_extract_string(payload, '$.event_params.approval_policy') AS approval_policy, + json_extract_string(payload, '$.event_params.approvals_reviewer') AS approvals_reviewer, + TRY_CAST(json_extract_string(payload, '$.event_params.sandbox_network_access') AS BOOLEAN) AS sandbox_network_access, + TRY_CAST(json_extract_string(payload, '$.event_params.num_input_images') AS BIGINT) AS num_input_images, + TRY_CAST(json_extract_string(payload, '$.event_params.is_first_turn') AS BOOLEAN) AS is_first_turn, + TRY_CAST(json_extract_string(payload, '$.event_params.ephemeral') AS BOOLEAN) AS ephemeral, + json_extract_string(payload, '$.event_params.initialization_mode') AS initialization_mode, + json_extract_string(payload, '$.event_params.workspace_kind') AS workspace_kind, + json_extract_string(payload, '$.event_params.submission_type') AS submission_type, + json_extract_string(payload, '$.event_params.model') AS model, + json_extract_string(payload, '$.event_params.sandbox_policy') AS sandbox_policy, + json_extract_string(payload, '$.event_params.reasoning_effort') AS reasoning_effort, + json_extract_string(payload, '$.event_params.reasoning_summary') AS reasoning_summary, + json_extract_string(payload, '$.event_params.collaboration_mode') AS collaboration_mode, + json_extract_string(payload, '$.event_params.personality') AS personality, + json_extract_string(payload, '$.event_params.status') AS status, + NULLIF(CAST(json_extract(payload, '$.event_params.turn_error') AS VARCHAR), 'null') AS turn_error, + TRY_CAST(json_extract_string(payload, '$.event_params.steer_count') AS BIGINT) AS steer_count, + TRY_CAST(json_extract_string(payload, '$.event_params.total_tool_call_count') AS BIGINT) AS total_tool_call_count, + TRY_CAST(json_extract_string(payload, '$.event_params.shell_command_count') AS BIGINT) AS shell_command_count, + TRY_CAST(json_extract_string(payload, '$.event_params.file_change_count') AS BIGINT) AS file_change_count, + TRY_CAST(json_extract_string(payload, '$.event_params.mcp_tool_call_count') AS BIGINT) AS mcp_tool_call_count, + TRY_CAST(json_extract_string(payload, '$.event_params.dynamic_tool_call_count') AS BIGINT) AS dynamic_tool_call_count, + TRY_CAST(json_extract_string(payload, '$.event_params.subagent_tool_call_count') AS BIGINT) AS subagent_tool_call_count, + TRY_CAST(json_extract_string(payload, '$.event_params.web_search_count') AS BIGINT) AS web_search_count, + TRY_CAST(json_extract_string(payload, '$.event_params.image_generation_count') AS BIGINT) AS image_generation_count, + TRY_CAST(json_extract_string(payload, '$.event_params.input_tokens') AS BIGINT) AS input_tokens, + TRY_CAST(json_extract_string(payload, '$.event_params.cached_input_tokens') AS BIGINT) AS cached_input_tokens, + TRY_CAST(json_extract_string(payload, '$.event_params.output_tokens') AS BIGINT) AS output_tokens, + TRY_CAST(json_extract_string(payload, '$.event_params.reasoning_output_tokens') AS BIGINT) AS reasoning_output_tokens, + TRY_CAST(json_extract_string(payload, '$.event_params.total_tokens') AS BIGINT) AS total_tokens, + TRY_CAST(json_extract_string(payload, '$.event_params.duration_ms') AS BIGINT) AS duration_ms, + TRY_CAST(json_extract_string(payload, '$.event_params.started_at') AS BIGINT) AS started_at_epoch_seconds, + TRY_CAST(json_extract_string(payload, '$.event_params.completed_at') AS BIGINT) AS completed_at_epoch_seconds +FROM codex_events +WHERE event_type = 'codex_turn_event' +QUALIFY row_number() OVER ( + PARTITION BY + json_extract_string(payload, '$.event_params.session_id'), + json_extract_string(payload, '$.event_params.thread_id'), + json_extract_string(payload, '$.event_params.turn_id') + ORDER BY sink_line_number DESC +) = 1; + +CREATE TEMP VIEW viewer_turn_tool_aggregates AS +SELECT + thread_id, + turn_id, + count(*) AS tool_calls_count, + coalesce(sum(TRY_CAST(json_extract_string(payload, '$.event_params.review_count') AS BIGINT)), 0) AS tool_calls_total_review_count, + coalesce(sum(TRY_CAST(json_extract_string(payload, '$.event_params.guardian_review_count') AS BIGINT)), 0) AS tool_calls_guardian_review_count, + coalesce(sum(TRY_CAST(json_extract_string(payload, '$.event_params.user_review_count') AS BIGINT)), 0) AS tool_calls_user_review_count, + sum(CASE WHEN json_extract_string(payload, '$.event_params.terminal_status') = 'failed' THEN 1 ELSE 0 END) AS tool_calls_failure_count, + sum(CASE WHEN json_extract_string(payload, '$.event_params.requested_network_access') = 'true' THEN 1 ELSE 0 END) AS tool_calls_requested_network_access_count, + sum(CASE WHEN json_extract_string(payload, '$.event_params.requested_additional_permissions') = 'true' THEN 1 ELSE 0 END) AS tool_calls_requested_additional_permissions_count, + coalesce(sum(TRY_CAST(json_extract_string(payload, '$.event_params.duration_ms') AS BIGINT)), 0) AS tool_calls_total_duration_ms +FROM codex_events +WHERE event_type IN ('codex_command_execution', 'codex_file_change', 'codex_mcp_tool_call', 'codex_dynamic_tool_call', 'codex_collab_agent_tool_call', 'codex_web_search', 'codex_image_generation') +GROUP BY thread_id, turn_id; + +CREATE TEMP VIEW viewer_turn_compaction_aggregates AS +SELECT + thread_id, + turn_id, + count(*) AS compactions_count, + sum(CASE WHEN json_extract_string(payload, '$.event_params.status') = 'completed' THEN 1 ELSE 0 END) AS compactions_completed_count, + sum(CASE WHEN json_extract_string(payload, '$.event_params.status') = 'failed' THEN 1 ELSE 0 END) AS compactions_failed_count, + bool_or(NULLIF(CAST(json_extract(payload, '$.event_params.error') AS VARCHAR), 'null') IS NOT NULL) AS compactions_any_error +FROM codex_events +WHERE event_type = 'codex_compaction' +GROUP BY thread_id, turn_id; + +CREATE TEMP VIEW viewer_turn_review_aggregates AS +SELECT + thread_id, + turn_id, + count(*) AS reviews_event_count +FROM codex_events +WHERE event_type = 'codex_review_event' +GROUP BY thread_id, turn_id; + +CREATE TEMP VIEW viewer_turn_guardian_review_aggregates AS +SELECT + thread_id, + turn_id, + count(*) AS guardian_reviews_event_count, + coalesce(sum(TRY_CAST(json_extract_string(payload, '$.event_params.tool_call_count') AS BIGINT)), 0) AS guardian_reviews_tool_call_count, + coalesce(sum(TRY_CAST(json_extract_string(payload, '$.event_params.completion_latency_ms') AS BIGINT)), 0) AS guardian_reviews_total_completion_latency_ms, + coalesce(sum(TRY_CAST(json_extract_string(payload, '$.event_params.total_tokens') AS BIGINT)), 0) AS guardian_reviews_total_tokens +FROM codex_events +WHERE event_type = 'codex_guardian_review' +GROUP BY thread_id, turn_id; + +CREATE TEMP VIEW viewer_turn_response_aggregates AS +SELECT + session_id, + thread_id, + turn_id, + count(*) AS responses_api_calls_total_count, + sum(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS responses_api_calls_failed_count, + sum(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) AS responses_api_calls_succeeded_count, + coalesce(sum(completed_at_epoch_millis - request_started_at_epoch_millis), 0) AS responses_api_calls_total_latency_ms +FROM viewer_responses_calls_v1 +GROUP BY session_id, thread_id, turn_id; + +CREATE TABLE viewer_turns_v1 AS +WITH turns_with_ordinal AS ( + SELECT + *, + row_number() OVER ( + PARTITION BY session_id, thread_id + ORDER BY started_at_epoch_seconds NULLS LAST, turn_id + ) AS turn_ordinal + FROM viewer_turn_base +) +SELECT + turns.session_id, + turns.thread_id, + turns.turn_id, + turns.turn_ordinal, + turns.parent_thread_id, + turns.thread_source, + turns.subagent_source, + turns.product_client_id, + turns.client_name, + turns.client_version, + turns.rpc_transport, + turns.experimental_api_enabled, + turns.codex_rs_version, + turns.runtime_os, + turns.runtime_os_version, + turns.runtime_arch, + turns.model_provider, + turns.service_tier, + turns.approval_policy, + turns.approvals_reviewer, + turns.sandbox_network_access, + turns.num_input_images, + turns.is_first_turn, + turns.ephemeral, + turns.initialization_mode, + turns.workspace_kind, + turns.submission_type, + turns.model, + turns.sandbox_policy, + turns.reasoning_effort, + turns.reasoning_summary, + turns.collaboration_mode, + turns.personality, + turns.status, + turns.turn_error, + turns.steer_count, + turns.total_tool_call_count, + turns.shell_command_count, + turns.file_change_count, + turns.mcp_tool_call_count, + turns.dynamic_tool_call_count, + turns.subagent_tool_call_count, + turns.web_search_count, + turns.image_generation_count, + turns.input_tokens, + turns.cached_input_tokens, + turns.output_tokens, + turns.reasoning_output_tokens, + turns.total_tokens, + turns.duration_ms, + turns.started_at_epoch_seconds, + turns.completed_at_epoch_seconds, + coalesce(compactions.compactions_count, 0) AS compactions_count, + coalesce(compactions.compactions_completed_count, 0) AS compactions_completed_count, + coalesce(compactions.compactions_failed_count, 0) AS compactions_failed_count, + coalesce(compactions.compactions_any_error, false) AS compactions_any_error, + coalesce(tools.tool_calls_count, 0) AS tool_calls_count, + coalesce(tools.tool_calls_total_review_count, 0) AS tool_calls_total_review_count, + coalesce(tools.tool_calls_guardian_review_count, 0) AS tool_calls_guardian_review_count, + coalesce(tools.tool_calls_user_review_count, 0) AS tool_calls_user_review_count, + coalesce(tools.tool_calls_failure_count, 0) AS tool_calls_failure_count, + coalesce(tools.tool_calls_requested_network_access_count, 0) AS tool_calls_requested_network_access_count, + coalesce(tools.tool_calls_requested_additional_permissions_count, 0) AS tool_calls_requested_additional_permissions_count, + coalesce(tools.tool_calls_total_duration_ms, 0) AS tool_calls_total_duration_ms, + coalesce(reviews.reviews_event_count, 0) AS reviews_event_count, + coalesce(guardian.guardian_reviews_event_count, 0) AS guardian_reviews_event_count, + coalesce(guardian.guardian_reviews_tool_call_count, 0) AS guardian_reviews_tool_call_count, + coalesce(guardian.guardian_reviews_total_completion_latency_ms, 0) AS guardian_reviews_total_completion_latency_ms, + coalesce(guardian.guardian_reviews_total_tokens, 0) AS guardian_reviews_total_tokens, + coalesce(responses.responses_api_calls_total_count, 0) AS responses_api_calls_total_count, + coalesce(responses.responses_api_calls_failed_count, 0) AS responses_api_calls_failed_count, + coalesce(responses.responses_api_calls_succeeded_count, 0) AS responses_api_calls_succeeded_count, + coalesce(responses.responses_api_calls_total_latency_ms, 0) AS responses_api_calls_total_latency_ms +FROM turns_with_ordinal turns +LEFT JOIN viewer_turn_compaction_aggregates compactions USING (thread_id, turn_id) +LEFT JOIN viewer_turn_tool_aggregates tools USING (thread_id, turn_id) +LEFT JOIN viewer_turn_review_aggregates reviews USING (thread_id, turn_id) +LEFT JOIN viewer_turn_guardian_review_aggregates guardian USING (thread_id, turn_id) +LEFT JOIN viewer_turn_response_aggregates responses USING (session_id, thread_id, turn_id); + +CREATE TEMP VIEW unique_thread_sessions AS +SELECT thread_id, min(session_id) AS session_id +FROM viewer_threads_v1 +GROUP BY thread_id +HAVING count(DISTINCT session_id) = 1; + +CREATE TABLE viewer_turn_events_v1 AS +WITH resolved_events AS ( + SELECT + coalesce(records.session_id, sessions.session_id) AS session_id, + records.thread_id, + records.turn_id, + records.sink_line_number, + records.recorded_at_epoch_millis, + records.record_type, + records.payload + FROM local_records records + LEFT JOIN unique_thread_sessions sessions USING (thread_id) + WHERE records.turn_id IS NOT NULL +) +SELECT + session_id, + thread_id, + turn_id, + row_number() OVER (PARTITION BY session_id, thread_id, turn_id ORDER BY sink_line_number) AS event_seq, + sink_line_number, + recorded_at_epoch_millis, + CASE WHEN record_type = 'responses_api_call' THEN 'responses_api' ELSE 'codex_analytics' END AS event_kind, + CASE WHEN record_type = 'responses_api_call' THEN 'responses_api_call' ELSE json_extract_string(payload, '$.event_type') END AS event_type, + CASE WHEN record_type = 'responses_api_call' THEN json_extract_string(payload, '$.responses_call_id') END AS responses_call_id, + CASE WHEN record_type = 'responses_api_call' THEN payload END AS event_summary_json, + CASE WHEN record_type = 'codex_analytics_event' THEN payload END AS analytics_event_json +FROM resolved_events; + +CREATE TABLE viewer_context_windows_v1 ( + session_id VARCHAR, + thread_id VARCHAR, + context_window_id VARCHAR, + context_window_ordinal BIGINT, + first_turn_id VARCHAR, + last_turn_id VARCHAR, + first_responses_call_id VARCHAR, + last_responses_call_id VARCHAR, + opened_at_epoch_millis BIGINT, + closed_at_epoch_millis BIGINT, + close_reason VARCHAR, + message_count BIGINT, + conversation_json VARCHAR, + conversation_source VARCHAR, + is_synthetic BOOLEAN +); +"#; + +pub fn default_output_path(input: impl AsRef) -> PathBuf { + input.as_ref().with_extension("duckdb") +} + +pub fn process_local_analytics(input: impl AsRef, output: impl AsRef) -> Result<()> { + let input = input.as_ref(); + let output = output.as_ref(); + if output.exists() { + fs::remove_file(output) + .with_context(|| format!("remove existing DuckDB file {}", output.display()))?; + } + + let connection = Connection::open(output) + .with_context(|| format!("open DuckDB file {}", output.display()))?; + create_raw_records_table(&connection)?; + insert_local_records(&connection, input)?; + connection.execute_batch(MATERIALIZE_SQL)?; + Ok(()) +} + +fn create_raw_records_table(connection: &Connection) -> Result<()> { + connection.execute_batch( + r#" + CREATE TABLE local_records ( + sink_line_number BIGINT, + schema_version BIGINT, + recorded_at_epoch_millis BIGINT, + record_type VARCHAR, + session_id VARCHAR, + thread_id VARCHAR, + turn_id VARCHAR, + payload VARCHAR + ); + "#, + )?; + Ok(()) +} + +fn insert_local_records(connection: &Connection, input: &Path) -> Result<()> { + let file = File::open(input) + .with_context(|| format!("open local analytics JSONL {}", input.display()))?; + for (line_index, line) in BufReader::new(file).lines().enumerate() { + let line_number = line_index + 1; + let line = line.with_context(|| format!("read JSONL line {line_number}"))?; + if line.trim().is_empty() { + continue; + } + let record: LocalAnalyticsRecord = serde_json::from_str(&line) + .with_context(|| format!("parse JSONL line {line_number}"))?; + if record.schema_version != LOCAL_ANALYTICS_SCHEMA_VERSION { + bail!( + "unsupported local analytics schema version {} on line {line_number}", + record.schema_version + ); + } + connection.execute( + "INSERT INTO local_records VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + params_from_iter([ + integer(i64::try_from(line_number)?), + integer(i64::from(record.schema_version)), + integer(to_i64(record.recorded_at_epoch_millis)), + text(record_type(record.record_type)), + optional_text(record.session_id), + optional_text(record.thread_id), + optional_text(record.turn_id), + text(serde_json::to_string(&record.payload)?), + ]), + )?; + } + Ok(()) +} + +fn record_type(record_type: LocalAnalyticsRecordType) -> String { + match record_type { + LocalAnalyticsRecordType::CodexAnalyticsEvent => "codex_analytics_event", + LocalAnalyticsRecordType::ResponsesApiCall => "responses_api_call", + } + .to_string() +} + +fn to_i64(value: u64) -> i64 { + i64::try_from(value).unwrap_or(i64::MAX) +} + +fn text(value: String) -> DuckValue { + DuckValue::Text(value) +} + +fn optional_text(value: Option) -> DuckValue { + value.map_or(DuckValue::Null, DuckValue::Text) +} + +fn integer(value: i64) -> DuckValue { + DuckValue::BigInt(value) +} + +#[cfg(test)] +#[path = "lib_tests.rs"] +mod tests; diff --git a/codex-rs/analytics-materializer/src/lib_tests.rs b/codex-rs/analytics-materializer/src/lib_tests.rs new file mode 100644 index 0000000000..54cc2d9745 --- /dev/null +++ b/codex-rs/analytics-materializer/src/lib_tests.rs @@ -0,0 +1,318 @@ +use super::default_output_path; +use super::process_local_analytics; +use codex_analytics::LOCAL_ANALYTICS_SCHEMA_VERSION; +use codex_analytics::LocalAnalyticsRecord; +use codex_analytics::LocalAnalyticsRecordType; +use duckdb::Connection; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::fs; +use std::path::PathBuf; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; + +static NEXT_TEST_PATH_ID: AtomicU64 = AtomicU64::new(0); + +#[test] +fn default_output_replaces_input_extension() { + assert_eq!( + default_output_path("/tmp/local-analytics.jsonl"), + PathBuf::from("/tmp/local-analytics.duckdb") + ); +} + +#[test] +fn materializes_threads_turns_events_responses_and_context_stub() { + let input = test_path("local-analytics.jsonl"); + let output = test_path("local-analytics.duckdb"); + let records = [ + local_record( + 1, + Some("session-1"), + Some("thread-root"), + None, + json!({ + "event_type": "codex_thread_initialized", + "event_params": { + "thread_id": "thread-root", + "session_id": "session-1", + "app_server_client": { + "product_client_id": "codex_cli_rs", + "client_name": "codex", + "client_version": "0.0.0", + "rpc_transport": "stdio", + "experimental_api_enabled": false + }, + "runtime": { + "codex_rs_version": "0.0.0", + "runtime_os": "macos", + "runtime_os_version": "14", + "runtime_arch": "aarch64" + }, + "model": "gpt-5", + "ephemeral": false, + "thread_source": "cli", + "initialization_mode": "new", + "subagent_source": null, + "parent_thread_id": null, + "forked_from_thread_id": null, + "created_at": 10 + } + }), + ), + local_record( + 2, + Some("session-1"), + Some("thread-child"), + None, + json!({ + "event_type": "codex_thread_initialized", + "event_params": { + "thread_id": "thread-child", + "session_id": "session-1", + "app_server_client": { + "product_client_id": "codex_cli_rs", + "client_name": "codex", + "client_version": "0.0.0", + "rpc_transport": "stdio", + "experimental_api_enabled": false + }, + "runtime": { + "codex_rs_version": "0.0.0", + "runtime_os": "macos", + "runtime_os_version": "14", + "runtime_arch": "aarch64" + }, + "model": "gpt-5", + "ephemeral": false, + "thread_source": "sub_agent", + "initialization_mode": "forked", + "subagent_source": "review", + "parent_thread_id": "thread-root", + "forked_from_thread_id": "thread-root", + "created_at": 11 + } + }), + ), + local_record( + 3, + Some("session-1"), + Some("thread-child"), + Some("turn-1"), + turn_event_payload(), + ), + local_record( + 4, + None, + Some("thread-child"), + Some("turn-1"), + json!({ + "event_type": "codex_command_execution", + "event_params": { + "thread_id": "thread-child", + "turn_id": "turn-1", + "review_count": 2, + "guardian_review_count": 1, + "user_review_count": 1, + "terminal_status": "failed", + "requested_network_access": true, + "requested_additional_permissions": false, + "duration_ms": 7 + } + }), + ), + responses_record( + 5, + json!({ + "responses_call_id": "call-1", + "transport": "http", + "status": "completed", + "request_started_at_epoch_millis": 100, + "completed_at_epoch_millis": 110, + "response_id": "response-1", + "upstream_request_id": "request-1", + "request_json": {"model": "gpt-5"}, + "response_json": {"output_items": []}, + "token_usage_json": {"total_tokens": 3}, + "error_json": null + }), + ), + ]; + write_jsonl(&input, &records); + + process_local_analytics(&input, &output).expect("materialization should succeed"); + + let connection = Connection::open(&output).expect("DuckDB should open"); + let child_thread: (String, bool) = connection + .query_row( + "SELECT root_thread_id, is_root FROM viewer_threads_v1 WHERE thread_id = 'thread-child'", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .expect("child thread should exist"); + assert_eq!(child_thread, ("thread-root".to_string(), false)); + let turn: (i64, i64, i64, i64, i64, i64) = connection + .query_row( + "SELECT turn_ordinal, tool_calls_count, tool_calls_failure_count, responses_api_calls_total_count, responses_api_calls_succeeded_count, responses_api_calls_total_latency_ms FROM viewer_turns_v1 WHERE turn_id = 'turn-1'", + [], + |row| { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + row.get(4)?, + row.get(5)?, + )) + }, + ) + .expect("turn should exist"); + assert_eq!(turn, (1, 1, 1, 1, 1, 10)); + let turn_event: (String, i64) = connection + .query_row( + "SELECT session_id, event_seq FROM viewer_turn_events_v1 WHERE event_type = 'codex_command_execution'", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .expect("tool event should exist"); + assert_eq!(turn_event, ("session-1".to_string(), 2)); + let response_call: (i64, String) = connection + .query_row( + "SELECT call_ordinal, request_json FROM viewer_responses_calls_v1", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .expect("responses call should exist"); + assert_eq!(response_call, (1, "{\"model\":\"gpt-5\"}".to_string())); + let context_rows: i64 = connection + .query_row( + "SELECT count(*) FROM viewer_context_windows_v1", + [], + |row| row.get(0), + ) + .expect("context window stub should exist"); + assert_eq!(context_rows, 0); +} + +fn turn_event_payload() -> serde_json::Value { + json!({ + "event_type": "codex_turn_event", + "event_params": { + "thread_id": "thread-child", + "session_id": "session-1", + "turn_id": "turn-1", + "submission_type": "user", + "app_server_client": { + "product_client_id": "codex_cli_rs", + "client_name": "codex", + "client_version": "0.0.0", + "rpc_transport": "stdio", + "experimental_api_enabled": false + }, + "runtime": { + "codex_rs_version": "0.0.0", + "runtime_os": "macos", + "runtime_os_version": "14", + "runtime_arch": "aarch64" + }, + "ephemeral": false, + "thread_source": "sub_agent", + "initialization_mode": "forked", + "subagent_source": "review", + "parent_thread_id": "thread-root", + "model": "gpt-5", + "model_provider": "openai", + "sandbox_policy": "workspace_write", + "reasoning_effort": "medium", + "reasoning_summary": "auto", + "service_tier": "default", + "approval_policy": "on_request", + "approvals_reviewer": "user", + "sandbox_network_access": false, + "collaboration_mode": null, + "personality": null, + "workspace_kind": "git", + "num_input_images": 0, + "is_first_turn": true, + "status": "completed", + "turn_error": null, + "codex_error_kind": null, + "codex_error_subreason": null, + "codex_error_http_status_code": null, + "steer_count": 0, + "total_tool_call_count": 1, + "shell_command_count": 1, + "file_change_count": 0, + "mcp_tool_call_count": 0, + "dynamic_tool_call_count": 0, + "subagent_tool_call_count": 0, + "web_search_count": 0, + "image_generation_count": 0, + "input_tokens": 1, + "cached_input_tokens": 0, + "output_tokens": 2, + "reasoning_output_tokens": 0, + "total_tokens": 3, + "before_first_sampling_ms": 0, + "sampling_ms": 0, + "between_sampling_overhead_ms": 0, + "tool_blocking_ms": 0, + "after_last_sampling_ms": 0, + "sampling_request_count": 1, + "sampling_retry_count": 0, + "duration_ms": 5, + "started_at": 12, + "completed_at": 13 + } + }) +} + +fn local_record( + recorded_at_epoch_millis: u64, + session_id: Option<&str>, + thread_id: Option<&str>, + turn_id: Option<&str>, + payload: serde_json::Value, +) -> LocalAnalyticsRecord { + LocalAnalyticsRecord { + schema_version: LOCAL_ANALYTICS_SCHEMA_VERSION, + recorded_at_epoch_millis, + record_type: LocalAnalyticsRecordType::CodexAnalyticsEvent, + session_id: session_id.map(std::string::ToString::to_string), + thread_id: thread_id.map(std::string::ToString::to_string), + turn_id: turn_id.map(std::string::ToString::to_string), + payload, + } +} + +fn responses_record( + recorded_at_epoch_millis: u64, + payload: serde_json::Value, +) -> LocalAnalyticsRecord { + LocalAnalyticsRecord { + schema_version: LOCAL_ANALYTICS_SCHEMA_VERSION, + recorded_at_epoch_millis, + record_type: LocalAnalyticsRecordType::ResponsesApiCall, + session_id: Some("session-1".to_string()), + thread_id: Some("thread-child".to_string()), + turn_id: Some("turn-1".to_string()), + payload, + } +} + +fn write_jsonl(path: &PathBuf, records: &[LocalAnalyticsRecord]) { + let contents = records + .iter() + .map(|record| serde_json::to_string(record).expect("record should serialize")) + .collect::>() + .join("\n"); + fs::write(path, format!("{contents}\n")).expect("fixture should write"); +} + +fn test_path(file_name: &str) -> PathBuf { + let test_id = NEXT_TEST_PATH_ID.fetch_add(1, Ordering::Relaxed); + let process_id = std::process::id(); + std::env::temp_dir().join(format!( + "codex-analytics-materializer-{process_id}-{test_id}-{file_name}" + )) +} diff --git a/codex-rs/analytics-materializer/src/main.rs b/codex-rs/analytics-materializer/src/main.rs new file mode 100644 index 0000000000..3c36a2a272 --- /dev/null +++ b/codex-rs/analytics-materializer/src/main.rs @@ -0,0 +1,30 @@ +use anyhow::Result; +use clap::Parser; +use std::path::PathBuf; + +#[derive(Debug, Parser)] +#[command( + name = "codex-analytics-materializer", + about = "Reduce local Codex analytics JSONL into a DuckDB file." +)] +struct Args { + /// Local analytics JSONL file to reduce. + input: PathBuf, + /// DuckDB file to write. Defaults to .duckdb. + #[arg(short, long, value_name = "FILE")] + output: Option, +} + +fn main() -> Result<()> { + let args = Args::parse(); + let output = args + .output + .unwrap_or_else(|| codex_analytics_materializer::default_output_path(&args.input)); + codex_analytics_materializer::process_local_analytics(&args.input, &output)?; + println!("{}", output.display()); + Ok(()) +} + +#[cfg(test)] +#[path = "main_tests.rs"] +mod tests; diff --git a/codex-rs/analytics-materializer/src/main_tests.rs b/codex-rs/analytics-materializer/src/main_tests.rs new file mode 100644 index 0000000000..7c3ac3899e --- /dev/null +++ b/codex-rs/analytics-materializer/src/main_tests.rs @@ -0,0 +1,21 @@ +use super::Args; +use clap::Parser; +use pretty_assertions::assert_eq; +use std::path::PathBuf; + +#[test] +fn parses_explicit_output_path() { + let args = Args::try_parse_from([ + "codex-analytics-materializer", + "/tmp/local-analytics.jsonl", + "--output", + "/tmp/session-viewer.duckdb", + ]) + .expect("arguments should parse"); + + assert_eq!(args.input, PathBuf::from("/tmp/local-analytics.jsonl")); + assert_eq!( + args.output, + Some(PathBuf::from("/tmp/session-viewer.duckdb")) + ); +} diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index f4cd18da1e..73c5655ad3 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -1322,6 +1322,7 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() { let (sender, _receiver) = mpsc::channel(1); let queue = AnalyticsEventsQueue { sender, + local_sink: None, app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), }; @@ -3048,6 +3049,7 @@ fn plugin_used_dedupe_is_keyed_by_turn_and_plugin() { let (sender, _receiver) = mpsc::channel(1); let queue = AnalyticsEventsQueue { sender, + local_sink: None, app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), }; diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index b99a2ec86f..e23f8c8d14 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -22,6 +22,12 @@ use crate::facts::TurnCodexErrorFact; use crate::facts::TurnProfileFact; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTokenUsageFact; +use crate::local_responses::AnalyticsQueueInput; +use crate::local_responses::LocalResponsesApiCallCapture; +use crate::local_responses::LocalResponsesApiCallReducer; +use crate::local_sink::append_codex_analytics_events_best_effort; +use crate::local_sink::append_local_analytics_record_best_effort; +use crate::local_sink::local_analytics_sink_from_env; use crate::reducer::AnalyticsReducer; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponsePayload; @@ -42,13 +48,19 @@ use std::sync::Mutex; use std::time::Duration; use tokio::sync::mpsc; +#[cfg(test)] +use crate::local_sink::local_analytics_sink_for_path; +#[cfg(test)] +use std::path::PathBuf; + const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256; const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10); const ANALYTICS_EVENT_DEDUPE_MAX_KEYS: usize = 4096; #[derive(Clone)] pub(crate) struct AnalyticsEventsQueue { - pub(crate) sender: mpsc::Sender, + pub(crate) sender: mpsc::Sender, + pub(crate) local_sink: Option, pub(crate) app_used_emitted_keys: Arc>>, pub(crate) plugin_used_emitted_keys: Arc>>, } @@ -59,24 +71,52 @@ pub struct AnalyticsEventsClient { } impl AnalyticsEventsQueue { - pub(crate) fn new(auth_manager: Arc, base_url: String) -> Self { + pub(crate) fn new( + auth_manager: Arc, + base_url: String, + backend_analytics_enabled: bool, + local_sink: Option, + ) -> Self { let (sender, mut receiver) = mpsc::channel(ANALYTICS_EVENTS_QUEUE_SIZE); + let worker_local_sink = local_sink.clone(); tokio::spawn(async move { let mut reducer = AnalyticsReducer::default(); + let mut local_responses_reducer = LocalResponsesApiCallReducer::default(); while let Some(input) = receiver.recv().await { - let mut events = Vec::new(); - reducer.ingest(input, &mut events).await; - send_track_events(&auth_manager, &base_url, events).await; + match input { + AnalyticsQueueInput::AnalyticsFact(input) => { + let mut events = Vec::new(); + reducer.ingest(input, &mut events).await; + if let Some(local_sink) = worker_local_sink.as_ref() { + append_codex_analytics_events_best_effort(local_sink, &events); + } + if backend_analytics_enabled { + send_track_events(&auth_manager, &base_url, events).await; + } + } + AnalyticsQueueInput::LocalResponsesApiCallStarted(started) => { + local_responses_reducer.ingest_started(started); + } + AnalyticsQueueInput::LocalResponsesApiCallTerminal(terminal) => { + let Some(record) = local_responses_reducer.ingest_terminal(terminal) else { + continue; + }; + if let Some(local_sink) = worker_local_sink.as_ref() { + append_local_analytics_record_best_effort(local_sink, &record); + } + } + } } }); Self { sender, + local_sink, app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), } } - fn try_send(&self, input: AnalyticsFact) { + fn try_send(&self, input: AnalyticsQueueInput) { if self.sender.try_send(input).is_err() { //TODO: add a metric for this tracing::warn!("dropping analytics events: queue is full"); @@ -123,16 +163,67 @@ impl AnalyticsEventsClient { base_url: String, analytics_enabled: Option, ) -> Self { - Self { - queue: (analytics_enabled != Some(false)) - .then(|| AnalyticsEventsQueue::new(Arc::clone(&auth_manager), base_url)), - } + Self::new_with_local_sink( + auth_manager, + base_url, + analytics_enabled, + local_analytics_sink_from_env(), + ) } pub fn disabled() -> Self { Self { queue: None } } + pub fn local_responses_api_call_capture( + &self, + session_id: String, + thread_id: String, + turn_id: String, + ) -> LocalResponsesApiCallCapture { + let Some(queue) = self.queue.as_ref() else { + return LocalResponsesApiCallCapture::disabled(); + }; + if queue.local_sink.is_none() { + return LocalResponsesApiCallCapture::disabled(); + } + LocalResponsesApiCallCapture::enabled(queue.sender.clone(), session_id, thread_id, turn_id) + } + + #[cfg(test)] + pub(crate) fn new_with_local_sink_path( + auth_manager: Arc, + base_url: String, + analytics_enabled: Option, + local_sink_path: Option, + ) -> Self { + Self::new_with_local_sink( + auth_manager, + base_url, + analytics_enabled, + local_sink_path.and_then(local_analytics_sink_for_path), + ) + } + + fn new_with_local_sink( + auth_manager: Arc, + base_url: String, + analytics_enabled: Option, + local_sink: Option, + ) -> Self { + let backend_analytics_enabled = analytics_enabled != Some(false); + Self { + queue: (backend_analytics_enabled || local_sink.is_some()).then(|| { + AnalyticsEventsQueue::new( + auth_manager, + base_url, + backend_analytics_enabled, + local_sink, + ) + }), + } + } + pub fn track_skill_invocations( &self, tracking: TrackEventsContext, @@ -308,7 +399,7 @@ impl AnalyticsEventsClient { pub(crate) fn record_fact(&self, input: AnalyticsFact) { if let Some(queue) = self.queue.as_ref() { - queue.try_send(input); + queue.try_send(AnalyticsQueueInput::AnalyticsFact(input)); } } diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index b7aa9f1c97..c026b61e68 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -1,6 +1,10 @@ use super::AnalyticsEventsClient; use super::AnalyticsEventsQueue; use super::track_event_request_batches; +use crate::LocalAnalyticsRecord; +use crate::LocalAnalyticsRecordType; +use crate::LocalResponsesApiTransport; +use crate::events::AppServerRpcTransport; use crate::events::CodexAcceptedLineFingerprintsEventParams; use crate::events::CodexAcceptedLineFingerprintsEventRequest; use crate::events::SkillInvocationEventParams; @@ -8,10 +12,14 @@ use crate::events::SkillInvocationEventRequest; use crate::events::TrackEventRequest; use crate::facts::AnalyticsFact; use crate::facts::InvocationType; +use crate::local_responses::AnalyticsQueueInput; use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; +use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponsePayload; +use codex_app_server_protocol::InitializeCapabilities; +use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; use codex_app_server_protocol::SessionSource as AppServerSessionSource; @@ -28,14 +36,23 @@ use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; use codex_app_server_protocol::TurnSteerParams; use codex_app_server_protocol::TurnSteerResponse; +use codex_login::AuthManager; +use codex_login::CodexAuth; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; +use serde_json::json; use std::collections::HashSet; +use std::fs; +use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; use tokio::sync::mpsc; use tokio::sync::mpsc::error::TryRecvError; +static NEXT_TEST_PATH_ID: AtomicU64 = AtomicU64::new(0); + fn sample_accepted_line_fingerprint_event(thread_id: &str) -> TrackEventRequest { TrackEventRequest::AcceptedLineFingerprints(Box::new( CodexAcceptedLineFingerprintsEventRequest { @@ -74,10 +91,11 @@ fn sample_regular_track_event(thread_id: &str) -> TrackEventRequest { }) } -fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver) { +fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver) { let (sender, receiver) = mpsc::channel(8); let queue = AnalyticsEventsQueue { sender, + local_sink: None, app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), }; @@ -217,6 +235,74 @@ fn sample_turn_steer_response() -> ClientResponsePayload { }) } +#[tokio::test] +async fn local_sink_reduces_events_when_backend_analytics_are_disabled() { + let path = test_sink_path("backend-disabled"); + let client = AnalyticsEventsClient::new_with_local_sink_path( + AuthManager::from_auth_for_testing(CodexAuth::from_api_key("test")), + "https://example.invalid".to_string(), + /*analytics_enabled*/ Some(false), + Some(path.clone()), + ); + + assert!(client.queue.is_some()); + client.track_initialize( + /*connection_id*/ 7, + InitializeParams { + client_info: ClientInfo { + name: "codex-tui".to_string(), + title: None, + version: "1.0.0".to_string(), + }, + capabilities: Some(InitializeCapabilities { + experimental_api: false, + request_attestation: false, + opt_out_notification_methods: None, + }), + }, + "codex".to_string(), + AppServerRpcTransport::Stdio, + ); + client.track_response( + /*connection_id*/ 7, + RequestId::Integer(1), + sample_thread_start_response(), + ); + + let records = wait_for_local_records(&path, 1).await; + assert_eq!(records.len(), 1); + assert_eq!(records[0].payload["event_type"], "codex_thread_initialized"); + assert_eq!(records[0].thread_id.as_deref(), Some("thread-1")); +} + +#[tokio::test] +async fn local_responses_capture_writes_reduced_terminal_record() { + let path = test_sink_path("local-responses"); + let client = AnalyticsEventsClient::new_with_local_sink_path( + AuthManager::from_auth_for_testing(CodexAuth::from_api_key("test")), + "https://example.invalid".to_string(), + Some(false), + Some(path.clone()), + ); + let capture = client.local_responses_api_call_capture( + "session-1".to_string(), + "thread-1".to_string(), + "turn-1".to_string(), + ); + let attempt = capture.start_attempt(LocalResponsesApiTransport::Http, &json!({"model": "gpt"})); + attempt.record_completed("response-1", Some("request-1"), &None, &[]); + + let records = wait_for_local_records(&path, 1).await; + assert_eq!(records.len(), 1); + assert_eq!( + records[0].record_type, + LocalAnalyticsRecordType::ResponsesApiCall + ); + assert_eq!(records[0].session_id.as_deref(), Some("session-1")); + assert_eq!(records[0].payload["status"], "completed"); + assert_eq!(records[0].payload["request_json"], json!({"model": "gpt"})); +} + #[test] fn track_request_only_enqueues_analytics_relevant_requests() { let (client, mut receiver) = client_with_receiver(); @@ -228,7 +314,9 @@ fn track_request_only_enqueues_analytics_relevant_requests() { client.track_request(/*connection_id*/ 7, request_id, &request); assert!(matches!( receiver.try_recv(), - Ok(AnalyticsFact::ClientRequest { .. }) + Ok(AnalyticsQueueInput::AnalyticsFact( + AnalyticsFact::ClientRequest { .. } + )) )); } @@ -255,7 +343,9 @@ fn track_response_only_enqueues_analytics_relevant_responses() { client.track_response(/*connection_id*/ 7, request_id, response); assert!(matches!( receiver.try_recv(), - Ok(AnalyticsFact::ClientResponse { .. }) + Ok(AnalyticsQueueInput::AnalyticsFact( + AnalyticsFact::ClientResponse { .. } + )) )); } @@ -286,3 +376,29 @@ fn track_event_request_batches_only_isolates_accepted_line_fingerprint_events() assert!(batches[1][0].should_send_in_isolated_request()); assert!(batches[2][0].should_send_in_isolated_request()); } + +async fn wait_for_local_records(path: &PathBuf, min_records: usize) -> Vec { + for _ in 0..100 { + if let Ok(contents) = fs::read_to_string(path) { + let records = contents + .lines() + .map(|line| serde_json::from_str::(line).expect("record")) + .collect::>(); + if records.len() >= min_records { + return records; + } + } + tokio::task::yield_now().await; + } + + panic!("timed out waiting for {min_records} local analytics records"); +} + +fn test_sink_path(label: &str) -> PathBuf { + let id = NEXT_TEST_PATH_ID.fetch_add(1, Ordering::Relaxed); + let process_id = std::process::id(); + let dir = + std::env::temp_dir().join(format!("codex-analytics-client-{process_id}-{label}-{id}")); + fs::create_dir_all(&dir).expect("create temp dir"); + dir.join("events.jsonl") +} diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index e2e16dfacd..87e7c866ae 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -2,6 +2,8 @@ mod accepted_lines; mod client; mod events; mod facts; +mod local_responses; +mod local_sink; mod reducer; use std::time::SystemTime; @@ -48,6 +50,12 @@ pub use facts::TurnSteerRequestError; pub use facts::TurnSteerResult; pub use facts::TurnTokenUsageFact; pub use facts::build_track_events_context; +pub use local_responses::LocalResponsesApiCallAttempt; +pub use local_responses::LocalResponsesApiCallCapture; +pub use local_responses::LocalResponsesApiTransport; +pub use local_sink::LOCAL_ANALYTICS_SCHEMA_VERSION; +pub use local_sink::LocalAnalyticsRecord; +pub use local_sink::LocalAnalyticsRecordType; #[cfg(test)] mod analytics_client_tests; diff --git a/codex-rs/analytics/src/local_responses.rs b/codex-rs/analytics/src/local_responses.rs new file mode 100644 index 0000000000..5f5fe6d2cc --- /dev/null +++ b/codex-rs/analytics/src/local_responses.rs @@ -0,0 +1,425 @@ +use crate::LOCAL_ANALYTICS_SCHEMA_VERSION; +use crate::LocalAnalyticsRecord; +use crate::LocalAnalyticsRecordType; +use crate::facts::AnalyticsFact; +use crate::now_unix_millis; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::TokenUsage; +use serde::Serialize; +use serde_json::Value as JsonValue; +use serde_json::json; +use std::collections::HashMap; +use std::fmt::Display; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; +use tokio::sync::mpsc; + +static NEXT_LOCAL_RESPONSES_CALL_ID: AtomicU64 = AtomicU64::new(0); + +pub(crate) enum AnalyticsQueueInput { + AnalyticsFact(AnalyticsFact), + LocalResponsesApiCallStarted(LocalResponsesApiCallStartedFact), + LocalResponsesApiCallTerminal(LocalResponsesApiCallTerminalFact), +} + +/// Transport used for one locally captured Responses API attempt. +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum LocalResponsesApiTransport { + Http, + Websocket, +} + +/// No-op-capable turn-local factory for locally captured Responses API attempts. +#[derive(Clone)] +pub struct LocalResponsesApiCallCapture { + sender: Option>, + session_id: String, + thread_id: String, + turn_id: String, +} + +/// One locally captured Responses API attempt. +pub struct LocalResponsesApiCallAttempt { + sender: Option>, + responses_call_id: String, + terminal_recorded: AtomicBool, +} + +#[derive(Debug)] +pub(crate) struct LocalResponsesApiCallStartedFact { + responses_call_id: String, + session_id: String, + thread_id: String, + turn_id: String, + transport: LocalResponsesApiTransport, + request_started_at_epoch_millis: u64, + request_json: JsonValue, +} + +#[derive(Debug)] +pub(crate) enum LocalResponsesApiCallTerminalFact { + Completed { + responses_call_id: String, + completed_at_epoch_millis: u64, + response_id: String, + upstream_request_id: Option, + token_usage_json: Option, + output_items: Vec, + }, + Failed { + responses_call_id: String, + completed_at_epoch_millis: u64, + upstream_request_id: Option, + error_json: JsonValue, + output_items: Vec, + }, + Cancelled { + responses_call_id: String, + completed_at_epoch_millis: u64, + upstream_request_id: Option, + error_json: JsonValue, + output_items: Vec, + }, +} + +#[derive(Default)] +pub(crate) struct LocalResponsesApiCallReducer { + started_calls: HashMap, +} + +#[derive(Serialize)] +struct LocalResponsesApiCallPayload { + responses_call_id: String, + transport: LocalResponsesApiTransport, + status: LocalResponsesApiCallStatus, + request_started_at_epoch_millis: u64, + completed_at_epoch_millis: u64, + response_id: Option, + upstream_request_id: Option, + request_json: JsonValue, + response_json: Option, + token_usage_json: Option, + error_json: Option, +} + +#[derive(Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +enum LocalResponsesApiCallStatus { + Completed, + Failed, + Cancelled, +} + +impl LocalResponsesApiCallCapture { + /// Builds a capture handle that accepts calls and records nothing. + pub fn disabled() -> Self { + Self { + sender: None, + session_id: String::new(), + thread_id: String::new(), + turn_id: String::new(), + } + } + + pub(crate) fn enabled( + sender: mpsc::Sender, + session_id: String, + thread_id: String, + turn_id: String, + ) -> Self { + Self { + sender: Some(sender), + session_id, + thread_id, + turn_id, + } + } + + /// Starts one locally captured Responses API attempt. + pub fn start_attempt( + &self, + transport: LocalResponsesApiTransport, + request: &impl Serialize, + ) -> LocalResponsesApiCallAttempt { + let Some(sender) = self.sender.as_ref() else { + return LocalResponsesApiCallAttempt::disabled(); + }; + let Some(request_json) = serialize_json_best_effort(request, "request") else { + return LocalResponsesApiCallAttempt::disabled(); + }; + let responses_call_id = next_local_responses_call_id(); + try_send_best_effort( + sender, + AnalyticsQueueInput::LocalResponsesApiCallStarted(LocalResponsesApiCallStartedFact { + responses_call_id: responses_call_id.clone(), + session_id: self.session_id.clone(), + thread_id: self.thread_id.clone(), + turn_id: self.turn_id.clone(), + transport, + request_started_at_epoch_millis: now_unix_millis(), + request_json, + }), + ); + LocalResponsesApiCallAttempt { + sender: Some(sender.clone()), + responses_call_id, + terminal_recorded: AtomicBool::new(false), + } + } +} + +impl LocalResponsesApiCallAttempt { + /// Builds an attempt that accepts terminal calls and records nothing. + pub fn disabled() -> Self { + Self { + sender: None, + responses_call_id: String::new(), + terminal_recorded: AtomicBool::new(false), + } + } + + /// Records successful provider completion and observed non-delta output items. + pub fn record_completed( + &self, + response_id: &str, + upstream_request_id: Option<&str>, + token_usage: &Option, + output_items: &[ResponseItem], + ) { + self.record_terminal(LocalResponsesApiCallTerminalFact::Completed { + responses_call_id: self.responses_call_id.clone(), + completed_at_epoch_millis: now_unix_millis(), + response_id: response_id.to_string(), + upstream_request_id: upstream_request_id.map(str::to_string), + token_usage_json: token_usage + .as_ref() + .and_then(|usage| serialize_json_best_effort(usage, "token usage")), + output_items: serialize_response_items(output_items), + }); + } + + /// Records a failed provider attempt and any observed non-delta output items. + pub fn record_failed( + &self, + error: impl Display, + upstream_request_id: Option<&str>, + output_items: &[ResponseItem], + ) { + self.record_terminal(LocalResponsesApiCallTerminalFact::Failed { + responses_call_id: self.responses_call_id.clone(), + completed_at_epoch_millis: now_unix_millis(), + upstream_request_id: upstream_request_id.map(str::to_string), + error_json: json!({ "message": error.to_string() }), + output_items: serialize_response_items(output_items), + }); + } + + /// Records a provider stream that Codex intentionally stopped consuming. + pub fn record_cancelled( + &self, + reason: impl Display, + upstream_request_id: Option<&str>, + output_items: &[ResponseItem], + ) { + self.record_terminal(LocalResponsesApiCallTerminalFact::Cancelled { + responses_call_id: self.responses_call_id.clone(), + completed_at_epoch_millis: now_unix_millis(), + upstream_request_id: upstream_request_id.map(str::to_string), + error_json: json!({ "reason": reason.to_string() }), + output_items: serialize_response_items(output_items), + }); + } + + fn record_terminal(&self, terminal: LocalResponsesApiCallTerminalFact) { + let Some(sender) = self.sender.as_ref() else { + return; + }; + if self.terminal_recorded.swap(true, Ordering::AcqRel) { + return; + } + try_send_best_effort( + sender, + AnalyticsQueueInput::LocalResponsesApiCallTerminal(terminal), + ); + } +} + +impl LocalResponsesApiCallReducer { + pub(crate) fn ingest_started(&mut self, started: LocalResponsesApiCallStartedFact) { + self.started_calls + .insert(started.responses_call_id.clone(), started); + } + + pub(crate) fn ingest_terminal( + &mut self, + terminal: LocalResponsesApiCallTerminalFact, + ) -> Option { + let responses_call_id = terminal.responses_call_id(); + let Some(started) = self.started_calls.remove(responses_call_id) else { + tracing::warn!( + responses_call_id, + "dropping local Responses terminal without matching start" + ); + return None; + }; + Some(terminal.into_record(started)) + } +} + +impl LocalResponsesApiCallTerminalFact { + fn responses_call_id(&self) -> &str { + match self { + Self::Completed { + responses_call_id, .. + } + | Self::Failed { + responses_call_id, .. + } + | Self::Cancelled { + responses_call_id, .. + } => responses_call_id, + } + } + + fn into_record(self, started: LocalResponsesApiCallStartedFact) -> LocalAnalyticsRecord { + let ( + status, + completed_at_epoch_millis, + response_id, + upstream_request_id, + response_json, + token_usage_json, + error_json, + ) = match self { + Self::Completed { + completed_at_epoch_millis, + response_id, + upstream_request_id, + token_usage_json, + output_items, + .. + } => ( + LocalResponsesApiCallStatus::Completed, + completed_at_epoch_millis, + Some(response_id.clone()), + upstream_request_id.clone(), + Some(response_summary( + Some(response_id), + upstream_request_id, + output_items, + )), + token_usage_json, + None, + ), + Self::Failed { + completed_at_epoch_millis, + upstream_request_id, + error_json, + output_items, + .. + } => ( + LocalResponsesApiCallStatus::Failed, + completed_at_epoch_millis, + None, + upstream_request_id.clone(), + partial_response_summary(upstream_request_id, output_items), + None, + Some(error_json), + ), + Self::Cancelled { + completed_at_epoch_millis, + upstream_request_id, + error_json, + output_items, + .. + } => ( + LocalResponsesApiCallStatus::Cancelled, + completed_at_epoch_millis, + None, + upstream_request_id.clone(), + partial_response_summary(upstream_request_id, output_items), + None, + Some(error_json), + ), + }; + let payload = LocalResponsesApiCallPayload { + responses_call_id: started.responses_call_id, + transport: started.transport, + status, + request_started_at_epoch_millis: started.request_started_at_epoch_millis, + completed_at_epoch_millis, + response_id, + upstream_request_id, + request_json: started.request_json, + response_json, + token_usage_json, + error_json, + }; + LocalAnalyticsRecord { + schema_version: LOCAL_ANALYTICS_SCHEMA_VERSION, + recorded_at_epoch_millis: completed_at_epoch_millis, + record_type: LocalAnalyticsRecordType::ResponsesApiCall, + session_id: Some(started.session_id), + thread_id: Some(started.thread_id), + turn_id: Some(started.turn_id), + payload: serde_json::to_value(payload).unwrap_or_else(|err| { + json!({ + "serialization_error": err.to_string(), + }) + }), + } + } +} + +fn response_summary( + response_id: Option, + upstream_request_id: Option, + output_items: Vec, +) -> JsonValue { + json!({ + "response_id": response_id, + "upstream_request_id": upstream_request_id, + "output_items": output_items, + }) +} + +fn partial_response_summary( + upstream_request_id: Option, + output_items: Vec, +) -> Option { + (!output_items.is_empty()).then(|| response_summary(None, upstream_request_id, output_items)) +} + +fn serialize_response_items(output_items: &[ResponseItem]) -> Vec { + output_items + .iter() + .filter_map(|item| serialize_json_best_effort(item, "response item")) + .collect() +} + +fn serialize_json_best_effort(value: &impl Serialize, kind: &str) -> Option { + match serde_json::to_value(value) { + Ok(value) => Some(value), + Err(err) => { + tracing::warn!(error = %err, "failed to serialize local Responses {kind}"); + None + } + } +} + +fn try_send_best_effort(sender: &mpsc::Sender, input: AnalyticsQueueInput) { + if sender.try_send(input).is_err() { + tracing::warn!("dropping local Responses analytics input: queue is full"); + } +} + +fn next_local_responses_call_id() -> String { + let sequence = NEXT_LOCAL_RESPONSES_CALL_ID.fetch_add(1, Ordering::Relaxed); + format!("local-responses-{}-{sequence}", now_unix_millis()) +} + +#[cfg(test)] +#[path = "local_responses_tests.rs"] +mod tests; diff --git a/codex-rs/analytics/src/local_responses_tests.rs b/codex-rs/analytics/src/local_responses_tests.rs new file mode 100644 index 0000000000..5901cca648 --- /dev/null +++ b/codex-rs/analytics/src/local_responses_tests.rs @@ -0,0 +1,58 @@ +use super::LocalResponsesApiCallReducer; +use super::LocalResponsesApiCallStartedFact; +use super::LocalResponsesApiCallTerminalFact; +use super::LocalResponsesApiTransport; +use crate::LocalAnalyticsRecordType; +use pretty_assertions::assert_eq; +use serde_json::json; + +#[test] +fn reducer_emits_one_responses_record_after_terminal_fact() { + let mut reducer = LocalResponsesApiCallReducer::default(); + reducer.ingest_started(LocalResponsesApiCallStartedFact { + responses_call_id: "call-1".to_string(), + session_id: "session-1".to_string(), + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + transport: LocalResponsesApiTransport::Http, + request_started_at_epoch_millis: 10, + request_json: json!({"model": "gpt-test"}), + }); + + let record = reducer + .ingest_terminal(LocalResponsesApiCallTerminalFact::Completed { + responses_call_id: "call-1".to_string(), + completed_at_epoch_millis: 20, + response_id: "response-1".to_string(), + upstream_request_id: Some("request-1".to_string()), + token_usage_json: Some(json!({"total_tokens": 3})), + output_items: vec![json!({"type": "message"})], + }) + .expect("terminal fact should reduce"); + + assert_eq!( + record.record_type, + LocalAnalyticsRecordType::ResponsesApiCall + ); + assert_eq!(record.session_id.as_deref(), Some("session-1")); + assert_eq!( + record.payload, + json!({ + "responses_call_id": "call-1", + "transport": "http", + "status": "completed", + "request_started_at_epoch_millis": 10, + "completed_at_epoch_millis": 20, + "response_id": "response-1", + "upstream_request_id": "request-1", + "request_json": {"model": "gpt-test"}, + "response_json": { + "response_id": "response-1", + "upstream_request_id": "request-1", + "output_items": [{"type": "message"}] + }, + "token_usage_json": {"total_tokens": 3}, + "error_json": null + }) + ); +} diff --git a/codex-rs/analytics/src/local_sink.rs b/codex-rs/analytics/src/local_sink.rs new file mode 100644 index 0000000000..5ab324d925 --- /dev/null +++ b/codex-rs/analytics/src/local_sink.rs @@ -0,0 +1,155 @@ +use crate::events::TrackEventRequest; +use crate::now_unix_millis; +use serde::Deserialize; +use serde::Serialize; +use serde_json::Value as JsonValue; +use std::collections::HashMap; +use std::fs::File; +use std::fs::OpenOptions; +use std::io::BufWriter; +use std::io::Write; +use std::path::Path; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::OnceLock; +use std::sync::Weak; + +pub const LOCAL_ANALYTICS_SCHEMA_VERSION: u32 = 1; +pub const LOCAL_ANALYTICS_SINK_PATH_ENV_VAR: &str = "CODEX_ANALYTICS_LOCAL_SINK_PATH"; + +pub(crate) type SharedLocalAnalyticsSink = Arc>; + +static PROCESS_LOCAL_SINKS: OnceLock>>>> = + OnceLock::new(); + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum LocalAnalyticsRecordType { + CodexAnalyticsEvent, + ResponsesApiCall, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +pub struct LocalAnalyticsRecord { + pub schema_version: u32, + pub recorded_at_epoch_millis: u64, + pub record_type: LocalAnalyticsRecordType, + pub session_id: Option, + pub thread_id: Option, + pub turn_id: Option, + pub payload: JsonValue, +} + +pub(crate) fn local_analytics_sink_from_env() -> Option { + let path = std::env::var_os(LOCAL_ANALYTICS_SINK_PATH_ENV_VAR)?; + if path.is_empty() { + return None; + } + + local_analytics_sink_for_path(PathBuf::from(path)) +} + +pub(crate) fn append_codex_analytics_events_best_effort( + sink: &SharedLocalAnalyticsSink, + events: &[TrackEventRequest], +) { + for event in events { + let Some(record) = LocalAnalyticsRecord::from_codex_analytics_event(event) else { + continue; + }; + append_local_analytics_record_best_effort(sink, &record); + } +} + +pub(crate) fn append_local_analytics_record_best_effort( + sink: &SharedLocalAnalyticsSink, + record: &LocalAnalyticsRecord, +) { + let result = sink + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .append(record); + if let Err(err) = result { + tracing::warn!(error = %err, "failed to append local analytics record"); + } +} + +pub(crate) fn local_analytics_sink_for_path(path: PathBuf) -> Option { + let sinks = PROCESS_LOCAL_SINKS.get_or_init(|| Mutex::new(HashMap::new())); + let mut sinks = sinks + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + sinks.retain(|_, sink| sink.strong_count() > 0); + if let Some(sink) = sinks.get(&path).and_then(Weak::upgrade) { + return Some(sink); + } + + match LocalAnalyticsSink::open(path.clone()) { + Ok(sink) => { + let sink = Arc::new(Mutex::new(sink)); + sinks.insert(path, Arc::downgrade(&sink)); + Some(sink) + } + Err(err) => { + tracing::warn!( + path = %path.display(), + error = %err, + "failed to initialize local analytics sink" + ); + None + } + } +} + +impl LocalAnalyticsRecord { + fn from_codex_analytics_event(event: &TrackEventRequest) -> Option { + let payload = match serde_json::to_value(event) { + Ok(payload) => payload, + Err(err) => { + tracing::warn!(error = %err, "failed to serialize local analytics event"); + return None; + } + }; + let event_params = payload.get("event_params"); + Some(Self { + schema_version: LOCAL_ANALYTICS_SCHEMA_VERSION, + recorded_at_epoch_millis: now_unix_millis(), + record_type: LocalAnalyticsRecordType::CodexAnalyticsEvent, + session_id: string_field(event_params, "session_id"), + thread_id: string_field(event_params, "thread_id"), + turn_id: string_field(event_params, "turn_id"), + payload, + }) + } +} + +pub(crate) struct LocalAnalyticsSink { + writer: BufWriter, +} + +impl LocalAnalyticsSink { + fn open(path: impl AsRef) -> std::io::Result { + let file = OpenOptions::new().create(true).append(true).open(path)?; + Ok(Self { + writer: BufWriter::new(file), + }) + } + + fn append(&mut self, record: &LocalAnalyticsRecord) -> std::io::Result<()> { + serde_json::to_writer(&mut self.writer, record)?; + self.writer.write_all(b"\n")?; + self.writer.flush() + } +} + +fn string_field(value: Option<&JsonValue>, field: &str) -> Option { + value? + .get(field)? + .as_str() + .map(std::string::ToString::to_string) +} + +#[cfg(test)] +#[path = "local_sink_tests.rs"] +mod tests; diff --git a/codex-rs/analytics/src/local_sink_tests.rs b/codex-rs/analytics/src/local_sink_tests.rs new file mode 100644 index 0000000000..dd05a8fafe --- /dev/null +++ b/codex-rs/analytics/src/local_sink_tests.rs @@ -0,0 +1,115 @@ +use super::LOCAL_ANALYTICS_SCHEMA_VERSION; +use super::LocalAnalyticsRecord; +use super::LocalAnalyticsRecordType; +use super::append_local_analytics_record_best_effort; +use super::local_analytics_sink_for_path; +use crate::events::SkillInvocationEventParams; +use crate::events::SkillInvocationEventRequest; +use crate::events::TrackEventRequest; +use crate::facts::InvocationType; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::fs; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; + +static NEXT_TEST_PATH_ID: AtomicU64 = AtomicU64::new(0); + +#[test] +fn codex_analytics_record_extracts_generic_envelope_metadata() { + let record = LocalAnalyticsRecord::from_codex_analytics_event(&sample_track_event()) + .expect("serialize local analytics event"); + + assert_eq!( + record, + LocalAnalyticsRecord { + schema_version: LOCAL_ANALYTICS_SCHEMA_VERSION, + recorded_at_epoch_millis: record.recorded_at_epoch_millis, + record_type: LocalAnalyticsRecordType::CodexAnalyticsEvent, + session_id: None, + thread_id: Some("thread-1".to_string()), + turn_id: Some("turn-1".to_string()), + payload: json!({ + "event_type": "skill_invocation", + "skill_id": "skill-1", + "skill_name": "doc", + "event_params": { + "product_client_id": null, + "skill_scope": null, + "plugin_id": null, + "repo_url": null, + "thread_id": "thread-1", + "turn_id": "turn-1", + "invoke_type": "explicit", + "model_slug": "gpt-5.1-codex" + } + }), + } + ); +} + +#[test] +fn process_global_sink_reuses_writer_for_same_path() { + let path = test_sink_path("shared"); + let first = local_analytics_sink_for_path(path.clone()).expect("first sink"); + let second = local_analytics_sink_for_path(path).expect("second sink"); + + assert!(Arc::ptr_eq(&first, &second)); +} + +#[test] +fn sink_appends_complete_jsonl_records() { + let path = test_sink_path("records"); + let sink = local_analytics_sink_for_path(path.clone()).expect("sink"); + let first = LocalAnalyticsRecord::from_codex_analytics_event(&sample_track_event()) + .expect("first record"); + let mut second = first.clone(); + second.turn_id = Some("turn-2".to_string()); + + append_local_analytics_record_best_effort(&sink, &first); + append_local_analytics_record_best_effort(&sink, &second); + + let contents = fs::read_to_string(path).expect("read sink"); + let records = contents + .lines() + .map(|line| serde_json::from_str::(line).expect("record")) + .collect::>(); + assert_eq!(records, vec![first, second]); +} + +#[test] +fn sink_initialization_failure_is_best_effort() { + let path = test_sink_path("missing-parent") + .join("missing") + .join("events.jsonl"); + + assert!(local_analytics_sink_for_path(path).is_none()); +} + +fn sample_track_event() -> TrackEventRequest { + TrackEventRequest::SkillInvocation(SkillInvocationEventRequest { + event_type: "skill_invocation", + skill_id: "skill-1".to_string(), + skill_name: "doc".to_string(), + event_params: SkillInvocationEventParams { + product_client_id: None, + skill_scope: None, + plugin_id: None, + repo_url: None, + thread_id: Some("thread-1".to_string()), + turn_id: Some("turn-1".to_string()), + invoke_type: Some(InvocationType::Explicit), + model_slug: Some("gpt-5.1-codex".to_string()), + }, + }) +} + +fn test_sink_path(label: &str) -> PathBuf { + let id = NEXT_TEST_PATH_ID.fetch_add(1, Ordering::Relaxed); + let process_id = std::process::id(); + let dir = std::env::temp_dir().join(format!("codex-analytics-{process_id}-{label}-{id}")); + fs::create_dir_all(&dir).expect("create temp dir"); + dir.join("events.jsonl") +} diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index dd7b9f400c..7d6337b7f1 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -31,6 +31,9 @@ use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicU64; use std::sync::atomic::Ordering; +use codex_analytics::LocalResponsesApiCallAttempt; +use codex_analytics::LocalResponsesApiCallCapture; +use codex_analytics::LocalResponsesApiTransport; use codex_api::ApiError; use codex_api::AuthProvider; use codex_api::CompactClient as ApiCompactClient; @@ -1259,6 +1262,7 @@ impl ModelClientSession { service_tier: Option, turn_metadata_header: Option<&str>, inference_trace: &InferenceTraceContext, + local_responses_api_call_capture: &LocalResponsesApiCallCapture, ) -> Result { let auth_manager = self.client.state.provider.auth_manager(); let mut auth_recovery = auth_manager @@ -1298,7 +1302,12 @@ impl ModelClientSession { )?; let inference_trace_attempt = inference_trace.start_attempt(); inference_trace_attempt.add_request_headers(&mut options.extra_headers); - inference_trace_attempt.record_started(&request); + let response_stream_attempts = ResponseStreamAttempts::started( + inference_trace_attempt, + local_responses_api_call_capture + .start_attempt(LocalResponsesApiTransport::Http, &request), + &request, + ); let client = ApiResponsesClient::new( transport, client_setup.api_provider, @@ -1312,7 +1321,7 @@ impl ModelClientSession { let (stream, _) = map_response_stream( stream, session_telemetry.clone(), - inference_trace_attempt, + response_stream_attempts, ); return Ok(stream); } @@ -1321,7 +1330,7 @@ impl ModelClientSession { )) if status == StatusCode::UNAUTHORIZED => { let response_debug_context = extract_response_debug_context(&unauthorized_transport); - inference_trace_attempt.record_failed( + response_stream_attempts.record_failed( &unauthorized_transport, response_debug_context.request_id.as_deref(), /*output_items*/ &[], @@ -1340,7 +1349,7 @@ impl ModelClientSession { let response_debug_context = extract_response_debug_context_from_api_error(&err); let err = map_api_error(err); - inference_trace_attempt.record_failed( + response_stream_attempts.record_failed( &err, response_debug_context.request_id.as_deref(), /*output_items*/ &[], @@ -1378,6 +1387,7 @@ impl ModelClientSession { warmup: bool, request_trace: Option, inference_trace: &InferenceTraceContext, + local_responses_api_call_capture: &LocalResponsesApiCallCapture, ) -> Result { let auth_manager = self.client.state.provider.auth_manager(); @@ -1469,14 +1479,24 @@ impl ModelClientSession { inference_trace.start_attempt() }; stamp_ws_stream_request_start_ms(&mut ws_request); - if previous_response_id_from_untraced_warmup { + let response_stream_attempts = if previous_response_id_from_untraced_warmup { // The transport can reuse an untraced warmup response id and omit the // already-sent input, but rollout replay needs the logical model-visible // request rather than the compressed websocket delta. - inference_trace_attempt.record_started(&request); + ResponseStreamAttempts::started( + inference_trace_attempt, + local_responses_api_call_capture + .start_attempt(LocalResponsesApiTransport::Websocket, &request), + &request, + ) } else { - inference_trace_attempt.record_started(&ws_request); - } + ResponseStreamAttempts::started( + inference_trace_attempt, + local_responses_api_call_capture + .start_attempt(LocalResponsesApiTransport::Websocket, &ws_request), + &ws_request, + ) + }; self.websocket_session.last_request = Some(request); self.websocket_session.last_response_from_untraced_warmup = warmup; let websocket_connection = @@ -1492,7 +1512,7 @@ impl ModelClientSession { let response_debug_context = extract_response_debug_context_from_api_error(&err); let err = map_api_error(err); - inference_trace_attempt.record_failed( + response_stream_attempts.record_failed( &err, response_debug_context.request_id.as_deref(), /*output_items*/ &[], @@ -1502,7 +1522,7 @@ impl ModelClientSession { let (stream, last_request_rx) = map_response_stream( stream_result, session_telemetry.clone(), - inference_trace_attempt, + response_stream_attempts, ); self.websocket_session.last_response_rx = Some(last_request_rx); return Ok(WebsocketStreamOutcome::Stream(stream)); @@ -1563,6 +1583,7 @@ impl ModelClientSession { } let disabled_trace = InferenceTraceContext::disabled(); + let disabled_local_responses_capture = LocalResponsesApiCallCapture::disabled(); match self .stream_responses_websocket( prompt, @@ -1575,6 +1596,7 @@ impl ModelClientSession { /*warmup*/ true, current_span_w3c_trace_context(), &disabled_trace, + &disabled_local_responses_capture, ) .await { @@ -1616,6 +1638,7 @@ impl ModelClientSession { service_tier: Option, turn_metadata_header: Option<&str>, inference_trace: &InferenceTraceContext, + local_responses_api_call_capture: &LocalResponsesApiCallCapture, ) -> Result { let wire_api = self.client.state.provider.info().wire_api; match wire_api { @@ -1634,6 +1657,7 @@ impl ModelClientSession { /*warmup*/ false, request_trace, inference_trace, + local_responses_api_call_capture, ) .await? { @@ -1653,6 +1677,7 @@ impl ModelClientSession { service_tier, turn_metadata_header, inference_trace, + local_responses_api_call_capture, ) .await } @@ -1769,10 +1794,84 @@ fn parent_thread_id_header_value(parent_thread_id: Option) -> Option Self { + inference_trace.record_started(request); + Self { + inference_trace, + local_responses_api_call, + } + } + + #[cfg(test)] + fn with_disabled_local_capture(inference_trace: InferenceTraceAttempt) -> Self { + Self { + inference_trace, + local_responses_api_call: LocalResponsesApiCallAttempt::disabled(), + } + } + + fn record_completed( + &self, + response_id: &str, + upstream_request_id: Option<&str>, + token_usage: &Option, + output_items: &[ResponseItem], + ) { + self.inference_trace.record_completed( + response_id, + upstream_request_id, + token_usage, + output_items, + ); + self.local_responses_api_call.record_completed( + response_id, + upstream_request_id, + token_usage, + output_items, + ); + } + + fn record_failed( + &self, + error: impl std::fmt::Display, + upstream_request_id: Option<&str>, + output_items: &[ResponseItem], + ) { + let error = error.to_string(); + self.inference_trace + .record_failed(&error, upstream_request_id, output_items); + self.local_responses_api_call + .record_failed(&error, upstream_request_id, output_items); + } + + fn record_cancelled( + &self, + reason: impl std::fmt::Display, + upstream_request_id: Option<&str>, + output_items: &[ResponseItem], + ) { + let reason = reason.to_string(); + self.inference_trace + .record_cancelled(&reason, upstream_request_id, output_items); + self.local_responses_api_call + .record_cancelled(&reason, upstream_request_id, output_items); + } +} + fn map_response_stream( api_stream: codex_api::ResponseStream, session_telemetry: SessionTelemetry, - inference_trace_attempt: InferenceTraceAttempt, + response_stream_attempts: ResponseStreamAttempts, ) -> (ResponseStream, oneshot::Receiver) { let codex_api::ResponseStream { rx_event, @@ -1786,7 +1885,7 @@ fn map_response_stream( upstream_request_id, api_stream, session_telemetry, - inference_trace_attempt, + response_stream_attempts, ) } @@ -1794,7 +1893,7 @@ fn map_response_events( upstream_request_id: Option, api_stream: S, session_telemetry: SessionTelemetry, - inference_trace_attempt: InferenceTraceAttempt, + response_stream_attempts: ResponseStreamAttempts, ) -> (ResponseStream, oneshot::Receiver) where S: futures::Stream> @@ -1820,7 +1919,7 @@ where loop { let event = tokio::select! { _ = consumer_dropped.cancelled() => { - inference_trace_attempt.record_cancelled( + response_stream_attempts.record_cancelled( STREAM_DROPPED_REASON, upstream_request_id, &items_added, @@ -1840,7 +1939,7 @@ where .await .is_err() { - inference_trace_attempt.record_cancelled( + response_stream_attempts.record_cancelled( STREAM_DROPPED_REASON, upstream_request_id, &items_added, @@ -1863,7 +1962,7 @@ where usage.total_tokens, ); } - inference_trace_attempt.record_completed( + response_stream_attempts.record_completed( &response_id, upstream_request_id, &token_usage, @@ -1889,7 +1988,7 @@ where } Ok(event) => { if tx_event.send(Ok(event)).await.is_err() { - inference_trace_attempt.record_cancelled( + response_stream_attempts.record_cancelled( STREAM_DROPPED_REASON, upstream_request_id, &items_added, @@ -1906,7 +2005,7 @@ where feedback_tags!(last_model_request_id = upstream_request_id); } let mapped = map_api_error(err); - inference_trace_attempt.record_failed( + response_stream_attempts.record_failed( &mapped, upstream_request_id, &items_added, @@ -1921,7 +2020,7 @@ where } } } - inference_trace_attempt.record_failed( + response_stream_attempts.record_failed( "stream closed before response.completed", upstream_request_id, &items_added, diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index 2ebffe3922..5c5137314a 100644 --- a/codex-rs/core/src/client_tests.rs +++ b/codex-rs/core/src/client_tests.rs @@ -359,7 +359,7 @@ async fn dropped_response_stream_traces_cancelled_partial_output() -> anyhow::Re /*upstream_request_id*/ None, api_stream, test_session_telemetry(), - attempt, + super::ResponseStreamAttempts::with_disabled_local_capture(attempt), ); let observed = stream @@ -408,7 +408,9 @@ async fn response_stream_records_last_model_feedback_ids() { Some("req-123".to_string()), api_stream, test_session_telemetry(), - InferenceTraceAttempt::disabled(), + super::ResponseStreamAttempts::with_disabled_local_capture( + InferenceTraceAttempt::disabled(), + ), ); while stream.next().await.is_some() {} @@ -449,7 +451,7 @@ async fn dropped_backpressured_response_stream_traces_cancelled_partial_output() /*upstream_request_id*/ None, api_stream, test_session_telemetry(), - attempt, + super::ResponseStreamAttempts::with_disabled_local_capture(attempt), ); // Fill the mapper channel with non-terminal events, then yield one output diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 18b5838292..2b7e541178 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -22,6 +22,7 @@ use codex_analytics::CompactionReason; use codex_analytics::CompactionStatus; use codex_analytics::CompactionStrategy; use codex_analytics::CompactionTrigger; +use codex_analytics::LocalResponsesApiCallCapture; use codex_analytics::now_unix_seconds; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; @@ -580,6 +581,7 @@ async fn drain_to_completed( // Rollout tracing currently models remote compaction only; local compaction streams // are left untraced until the reducer has a first-class local compaction lifecycle. &InferenceTraceContext::disabled(), + &LocalResponsesApiCallCapture::disabled(), ) .await?; loop { diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 5f585862dd..fb06c6917f 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -26,6 +26,7 @@ use codex_analytics::CompactionImplementation; use codex_analytics::CompactionPhase; use codex_analytics::CompactionReason; use codex_analytics::CompactionTrigger; +use codex_analytics::LocalResponsesApiCallCapture; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; use codex_protocol::items::ContextCompactionItem; @@ -344,6 +345,7 @@ async fn run_remote_compaction_request_v2( turn_context.config.service_tier.clone(), turn_metadata_header, &InferenceTraceContext::disabled(), + &LocalResponsesApiCallCapture::disabled(), ) .await { diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 9d4e7d1e82..adba06bc82 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -1771,6 +1771,14 @@ async fn try_run_sampling_request( turn_context.model_info.slug.as_str(), turn_context.provider.info().name.as_str(), ); + let local_responses_api_call_capture = sess + .services + .analytics_events_client + .local_responses_api_call_capture( + sess.session_id().to_string(), + sess.thread_id().to_string(), + turn_context.sub_id.clone(), + ); let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling(); let mut stream = client_session .stream( @@ -1782,6 +1790,7 @@ async fn try_run_sampling_request( turn_context.config.service_tier.clone(), turn_metadata_header, &inference_trace, + &local_responses_api_call_capture, ) .instrument(trace_span!("stream_request")) .or_cancel(&cancellation_token) diff --git a/codex-rs/core/tests/responses_headers.rs b/codex-rs/core/tests/responses_headers.rs index 75649fa4bc..aaaee3db82 100644 --- a/codex-rs/core/tests/responses_headers.rs +++ b/codex-rs/core/tests/responses_headers.rs @@ -134,6 +134,7 @@ async fn responses_stream_includes_subagent_header_on_review() { /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("stream failed"); @@ -263,6 +264,7 @@ async fn responses_stream_includes_subagent_header_on_other() { /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("stream failed"); @@ -381,6 +383,7 @@ async fn responses_respects_model_info_overrides_from_config() { /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("stream failed"); diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 2fcb73e25a..2a6f5824e1 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -933,6 +933,7 @@ async fn send_provider_auth_request(server: &MockServer, auth: ModelProviderAuth /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("responses stream to start"); @@ -2499,6 +2500,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() { /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("responses stream to start"); diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index 3e1ed23261..12d6078f37 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -449,6 +449,7 @@ async fn responses_websocket_request_prewarm_traces_logical_request() { /*service_tier*/ None, /*turn_metadata_header*/ None, &inference_trace, + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("websocket stream failed"); @@ -620,6 +621,7 @@ async fn responses_websocket_preconnect_is_reused_even_with_header_changes() { /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("websocket stream failed"); @@ -671,6 +673,7 @@ async fn responses_websocket_request_prewarm_is_reused_even_with_header_changes( /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("websocket stream failed"); @@ -1074,6 +1077,7 @@ async fn responses_websocket_emits_reasoning_included_event() { /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("websocket stream failed"); @@ -1148,6 +1152,7 @@ async fn responses_websocket_emits_rate_limit_events() { /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("websocket stream failed"); @@ -1804,6 +1809,7 @@ async fn responses_websocket_v2_after_error_uses_full_create_without_previous_re /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("websocket stream failed"); @@ -1892,6 +1898,7 @@ async fn responses_websocket_v2_surfaces_terminal_error_without_close_handshake( /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("websocket stream failed"); @@ -2126,6 +2133,7 @@ async fn stream_until_complete_with_model_info( /*service_tier*/ None, /*turn_metadata_header*/ None, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("websocket stream failed"); @@ -2193,6 +2201,7 @@ async fn stream_until_complete_with_request_metadata( service_tier.map(|service_tier| service_tier.request_value().to_string()), turn_metadata_header, &codex_rollout_trace::InferenceTraceContext::disabled(), + &codex_analytics::LocalResponsesApiCallCapture::disabled(), ) .await .expect("websocket stream failed"); diff --git a/codex-rs/local_analytics_plan.md b/codex-rs/local_analytics_plan.md index 12b4653ab3..1488cf4a52 100644 --- a/codex-rs/local_analytics_plan.md +++ b/codex-rs/local_analytics_plan.md @@ -18,10 +18,10 @@ The helper reduces the JSONL sink into a DuckDB artifact containing the first se - `viewer_threads_v1` - `viewer_turns_v1` - `viewer_turn_events_v1` -- `viewer_responses_calls_v1` as an empty schema-only stub until the Responses follow-up lands +- `viewer_responses_calls_v1` - `viewer_context_windows_v1` as an empty schema-only stub -Responses API call capture remains part of the intended design, but it is a follow-up slice after the reduced analytics JSONL sink and standalone DuckDB materializer work end to end. This pass does not change the React/FastAPI session viewer, does not upload artifacts to blob store, does not synthesize `chat.Conversation` snapshots, and does not add DuckDB to the main `codex` binary. +This pass also captures local-only Responses API attempt summaries needed to populate `viewer_responses_calls_v1`. It does not change the React/FastAPI session viewer, upload artifacts to blob store, synthesize `chat.Conversation` snapshots, or add DuckDB to the main `codex` binary. ## Motivation @@ -31,7 +31,7 @@ This proposal adds a local-only analytics sink that mirrors the intended product We intentionally use `codex-analytics` as the entry point because it already owns the normalized Codex thread, turn, tool, compaction, steering, app, plugin, hook, and review events needed for the first viewer tables. We are not building on `codex-rollout-trace`, whose rollout-scoped semantic graph is useful for a different debugging problem but does not match the intended long-lived session/thread/turn warehouse model. -This pass is intentionally narrow: generate believable local tables from Codex analytics events first, then add local Responses API call records in a follow-up. React integration, blob-backed sharing, production ingestion, and canonical `chat.Conversation` / context-window materialization are deferred. +This pass is intentionally narrow: generate believable local tables from Codex analytics events plus local Responses API attempt summaries. React integration, blob-backed sharing, production ingestion, and canonical `chat.Conversation` / context-window materialization are deferred. ## Goals @@ -39,6 +39,7 @@ This pass is intentionally narrow: generate believable local tables from Codex a - Keep the sink process-global, not session-scoped: one JSONL file may contain events for many sessions, threads, and turns handled by one Codex process. - Preserve existing analytics behavior; the local sink is an additional best-effort side effect. - Capture enough reduced analytics events locally to materialize believable thread, turn, and turn-event rows. +- Capture local-only Responses API attempt summaries without routing raw payloads through backend analytics. - Materialize a local DuckDB file manually from the JSONL sink for later viewer/UI work. ## Delivery Strategy @@ -47,13 +48,13 @@ Land the smallest useful vertical slice first: 1. Add the reduced analytics JSONL sink in `codex-analytics`. 2. Add the standalone `codex-analytics-materializer` helper binary that reads that JSONL file and writes DuckDB. -3. Follow up with local-only Responses API call capture and populate `viewer_responses_calls_v1`. +3. Add local-only Responses API call capture and populate `viewer_responses_calls_v1`. Keep these boundaries explicit: - Local JSONL capture is independent from backend analytics POSTing. If `analytics.enabled = false` but `CODEX_ANALYTICS_LOCAL_SINK_PATH` is set, Codex should still reduce analytics facts and append local JSONL records without POSTing them. - The sink is process-global. Multiple `AnalyticsEventsClient` instances in one Codex process should resolve the same optional local sink handle instead of each opening their own unrelated writer. -- Local-only Responses API payloads must not be modeled as normal backend analytics facts. When that follow-up lands, use a private queue input variant for local-only Responses facts so raw request/response payloads cannot accidentally reach `send_track_events`. +- Local-only Responses API payloads must not be modeled as normal backend analytics facts. Use a private queue input variant for local-only Responses facts so raw request/response payloads cannot accidentally reach `send_track_events`. - DuckDB stays out of the main `codex` binary build graph. Do not add `codex-analytics-materializer` as a dependency of `codex-cli` in v0. ## Out Of Scope @@ -64,8 +65,6 @@ Keep these boundaries explicit: - No blob-store upload/download path. - No `chat.Conversation` synthesis. - No populated `viewer_context_windows_v1`. -- No populated `viewer_responses_calls_v1` in the first JSONL -> DuckDB milestone. -- No Responses API call capture in the first JSONL -> DuckDB milestone. - No `codex debug process-local-analytics` wiring in the main `codex` binary. - No `--analytics-local-sink` CLI flag in v0; the env var is enough for the first slice. - No rollout-trace-style semantic graph, interaction edges, code cells, or terminal-session modeling. @@ -89,7 +88,7 @@ Behavior: - When set, Codex appends one JSON object per line to the configured file. - The file may contain all sessions launched by one Codex process with that setting. - Pointing multiple Codex processes at the same sink path is unsupported in v0. -- The sink is local-only and may contain sensitive tool metadata, paths, errors, and later Responses API payloads. +- The sink is local-only and may contain sensitive tool metadata, paths, errors, and Responses API payloads. - Sink initialization and writes are best-effort; failures log a warning and must not fail the Codex session. ### Manual Materialization @@ -125,18 +124,16 @@ Codex runtime facts -> existing POST to codex-backend when backend analytics are enabled -> optional append to local JSONL sink -Later: - codex-analytics-materializer helper binary - -> codex-analytics-materializer reads JSONL - -> reduce/group by session/thread/turn - -> write materialized.duckdb +Responses API call lifecycle + -> local-only Responses API start/terminal facts + -> AnalyticsEventsQueue worker + -> codex-analytics local reducer + -> one optional responses_api_call JSONL record -Follow-up: - Responses API call lifecycle - -> local-only Responses API start/terminal facts - -> AnalyticsEventsQueue worker - -> codex-analytics local reducer - -> one optional responses_api_call JSONL record +codex-analytics-materializer helper binary + -> codex-analytics-materializer reads JSONL + -> reduce/group by session/thread/turn/call + -> write materialized.duckdb ``` `codex-analytics` is the right sink entry point because it already owns the thread, turn, resolved-config, compaction, tool, app, plugin, hook, steering, review, and subagent analytics semantics that the initial viewer tables depend on. `codex-analytics-materializer` owns only the local DuckDB reduction path and depends on the generic local JSONL envelope exported by `codex-analytics`. @@ -158,7 +155,7 @@ It owns: The existing `AnalyticsEventsQueue` worker still owns analytics reduction and is the only caller that emits analytics-derived JSONL records. The optional sink itself is process-global and shared by every `AnalyticsEventsClient` instance in one Codex process, so final file writes are serialized by one shared local sink handle instead of depending on one session-scoped queue existing. -`AnalyticsEventsClient` should create a queue whenever either backend analytics are enabled or the local sink is enabled. The worker should tee reduced `TrackEventRequest` values to the local sink when present, then preserve existing network analytics sending only when backend analytics are enabled. Later, when Responses API capture lands, the queue should accept a private local-only Responses input variant alongside normal `AnalyticsFact` values; those local-only payloads must never enter the backend analytics send path. +`AnalyticsEventsClient` should create a queue whenever either backend analytics are enabled or the local sink is enabled. The worker should tee reduced `TrackEventRequest` values to the local sink when present, then preserve existing network analytics sending only when backend analytics are enabled. The queue also accepts a private local-only Responses input variant alongside normal `AnalyticsFact` values; those local-only payloads must never enter the backend analytics send path. `codex-analytics` exports only the generic local sink envelope needed by downstream readers: @@ -216,7 +213,7 @@ Common fields: Record type contract: - `codex_analytics_event` -- `responses_api_call` after the Responses follow-up lands +- `responses_api_call` Do not add a separate manifest record in v0. The reducer can infer covered sessions from the event stream. @@ -251,15 +248,13 @@ Preserve current network send behavior: - local sink write failure: warn and continue existing POST behavior - analytics POST failure: existing behavior unchanged -## Follow-Up: Responses API Call Capture - -This section is intentionally deferred from the first JSONL -> DuckDB milestone. The first milestone should not block on `core/src/client.rs` plumbing; it should ship once reduced analytics events can be captured locally and materialized by the standalone helper. +## Responses API Call Capture ### Why It Lives Here Responses API calls are not Codex analytics events today, and raw request/response payloads must not be sent as normal analytics telemetry. But they are required for the local viewer-table prototype, so add local-only Responses API facts and reduce them into one local sink event inside `codex-analytics`. -Suggested local-only facts: +Local-only facts: ```rust struct LocalResponsesApiCallStartedFact { ... } @@ -271,7 +266,7 @@ enum LocalResponsesApiCallTerminalFact { } ``` -Suggested no-op-capable capture API, parallel to rollout trace's context/attempt shape: +No-op-capable capture API, parallel to rollout trace's context/attempt shape: ```rust pub struct LocalResponsesApiCallCapture { ... } @@ -321,7 +316,7 @@ Started fact: Terminal fact: - `responses_call_id` -- `terminal_at_epoch_millis` +- `completed_at_epoch_millis` - `status`: `completed`, `failed`, or `cancelled` - completed: `response_id`, `upstream_request_id`, `token_usage_json`, `output_items` - failed/cancelled: `upstream_request_id`, `error` or `reason`, `output_items` @@ -495,7 +490,7 @@ responses_api_calls_total_latency_ms `turn_ordinal` is derived per thread by start time, then `turn_id` as a stable tie-breaker. -Before the Responses follow-up lands, the `responses_api_calls_*` aggregates should materialize as zero values. +Turns with no reduced Responses records should materialize `responses_api_calls_*` aggregates as zero values. #### `viewer_turn_events_v1` @@ -522,12 +517,12 @@ Rules: - `event_seq` is per `(session_id, thread_id, turn_id)` ordered by input line number. - `event_kind` is `codex_analytics` or `responses_api`. - Codex analytics rows store the original reduced analytics event in `analytics_event_json`. -- Responses rows store lifecycle summaries in `event_summary_json` after the Responses follow-up lands. +- Responses rows store lifecycle summaries in `event_summary_json`. - Keep this table append-only and paginated-friendly. #### `viewer_responses_calls_v1` -Create this table in the first materializer milestone but leave it empty. Populate it after the Responses follow-up from reduced local `responses_api_call` sink records. +Populate this table from reduced local `responses_api_call` sink records. Columns: @@ -615,11 +610,11 @@ Behavior: ### Phase 2: Standalone DuckDB Materializer 5. Add `codex-analytics-materializer` with DuckDB schema creation and its own helper binary. -6. Implement reductions for threads, turns, and turn events; create empty `viewer_responses_calls_v1` and `viewer_context_windows_v1` tables. +6. Implement reductions for threads, turns, and turn events; create `viewer_responses_calls_v1` and empty `viewer_context_windows_v1` tables. 7. Add helper-binary argument parsing, default output behavior, and clear errors. 8. Add tests and run scoped formatting/tests. -### Phase 3: Responses API Follow-Up +### Phase 3: Responses API Capture 9. Add a private analytics queue input variant for local-only Responses API facts plus local reducer state in `codex-analytics`. 10. Build `LocalResponsesApiCallCapture` at the sampling callsite, pass it into `ModelClientSession::stream(...)`, and wire HTTP and WebSocket Responses hook points in core to start and finish `LocalResponsesApiCallAttempt` values. @@ -661,8 +656,7 @@ Behavior: - `viewer_threads_v1` contains root and child thread metadata - `viewer_turns_v1` contains expected resolved turn row - `viewer_turn_events_v1` preserves per-turn line order -- `viewer_responses_calls_v1` exists and is empty before the Responses follow-up -- `viewer_responses_calls_v1` materializes reduced `responses_api_call` sink records after the Responses follow-up +- `viewer_responses_calls_v1` materializes reduced `responses_api_call` sink records - `viewer_context_windows_v1` exists and is empty - malformed JSONL reports line number - unsupported `schema_version` fails clearly @@ -684,7 +678,6 @@ Behavior: ## Future Follow-Ups -- Populate `viewer_responses_calls_v1` from local Responses API capture once the first JSONL -> DuckDB milestone is working. - Populate `viewer_context_windows_v1` from canonical Responses API `chat.Conversation` Kafka payloads. - Add synthetic local `chat.Conversation` generation only if needed before upstream Kafka support lands. - Add session viewer local DuckDB drag-and-drop mode. diff --git a/codex-rs/memories/write/Cargo.toml b/codex-rs/memories/write/Cargo.toml index ebe6b0f2a8..cc83466e4e 100644 --- a/codex-rs/memories/write/Cargo.toml +++ b/codex-rs/memories/write/Cargo.toml @@ -16,6 +16,7 @@ workspace = true anyhow = { workspace = true } chrono = { workspace = true } codex-backend-client = { workspace = true } +codex-analytics = { workspace = true } codex-core = { workspace = true } codex-config = { workspace = true } codex-features = { workspace = true } diff --git a/codex-rs/memories/write/src/runtime.rs b/codex-rs/memories/write/src/runtime.rs index 58bdc9dfc4..16c7164de9 100644 --- a/codex-rs/memories/write/src/runtime.rs +++ b/codex-rs/memories/write/src/runtime.rs @@ -1,3 +1,4 @@ +use codex_analytics::LocalResponsesApiCallCapture; use codex_core::CodexThread; use codex_core::ModelClient; use codex_core::NewThread; @@ -199,6 +200,7 @@ impl MemoryStartupContext { context.service_tier.clone(), context.turn_metadata_header.as_deref(), &InferenceTraceContext::disabled(), + &LocalResponsesApiCallCapture::disabled(), ) .await?;