From bb40d0f1af189115ac7127498d315df30b2ea401 Mon Sep 17 00:00:00 2001 From: Greg Clark Date: Thu, 16 Jul 2026 19:39:06 -0400 Subject: [PATCH 1/3] feat(switchyard-translation): encode / decode helpers Signed-off-by: Greg Clark --- Cargo.lock | 379 +++++++------------ crates/switchyard-translation/Cargo.toml | 2 + crates/switchyard-translation/src/helpers.rs | 274 ++++++++++++++ crates/switchyard-translation/src/lib.rs | 4 + crates/switchyard-translation/src/sse.rs | 141 +++++++ 5 files changed, 567 insertions(+), 233 deletions(-) create mode 100644 crates/switchyard-translation/src/helpers.rs create mode 100644 crates/switchyard-translation/src/sse.rs diff --git a/Cargo.lock b/Cargo.lock index 18749f29..642b93ae 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -217,27 +217,27 @@ checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" [[package]] name = "bitflags" -version = "2.11.1" +version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" +checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" [[package]] name = "bumpalo" -version = "3.20.2" +version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" [[package]] name = "bytes" -version = "1.11.1" +version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" -version = "1.2.62" +version = "1.2.67" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1dce859f0832a7d088c4f1119888ab94ef4b5d6795d1ce05afb7fe159d79f98" +checksum = "e17dd265a7d0f31ef544e1b20e03add05d3b45b491b633b10d67145d2acc1a38" dependencies = [ "find-msvc-tools", "jobserver", @@ -253,15 +253,26 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "cfg_aliases" -version = "0.2.1" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + +[[package]] +name = "chacha20" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core 0.10.1", +] [[package]] name = "clap" -version = "4.6.1" +version = "4.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51" +checksum = "dd059f9da4f5c36b3787f65d38ccaab1cc315f07b01f89abc8359ee6a8205011" dependencies = [ "clap_builder", "clap_derive", @@ -269,9 +280,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.6.0" +version = "4.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f" +checksum = "f09628afdcc538b57f3c6341e9c8e9970f18e4a481690a64974d7023bd33548b" dependencies = [ "anstream", "anstyle", @@ -328,6 +339,15 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + [[package]] name = "diff" version = "0.1.13" @@ -336,9 +356,9 @@ checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" [[package]] name = "displaydoc" -version = "0.2.5" +version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" +checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" dependencies = [ "proc-macro2", "quote", @@ -517,20 +537,6 @@ dependencies = [ "wasm-bindgen", ] -[[package]] -name = "getrandom" -version = "0.3.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" -dependencies = [ - "cfg-if", - "js-sys", - "libc", - "r-efi 5.3.0", - "wasip2", - "wasm-bindgen", -] - [[package]] name = "getrandom" version = "0.4.3" @@ -538,8 +544,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", + "js-sys", "libc", - "r-efi 6.0.0", + "r-efi", + "rand_core 0.10.1", + "wasm-bindgen", ] [[package]] @@ -586,9 +595,9 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "http" -version = "1.4.0" +version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a" +checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" dependencies = [ "bytes", "itoa", @@ -596,9 +605,9 @@ dependencies = [ [[package]] name = "http-body" -version = "1.0.1" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" dependencies = [ "bytes", "http", @@ -606,9 +615,9 @@ dependencies = [ [[package]] name = "http-body-util" -version = "0.1.3" +version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" +checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2" dependencies = [ "bytes", "futures-core", @@ -631,9 +640,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hyper" -version = "1.9.0" +version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6299f016b246a94207e63da54dbe807655bf9e00044f73ded42c3ac5305fbcca" +checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" dependencies = [ "atomic-waker", "bytes", @@ -833,13 +842,12 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.98" +version = "0.3.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67df7112613f8bfd9150013a0314e196f4800d3201ae742489d999db2f979f08" +checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" dependencies = [ "cfg-if", "futures-util", - "once_cell", "wasm-bindgen", ] @@ -855,7 +863,7 @@ version = "0.1.0" dependencies = [ "async-trait", "futures", - "rand 0.8.6", + "rand 0.8.7", "serde_json", "switchyard-protocol", "tokio", @@ -897,9 +905,9 @@ dependencies = [ [[package]] name = "log" -version = "0.4.29" +version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" [[package]] name = "lru" @@ -924,9 +932,9 @@ checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" [[package]] name = "memchr" -version = "2.8.0" +version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "mime" @@ -936,9 +944,9 @@ checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" [[package]] name = "mio" -version = "1.2.0" +version = "1.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" dependencies = [ "libc", "wasi", @@ -1131,9 +1139,9 @@ dependencies = [ [[package]] name = "quinn" -version = "0.11.9" +version = "0.11.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9e20a958963c291dc322d98411f541009df2ced7b5a4f2bd52337638cfccf20" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" dependencies = [ "bytes", "cfg_aliases", @@ -1151,14 +1159,15 @@ dependencies = [ [[package]] name = "quinn-proto" -version = "0.11.14" +version = "0.11.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098" +checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" dependencies = [ "bytes", - "getrandom 0.3.4", + "getrandom 0.4.3", "lru-slab", - "rand 0.9.4", + "rand 0.10.2", + "rand_pcg", "ring", "rustc-hash", "rustls", @@ -1172,33 +1181,27 @@ dependencies = [ [[package]] name = "quinn-udp" -version = "0.5.14" +version = "0.5.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "addec6a0dcad8a8d96a771f815f0eaf55f9d1805756410b39f5fa81332574cbd" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" dependencies = [ "cfg_aliases", "libc", "once_cell", "socket2", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] name = "quote" -version = "1.0.45" +version = "1.0.46" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41f2619966050689382d2b44f664f4bc593e129785a36d6ee376ddf37259b924" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" dependencies = [ "proc-macro2", ] -[[package]] -name = "r-efi" -version = "5.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" - [[package]] name = "r-efi" version = "6.0.0" @@ -1207,23 +1210,24 @@ checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" [[package]] name = "rand" -version = "0.8.6" +version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" +checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" dependencies = [ "libc", - "rand_chacha 0.3.1", + "rand_chacha", "rand_core 0.6.4", ] [[package]] name = "rand" -version = "0.9.4" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "rand_chacha 0.9.0", - "rand_core 0.9.5", + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", ] [[package]] @@ -1236,16 +1240,6 @@ dependencies = [ "rand_core 0.6.4", ] -[[package]] -name = "rand_chacha" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" -dependencies = [ - "ppv-lite86", - "rand_core 0.9.5", -] - [[package]] name = "rand_core" version = "0.6.4" @@ -1257,11 +1251,17 @@ dependencies = [ [[package]] name = "rand_core" -version = "0.9.5" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" dependencies = [ - "getrandom 0.3.4", + "rand_core 0.10.1", ] [[package]] @@ -1330,15 +1330,15 @@ dependencies = [ [[package]] name = "rustc-hash" -version = "2.1.2" +version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" [[package]] name = "rustls" -version = "0.23.40" +version = "0.23.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" +checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" dependencies = [ "aws-lc-rs", "log", @@ -1352,9 +1352,9 @@ dependencies = [ [[package]] name = "rustls-native-certs" -version = "0.8.3" +version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63" +checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" dependencies = [ "openssl-probe", "rustls-pki-types", @@ -1364,9 +1364,9 @@ dependencies = [ [[package]] name = "rustls-pki-types" -version = "1.14.1" +version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9" +checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" dependencies = [ "web-time", "zeroize", @@ -1386,9 +1386,9 @@ dependencies = [ [[package]] name = "rustversion" -version = "1.0.22" +version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" [[package]] name = "ryu" @@ -1466,9 +1466,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.149" +version = "1.0.150" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" dependencies = [ "itoa", "memchr", @@ -1511,9 +1511,9 @@ dependencies = [ [[package]] name = "shlex" -version = "1.3.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" [[package]] name = "signal-hook-registry" @@ -1533,15 +1533,15 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "smallvec" -version = "1.15.1" +version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" [[package]] name = "socket2" -version = "0.6.3" +version = "0.6.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", "windows-sys 0.61.2", @@ -1573,7 +1573,7 @@ dependencies = [ "async-trait", "futures-util", "parking_lot", - "rand 0.8.6", + "rand 0.8.7", "reqwest", "serde", "serde_json", @@ -1592,7 +1592,7 @@ dependencies = [ "futures-core", "futures-util", "parking_lot", - "rand 0.8.6", + "rand 0.8.7", "reqwest", "serde", "serde_json", @@ -1621,7 +1621,7 @@ dependencies = [ "async-trait", "futures-core", "lru", - "rand 0.8.6", + "rand 0.8.7", "serde", "serde_json", "thiserror", @@ -1696,6 +1696,8 @@ dependencies = [ name = "switchyard-translation" version = "0.1.0" dependencies = [ + "async-stream", + "futures", "pretty_assertions", "serde", "serde_json", @@ -1705,9 +1707,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.117" +version = "2.0.119" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" dependencies = [ "proc-macro2", "quote", @@ -1772,9 +1774,9 @@ dependencies = [ [[package]] name = "tinyvec" -version = "1.11.0" +version = "1.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3e61e67053d25a4e82c844e8424039d9745781b3fc4f32b8d55ed50f5f667ef3" +checksum = "bb4ebadaa0af04fab11ae01eb5f9fdb5f9c5b875506e210e71c07873528baa7f" dependencies = [ "tinyvec_macros", ] @@ -1787,9 +1789,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.52.3" +version = "1.52.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +checksum = "317fafbbe3f02fc663dad00ea6186197de963cd4190e86a26d8d0fae095539af" dependencies = [ "bytes", "libc", @@ -2011,20 +2013,11 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" -[[package]] -name = "wasip2" -version = "1.0.3+wasi-0.2.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6" -dependencies = [ - "wit-bindgen", -] - [[package]] name = "wasm-bindgen" -version = "0.2.121" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49ace1d07c165b0864824eee619580c4689389afa9dc9ed3a4c75040d82e6790" +checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" dependencies = [ "cfg-if", "once_cell", @@ -2035,9 +2028,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-futures" -version = "0.4.71" +version = "0.4.76" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96492d0d3ffba25305a7dc88720d250b1401d7edca02cc3bcd50633b424673b8" +checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" dependencies = [ "js-sys", "wasm-bindgen", @@ -2045,9 +2038,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.121" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e68e6f4afd367a562002c05637acb8578ff2dea1943df76afb9e83d177c8578" +checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -2055,9 +2048,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.121" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d95a9ec35c64b2a7cb35d3fead40c4238d0940c86d107136999567a4703259f2" +checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" dependencies = [ "bumpalo", "proc-macro2", @@ -2068,9 +2061,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.121" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4e0100b01e9f0d03189a92b96772a1fb998639d981193d7dbab487302513441" +checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" dependencies = [ "unicode-ident", ] @@ -2090,9 +2083,9 @@ dependencies = [ [[package]] name = "web-sys" -version = "0.3.98" +version = "0.3.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b572dff8bcf38bad0fa19729c89bb5748b2b9b1d8be70cf90df697e3a8f32aa" +checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141" dependencies = [ "js-sys", "wasm-bindgen", @@ -2120,16 +2113,7 @@ version = "0.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" dependencies = [ - "windows-targets 0.52.6", -] - -[[package]] -name = "windows-sys" -version = "0.60.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" -dependencies = [ - "windows-targets 0.53.5", + "windows-targets", ] [[package]] @@ -2147,31 +2131,14 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ - "windows_aarch64_gnullvm 0.52.6", - "windows_aarch64_msvc 0.52.6", - "windows_i686_gnu 0.52.6", - "windows_i686_gnullvm 0.52.6", - "windows_i686_msvc 0.52.6", - "windows_x86_64_gnu 0.52.6", - "windows_x86_64_gnullvm 0.52.6", - "windows_x86_64_msvc 0.52.6", -] - -[[package]] -name = "windows-targets" -version = "0.53.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" -dependencies = [ - "windows-link", - "windows_aarch64_gnullvm 0.53.1", - "windows_aarch64_msvc 0.53.1", - "windows_i686_gnu 0.53.1", - "windows_i686_gnullvm 0.53.1", - "windows_i686_msvc 0.53.1", - "windows_x86_64_gnu 0.53.1", - "windows_x86_64_gnullvm 0.53.1", - "windows_x86_64_msvc 0.53.1", + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", ] [[package]] @@ -2180,96 +2147,48 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" -[[package]] -name = "windows_aarch64_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" - [[package]] name = "windows_aarch64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" -[[package]] -name = "windows_aarch64_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" - [[package]] name = "windows_i686_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" -[[package]] -name = "windows_i686_gnu" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" - [[package]] name = "windows_i686_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" -[[package]] -name = "windows_i686_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" - [[package]] name = "windows_i686_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" -[[package]] -name = "windows_i686_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" - [[package]] name = "windows_x86_64_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" -[[package]] -name = "windows_x86_64_gnu" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" - [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" -[[package]] -name = "windows_x86_64_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" - [[package]] name = "windows_x86_64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" -[[package]] -name = "windows_x86_64_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" - [[package]] name = "winnow" version = "0.7.15" @@ -2279,12 +2198,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "wit-bindgen" -version = "0.57.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" - [[package]] name = "writeable" version = "0.6.3" @@ -2312,9 +2225,9 @@ checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" [[package]] name = "yoke" -version = "0.8.2" +version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "abe8c5fda708d9ca3df187cae8bfb9ceda00dd96231bed36e445a1a48e66f9ca" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" dependencies = [ "stable_deref_trait", "yoke-derive", @@ -2335,18 +2248,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.48" +version = "0.8.54" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9" +checksum = "b7cbbc0a705a0fd05cc3676525980d2bf5a9bc4adac6d6475209a7887cf59d19" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.48" +version = "0.8.54" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4" +checksum = "e2e817b7b52d0c7358d3246da9d69935ebb18116b2b102b4230dac079b4862f5" dependencies = [ "proc-macro2", "quote", @@ -2376,9 +2289,9 @@ dependencies = [ [[package]] name = "zeroize" -version = "1.8.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" [[package]] name = "zerotrie" @@ -2415,6 +2328,6 @@ dependencies = [ [[package]] name = "zmij" -version = "1.0.21" +version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/crates/switchyard-translation/Cargo.toml b/crates/switchyard-translation/Cargo.toml index 6f762904..9e55e7e9 100644 --- a/crates/switchyard-translation/Cargo.toml +++ b/crates/switchyard-translation/Cargo.toml @@ -21,6 +21,8 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" switchyard-protocol = { path = "../libsy-protocol", version = "0.1.0" } thiserror = "2" +futures = "0.3" +async-stream = "0.3" [dev-dependencies] pretty_assertions = "1" diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs new file mode 100644 index 00000000..a5e644e0 --- /dev/null +++ b/crates/switchyard-translation/src/helpers.rs @@ -0,0 +1,274 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Convenience wrappers over the default [`TranslationEngine`] — decode a wire +//! request/response to the neutral IR, encode the IR back, and decode/encode a +//! streamed response — so callers can translate without threading an engine and +//! policy through every call. + +use std::pin::Pin; +use std::sync::LazyLock; + +use async_stream::try_stream; +use futures::{Stream, StreamExt}; +use serde_json::Value; + +use crate::sse; +use crate::{ + AggLlmResponse, LlmRequest, LlmResponseStream, Result, StreamCodecRegistry, + StreamTranslationState, TranslationEngine, TranslationPolicy, WireFormat, +}; + +static DEFAULT_TRANSLATION_POLICY: LazyLock = + LazyLock::new(TranslationPolicy::default); +static DEFAULT_TRANSLATION_ENGINE: LazyLock = + LazyLock::new(TranslationEngine::default); + +/// Decodes a `wire_format` request body into the neutral IR. +pub fn decode_request(wire_format: WireFormat, body: &Value) -> Result { + Ok(DEFAULT_TRANSLATION_ENGINE + .decode_request(wire_format, body, &DEFAULT_TRANSLATION_POLICY)? + .request) +} + +/// Encodes a normalized request into `wire_format`'s JSON body. +pub fn encode_request(request: &LlmRequest, wire_format: WireFormat) -> Result { + Ok(DEFAULT_TRANSLATION_ENGINE + .encode_request(wire_format, request, &DEFAULT_TRANSLATION_POLICY)? + .body) +} + +/// Decodes a buffered `wire_format` response body into the neutral aggregate. +pub fn decode_buffered_response(body: &Value, wire_format: WireFormat) -> Result { + Ok(DEFAULT_TRANSLATION_ENGINE + .decode_response(wire_format, body, &DEFAULT_TRANSLATION_POLICY)? + .response) +} + +/// Encodes a buffered aggregate into `wire_format`'s JSON body, stamping +/// `requested_model` over the upstream id so the caller sees the model it asked for. +pub fn encode_buffered_response( + agg: &AggLlmResponse, + wire_format: WireFormat, + requested_model: Option<&str>, +) -> Result { + let mut body = DEFAULT_TRANSLATION_ENGINE + .encode_response(wire_format, agg, &DEFAULT_TRANSLATION_POLICY)? + .body; + if let (Some(model), Value::Object(object)) = (requested_model, &mut body) { + object.insert("model".to_string(), Value::String(model.to_string())); + } + Ok(body) +} + +/// A stream of wire-format event objects in one format — the unframed body of an +/// SSE response. The serving layer frames each `Value` (e.g. as an SSE +/// `data:`/`event:` block). +pub type RawEventStream = Pin< + Box< + dyn Stream>> + + Send, + >, +>; + +/// Encodes a stream of IR chunks into a stream of target-format wire events. +/// +/// `requested_model` is exposed as the response model (via the stream state's +/// `target_model`). The target stream codec is resolved once and reused per chunk; +/// terminal events (`message_stop` / `response.completed`) come from `finish`. +pub fn encode_stream( + chunks: LlmResponseStream, + target: WireFormat, + requested_model: Option, +) -> RawEventStream { + // The target is always a built-in wire format, so this lookup cannot fail; a + // failure surfaces as a single error item rather than a panic. + let codec = match StreamCodecRegistry::with_builtins().codec(target) { + Ok(codec) => codec, + Err(error) => { + let message = error.to_string(); + return Box::pin(futures::stream::once(async move { + Err(Box::new(std::io::Error::other(message)) + as Box) + })); + } + }; + + let events = try_stream! { + let mut state = StreamTranslationState { + target: Some(target.into()), + target_model: requested_model, + ..Default::default() + }; + + let mut chunks = chunks; + while let Some(item) = chunks.next().await { + let chunk = item?; + for value in codec.encode_event(&mut state, chunk) { + yield value; + } + } + for value in codec.finish(&mut state) { + yield value; + } + }; + + Box::pin(events) +} + +/// Decodes a byte stream of `source`-format SSE frames into neutral IR chunks. +/// +/// Operates on raw bytes, not any HTTP client type: the caller adapts its +/// transport's body stream into `Stream, _>>`. Frames are +/// buffered across chunks (a partial frame waits for its boundary); the source +/// stream codec is resolved once and reused for every frame. +pub fn decode_stream(bytes: S, source: WireFormat) -> LlmResponseStream +where + S: Stream, Box>> + + Send + + 'static, +{ + let marker = sse::done_marker(source); + // The source is always a built-in wire format, so this lookup cannot fail; a + // failure still surfaces as a single error item rather than a panic. + let codec = match StreamCodecRegistry::with_builtins().codec(source) { + Ok(codec) => codec, + Err(error) => { + let boxed: Box = error.to_string().into(); + return Box::pin(futures::stream::once(async move { Err(boxed) })); + } + }; + Box::pin(try_stream! { + let mut state = StreamTranslationState::default(); + let mut buffer = Vec::new(); + futures::pin_mut!(bytes); + + while let Some(chunk) = bytes.next().await { + buffer.extend_from_slice(&chunk?); + while let Some(frame) = sse::drain_next_sse_frame(&mut buffer)? { + match sse::parse_json_sse_frame(&frame, marker)? { + sse::ParsedSseFrame::Json(value) => { + for event in codec.decode_event(&mut state, &value) { + yield event; + } + } + sse::ParsedSseFrame::Done => return, + sse::ParsedSseFrame::Empty => {} + } + } + } + + // A non-standard upstream might omit the final blank line; parse a trailing + // complete frame instead of losing its last chunk. + if sse::has_non_whitespace_bytes(&buffer) { + let frame = sse::decode_sse_frame(&buffer)?; + if let sse::ParsedSseFrame::Json(value) = sse::parse_json_sse_frame(&frame, marker)? { + for event in codec.decode_event(&mut state, &value) { + yield event; + } + } + } + }) +} + +#[cfg(test)] +mod tests { + use futures::executor::block_on; + use futures::{stream, StreamExt}; + use serde_json::{json, Value}; + use switchyard_protocol::{completion_text, LlmResponseChunk}; + + use super::{ + decode_buffered_response, decode_request, decode_stream, encode_buffered_response, + encode_request, encode_stream, + }; + use crate::{LlmResponseStream, WireFormat}; + + // A boxed stream item error, matching the streamed IR contract. + type BoxError = Box; + + #[test] + fn request_round_trips_through_openai_chat() { + let body = json!({"model": "gpt", "messages": [{"role": "user", "content": "hi"}]}); + let request = decode_request(WireFormat::OpenAiChat, &body).unwrap(); + assert_eq!(request.model.as_deref(), Some("gpt")); + + let encoded = encode_request(&request, WireFormat::OpenAiChat).unwrap(); + assert_eq!(encoded["model"], "gpt"); + assert_eq!(encoded["messages"][0]["content"], "hi"); + } + + #[test] + fn buffered_response_round_trips_and_restamps_model() { + let body = json!({ + "id": "1", + "model": "upstream", + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "Hi there"}, + "finish_reason": "stop" + }], + "usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3} + }); + let agg = decode_buffered_response(&body, WireFormat::OpenAiChat).unwrap(); + assert_eq!(completion_text(&agg), "Hi there"); + + // `requested_model` overrides the encoded model. + let encoded = + encode_buffered_response(&agg, WireFormat::OpenAiChat, Some("client-facing")).unwrap(); + assert_eq!(encoded["model"], "client-facing"); + assert_eq!(encoded["choices"][0]["message"]["content"], "Hi there"); + } + + #[test] + fn encode_stream_emits_openai_chunks_whose_deltas_reassemble() { + let chunks: LlmResponseStream = stream::iter(vec![ + Ok(LlmResponseChunk::TextDelta { + index: 0, + text: "Hello".to_string(), + }), + Ok(LlmResponseChunk::TextDelta { + index: 0, + text: " world".to_string(), + }), + Ok(LlmResponseChunk::MessageStop { + reason: Some("stop".to_string()), + }), + ]) + .boxed(); + + let events: Vec = block_on( + encode_stream(chunks, WireFormat::OpenAiChat, Some("m".to_string())) + .map(|item| item.unwrap()) + .collect(), + ); + let content: String = events + .iter() + .filter_map(|event| event["choices"][0]["delta"]["content"].as_str()) + .collect(); + assert_eq!(content, "Hello world"); + } + + #[test] + fn decode_stream_parses_sse_bytes_into_ir_chunks() { + let sse = b"data: {\"choices\":[{\"delta\":{\"content\":\"Hello\"}}]}\n\n\ + data: {\"choices\":[{\"delta\":{\"content\":\" world\"}}]}\n\n\ + data: [DONE]\n\n" + .to_vec(); + let bytes = stream::once(async move { Ok::, BoxError>(sse) }); + + let chunks: Vec = block_on( + decode_stream(bytes, WireFormat::OpenAiChat) + .map(|item| item.unwrap()) + .collect(), + ); + let content: String = chunks + .iter() + .filter_map(|chunk| match chunk { + LlmResponseChunk::TextDelta { text, .. } => Some(text.as_str()), + _ => None, + }) + .collect(); + assert_eq!(content, "Hello world"); + } +} diff --git a/crates/switchyard-translation/src/lib.rs b/crates/switchyard-translation/src/lib.rs index 3eaaf21a..b54296d8 100644 --- a/crates/switchyard-translation/src/lib.rs +++ b/crates/switchyard-translation/src/lib.rs @@ -11,17 +11,21 @@ pub mod codecs; pub mod diagnostic; pub mod engine; pub mod error; +mod helpers; pub mod policy; +mod sse; pub mod stream; pub mod util; pub use switchyard_protocol::stream::LlmResponseChunk; +pub use switchyard_protocol::stream::LlmResponseStream; pub use switchyard_protocol::{format, llm}; pub use diagnostic::*; pub use engine::*; pub use error::*; pub use format::*; +pub use helpers::*; pub use llm::*; pub use policy::*; pub use stream::*; diff --git a/crates/switchyard-translation/src/sse.rs b/crates/switchyard-translation/src/sse.rs new file mode 100644 index 00000000..443f062e --- /dev/null +++ b/crates/switchyard-translation/src/sse.rs @@ -0,0 +1,141 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Minimal SSE frame parser for decoding streamed provider responses. +//! +//! One copy backs all neutral-IR stream decoding ([`decode_stream`](crate::decode_stream)). +//! Errors are boxed `std::error::Error`s — the item error type of a streamed +//! response — so this module stays free of any HTTP client or server types. + +use serde_json::Value; + +use crate::WireFormat; + +/// Boxed, thread-safe error carried by a streamed item. +pub(crate) type BoxError = Box; + +/// The outcome of parsing one SSE frame's data lines. +pub(crate) enum ParsedSseFrame { + /// Frame contained a JSON payload. + Json(Value), + /// Frame contained the provider's terminal marker (e.g. `[DONE]`). + Done, + /// Frame had no data payload. + Empty, +} + +/// The terminal SSE marker for `format`, if any. OpenAI Chat and Responses use +/// `[DONE]`; Anthropic ends on its `message_stop` event with no marker. +pub(crate) fn done_marker(format: WireFormat) -> Option<&'static str> { + match format { + WireFormat::OpenAiChat | WireFormat::OpenAiResponses => Some("[DONE]"), + WireFormat::AnthropicMessages => None, + } +} + +/// Drains one complete SSE frame from the buffer when a boundary is present. +pub(crate) fn drain_next_sse_frame(buffer: &mut Vec) -> Result, BoxError> { + let Some((index, separator_len)) = next_sse_boundary(buffer) else { + return Ok(None); + }; + let frame = decode_sse_frame(&buffer[..index])?; + buffer.drain(..index + separator_len); + Ok(Some(frame)) +} + +/// Decodes one raw SSE frame as UTF-8. +pub(crate) fn decode_sse_frame(frame: &[u8]) -> Result { + std::str::from_utf8(frame) + .map(str::to_string) + .map_err(|error| format!("stream emitted invalid UTF-8 frame: {error}").into()) +} + +/// Returns whether the buffer has any non-whitespace bytes. +pub(crate) fn has_non_whitespace_bytes(buffer: &[u8]) -> bool { + buffer.iter().any(|byte| !byte.is_ascii_whitespace()) +} + +/// Parses data lines from one SSE frame into JSON, terminal, or empty states. +pub(crate) fn parse_json_sse_frame( + frame: &str, + done_marker: Option<&str>, +) -> Result { + let mut data_lines = Vec::new(); + for line in frame.lines() { + // SSE comments and blank lines do not contribute data. + if line.is_empty() || line.starts_with(':') { + continue; + } + if let Some(data) = line.strip_prefix("data:") { + data_lines.push(data.trim_start().to_string()); + } + } + + if data_lines.is_empty() { + return Ok(ParsedSseFrame::Empty); + } + + let data = data_lines.join("\n"); + if done_marker.is_some_and(|marker| data.trim() == marker) { + return Ok(ParsedSseFrame::Done); + } + + let value = serde_json::from_str::(&data) + .map_err(|error| format!("stream emitted invalid JSON frame: {error}"))?; + Ok(ParsedSseFrame::Json(value)) +} + +/// Finds the next CRLF or LF SSE frame boundary. +fn next_sse_boundary(buffer: &[u8]) -> Option<(usize, usize)> { + match (find_bytes(buffer, b"\r\n\r\n"), find_bytes(buffer, b"\n\n")) { + (Some(crlf), Some(lf)) if crlf < lf => Some((crlf, 4)), + (Some(_), Some(lf)) => Some((lf, 2)), + (Some(crlf), None) => Some((crlf, 4)), + (None, Some(lf)) => Some((lf, 2)), + (None, None) => None, + } +} + +/// Finds a byte needle inside a byte haystack. +fn find_bytes(haystack: &[u8], needle: &[u8]) -> Option { + haystack + .windows(needle.len()) + .position(|window| window == needle) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + // Multi-byte UTF-8 split across network chunks should wait for a full frame. + #[test] + fn buffers_incomplete_utf8_until_a_complete_sse_frame_arrives() -> Result<(), BoxError> { + let mut buffer = b"data: {\"text\":\"".to_vec(); + let multibyte = "é".as_bytes(); + buffer.extend_from_slice(&multibyte[..1]); + assert!(drain_next_sse_frame(&mut buffer)?.is_none()); + + buffer.extend_from_slice(&multibyte[1..]); + buffer.extend_from_slice(b"\"}\n\n"); + + let Some(frame) = drain_next_sse_frame(&mut buffer)? else { + return Err("complete SSE frame should be drained".into()); + }; + let ParsedSseFrame::Json(value) = parse_json_sse_frame(&frame, None)? else { + return Err("SSE frame should parse as JSON".into()); + }; + assert_eq!(value, json!({"text": "é"})); + assert!(buffer.is_empty()); + Ok(()) + } + + #[test] + fn recognizes_done_marker() -> Result<(), BoxError> { + let ParsedSseFrame::Done = parse_json_sse_frame("data: [DONE]\n", Some("[DONE]"))? else { + return Err("DONE frame should stop".into()); + }; + Ok(()) + } +} From 714824c19dd0ff7877d770f54e9e211a110ac356 Mon Sep 17 00:00:00 2001 From: Greg Clark Date: Fri, 17 Jul 2026 14:27:35 -0400 Subject: [PATCH 2/3] chore: review feedback Signed-off-by: Greg Clark --- crates/switchyard-translation/src/helpers.rs | 16 ++--- crates/switchyard-translation/src/sse.rs | 75 +++++++------------- 2 files changed, 30 insertions(+), 61 deletions(-) diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index a5e644e0..fa8bb4b6 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -146,23 +146,19 @@ where while let Some(chunk) = bytes.next().await { buffer.extend_from_slice(&chunk?); while let Some(frame) = sse::drain_next_sse_frame(&mut buffer)? { - match sse::parse_json_sse_frame(&frame, marker)? { - sse::ParsedSseFrame::Json(value) => { - for event in codec.decode_event(&mut state, &value) { - yield event; - } + if let Some(value) = sse::parse_json_sse_frame(&frame, marker)? { + for event in codec.decode_event(&mut state, &value) { + yield event; } - sse::ParsedSseFrame::Done => return, - sse::ParsedSseFrame::Empty => {} } } } // A non-standard upstream might omit the final blank line; parse a trailing // complete frame instead of losing its last chunk. - if sse::has_non_whitespace_bytes(&buffer) { - let frame = sse::decode_sse_frame(&buffer)?; - if let sse::ParsedSseFrame::Json(value) = sse::parse_json_sse_frame(&frame, marker)? { + if !buffer.trim_ascii().is_empty() { + let frame: String = String::from_utf8_lossy(&buffer).to_string(); + if let Some(value) = sse::parse_json_sse_frame(&frame, marker)? { for event in codec.decode_event(&mut state, &value) { yield event; } diff --git a/crates/switchyard-translation/src/sse.rs b/crates/switchyard-translation/src/sse.rs index 443f062e..8643dd4f 100644 --- a/crates/switchyard-translation/src/sse.rs +++ b/crates/switchyard-translation/src/sse.rs @@ -14,16 +14,6 @@ use crate::WireFormat; /// Boxed, thread-safe error carried by a streamed item. pub(crate) type BoxError = Box; -/// The outcome of parsing one SSE frame's data lines. -pub(crate) enum ParsedSseFrame { - /// Frame contained a JSON payload. - Json(Value), - /// Frame contained the provider's terminal marker (e.g. `[DONE]`). - Done, - /// Frame had no data payload. - Empty, -} - /// The terminal SSE marker for `format`, if any. OpenAI Chat and Responses use /// `[DONE]`; Anthropic ends on its `message_stop` event with no marker. pub(crate) fn done_marker(format: WireFormat) -> Option<&'static str> { @@ -38,51 +28,33 @@ pub(crate) fn drain_next_sse_frame(buffer: &mut Vec) -> Result Result { - std::str::from_utf8(frame) - .map(str::to_string) - .map_err(|error| format!("stream emitted invalid UTF-8 frame: {error}").into()) -} - -/// Returns whether the buffer has any non-whitespace bytes. -pub(crate) fn has_non_whitespace_bytes(buffer: &[u8]) -> bool { - buffer.iter().any(|byte| !byte.is_ascii_whitespace()) -} - -/// Parses data lines from one SSE frame into JSON, terminal, or empty states. pub(crate) fn parse_json_sse_frame( frame: &str, done_marker: Option<&str>, -) -> Result { - let mut data_lines = Vec::new(); - for line in frame.lines() { - // SSE comments and blank lines do not contribute data. - if line.is_empty() || line.starts_with(':') { - continue; - } - if let Some(data) = line.strip_prefix("data:") { - data_lines.push(data.trim_start().to_string()); - } - } - - if data_lines.is_empty() { - return Ok(ParsedSseFrame::Empty); - } - - let data = data_lines.join("\n"); - if done_marker.is_some_and(|marker| data.trim() == marker) { - return Ok(ParsedSseFrame::Done); +) -> Result, BoxError> { + let data = frame + .lines() + .filter(|line| !line.is_empty() && !line.starts_with(':')) + .filter_map(|line| line.strip_prefix("data: ").map(|l| l.to_string())) + .fold(String::new(), |mut a, b| { + a.reserve(b.len() + 1); + a.push_str(&b); + a.push('\n'); + a + }); + let data = data.trim_end(); + + // No data payload, or the terminal marker: nothing to decode. + if data.is_empty() || done_marker.is_some_and(|marker| data == marker) { + return Ok(None); } - - let value = serde_json::from_str::(&data) - .map_err(|error| format!("stream emitted invalid JSON frame: {error}"))?; - Ok(ParsedSseFrame::Json(value)) + let value = serde_json::from_str::(data)?; + Ok(Some(value)) } /// Finds the next CRLF or LF SSE frame boundary. @@ -123,7 +95,7 @@ mod tests { let Some(frame) = drain_next_sse_frame(&mut buffer)? else { return Err("complete SSE frame should be drained".into()); }; - let ParsedSseFrame::Json(value) = parse_json_sse_frame(&frame, None)? else { + let Some(value) = parse_json_sse_frame(&frame, None)? else { return Err("SSE frame should parse as JSON".into()); }; assert_eq!(value, json!({"text": "é"})); @@ -133,9 +105,10 @@ mod tests { #[test] fn recognizes_done_marker() -> Result<(), BoxError> { - let ParsedSseFrame::Done = parse_json_sse_frame("data: [DONE]\n", Some("[DONE]"))? else { - return Err("DONE frame should stop".into()); - }; + // The terminal marker carries no JSON payload. + if parse_json_sse_frame("data: [DONE]\n", Some("[DONE]"))?.is_some() { + return Err("DONE frame should not yield a JSON value".into()); + } Ok(()) } } From 49c76966308a91505c01b86f13141b11ff1e70bb Mon Sep 17 00:00:00 2001 From: Greg Clark Date: Fri, 17 Jul 2026 16:26:12 -0400 Subject: [PATCH 3/3] chore: review feedback Signed-off-by: Greg Clark --- crates/switchyard-translation/src/helpers.rs | 224 +++++++++++++------ crates/switchyard-translation/src/sse.rs | 100 ++++----- 2 files changed, 204 insertions(+), 120 deletions(-) diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index fa8bb4b6..15f9c185 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -10,7 +10,8 @@ use std::pin::Pin; use std::sync::LazyLock; use async_stream::try_stream; -use futures::{Stream, StreamExt}; +use futures::io::AsyncBufReadExt; +use futures::{Stream, StreamExt, TryStreamExt}; use serde_json::Value; use crate::sse; @@ -39,7 +40,7 @@ pub fn encode_request(request: &LlmRequest, wire_format: WireFormat) -> Result Result { +pub fn decode_aggregated_response(body: &Value, wire_format: WireFormat) -> Result { Ok(DEFAULT_TRANSLATION_ENGINE .decode_response(wire_format, body, &DEFAULT_TRANSLATION_POLICY)? .response) @@ -47,7 +48,7 @@ pub fn decode_buffered_response(body: &Value, wire_format: WireFormat) -> Result /// Encodes a buffered aggregate into `wire_format`'s JSON body, stamping /// `requested_model` over the upstream id so the caller sees the model it asked for. -pub fn encode_buffered_response( +pub fn encode_aggregated_response( agg: &AggLlmResponse, wire_format: WireFormat, requested_model: Option<&str>, @@ -80,28 +81,19 @@ pub fn encode_stream( chunks: LlmResponseStream, target: WireFormat, requested_model: Option, -) -> RawEventStream { +) -> std::result::Result> { // The target is always a built-in wire format, so this lookup cannot fail; a - // failure surfaces as a single error item rather than a panic. - let codec = match StreamCodecRegistry::with_builtins().codec(target) { - Ok(codec) => codec, - Err(error) => { - let message = error.to_string(); - return Box::pin(futures::stream::once(async move { - Err(Box::new(std::io::Error::other(message)) - as Box) - })); - } + // failure returns as an `Err` rather than a panic. + let codec = StreamCodecRegistry::with_builtins().codec(target)?; + + let mut state = StreamTranslationState { + target: Some(target.into()), + target_model: requested_model, + ..Default::default() }; + let mut chunks = chunks; let events = try_stream! { - let mut state = StreamTranslationState { - target: Some(target.into()), - target_model: requested_model, - ..Default::default() - }; - - let mut chunks = chunks; while let Some(item) = chunks.next().await { let chunk = item?; for value in codec.encode_event(&mut state, chunk) { @@ -113,7 +105,7 @@ pub fn encode_stream( } }; - Box::pin(events) + Ok(Box::pin(events)) } /// Decodes a byte stream of `source`-format SSE frames into neutral IR chunks. @@ -122,7 +114,10 @@ pub fn encode_stream( /// transport's body stream into `Stream, _>>`. Frames are /// buffered across chunks (a partial frame waits for its boundary); the source /// stream codec is resolved once and reused for every frame. -pub fn decode_stream(bytes: S, source: WireFormat) -> LlmResponseStream +pub fn decode_stream( + bytes: S, + source: WireFormat, +) -> std::result::Result> where S: Stream, Box>> + Send @@ -130,52 +125,58 @@ where { let marker = sse::done_marker(source); // The source is always a built-in wire format, so this lookup cannot fail; a - // failure still surfaces as a single error item rather than a panic. - let codec = match StreamCodecRegistry::with_builtins().codec(source) { - Ok(codec) => codec, - Err(error) => { - let boxed: Box = error.to_string().into(); - return Box::pin(futures::stream::once(async move { Err(boxed) })); - } - }; - Box::pin(try_stream! { - let mut state = StreamTranslationState::default(); - let mut buffer = Vec::new(); - futures::pin_mut!(bytes); - - while let Some(chunk) = bytes.next().await { - buffer.extend_from_slice(&chunk?); - while let Some(frame) = sse::drain_next_sse_frame(&mut buffer)? { + // failure returns as an `Err` rather than a panic. + let codec = StreamCodecRegistry::with_builtins().codec(source)?; + // Adapt the byte-chunk stream into an async line reader. The BufReader + // reassembles data split across network chunks (including multi-byte UTF-8), + // and `lines()` yields one SSE field line at a time. The stream is boxed to + // an `io::Error` item so `into_async_read`'s error bound resolves cleanly. + let io_bytes: Pin>> + Send>> = + Box::pin(bytes.map(|item| item.map_err(std::io::Error::other))); + let lines = futures::io::BufReader::new(io_bytes.into_async_read()).lines(); + + let mut state = StreamTranslationState::default(); + let mut frame = String::new(); + let stream = Box::pin(try_stream! { + futures::pin_mut!(lines); + while let Some(line) = lines.next().await { + let line = line?; + // A blank line (allowing a bare CR for CRLF streams) ends the frame. + if line.trim_end().is_empty() { if let Some(value) = sse::parse_json_sse_frame(&frame, marker)? { for event in codec.decode_event(&mut state, &value) { yield event; } } + frame.clear(); + } else { + frame.push_str(&line); + frame.push('\n'); } } // A non-standard upstream might omit the final blank line; parse a trailing // complete frame instead of losing its last chunk. - if !buffer.trim_ascii().is_empty() { - let frame: String = String::from_utf8_lossy(&buffer).to_string(); + if !frame.trim_end().is_empty() { if let Some(value) = sse::parse_json_sse_frame(&frame, marker)? { for event in codec.decode_event(&mut state, &value) { yield event; } } } - }) + }); + Ok(stream) } #[cfg(test)] mod tests { use futures::executor::block_on; - use futures::{stream, StreamExt}; + use futures::{stream, Stream, StreamExt}; use serde_json::{json, Value}; use switchyard_protocol::{completion_text, LlmResponseChunk}; use super::{ - decode_buffered_response, decode_request, decode_stream, encode_buffered_response, + decode_aggregated_response, decode_request, decode_stream, encode_aggregated_response, encode_request, encode_stream, }; use crate::{LlmResponseStream, WireFormat}; @@ -183,19 +184,41 @@ mod tests { // A boxed stream item error, matching the streamed IR contract. type BoxError = Box; + // Collects a decoded IR stream, surfacing the first error instead of panicking. + fn decode_all( + bytes: impl Stream, BoxError>> + Send + 'static, + source: WireFormat, + ) -> Result, BoxError> { + block_on(decode_stream(bytes, source)?.collect::>()) + .into_iter() + .collect() + } + + // Concatenates the text of every `TextDelta` chunk. + fn text_of(chunks: &[LlmResponseChunk]) -> String { + chunks + .iter() + .filter_map(|chunk| match chunk { + LlmResponseChunk::TextDelta { text, .. } => Some(text.as_str()), + _ => None, + }) + .collect() + } + #[test] - fn request_round_trips_through_openai_chat() { + fn request_round_trips_through_openai_chat() -> Result<(), BoxError> { let body = json!({"model": "gpt", "messages": [{"role": "user", "content": "hi"}]}); - let request = decode_request(WireFormat::OpenAiChat, &body).unwrap(); + let request = decode_request(WireFormat::OpenAiChat, &body)?; assert_eq!(request.model.as_deref(), Some("gpt")); - let encoded = encode_request(&request, WireFormat::OpenAiChat).unwrap(); + let encoded = encode_request(&request, WireFormat::OpenAiChat)?; assert_eq!(encoded["model"], "gpt"); assert_eq!(encoded["messages"][0]["content"], "hi"); + Ok(()) } #[test] - fn buffered_response_round_trips_and_restamps_model() { + fn aggregated_response_round_trips_and_restamps_model() -> Result<(), BoxError> { let body = json!({ "id": "1", "model": "upstream", @@ -206,18 +229,19 @@ mod tests { }], "usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3} }); - let agg = decode_buffered_response(&body, WireFormat::OpenAiChat).unwrap(); + let agg = decode_aggregated_response(&body, WireFormat::OpenAiChat)?; assert_eq!(completion_text(&agg), "Hi there"); // `requested_model` overrides the encoded model. let encoded = - encode_buffered_response(&agg, WireFormat::OpenAiChat, Some("client-facing")).unwrap(); + encode_aggregated_response(&agg, WireFormat::OpenAiChat, Some("client-facing"))?; assert_eq!(encoded["model"], "client-facing"); assert_eq!(encoded["choices"][0]["message"]["content"], "Hi there"); + Ok(()) } #[test] - fn encode_stream_emits_openai_chunks_whose_deltas_reassemble() { + fn encode_stream_reassembles_deltas_and_finishes() -> Result<(), BoxError> { let chunks: LlmResponseStream = stream::iter(vec![ Ok(LlmResponseChunk::TextDelta { index: 0, @@ -233,38 +257,100 @@ mod tests { ]) .boxed(); - let events: Vec = block_on( - encode_stream(chunks, WireFormat::OpenAiChat, Some("m".to_string())) - .map(|item| item.unwrap()) - .collect(), - ); + let events = block_on( + encode_stream(chunks, WireFormat::OpenAiChat, Some("m".to_string()))? + .collect::>(), + ) + .into_iter() + .collect::, BoxError>>()?; + let content: String = events .iter() .filter_map(|event| event["choices"][0]["delta"]["content"].as_str()) .collect(); assert_eq!(content, "Hello world"); + // The terminal chunk carries the stop reason through to the wire events. + assert!(events + .iter() + .any(|event| event["choices"][0]["finish_reason"] == "stop")); + Ok(()) + } + + #[test] + fn encode_stream_propagates_chunk_errors() -> Result<(), BoxError> { + let chunks: LlmResponseStream = stream::iter(vec![Err::( + "chunk exploded".into(), + )]) + .boxed(); + let results = + block_on(encode_stream(chunks, WireFormat::OpenAiChat, None)?.collect::>()); + assert!(results.iter().any(Result::is_err)); + Ok(()) } #[test] - fn decode_stream_parses_sse_bytes_into_ir_chunks() { + fn decode_stream_parses_sse_bytes_into_ir_chunks() -> Result<(), BoxError> { let sse = b"data: {\"choices\":[{\"delta\":{\"content\":\"Hello\"}}]}\n\n\ data: {\"choices\":[{\"delta\":{\"content\":\" world\"}}]}\n\n\ data: [DONE]\n\n" .to_vec(); let bytes = stream::once(async move { Ok::, BoxError>(sse) }); + let chunks = decode_all(bytes, WireFormat::OpenAiChat)?; + assert_eq!(text_of(&chunks), "Hello world"); + Ok(()) + } - let chunks: Vec = block_on( - decode_stream(bytes, WireFormat::OpenAiChat) - .map(|item| item.unwrap()) - .collect(), + #[test] + fn decode_stream_reassembles_frames_split_across_chunks() -> Result<(), BoxError> { + // A multi-byte codepoint and the frame boundaries are split across + // one-byte chunks; the BufReader must reassemble them losslessly. + let payload = json!({"choices": [{"delta": {"content": "café"}}]}); + let sse = format!("data: {payload}\n\ndata: [DONE]\n\n"); + let bytes = stream::iter( + sse.into_bytes() + .into_iter() + .map(|byte| Ok::, BoxError>(vec![byte])), ); - let content: String = chunks - .iter() - .filter_map(|chunk| match chunk { - LlmResponseChunk::TextDelta { text, .. } => Some(text.as_str()), - _ => None, - }) - .collect(); - assert_eq!(content, "Hello world"); + let chunks = decode_all(bytes, WireFormat::OpenAiChat)?; + assert_eq!(text_of(&chunks), "café"); + Ok(()) + } + + #[test] + fn decode_stream_decodes_trailing_frame_without_blank_line() -> Result<(), BoxError> { + // A non-standard upstream omits the final blank line; the last frame + // must still be decoded rather than dropped. + let sse = b"data: {\"choices\":[{\"delta\":{\"content\":\"tail\"}}]}".to_vec(); + let bytes = stream::once(async move { Ok::, BoxError>(sse) }); + let chunks = decode_all(bytes, WireFormat::OpenAiChat)?; + assert_eq!(text_of(&chunks), "tail"); + Ok(()) + } + + #[test] + fn decode_stream_decodes_crlf_delimited_frames() -> Result<(), BoxError> { + // CRLF framing: blank lines are `\r\n\r\n` and the bare `\r` must not + // block the frame boundary. + let sse = + b"data: {\"choices\":[{\"delta\":{\"content\":\"crlf\"}}]}\r\n\r\ndata: [DONE]\r\n\r\n" + .to_vec(); + let bytes = stream::once(async move { Ok::, BoxError>(sse) }); + let chunks = decode_all(bytes, WireFormat::OpenAiChat)?; + assert_eq!(text_of(&chunks), "crlf"); + Ok(()) + } + + #[test] + fn decode_stream_propagates_source_errors() -> Result<(), BoxError> { + // A transport error mid-stream surfaces as an error item, not a panic. + let bytes = stream::iter(vec![ + Ok::, BoxError>( + b"data: {\"choices\":[{\"delta\":{\"content\":\"x\"}}]}\n\n".to_vec(), + ), + Err::, BoxError>("upstream exploded".into()), + ]); + let results = block_on(decode_stream(bytes, WireFormat::OpenAiChat)?.collect::>()); + assert!(results.iter().any(Result::is_err)); + Ok(()) } } diff --git a/crates/switchyard-translation/src/sse.rs b/crates/switchyard-translation/src/sse.rs index 8643dd4f..06864baf 100644 --- a/crates/switchyard-translation/src/sse.rs +++ b/crates/switchyard-translation/src/sse.rs @@ -16,6 +16,7 @@ pub(crate) type BoxError = Box; /// The terminal SSE marker for `format`, if any. OpenAI Chat and Responses use /// `[DONE]`; Anthropic ends on its `message_stop` event with no marker. +#[inline] pub(crate) fn done_marker(format: WireFormat) -> Option<&'static str> { match format { WireFormat::OpenAiChat | WireFormat::OpenAiResponses => Some("[DONE]"), @@ -23,16 +24,6 @@ pub(crate) fn done_marker(format: WireFormat) -> Option<&'static str> { } } -/// Drains one complete SSE frame from the buffer when a boundary is present. -pub(crate) fn drain_next_sse_frame(buffer: &mut Vec) -> Result, BoxError> { - let Some((index, separator_len)) = next_sse_boundary(buffer) else { - return Ok(None); - }; - let frame: String = String::from_utf8_lossy(&buffer[..index]).to_string(); - buffer.drain(..index + separator_len); - Ok(Some(frame)) -} - pub(crate) fn parse_json_sse_frame( frame: &str, done_marker: Option<&str>, @@ -57,58 +48,65 @@ pub(crate) fn parse_json_sse_frame( Ok(Some(value)) } -/// Finds the next CRLF or LF SSE frame boundary. -fn next_sse_boundary(buffer: &[u8]) -> Option<(usize, usize)> { - match (find_bytes(buffer, b"\r\n\r\n"), find_bytes(buffer, b"\n\n")) { - (Some(crlf), Some(lf)) if crlf < lf => Some((crlf, 4)), - (Some(_), Some(lf)) => Some((lf, 2)), - (Some(crlf), None) => Some((crlf, 4)), - (None, Some(lf)) => Some((lf, 2)), - (None, None) => None, - } -} - -/// Finds a byte needle inside a byte haystack. -fn find_bytes(haystack: &[u8], needle: &[u8]) -> Option { - haystack - .windows(needle.len()) - .position(|window| window == needle) -} - #[cfg(test)] mod tests { use serde_json::json; use super::*; - // Multi-byte UTF-8 split across network chunks should wait for a full frame. + const DONE: Option<&str> = Some("[DONE]"); + #[test] - fn buffers_incomplete_utf8_until_a_complete_sse_frame_arrives() -> Result<(), BoxError> { - let mut buffer = b"data: {\"text\":\"".to_vec(); - let multibyte = "é".as_bytes(); - buffer.extend_from_slice(&multibyte[..1]); - assert!(drain_next_sse_frame(&mut buffer)?.is_none()); - - buffer.extend_from_slice(&multibyte[1..]); - buffer.extend_from_slice(b"\"}\n\n"); - - let Some(frame) = drain_next_sse_frame(&mut buffer)? else { - return Err("complete SSE frame should be drained".into()); - }; - let Some(value) = parse_json_sse_frame(&frame, None)? else { - return Err("SSE frame should parse as JSON".into()); - }; - assert_eq!(value, json!({"text": "é"})); - assert!(buffer.is_empty()); + fn parses_a_data_line_as_json() -> Result<(), BoxError> { + let value = + parse_json_sse_frame("data: {\"text\":\"hi\"}\n", DONE)?.ok_or("expected a payload")?; + assert_eq!(value, json!({"text": "hi"})); Ok(()) } #[test] - fn recognizes_done_marker() -> Result<(), BoxError> { - // The terminal marker carries no JSON payload. - if parse_json_sse_frame("data: [DONE]\n", Some("[DONE]"))?.is_some() { - return Err("DONE frame should not yield a JSON value".into()); - } + fn ignores_comment_and_non_data_fields() -> Result<(), BoxError> { + // Only `data:` fields contribute; comments (`:`) and other fields (`event:`) are dropped. + let frame = ": keep-alive\nevent: message\ndata: {\"n\":1}\n"; + let value = parse_json_sse_frame(frame, DONE)?.ok_or("expected a payload")?; + assert_eq!(value, json!({"n": 1})); Ok(()) } + + #[test] + fn done_marker_yields_no_payload() -> Result<(), BoxError> { + assert!(parse_json_sse_frame("data: [DONE]\n", DONE)?.is_none()); + Ok(()) + } + + #[test] + fn frame_without_data_yields_no_payload() -> Result<(), BoxError> { + // A comment-only frame carries no data payload. + assert!(parse_json_sse_frame(": keep-alive\n", DONE)?.is_none()); + // An empty frame likewise decodes to nothing. + assert!(parse_json_sse_frame("", DONE)?.is_none()); + Ok(()) + } + + #[test] + fn marker_is_only_terminal_when_configured() { + // Without a configured marker, `[DONE]` is treated as (invalid) JSON data. + assert!(parse_json_sse_frame("data: [DONE]\n", None).is_err()); + } + + #[test] + fn invalid_json_is_an_error() { + assert!(parse_json_sse_frame("data: {not json}\n", DONE).is_err()); + } + + #[test] + fn openai_formats_terminate_on_done() { + assert_eq!(done_marker(WireFormat::OpenAiChat), Some("[DONE]")); + assert_eq!(done_marker(WireFormat::OpenAiResponses), Some("[DONE]")); + } + + #[test] + fn anthropic_has_no_done_marker() { + assert_eq!(done_marker(WireFormat::AnthropicMessages), None); + } }