diff --git a/AGENTS.md b/AGENTS.md index 4a58654..29080f3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,6 +26,8 @@ navigation, invariants, verification, and handoff. - `src/storage/`: self-contained run directories, one self-contained message JSONL (including compacted-state boundaries), event JSONL, status, and final-result persistence. +- `src/storage/message_log/transcript.rs`: checkpoint-safe tail-first timeline + and exact NDJSON source used by the read-only inspector. - `src/trajectory.rs` and `src/trajectory/`: provider-neutral compacted-history search/read contracts plus the local message and artifact reader. - `src/skills/`: Agent Skills discovery and progressive `SKILL.md` loading. @@ -157,9 +159,11 @@ navigation, invariants, verification, and handoff. protects. For rare process crashes, prefer one durable authority, complete checkpoints, an explicit `interrupted` result, and model-directed retry over a cross-file state machine for transparent continuation. -- Do not add a TUI, frontend framework, built-in scheduler, vector database, - native dynamic plugin ABI, or distributed worker system without a concrete - request. +- Keep the requested transcript TUI behind fmtview's embedding facade. Fiasco + owns the checkpoint-aware source and command routing, not terminal rendering, + viewport/search logic, or the crossterm event loop. Do not add another + frontend framework, built-in scheduler, vector database, native dynamic + plugin ABI, or distributed worker system without a concrete request. - Prefer a readable module over speculative framework layers or defensive code for states the program cannot produce. diff --git a/Cargo.lock b/Cargo.lock index 1aaffce..ef1e757 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -82,6 +88,15 @@ version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" +[[package]] +name = "approx" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cab112f0a86d568ea0e627cc1d6be74a1e9cd55214684db5561995f6dad897c6" +dependencies = [ + "num-traits", +] + [[package]] name = "assert-json-diff" version = "2.0.2" @@ -165,6 +180,12 @@ version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "by_address" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64fa3c856b712db6612c019f14756e64e4bcea13337a6b33b696333a9eaa2d06" + [[package]] name = "bytemuck" version = "1.25.2" @@ -183,6 +204,15 @@ version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +[[package]] +name = "castaway" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dec551ab6e7578819132c713a93c022a05d60159dc86e7a7050223577484c55a" +dependencies = [ + "rustversion", +] + [[package]] name = "cc" version = "1.2.67" @@ -303,6 +333,29 @@ dependencies = [ "memchr", ] +[[package]] +name = "compact_str" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dfdd1c2274d9aa354115b09dc9a901d6c5576818cdf70d14cae2bdb47df00ab" +dependencies = [ + "castaway", + "cfg-if", + "itoa", + "rustversion", + "ryu", + "static_assertions", +] + +[[package]] +name = "convert_case" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "633458d4ef8c78b72454de2d54fd6ab2e60f9e02be22f3c6104cdc8a4e0fceb9" +dependencies = [ + "unicode-segmentation", +] + [[package]] name = "core-foundation" version = "0.10.1" @@ -346,6 +399,33 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossterm" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8b9f2e4c67f833b660cdb0a3523065869fb35570177239812ed4c905aeff87b" +dependencies = [ + "bitflags", + "crossterm_winapi", + "derive_more", + "document-features", + "mio", + "parking_lot", + "rustix", + "signal-hook", + "signal-hook-mio", + "winapi", +] + +[[package]] +name = "crossterm_winapi" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "acdd7c62a3665c7f6830a51635d9ac9b23ed385797f70a83bb8bafe9c572ab2b" +dependencies = [ + "winapi", +] + [[package]] name = "crypto-common" version = "0.1.7" @@ -408,6 +488,34 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b" +[[package]] +name = "deranged" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" + +[[package]] +name = "derive_more" +version = "2.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d751e9e49156b02b44f9c1815bcb94b984cdcc4396ecc32521c739452808b134" +dependencies = [ + "derive_more-impl", +] + +[[package]] +name = "derive_more-impl" +version = "2.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "799a97264921d8623a957f6c3b9011f3b5492f557bbb7a5a19b7fa6d06ba8dcb" +dependencies = [ + "convert_case", + "proc-macro2", + "quote", + "rustc_version", + "syn", +] + [[package]] name = "digest" version = "0.10.7" @@ -429,6 +537,15 @@ dependencies = [ "syn", ] +[[package]] +name = "document-features" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4b8a88685455ed29a21542a33abd9cb6510b6b129abadabdcef0f4c55bc8f61" +dependencies = [ + "litrs", +] + [[package]] name = "dunce" version = "1.0.5" @@ -441,6 +558,12 @@ version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" +[[package]] +name = "either" +version = "1.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" + [[package]] name = "equivalent" version = "1.0.2" @@ -468,6 +591,12 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "fast-srgb8" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd2e7510819d6fbf51a5545c8f922716ecfb14df168a3242f7d33e0239efe6a1" + [[package]] name = "fastrand" version = "2.4.1" @@ -493,6 +622,7 @@ dependencies = [ "chrono", "clap", "eventsource-stream", + "fmtview", "futures-util", "image", "libc", @@ -531,12 +661,46 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "fmtview" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d2798863a1347f246f212fbf7e294de1728d84d692c8496d9f169632b04f54b" +dependencies = [ + "anyhow", + "clap", + "crossterm", + "fmtview-core", + "ratatui", + "tempfile", +] + +[[package]] +name = "fmtview-core" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c281ad6eb69f32992a66bd07ee2c3c203faaeb17a43aad9bfb7166985404e02" +dependencies = [ + "anyhow", + "memchr", + "quick-xml", + "ratatui", + "tempfile", + "unicode-width", +] + [[package]] name = "fnv" version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -718,11 +882,27 @@ dependencies = [ "tracing", ] +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] + [[package]] name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] [[package]] name = "heck" @@ -996,7 +1176,29 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", - "hashbrown", + "hashbrown 0.17.1", +] + +[[package]] +name = "indoc" +version = "2.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79cf5c93f93228cf8efb3ba362535fb11199ac548a09ce117c9b1adc3030d706" +dependencies = [ + "rustversion", +] + +[[package]] +name = "instability" +version = "0.3.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eb2d60ef19920a3a9193c3e371f726ec1dafc045dac788d0fb3704272458971" +dependencies = [ + "darling", + "indoc", + "proc-macro2", + "quote", + "syn", ] [[package]] @@ -1011,6 +1213,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -1087,6 +1298,17 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "kasuari" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bde5057d6143cc94e861d90f591b9303d6716c6b9602309150bd068853c10899" +dependencies = [ + "hashbrown 0.16.1", + "portable-atomic", + "thiserror", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -1099,6 +1321,21 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libm" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" + +[[package]] +name = "line-clipping" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f50e8f47623268b5407192d26876c4d7f89d686ca130fdc53bced4814cd29f8" +dependencies = [ + "bitflags", +] + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -1111,6 +1348,12 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +[[package]] +name = "litrs" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11d3d7f243d5c5a8b9bb5d6dd2b1602c0cb0b9db1621bafc7ed66e35ff9fe092" + [[package]] name = "lock_api" version = "0.4.14" @@ -1126,6 +1369,15 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "lru" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b6180140927ee907000b0aa540091f6ea512ead4447c92b8fc35bc72788a5a6" +dependencies = [ + "hashbrown 0.17.1", +] + [[package]] name = "lru-slab" version = "0.1.2" @@ -1170,6 +1422,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" dependencies = [ "libc", + "log", "wasi", "windows-sys 0.61.2", ] @@ -1215,6 +1468,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-conv" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" + [[package]] name = "num-traits" version = "0.2.19" @@ -1234,6 +1493,15 @@ dependencies = [ "libc", ] +[[package]] +name = "num_threads" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c7398b9c8b70908f6371f47ed36737907c87c52af34c268fed0bf0ceb92ead9" +dependencies = [ + "libc", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -1252,6 +1520,30 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "palette" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cbf71184cc5ecc2e4e1baccdb21026c20e5fc3dcf63028a086131b3ab00b6e6" +dependencies = [ + "approx", + "fast-srgb8", + "libm", + "palette_derive", +] + +[[package]] +name = "palette_derive" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f5030daf005bface118c096f510ffb781fc28f9ab6a32ab224d8631be6851d30" +dependencies = [ + "by_address", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "parking_lot" version = "0.12.5" @@ -1312,6 +1604,12 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "portable-atomic" +version = "1.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d20d5497ef88037a52ff98267d066e7f11fcc5e99bbfbd58a42336193aacec3" + [[package]] name = "potential_utf" version = "0.1.5" @@ -1321,6 +1619,12 @@ dependencies = [ "zerovec", ] +[[package]] +name = "powerfmt" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -1359,6 +1663,15 @@ version = "0.1.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d55d956fa96f5ec02be2e13af0e20391a5aa83d6a074e3ad368959d0fab299ea" +[[package]] +name = "quick-xml" +version = "0.39.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdcc8dd4e2f670d309a5f0e83fe36dfdc05af317008fea29144da1a2ac858e5e" +dependencies = [ + "memchr", +] + [[package]] name = "quinn" version = "0.11.11" @@ -1492,6 +1805,72 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "ratatui" +version = "0.30.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3274ba0a2c5e1bcad2a2005d20f4dc59dad26b2eb0940fb094500dba4099d57d" +dependencies = [ + "instability", + "ratatui-core", + "ratatui-crossterm", + "ratatui-widgets", + "serde", +] + +[[package]] +name = "ratatui-core" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cbb175c433c8e28a809d1f5773a2ae96e68c0ce40db865cbab1020bf33ae479c" +dependencies = [ + "bitflags", + "compact_str", + "hashbrown 0.17.1", + "itertools", + "kasuari", + "lru", + "palette", + "serde", + "strum", + "thiserror", + "unicode-segmentation", + "unicode-truncate", + "unicode-width", +] + +[[package]] +name = "ratatui-crossterm" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "567584a3b0e6a8203c23de40b4861497266725eb5363dbfd18a1edd603cca9f0" +dependencies = [ + "cfg-if", + "crossterm", + "instability", + "ratatui-core", +] + +[[package]] +name = "ratatui-widgets" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66e3d19bcc9130ca376277d93b60767ff121ace3be06f5f95f81dd68956407d1" +dependencies = [ + "bitflags", + "hashbrown 0.17.1", + "indoc", + "instability", + "itertools", + "line-clipping", + "ratatui-core", + "serde", + "strum", + "time", + "unicode-segmentation", + "unicode-width", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1943,6 +2322,27 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-mio" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b75a19a7a740b25bc7944bdee6172368f988763b744e3d4dfe753f6b4ece40cc" +dependencies = [ + "libc", + "mio", + "signal-hook", +] + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -2003,12 +2403,39 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "static_assertions" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" + [[package]] name = "strsim" version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" +[[package]] +name = "strum" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +dependencies = [ + "strum_macros", +] + +[[package]] +name = "strum_macros" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "subtle" version = "2.6.1" @@ -2088,6 +2515,27 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "time" +version = "0.3.54" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e1d5e639ff6bab73cb6885cc7e7b1de96c3f32c68ec55f3952614bec1092244" +dependencies = [ + "deranged", + "libc", + "num-conv", + "num_threads", + "powerfmt", + "serde_core", + "time-core", +] + +[[package]] +name = "time-core" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" + [[package]] name = "tinystr" version = "0.8.3" @@ -2349,6 +2797,29 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-segmentation" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" + +[[package]] +name = "unicode-truncate" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16b380a1238663e5f8a691f9039c73e1cdae598a30e9855f541d29b08b53e9a5" +dependencies = [ + "itertools", + "unicode-segmentation", + "unicode-width", +] + +[[package]] +name = "unicode-width" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" + [[package]] name = "unsafe-libyaml" version = "0.2.11" @@ -2534,6 +3005,22 @@ version = "0.1.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a28ac98ddc8b9274cb41bb4d9d4d5c425b6020c50c46f25559911905610b4a88" +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + [[package]] name = "winapi-util" version = "0.1.11" @@ -2543,6 +3030,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows" version = "0.62.2" diff --git a/Cargo.toml b/Cargo.toml index 1ac2b67..9ad91c1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ chrono = { version = "0.4", features = ["serde"] } clap = { version = "4", features = ["derive", "env"] } eventsource-stream = "0.2" futures-util = "0.3" +fmtview = "0.6.2" image = { version = "0.25", default-features = false, features = ["bmp", "gif", "png"] } libc = "0.2" regex = "1" diff --git a/README.md b/README.md index 188c86b..cdf498f 100644 --- a/README.md +++ b/README.md @@ -7,9 +7,10 @@ delegated agents and long-running background jobs through one durable runtime for local automation and cloud workloads. The runtime deliberately shares one `AgentRunner` implementation and one tool -contract across root and child agents. Runs are portable, resumable, and -inspectable without an embedded TUI, database, distributed scheduler, sandbox, -or approval system. +contract across root and child agents. Runs are portable and resumable without +a database, distributed scheduler, sandbox, or approval system. Transcript +inspection embeds fmtview behind a storage-neutral boundary; the agent runtime +itself remains headless. ## Features @@ -33,6 +34,7 @@ or approval system. - run-local YAML planning graphs maintained with ordinary file tools - ordinary Markdown user/project memory maintained with normal file tools - self-contained run directories and optional NDJSON events +- tail-first transcript inspection for completed and running runs ## Important Security Boundary @@ -79,10 +81,26 @@ Runtime output is stored beneath the current project: tasks/ ``` -Inspect a previous result: +On an interactive terminal, inspect opens the current committed tail and loads +older checkpoints lazily: ```bash fiasco inspect +fiasco inspect --follow +``` + +`--follow` refreshes newly committed checkpoints while preserving fmtview's +attached, detached, and paused browsing states. A redirected `--follow` is +rejected because this mode requires an interactive terminal. + +When stdout is redirected, inspect instead writes exact checkpoint-safe JSONL. +The explicit form is useful in scripts. The legacy metadata/final-output view +remains available separately: + +```bash +fiasco inspect | jq -c . +fiasco inspect --output ndjson +fiasco inspect --summary ``` Resume an interrupted or failed run from its last complete checkpoint: @@ -140,6 +158,14 @@ before resuming appends. `run.json` declares freezes the stored profile plus remaining delegation depth. Compaction never rewrites or deletes committed trajectory records. +The interactive inspector delegates terminal lifecycle, JSON/chat rendering, +search, navigation, wrap, and follow state to the released `fmtview` embedding +facade. Fiasco owns only run lookup and a checkpoint-aware `RecordTimeline`; +the inspector never introduces ratatui/crossterm rendering or event-loop logic +into the runtime. Tail discovery scans backward from EOF, older loads move by +whole checkpoints, and refresh continues the shared incremental checkpoint +decoder from the last committed byte boundary. + Stable agent instructions are folded scalar values in the typed, compile-time `prompts/agents.yaml` registry. Every local model-facing tool adapter has a typed `tool.yaml` beside it; standalone tools live under `src/tools//`, diff --git a/docs/adr/0037-embed-fmtview-over-checkpoint-timeline.md b/docs/adr/0037-embed-fmtview-over-checkpoint-timeline.md new file mode 100644 index 0000000..b3bb15a --- /dev/null +++ b/docs/adr/0037-embed-fmtview-over-checkpoint-timeline.md @@ -0,0 +1,91 @@ +# ADR 0037: Embed fmtview Over a Checkpoint Timeline + +- Status: Accepted +- Date: 2026-07-22 +- Refines: ADR 0032 (read-only message-log inspection) +- Refines: ADR 0034 (checkpoint visibility boundary) + +## Context + +Operators need to inspect both completed and actively growing run transcripts. +Large histories must open at the useful tail without indexing or formatting the +whole file. A live viewer must not expose a torn physical line or any member of +an incomplete logical checkpoint, and writer recovery must not have a second, +slightly different definition of committed messages. + +fmtview already owns lazy formatting, search, structured navigation, wrapping, +follow attachment state, and terminal lifecycle. Copying those behaviors into +Fiasco would couple the orchestration runtime to ratatui/crossterm details. +Teaching fmtview about Fiasco run directories or checkpoint metadata would move +the storage boundary in the other direction. + +## Decision + +- Fiasco implements fmtview's backend-neutral `RecordTimeline` for one + `messages.jsonl`. It owns run lookup, run-state liveness, checkpoint-aware + loading, raw bytes, physical offsets, and reset epochs. +- fmtview owns all formatting, raw/formatted spools, viewport, search, + navigation, follow state, rendering, input polling, and terminal cleanup. + Fiasco depends on the released `fmtview` crate only, not `fmtview-core` and + not a path or Git revision. +- The source opens at the committed tail. It discovers the last complete group + by reverse physical-line scanning, then feeds that candidate forward through + the same `CheckpointDecoder` used by trajectory loading and append recovery. + Older reads repeat this by whole groups; newer reads continue forward from a + known sequence and byte boundary. A checkpoint can cross a load budget only + as the first group in a batch and is never partially published. +- Follow refresh retains the pending decoder, torn-line bytes, and scan cursor. + Bounded samples detect committed-prefix and pending-suffix changes. Rewriting + only an uncommitted tail rebuilds pending state from the same committed + boundary without duplicating records. Refresh retries a concurrently changing + file without publishing partial decoder progress. Committed truncation or + replacement starts a new record-id epoch. +- Queued, running, and idle run states are live. Completed, failed, cancelled, + and closed states are terminal. +- On a TTY, `fiasco inspect ` opens a tail-first snapshot and + `--follow` enables refresh. Redirected stdout defaults to exact committed + NDJSON; `--output ndjson` selects it explicitly. `--summary` keeps the former + metadata and final-output view. Redirected follow is rejected. +- Inspect dispatch occurs before application config/provider initialization. + Transcript reads do not require provider credentials, tools, MCP servers, + hooks, or a valid provider config. +- Events, child-run trees, and task control are not part of this source or the + first inspector surface. + +## Consequences + +- Completed and growing runs use one viewer without adding terminal backend + code to Fiasco or Fiasco storage types to fmtview. +- First screen work is proportional to the EOF checkpoint/suffix rather than + total history. Search and older navigation may validate earlier checkpoints + lazily as they are requested. +- NDJSON is safe for pipes and preserves provider tool-argument strings and all + other raw record bytes exactly. +- Reverse lazy validation can discover an old malformed checkpoint only when a + user navigates or searches into that portion. This is intentional; resume and + whole-trajectory loading continue to validate the complete committed prefix. +- Bounded identity samples are mutation detection for the supported sole-writer + workflow, not a cryptographic integrity scheme for hostile run directories. + +## Alternatives Considered + +- **Build a Fiasco-specific ratatui viewer.** Rejected because it duplicates + fmtview's rendering, search, navigation, follow state, and terminal cleanup. +- **Teach fmtview to decode `.fiasco` runs.** Rejected because a general viewer + should consume records, not own orchestration storage semantics. +- **Read the entire trajectory before opening.** Rejected because startup and + memory would scale with total history and defeat tail-first follow. +- **Expose newline-complete records immediately.** Rejected because a logical + assistant/tool turn would become visible in a partial state and disagree with + resume. +- **Create a separate persistent transcript index.** Rejected because reverse + checkpoint scans and lazy directional cursors meet the current access pattern + without another durability/recovery contract. + +## Related Documents + +- [Architecture](../architecture.md) +- [Entrypoints](../entrypoints.md) +- [Source map](../source-map.md) +- [ADR 0032](0032-self-contained-message-log.md) +- [ADR 0034](0034-atomic-turn-checkpoints.md) diff --git a/docs/adr/README.md b/docs/adr/README.md index fb85f0c..dc928bf 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -107,3 +107,5 @@ Use these sections: - [ADR 0035: Model delegated agents as reusable tasks](0035-reusable-agent-tasks.md) - [ADR 0036: Interrupt agent activities on process restart](0036-interrupt-agent-activities-on-restart.md) +- [ADR 0037: Embed fmtview over a checkpoint + timeline](0037-embed-fmtview-over-checkpoint-timeline.md) diff --git a/docs/architecture.md b/docs/architecture.md index e0b28bc..81f5eec 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -181,6 +181,48 @@ final line across EOF and streams forward without reading the whole file; trajectory loading collects its committed records, while append recovery only counts them and uses the same committed byte offset. +### Transcript inspection + +`fiasco inspect` is a read-only adapter over the same durable boundary: + +```text +inspect command + -> TranscriptTimeline (Fiasco storage/checkpoints/run state) + -> fmtview::view::RecordTimeline + -> fmtview (tail loading/search/navigation/render/terminal lifecycle) +``` + +The command is selected before application config and provider composition. +Summary and exact NDJSON paths therefore need only the run directory, and the +interactive path constructs no provider, MCP client, hooks, tools, or runner. +Fiasco depends on the released `fmtview` facade and does not depend directly on +`fmtview-core` or own ratatui/crossterm behavior. + +The timeline opens at the last committed checkpoint by scanning physical lines +backward from EOF. It loads older checkpoints backward and newer checkpoints +forward; one logical checkpoint may exceed a requested record/byte budget when +it is the first group in a batch, but is never split across published batches. +Prefix probes are bounded. The initial tail path does not index or validate +every earlier message, so a large history can show its first screen without a +forward scan. + +Follow refresh retains the shared `CheckpointDecoder`, its pending checkpoint, +a torn-line buffer, and the scanned suffix cursor. An unchanged large pending +checkpoint therefore costs only bounded head/middle/tail probes per refresh. +Rewriting only the uncommitted suffix invalidates and rebuilds this tracker from +the unchanged committed boundary without changing the epoch. Reads from a +concurrently shrinking or rewritten suffix are retried from a clean working +tracker and published only after one coherent file observation. Truncating a +committed prefix, replacing the file identity, or changing bounded committed +prefix probes starts a new epoch so fmtview discards old record identities. +Queued, running, and idle runs report a live boundary; completed, failed, +cancelled, and closed runs report a terminal boundary. + +Interactive snapshot and follow modes are TTY-only. Redirected stdout defaults +to exact committed NDJSON, preserving every raw record byte and LF without +reserialization; `--summary` retains the metadata/final-output view. Events, +child-run trees, and task controls are outside this transcript source. + The message file is created and directory-synced with the run. The writer's cached next sequence is invalidated before cancellable I/O and restored only after a complete record has synced. Multiple independent writers for one run @@ -436,17 +478,18 @@ root, `run_end` follows terminal completion; for a reusable child it follows each completed activity after the child becomes idle, not explicit close. Hooks do not define a second execution path. -## Headless Surface +## Runtime And Inspector Surfaces -The binary emits NDJSON runtime events for machines and a compact final result -for humans. There is no TUI or embedded web frontend. A future API or web client -should consume the same runtime events and run artifacts rather than introduce -model logic in the transport. +The execution path emits NDJSON runtime events for machines and a compact final +result for humans. The separate read-only inspect path can embed fmtview's TUI, +but no terminal state enters `AgentRunner` or storage contracts. A future API or +web client should consume the same runtime events and run artifacts rather than +introduce model logic in the transport. ## Deliberate Launch Omissions - OS sandbox and interactive approvals -- TUI or browser frontend +- browser frontend or a second transcript rendering stack - built-in scheduler - vector search - provider/server-side compaction diff --git a/docs/design-choices.md b/docs/design-choices.md index 66057c5..9d1c89e 100644 --- a/docs/design-choices.md +++ b/docs/design-choices.md @@ -33,13 +33,16 @@ healthy reasoning response; and an idle timeout alone, because heartbeat or trickle traffic could retain a model slot indefinitely. See [ADR 0011](adr/0011-model-stream-idle-timeout-and-request-deadline.md). -## Headless First +## Headless Runtime, Separate Inspector -The runtime emits structured events and portable artifacts. It does not contain -a TUI or web frontend. +The agent runtime emits structured events and portable artifacts without UI +state. The `inspect` command is a separate read-only adapter that embeds +fmtview's terminal frontend over a backend-neutral checkpoint timeline. Fiasco +does not own terminal rendering, viewport, search, or event-loop behavior. Rejected: preserving the legacy Bun/Hono/Ink surfaces. A future service can wrap -the Rust library and event stream without owning agent behavior. +the Rust library and event stream without owning agent behavior. Also rejected: +duplicating ratatui/crossterm logic inside the orchestration runtime. ## File-Based Runs diff --git a/docs/entrypoints.md b/docs/entrypoints.md index c3ca5e5..3353d03 100644 --- a/docs/entrypoints.md +++ b/docs/entrypoints.md @@ -6,7 +6,12 @@ reusable library. ## Commands - `fiasco run`: execute a task and persist its run directory. -- `fiasco inspect`: print run metadata and final output. +- `fiasco inspect`: open a tail-first snapshot of committed transcript records + on a TTY; redirected stdout writes exact committed NDJSON. +- `fiasco inspect --follow`: refresh a running transcript in the interactive + viewer. It requires a TTY. +- `fiasco inspect --output ndjson`: explicitly select checkpoint-safe NDJSON. +- `fiasco inspect --summary`: print run metadata and final output. - `fiasco auth login`: OpenAI OAuth device login. - `fiasco memory consolidate`: run model-driven user/project memory maintenance. - `fiasco skills list`: inspect discovered skill metadata. @@ -14,3 +19,9 @@ reusable library. `fiasco run --output ndjson` emits transport-neutral runtime events. A future HTTP or queue worker should call the library and forward the same events. It should not create another model/tool loop. + +Inspect is dispatched after workspace resolution but before application config, +provider, MCP, skill, hook, or model initialization. Reading a portable run +therefore does not require usable provider credentials or configuration. The +interactive path embeds fmtview through `fmtview::view`; Fiasco implements only +the backend-neutral record timeline and command selection. diff --git a/docs/source-map.md b/docs/source-map.md index 04581d8..84f75fa 100644 --- a/docs/source-map.md +++ b/docs/source-map.md @@ -55,6 +55,9 @@ - `src/storage/message_log/decoder.rs`: synchronous incremental checkpoint validation plus the forward async reader; it preserves exact raw lines and committed byte offsets without exposing partial groups. +- `src/storage/message_log/transcript.rs`: tail-first, bidirectional fmtview + timeline over complete checkpoints, incremental follow refresh, reset + detection, and exact committed-NDJSON streaming. - `src/storage/trajectory.rs`: classified append-only messages and compacted-history loading. - `src/skills/mod.rs`: Agent Skills metadata discovery and body/path loading; @@ -65,4 +68,5 @@ - `src/config.rs`: TOML configuration. - `src/events.rs`: runtime event contract and sinks. - `src/cli.rs`: CLI command schema. -- `src/main.rs`: headless composition root. +- `src/main.rs`: command composition root; inspect dispatch occurs before + provider/runtime initialization and embeds only the public fmtview facade. diff --git a/src/cli.rs b/src/cli.rs index 867d372..9e1d491 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -31,8 +31,19 @@ pub(crate) enum Command { #[arg(long, value_enum, default_value = "text")] output: OutputFormat, }, - /// Print persisted metadata and the final output for a run. - Inspect { run_id: String }, + /// Inspect a run's committed transcript or persisted summary. + Inspect { + run_id: String, + /// Continue following newly committed checkpoints. + #[arg(long, conflicts_with_all = ["output", "summary"])] + follow: bool, + /// Write checkpoint-safe records instead of opening the viewer. + #[arg(long, value_enum, conflicts_with_all = ["follow", "summary"])] + output: Option, + /// Print persisted metadata and final output. + #[arg(long, conflicts_with_all = ["follow", "output"])] + summary: bool, + }, /// Authenticate an OpenAI OAuth provider with the device-code flow. Auth { #[command(subcommand)] @@ -56,6 +67,11 @@ pub(crate) enum OutputFormat { Ndjson, } +#[derive(Debug, Clone, Copy, ValueEnum)] +pub(crate) enum InspectOutput { + Ndjson, +} + #[derive(Subcommand)] pub(crate) enum AuthCommand { Login, diff --git a/src/main.rs b/src/main.rs index 7c250bf..37a696f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,5 +1,6 @@ use std::{ env, + io::IsTerminal, path::{Path, PathBuf}, sync::Arc, }; @@ -22,13 +23,13 @@ use fiasco::{ openai_oauth::{DEFAULT_OPENAI_OAUTH_BASE_URL, OpenAiOAuthProvider}, }, skills::SkillRegistry, - storage::RunDirStore, + storage::{RunDirStore, TranscriptTimeline}, tools::{WebSearchTool, build_app_tools}, }; mod cli; -use cli::{AuthCommand, Cli, Command, MemoryCommand, OutputFormat, SkillsCommand}; +use cli::{AuthCommand, Cli, Command, InspectOutput, MemoryCommand, OutputFormat, SkillsCommand}; #[tokio::main] async fn main() -> Result<()> { @@ -38,6 +39,23 @@ async fn main() -> Result<()> { .init(); let cli = Cli::parse(); let workspace = dunce_canonicalize(&cli.workspace)?; + if let Command::Inspect { + run_id, + follow, + output, + summary, + } = &cli.command + { + return inspect_run( + &workspace, + run_id, + *follow, + *output, + *summary, + std::io::stdout().is_terminal(), + ) + .await; + } let config = AppConfig::load(&workspace, cli.config.as_deref())?; let fiasco_home = fiasco_home()?; @@ -48,7 +66,7 @@ async fn main() -> Result<()> { Command::Resume { run_id, output } => { resume_task(&workspace, &fiasco_home, config, run_id, output).await } - Command::Inspect { run_id } => inspect_run(&workspace, &run_id).await, + Command::Inspect { .. } => unreachable!("inspect is dispatched before config loading"), Command::Auth { command: AuthCommand::Login, } => login(&fiasco_home, &config).await, @@ -310,8 +328,32 @@ fn build_hooks(config: &AppConfig, workspace: &Path) -> Result { Ok(pipeline) } -async fn inspect_run(workspace: &Path, run_id: &str) -> Result<()> { +async fn inspect_run( + workspace: &Path, + run_id: &str, + follow: bool, + output: Option, + summary: bool, + stdout_is_terminal: bool, +) -> Result<()> { let store = RunDirStore::new(workspace); + if summary { + return inspect_summary(&store, run_id).await; + } + if follow && !stdout_is_terminal { + bail!("`fiasco inspect --follow` requires an interactive terminal on stdout"); + } + if output.is_some() || !stdout_is_terminal { + let mut stdout = tokio::io::stdout(); + return store.write_committed_ndjson(run_id, &mut stdout).await; + } + let source = TranscriptTimeline::open(&store, run_id)?; + let mut options = fmtview::view::ViewOptions::default(); + options.follow = follow; + fmtview::view::run(Box::new(source), options) +} + +async fn inspect_summary(store: &RunDirStore, run_id: &str) -> Result<()> { let run = store.load_run(run_id).await?; println!("{}", serde_json::to_string_pretty(&run)?); let final_path = store.paths(run_id).final_output; diff --git a/src/storage/message_log.rs b/src/storage/message_log.rs index 8eb4238..41cc931 100644 --- a/src/storage/message_log.rs +++ b/src/storage/message_log.rs @@ -14,6 +14,9 @@ use crate::{ }; pub(crate) mod decoder; +mod transcript; + +pub use transcript::TranscriptTimeline; use decoder::{CommittedCheckpointReader, CommittedRecord, DecodeResult}; diff --git a/src/storage/message_log/decoder.rs b/src/storage/message_log/decoder.rs index 435d863..71025a8 100644 --- a/src/storage/message_log/decoder.rs +++ b/src/storage/message_log/decoder.rs @@ -69,6 +69,33 @@ impl CheckpointDecoder { self.committed_end } + pub(crate) fn next_seq(&self) -> u64 { + self.next_seq + } + + /// Validate one complete line as the first record of the next checkpoint + /// without advancing decoder state. Directional readers use this to stop + /// before a checkpoint that cannot fit the remainder of a non-empty batch. + pub(crate) fn preflight_checkpoint_start( + &self, + path: &std::path::Path, + line_with_newline: &[u8], + ) -> Result { + ensure!( + self.pending.is_none(), + "checkpoint preflight requires a checkpoint boundary" + ); + ensure!( + line_with_newline.ends_with(b"\n"), + "checkpoint preflight requires a newline-terminated record" + ); + let stored = parse_stored_line(path, line_with_newline)?; + let checkpoint = stored.local.checkpoint.clone(); + self.validate_checkpoint_start(&stored, checkpoint.as_ref())?; + let _ = trajectory_record(stored, self.next_seq)?; + Ok(checkpoint.map_or(1, |checkpoint| checkpoint.count)) + } + pub(crate) fn push_complete_line( &mut self, path: &std::path::Path, @@ -109,28 +136,8 @@ impl CheckpointDecoder { "message `{}` has inconsistent checkpoint metadata", stored.message_ref ); - } else if let Some(checkpoint) = checkpoint.as_ref() { - ensure!( - checkpoint.count > 0, - "message checkpoint count must be positive" - ); - ensure!( - checkpoint.index == 0, - "message checkpoint `{}` starts at index {} instead of 0", - checkpoint.first_message_ref, - checkpoint.index - ); - ensure!( - checkpoint.first_message_ref == stored.message_ref, - "message checkpoint `{}` starts with message `{}`", - checkpoint.first_message_ref, - stored.message_ref - ); - usize::try_from(checkpoint.count) - .context("message checkpoint count does not fit in memory")?; - self.next_seq - .checked_add(checkpoint.count) - .context("message sequence overflow after checkpoint")?; + } else { + self.validate_checkpoint_start(&stored, checkpoint.as_ref())?; } let expected_seq = self .next_seq @@ -179,6 +186,37 @@ impl CheckpointDecoder { } } + fn validate_checkpoint_start( + &self, + stored: &super::StoredMessage, + checkpoint: Option<&MessageCheckpoint>, + ) -> Result<()> { + if let Some(checkpoint) = checkpoint { + ensure!( + checkpoint.count > 0, + "message checkpoint count must be positive" + ); + ensure!( + checkpoint.index == 0, + "message checkpoint `{}` starts at index {} instead of 0", + checkpoint.first_message_ref, + checkpoint.index + ); + ensure!( + checkpoint.first_message_ref == stored.message_ref, + "message checkpoint `{}` starts with message `{}`", + checkpoint.first_message_ref, + stored.message_ref + ); + usize::try_from(checkpoint.count) + .context("message checkpoint count does not fit in memory")?; + self.next_seq + .checked_add(checkpoint.count) + .context("message sequence overflow after checkpoint")?; + } + Ok(()) + } + fn commit_pending(&mut self, committed_end: u64) -> Result { let pending = self .pending diff --git a/src/storage/message_log/transcript.rs b/src/storage/message_log/transcript.rs new file mode 100644 index 0000000..501bd3b --- /dev/null +++ b/src/storage/message_log/transcript.rs @@ -0,0 +1,350 @@ +use std::{ + fs::File, + path::{Path, PathBuf}, +}; + +use anyhow::{Context, Result, bail, ensure}; +use fmtview::view::{ + RecordLoadLimit, RecordTimeline, TimelineRead, TimelineReadNext, TimelineRefresh, + TimelineSnapshot, +}; +use tokio::io::{AsyncWrite, AsyncWriteExt, BufReader as AsyncBufReader}; + +use super::{CommittedCheckpointReader, DecodeResult}; +use crate::storage::{RunDirStore, RunRecord, RunState, ensure_run_exists, validate_loaded_run}; + +mod refresh; +mod scan; + +use refresh::{FileIdentity, PrefixSample, SuffixTracker}; +use scan::{ + checkpoint_first_seq, checkpoint_next_seq, checkpoint_raw_bytes, decode_group_with_limit, + find_committed_tail, normalized_limit, read_forward_batch, timeline_records, +}; + +/// A checkpoint-safe, bidirectional view over one Fiasco transcript. +/// +/// Construction reads only the physical EOF suffix and the last logical +/// checkpoint. Both directional cursors start at that committed tail. +pub struct TranscriptTimeline { + label: String, + metadata_path: PathBuf, + messages_path: PathBuf, + file: File, + identity: FileIdentity, + epoch: u64, + state: RunState, + committed_end: u64, + committed_next_seq: u64, + observed_end: u64, + older_cursor: u64, + older_next_seq: u64, + newer_cursor: u64, + newer_next_seq: u64, + prefix_sample: PrefixSample, + suffix_tracker: SuffixTracker, + instrumentation: TranscriptInstrumentation, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +struct TranscriptInstrumentation { + bytes_read: u64, + read_operations: u64, + records_yielded: u64, +} + +impl TranscriptTimeline { + pub fn open(store: &RunDirStore, run_id: &str) -> Result { + let paths = store.paths(run_id); + Self::open_paths( + format!("fiasco run {run_id}"), + paths.metadata, + paths.messages, + 1, + ) + } + + fn open_paths( + label: String, + metadata_path: PathBuf, + messages_path: PathBuf, + epoch: u64, + ) -> Result { + let run = read_run(&metadata_path)?; + let mut file = File::open(&messages_path) + .with_context(|| format!("open transcript {}", messages_path.display()))?; + let metadata = file + .metadata() + .with_context(|| format!("stat transcript {}", messages_path.display()))?; + ensure!( + metadata.is_file(), + "transcript is not a file: {}", + messages_path.display() + ); + let observed_end = metadata.len(); + let mut instrumentation = TranscriptInstrumentation::default(); + let tail = find_committed_tail( + &mut file, + &messages_path, + observed_end, + &mut instrumentation, + )?; + let prefix_sample = + PrefixSample::read(&mut file, tail.committed_end, &mut instrumentation, &label)?; + let mut suffix_tracker = SuffixTracker::new(tail.committed_end, tail.next_seq); + suffix_tracker.scan_to( + &mut file, + &messages_path, + observed_end, + &mut instrumentation, + &label, + )?; + ensure!( + suffix_tracker.committed_end() == tail.committed_end, + "tail discovery and forward suffix validation disagree" + ); + + Ok(Self { + label, + metadata_path, + messages_path, + file, + identity: FileIdentity::from_metadata(&metadata), + epoch, + state: run.state, + committed_end: tail.committed_end, + committed_next_seq: tail.next_seq, + observed_end, + older_cursor: tail.committed_end, + older_next_seq: tail.next_seq, + newer_cursor: tail.committed_end, + newer_next_seq: tail.next_seq, + prefix_sample, + suffix_tracker, + instrumentation, + }) + } + + fn is_terminal(&self) -> bool { + is_terminal(self.state) + } + + fn boundary_next(&self) -> TimelineReadNext { + if self.is_terminal() { + TimelineReadNext::End + } else { + TimelineReadNext::Pending + } + } + + fn empty_boundary(&self) -> TimelineRead { + if self.is_terminal() { + TimelineRead::End + } else { + TimelineRead::Pending + } + } + + #[cfg(test)] + fn instrumentation(&self) -> TranscriptInstrumentation { + self.instrumentation + } +} + +impl RecordTimeline for TranscriptTimeline { + fn label(&self) -> &str { + &self.label + } + + fn snapshot(&self) -> TimelineSnapshot { + TimelineSnapshot { + epoch: self.epoch, + committed_end: self.committed_end, + observed_end: self.observed_end, + pending_bytes: self.observed_end.saturating_sub(self.committed_end), + } + } + + fn probe_prefix(&mut self, limit: RecordLoadLimit) -> Result { + if self.committed_end == 0 { + return Ok(self.empty_boundary()); + } + let batch = read_forward_batch( + &mut self.file, + &self.messages_path, + self.epoch, + 0, + 1, + self.committed_end, + normalized_limit(limit), + &mut self.instrumentation, + &self.label, + )?; + Ok(TimelineRead::Records { + records: batch.records, + next: if batch.cursor < self.committed_end { + TimelineReadNext::More + } else { + self.boundary_next() + }, + }) + } + + fn load_older(&mut self, limit: RecordLoadLimit) -> Result { + if self.older_cursor == 0 { + return Ok(TimelineRead::End); + } + let limit = normalized_limit(limit); + let mut cursor = self.older_cursor; + let mut expected_next_seq = self.older_next_seq; + let mut groups = Vec::new(); + let mut record_count = 0_usize; + let mut byte_count = 0_usize; + + while cursor > 0 { + let remaining = (!groups.is_empty()).then(|| { + RecordLoadLimit::new( + limit.max_records.saturating_sub(record_count), + limit.max_bytes.saturating_sub(byte_count), + ) + }); + let Some(group) = decode_group_with_limit( + &mut self.file, + &self.messages_path, + cursor, + remaining, + &mut self.instrumentation, + &self.label, + )? + else { + break; + }; + let checkpoint = match group.decode { + DecodeResult::Checkpoint(checkpoint) => checkpoint, + DecodeResult::NeedMore => { + bail!("incomplete checkpoint ends inside committed transcript at byte {cursor}") + } + }; + let first_seq = checkpoint_first_seq(&checkpoint)?; + let next_seq = checkpoint_next_seq(&checkpoint)?; + ensure!( + next_seq == expected_next_seq, + "checkpoint before byte {cursor} ends at m{}, expected m{}", + next_seq.saturating_sub(1), + expected_next_seq.saturating_sub(1) + ); + if group.start == 0 { + ensure!( + first_seq == 1, + "transcript starts at m{first_seq} instead of m1" + ); + } + let group_records = checkpoint.records.len(); + let group_bytes = checkpoint_raw_bytes(&checkpoint); + record_count = record_count.saturating_add(group_records); + byte_count = byte_count.saturating_add(group_bytes); + cursor = group.start; + expected_next_seq = first_seq; + groups.push(checkpoint); + if record_count >= limit.max_records || byte_count >= limit.max_bytes { + break; + } + } + + let mut records = Vec::with_capacity(record_count); + for checkpoint in groups.into_iter().rev() { + records.extend(timeline_records(self.epoch, checkpoint)); + } + self.older_cursor = cursor; + self.older_next_seq = expected_next_seq; + self.instrumentation.records_yielded = self + .instrumentation + .records_yielded + .saturating_add(records.len() as u64); + Ok(TimelineRead::Records { + records, + next: if cursor == 0 { + TimelineReadNext::End + } else { + TimelineReadNext::More + }, + }) + } + + fn load_newer(&mut self, limit: RecordLoadLimit) -> Result { + if self.newer_cursor >= self.committed_end { + return Ok(self.empty_boundary()); + } + let batch = read_forward_batch( + &mut self.file, + &self.messages_path, + self.epoch, + self.newer_cursor, + self.newer_next_seq, + self.committed_end, + normalized_limit(limit), + &mut self.instrumentation, + &self.label, + )?; + self.newer_cursor = batch.cursor; + self.newer_next_seq = batch.next_seq; + Ok(TimelineRead::Records { + records: batch.records, + next: if self.newer_cursor < self.committed_end { + TimelineReadNext::More + } else { + self.boundary_next() + }, + }) + } + + fn refresh(&mut self) -> Result { + self.refresh_timeline() + } +} + +impl RunDirStore { + /// Copy only complete committed checkpoints to an NDJSON sink. + /// + /// Record bytes, including their LF delimiters, are not reserialized. + pub async fn write_committed_ndjson(&self, run_id: &str, output: &mut W) -> Result<()> + where + W: AsyncWrite + Unpin, + { + self.load_run(run_id).await?; + let paths = self.paths(run_id); + ensure_run_exists(&paths).await?; + let file = tokio::fs::File::open(&paths.messages) + .await + .with_context(|| format!("open transcript {}", paths.messages.display()))?; + let mut reader = + CommittedCheckpointReader::new(AsyncBufReader::new(file), paths.messages.clone()); + while let DecodeResult::Checkpoint(checkpoint) = reader.next_checkpoint().await? { + for record in checkpoint.records { + output + .write_all(&record.raw) + .await + .context("write committed transcript NDJSON")?; + } + } + output.flush().await.context("flush transcript NDJSON") + } +} + +fn read_run(path: &Path) -> Result { + let bytes = std::fs::read(path).with_context(|| format!("read {}", path.display()))?; + let run: RunRecord = + serde_json::from_slice(&bytes).with_context(|| format!("parse {}", path.display()))?; + validate_loaded_run(&run)?; + Ok(run) +} + +fn is_terminal(state: RunState) -> bool { + matches!( + state, + RunState::Completed | RunState::Failed | RunState::Cancelled | RunState::Closed + ) +} + +#[cfg(test)] +mod tests; diff --git a/src/storage/message_log/transcript/refresh.rs b/src/storage/message_log/transcript/refresh.rs new file mode 100644 index 0000000..77d1a97 --- /dev/null +++ b/src/storage/message_log/transcript/refresh.rs @@ -0,0 +1,458 @@ +use std::{ + fs::{File, Metadata}, + io::{self, BufRead, BufReader, Read, Seek, SeekFrom}, + path::Path, + time::SystemTime, +}; + +use anyhow::{Context, Result, ensure}; +use fmtview::view::{RecordTimeline, TimelineRefresh, TimelineResetReason}; + +use super::{TranscriptInstrumentation, TranscriptTimeline, read_run, scan::read_range}; +use crate::storage::message_log::decoder::CheckpointDecoder; + +const SAMPLE_BYTES: usize = 64; +const REFRESH_SHORT_READ_ATTEMPTS: usize = 3; + +impl TranscriptTimeline { + pub(super) fn refresh_timeline(&mut self) -> Result { + self.refresh_timeline_with_hook(|_| Ok(())) + } + + pub(super) fn refresh_timeline_with_hook( + &mut self, + mut after_stat: impl FnMut(usize) -> Result<()>, + ) -> Result { + for attempt in 1..=REFRESH_SHORT_READ_ATTEMPTS { + let mut observation = None; + match self.refresh_once(&mut observation, || after_stat(attempt)) { + Ok(refresh) => return Ok(refresh), + Err(error) if is_unexpected_eof(&error) => { + let changed = observation.is_some_and(|previous| { + current_observation(&self.messages_path) + .is_ok_and(|current| current != previous) + }); + if changed { + if attempt < REFRESH_SHORT_READ_ATTEMPTS { + continue; + } + return Ok(TimelineRefresh::NoChange(self.snapshot())); + } + return Err(error); + } + Err(error) => return Err(error), + } + } + unreachable!("refresh retry loop always returns") + } + + fn refresh_once( + &mut self, + observation: &mut Option, + after_stat: impl FnOnce() -> Result<()>, + ) -> Result { + let run = read_run(&self.metadata_path)?; + let mut file = File::open(&self.messages_path) + .with_context(|| format!("reopen transcript {}", self.messages_path.display()))?; + let metadata = file + .metadata() + .with_context(|| format!("stat transcript {}", self.messages_path.display()))?; + *observation = Some(FileObservation::from_metadata(&metadata)); + after_stat()?; + let identity = FileIdentity::from_metadata(&metadata); + if identity != self.identity { + return self.reset(TimelineResetReason::IdentityChanged); + } + if metadata.len() < self.committed_end { + return self.reset(TimelineResetReason::Truncated); + } + if !self.prefix_sample.matches( + &mut file, + self.committed_end, + &mut self.instrumentation, + &self.label, + )? { + return self.reset(TimelineResetReason::Replaced); + } + + let old_committed_end = self.committed_end; + let suffix_matches = self.suffix_tracker.matches( + &mut file, + metadata.len(), + &mut self.instrumentation, + &self.label, + )?; + let mut suffix_tracker = if suffix_matches { + std::mem::replace( + &mut self.suffix_tracker, + SuffixTracker::new(self.committed_end, self.committed_next_seq), + ) + } else { + SuffixTracker::new(self.committed_end, self.committed_next_seq) + }; + if let Err(error) = suffix_tracker.scan_to( + &mut file, + &self.messages_path, + metadata.len(), + &mut self.instrumentation, + &self.label, + ) { + // The working tracker may have consumed bytes from a concurrent + // truncate. Keep only a clean committed-boundary tracker so a + // retry cannot inherit its partial cursor or decoder state. + self.suffix_tracker = SuffixTracker::new(self.committed_end, self.committed_next_seq); + return Err(error); + } + let committed_end = suffix_tracker.committed_end(); + let committed_next_seq = suffix_tracker.next_seq(); + let prefix_sample = if committed_end != old_committed_end { + Some(PrefixSample::read( + &mut file, + committed_end, + &mut self.instrumentation, + &self.label, + )?) + } else { + None + }; + let after_reads = current_observation(&self.messages_path)?; + if observation.is_some_and(|before_reads| after_reads != before_reads) { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!("transcript {} changed during refresh", self.label), + ) + .into()); + } + + // Publish the new snapshot only after every read against the statted + // length succeeds. A concurrent shrink leaves the prior decoder, + // partial line, cursor, and samples untouched for a clean retry. + self.file = file; + self.state = run.state; + self.observed_end = metadata.len(); + self.committed_end = committed_end; + self.committed_next_seq = committed_next_seq; + self.suffix_tracker = suffix_tracker; + if self.committed_end != old_committed_end { + self.prefix_sample = + prefix_sample.context("appended transcript lost its prefix sample")?; + return Ok(TimelineRefresh::Appended(self.snapshot())); + } + if self.is_terminal() { + Ok(TimelineRefresh::End(self.snapshot())) + } else if self.observed_end > self.committed_end { + Ok(TimelineRefresh::Pending(self.snapshot())) + } else { + Ok(TimelineRefresh::NoChange(self.snapshot())) + } + } + + fn reset(&mut self, reason: TimelineResetReason) -> Result { + let epoch = self + .epoch + .checked_add(1) + .context("transcript identity epoch overflow")?; + let replacement = Self::open_paths( + self.label.clone(), + self.metadata_path.clone(), + self.messages_path.clone(), + epoch, + )?; + *self = replacement; + Ok(TimelineRefresh::Reset { + reason, + snapshot: self.snapshot(), + }) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct FileObservation { + identity: FileIdentity, + len: u64, + modified: Option, +} + +impl FileObservation { + fn from_metadata(metadata: &Metadata) -> Self { + Self { + identity: FileIdentity::from_metadata(metadata), + len: metadata.len(), + modified: metadata.modified().ok(), + } + } +} + +fn current_observation(path: &Path) -> Result { + let metadata = std::fs::metadata(path) + .with_context(|| format!("restat transcript {} after short read", path.display()))?; + Ok(FileObservation::from_metadata(&metadata)) +} + +fn is_unexpected_eof(error: &anyhow::Error) -> bool { + error.chain().any(|cause| { + cause + .downcast_ref::() + .is_some_and(|error| error.kind() == io::ErrorKind::UnexpectedEof) + }) +} + +pub(super) struct SuffixTracker { + decoder: CheckpointDecoder, + scan_cursor: u64, + partial_line: Vec, + sample: RangeSample, +} + +impl SuffixTracker { + pub(super) fn new(committed_end: u64, next_seq: u64) -> Self { + Self { + decoder: CheckpointDecoder::new(next_seq, committed_end), + scan_cursor: committed_end, + partial_line: Vec::new(), + sample: RangeSample::empty(committed_end), + } + } + + pub(super) fn committed_end(&self) -> u64 { + self.decoder.committed_end() + } + + pub(super) fn next_seq(&self) -> u64 { + self.decoder.next_seq() + } + + fn matches( + &self, + file: &mut File, + observed_end: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, + ) -> Result { + if observed_end < self.scan_cursor { + return Ok(false); + } + self.sample.matches(file, instrumentation, label) + } + + pub(super) fn scan_to( + &mut self, + file: &mut File, + path: &Path, + observed_end: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, + ) -> Result<()> { + ensure!( + observed_end >= self.scan_cursor, + "transcript shrank below the pending suffix cursor" + ); + if observed_end > self.scan_cursor { + let mut reader = BufReader::new( + file.try_clone() + .with_context(|| format!("clone transcript {label}"))?, + ); + reader + .seek(SeekFrom::Start(self.scan_cursor)) + .with_context(|| format!("seek transcript {label}"))?; + while self.scan_cursor < observed_end { + let remaining = observed_end - self.scan_cursor; + let read = reader + .by_ref() + .take(remaining) + .read_until(b'\n', &mut self.partial_line) + .with_context(|| format!("refresh transcript {label}"))?; + instrumentation.read_operations = instrumentation.read_operations.saturating_add(1); + instrumentation.bytes_read = instrumentation.bytes_read.saturating_add(read as u64); + if read == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!( + "transcript {label} ended at byte {} before statted byte {observed_end}", + self.scan_cursor + ), + ) + .into()); + } + self.scan_cursor = self + .scan_cursor + .checked_add(read as u64) + .context("transcript byte offset overflow")?; + if !self.partial_line.ends_with(b"\n") { + break; + } + let line = std::mem::take(&mut self.partial_line); + let _ = self + .decoder + .push_complete_line(path, line, self.scan_cursor)?; + } + } + self.sample = RangeSample::read( + file, + self.decoder.committed_end(), + self.scan_cursor, + instrumentation, + label, + )?; + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct PrefixSample { + range: RangeSample, +} + +impl PrefixSample { + pub(super) fn read( + file: &mut File, + committed_end: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, + ) -> Result { + Ok(Self { + range: RangeSample::read(file, 0, committed_end, instrumentation, label)?, + }) + } + + fn matches( + &self, + file: &mut File, + committed_end: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, + ) -> Result { + ensure!( + self.range.end == committed_end, + "committed prefix sample boundary changed without refresh" + ); + self.range.matches(file, instrumentation, label) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct RangeSample { + end: u64, + head: Sample, + middle: Sample, + tail: Sample, +} + +impl RangeSample { + fn empty(offset: u64) -> Self { + let sample = Sample { + offset, + bytes: Vec::new(), + }; + Self { + end: offset, + head: sample.clone(), + middle: sample.clone(), + tail: sample, + } + } + + fn read( + file: &mut File, + start: u64, + end: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, + ) -> Result { + ensure!(end >= start, "sample range ends before it starts"); + if start == end { + return Ok(Self::empty(start)); + } + let len = end - start; + let sample_len = len.min(SAMPLE_BYTES as u64); + let middle_offset = start + len.saturating_sub(sample_len) / 2; + let tail_offset = end - sample_len; + Ok(Self { + end, + head: Sample::read(file, start, sample_len, instrumentation, label)?, + middle: Sample::read(file, middle_offset, sample_len, instrumentation, label)?, + tail: Sample::read(file, tail_offset, sample_len, instrumentation, label)?, + }) + } + + fn matches( + &self, + file: &mut File, + instrumentation: &mut TranscriptInstrumentation, + label: &str, + ) -> Result { + Ok(self.head.matches(file, instrumentation, label)? + && self.middle.matches(file, instrumentation, label)? + && self.tail.matches(file, instrumentation, label)?) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct Sample { + offset: u64, + bytes: Vec, +} + +impl Sample { + fn read( + file: &mut File, + offset: u64, + len: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, + ) -> Result { + Ok(Self { + offset, + bytes: read_range(file, offset, offset + len, instrumentation, label)?, + }) + } + + fn matches( + &self, + file: &mut File, + instrumentation: &mut TranscriptInstrumentation, + label: &str, + ) -> Result { + Ok(self.bytes + == read_range( + file, + self.offset, + self.offset + self.bytes.len() as u64, + instrumentation, + label, + )?) + } +} + +#[cfg(unix)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) struct FileIdentity { + device: u64, + inode: u64, +} + +#[cfg(unix)] +impl FileIdentity { + pub(super) fn from_metadata(metadata: &Metadata) -> Self { + use std::os::unix::fs::MetadataExt; + Self { + device: metadata.dev(), + inode: metadata.ino(), + } + } +} + +#[cfg(not(unix))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) struct FileIdentity { + created: Option, +} + +#[cfg(not(unix))] +impl FileIdentity { + pub(super) fn from_metadata(metadata: &Metadata) -> Self { + Self { + created: metadata.created().ok(), + } + } +} diff --git a/src/storage/message_log/transcript/scan.rs b/src/storage/message_log/transcript/scan.rs new file mode 100644 index 0000000..90da5bd --- /dev/null +++ b/src/storage/message_log/transcript/scan.rs @@ -0,0 +1,490 @@ +use std::{ + fs::File, + io::{BufRead, BufReader, Read, Seek, SeekFrom}, + path::Path, +}; + +use anyhow::{Context, Result, bail, ensure}; +use fmtview::view::{RecordId, RecordLoadLimit, TimelineRecord}; + +use crate::trajectory::message_ref_seq; + +use super::TranscriptInstrumentation; +use crate::storage::message_log::{ + DecodeResult, + decoder::{CheckpointDecoder, CommittedCheckpoint}, + parse_stored_line, +}; + +const REVERSE_SCAN_CHUNK_BYTES: usize = 64 * 1024; + +#[derive(Debug, Clone, Copy)] +pub(super) struct CommittedTail { + pub(super) committed_end: u64, + pub(super) next_seq: u64, +} + +pub(super) fn find_committed_tail( + file: &mut File, + path: &Path, + observed_end: u64, + instrumentation: &mut TranscriptInstrumentation, +) -> Result { + let mut cursor = find_complete_physical_end(file, observed_end, instrumentation, "transcript")?; + let mut incomplete_first_seq = None; + if cursor == 0 { + return Ok(CommittedTail { + committed_end: 0, + next_seq: 1, + }); + } + loop { + let group = decode_group_ending_at(file, path, cursor, instrumentation, "transcript")?; + match group.decode { + DecodeResult::Checkpoint(checkpoint) => { + let first_seq = checkpoint_first_seq(&checkpoint)?; + let next_seq = checkpoint_next_seq(&checkpoint)?; + if let Some(incomplete_first_seq) = incomplete_first_seq { + ensure!( + next_seq == incomplete_first_seq, + "incomplete tail starts at m{incomplete_first_seq} after m{}", + next_seq.saturating_sub(1) + ); + } + if group.start == 0 { + ensure!( + first_seq == 1, + "transcript starts at m{first_seq} instead of m1" + ); + } + return Ok(CommittedTail { + committed_end: cursor, + next_seq, + }); + } + DecodeResult::NeedMore => { + ensure!( + incomplete_first_seq.is_none(), + "more than one incomplete checkpoint appears at the transcript tail" + ); + incomplete_first_seq = Some(group.first_seq); + cursor = group.start; + if cursor == 0 { + ensure!( + group.first_seq == 1, + "transcript starts at m{} instead of m1", + group.first_seq + ); + return Ok(CommittedTail { + committed_end: 0, + next_seq: 1, + }); + } + } + } + } +} + +pub(super) struct DecodedGroup { + pub(super) start: u64, + pub(super) first_seq: u64, + pub(super) decode: DecodeResult, +} + +pub(super) fn decode_group_ending_at( + file: &mut File, + path: &Path, + end: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result { + decode_group_with_limit(file, path, end, None, instrumentation, label)? + .context("unlimited checkpoint decode was skipped") +} + +pub(super) fn decode_group_with_limit( + file: &mut File, + path: &Path, + end: u64, + remaining: Option, + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result> { + ensure!(end > 0, "cannot decode a checkpoint ending at byte zero"); + let last_start = find_start_of_lines(file, end, 1, instrumentation, label)?; + let last_line = read_range(file, last_start, end, instrumentation, label)?; + ensure!( + last_line.ends_with(b"\n"), + "checkpoint candidate at byte {end} lacks a terminating newline" + ); + let stored = parse_stored_line(path, &last_line)?; + let (group_line_count, declared_records) = match stored.local.checkpoint.as_ref() { + Some(checkpoint) => ( + checkpoint + .index + .checked_add(1) + .context("checkpoint index overflow")?, + checkpoint.count, + ), + None => (1, 1), + }; + if remaining.is_some_and(|limit| { + declared_records > u64::try_from(limit.max_records).unwrap_or(u64::MAX) + }) { + return Ok(None); + } + let start = find_start_of_lines(file, end, group_line_count, instrumentation, label)?; + if remaining + .is_some_and(|limit| end - start > u64::try_from(limit.max_bytes).unwrap_or(u64::MAX)) + { + return Ok(None); + } + let first_line_end = find_next_newline(file, start, end, instrumentation, label)?; + let first_line = read_range(file, start, first_line_end, instrumentation, label)?; + let first = parse_stored_line(path, &first_line)?; + let first_seq = message_ref_seq(&first.message_ref) + .with_context(|| format!("stored message has invalid ref `{}`", first.message_ref))?; + let mut decoder = CheckpointDecoder::new(first_seq, start); + let mut reader = BufReader::new( + file.try_clone() + .with_context(|| format!("clone transcript {label}"))?, + ); + reader + .seek(SeekFrom::Start(start)) + .with_context(|| format!("seek transcript {label}"))?; + let mut cursor = start; + let mut final_decode = DecodeResult::NeedMore; + while cursor < end { + let remaining = end - cursor; + let mut line = Vec::new(); + let read = reader + .by_ref() + .take(remaining) + .read_until(b'\n', &mut line) + .with_context(|| format!("read checkpoint from {label}"))?; + instrumentation.read_operations = instrumentation.read_operations.saturating_add(1); + instrumentation.bytes_read = instrumentation.bytes_read.saturating_add(read as u64); + ensure!( + read > 0, + "checkpoint at byte {cursor} made no read progress" + ); + cursor = cursor + .checked_add(read as u64) + .context("checkpoint byte offset overflow")?; + ensure!( + line.ends_with(b"\n"), + "checkpoint line ending at byte {cursor} is torn" + ); + final_decode = decoder.push_complete_line(path, line, cursor)?; + if matches!(final_decode, DecodeResult::Checkpoint(_)) { + ensure!( + cursor == end, + "checkpoint ending at byte {cursor} overlaps a later physical record" + ); + } + } + Ok(Some(DecodedGroup { + start, + first_seq, + decode: final_decode, + })) +} + +pub(super) struct ForwardBatch { + pub(super) records: Vec, + pub(super) cursor: u64, + pub(super) next_seq: u64, +} + +#[allow(clippy::too_many_arguments)] +pub(super) fn read_forward_batch( + file: &mut File, + path: &Path, + epoch: u64, + start: u64, + next_seq: u64, + end: u64, + limit: RecordLoadLimit, + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result { + let mut reader = BufReader::new( + file.try_clone() + .with_context(|| format!("clone transcript {label}"))?, + ); + reader + .seek(SeekFrom::Start(start)) + .with_context(|| format!("seek transcript {label}"))?; + let mut decoder = CheckpointDecoder::new(next_seq, start); + let mut cursor = start; + let mut records = Vec::new(); + let mut bytes = 0_usize; + while cursor < end { + let remaining = end - cursor; + let mut line = Vec::new(); + let read = reader + .by_ref() + .take(remaining) + .read_until(b'\n', &mut line) + .with_context(|| format!("read transcript {label}"))?; + instrumentation.read_operations = instrumentation.read_operations.saturating_add(1); + instrumentation.bytes_read = instrumentation.bytes_read.saturating_add(read as u64); + ensure!( + read > 0, + "transcript at byte {cursor} made no read progress" + ); + let line_end = cursor + .checked_add(read as u64) + .context("transcript byte offset overflow")?; + ensure!(line.ends_with(b"\n"), "committed transcript line is torn"); + if !records.is_empty() && decoder.committed_end() == cursor { + let declared_records = decoder.preflight_checkpoint_start(path, &line)?; + let remaining_records = limit.max_records.saturating_sub(records.len()); + let remaining_bytes = limit.max_bytes.saturating_sub(bytes); + if declared_records > u64::try_from(remaining_records).unwrap_or(u64::MAX) + || line.len() > remaining_bytes + { + break; + } + if !remaining_checkpoint_lines_fit( + file, + line_end, + end, + declared_records.saturating_sub(1), + remaining_bytes.saturating_sub(line.len()), + instrumentation, + label, + )? { + break; + } + // `File::try_clone` may share the underlying file offset with + // `file`. The bounded preflight seeks that handle, so discard any + // BufReader read-ahead and restore this reader's logical position. + reader.seek(SeekFrom::Start(line_end)).with_context(|| { + format!("restore transcript {label} after checkpoint preflight") + })?; + } + cursor = line_end; + if let DecodeResult::Checkpoint(checkpoint) = + decoder.push_complete_line(path, line, cursor)? + { + bytes = bytes.saturating_add(checkpoint_raw_bytes(&checkpoint)); + records.extend(timeline_records(epoch, checkpoint)); + if records.len() >= limit.max_records || bytes >= limit.max_bytes { + break; + } + } + } + ensure!( + decoder.committed_end() == cursor, + "committed transcript boundary splits a checkpoint at byte {cursor}" + ); + instrumentation.records_yielded = instrumentation + .records_yielded + .saturating_add(records.len() as u64); + Ok(ForwardBatch { + records, + cursor, + next_seq: decoder.next_seq(), + }) +} + +fn remaining_checkpoint_lines_fit( + file: &mut File, + start: u64, + committed_end: u64, + mut remaining_lines: u64, + byte_budget: usize, + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result { + if remaining_lines == 0 { + return Ok(true); + } + if start >= committed_end { + // Force the decoder to report the incomplete checkpoint instead of + // treating malformed metadata as a budget-limited boundary. + return Ok(true); + } + let budget_end = start + .saturating_add(u64::try_from(byte_budget).unwrap_or(u64::MAX)) + .min(committed_end); + let mut cursor = start; + let mut buffer = vec![0_u8; REVERSE_SCAN_CHUNK_BYTES]; + while cursor < budget_end { + let count = + usize::try_from((budget_end - cursor).min(buffer.len() as u64)).unwrap_or(buffer.len()); + read_exact_at(file, cursor, &mut buffer[..count], instrumentation, label)?; + for byte in &buffer[..count] { + if *byte == b'\n' { + remaining_lines -= 1; + if remaining_lines == 0 { + return Ok(true); + } + } + } + cursor = cursor + .checked_add(count as u64) + .context("transcript byte offset overflow")?; + } + if budget_end == committed_end { + // The declaration runs past the committed boundary. Decode it so the + // shared validator returns the structural error on this call. + Ok(true) + } else { + Ok(false) + } +} + +pub(super) fn timeline_records(epoch: u64, checkpoint: CommittedCheckpoint) -> Vec { + checkpoint + .records + .into_iter() + .map(|record| { + let end_offset = record.source_offset.saturating_add(record.raw.len() as u64); + TimelineRecord { + id: RecordId { + epoch, + start_offset: record.source_offset, + end_offset, + }, + raw: record.raw, + } + }) + .collect() +} + +pub(super) fn checkpoint_first_seq(checkpoint: &CommittedCheckpoint) -> Result { + let first = checkpoint + .records + .first() + .context("committed checkpoint contains no records")?; + Ok(first.trajectory.seq) +} + +pub(super) fn checkpoint_next_seq(checkpoint: &CommittedCheckpoint) -> Result { + let last = checkpoint + .records + .last() + .context("committed checkpoint contains no records")?; + last.trajectory + .seq + .checked_add(1) + .context("message sequence overflow after committed checkpoint") +} + +pub(super) fn checkpoint_raw_bytes(checkpoint: &CommittedCheckpoint) -> usize { + checkpoint + .records + .iter() + .fold(0_usize, |sum, record| sum.saturating_add(record.raw.len())) +} + +pub(super) fn normalized_limit(limit: RecordLoadLimit) -> RecordLoadLimit { + RecordLoadLimit::new(limit.max_records.max(1), limit.max_bytes.max(1)) +} + +fn find_complete_physical_end( + file: &mut File, + observed_end: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result { + let mut cursor = observed_end; + let mut buffer = vec![0_u8; REVERSE_SCAN_CHUNK_BYTES]; + while cursor > 0 { + let start = cursor.saturating_sub(buffer.len() as u64); + let count = usize::try_from(cursor - start).unwrap_or(buffer.len()); + read_exact_at(file, start, &mut buffer[..count], instrumentation, label)?; + if let Some(index) = buffer[..count].iter().rposition(|byte| *byte == b'\n') { + return Ok(start + index as u64 + 1); + } + cursor = start; + } + Ok(0) +} + +fn find_start_of_lines( + file: &mut File, + end: u64, + line_count: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result { + ensure!(line_count > 0, "checkpoint line count must be positive"); + let mut remaining = line_count; + let mut cursor = end + .checked_sub(1) + .context("checkpoint end precedes its terminating newline")?; + let mut buffer = vec![0_u8; REVERSE_SCAN_CHUNK_BYTES]; + while cursor > 0 { + let start = cursor.saturating_sub(buffer.len() as u64); + let count = usize::try_from(cursor - start).unwrap_or(buffer.len()); + read_exact_at(file, start, &mut buffer[..count], instrumentation, label)?; + for index in (0..count).rev() { + if buffer[index] == b'\n' { + remaining -= 1; + if remaining == 0 { + return Ok(start + index as u64 + 1); + } + } + } + cursor = start; + } + Ok(0) +} + +fn find_next_newline( + file: &mut File, + start: u64, + end: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result { + let mut cursor = start; + let mut buffer = vec![0_u8; REVERSE_SCAN_CHUNK_BYTES]; + while cursor < end { + let count = + usize::try_from((end - cursor).min(buffer.len() as u64)).unwrap_or(buffer.len()); + read_exact_at(file, cursor, &mut buffer[..count], instrumentation, label)?; + if let Some(index) = buffer[..count].iter().position(|byte| *byte == b'\n') { + return Ok(cursor + index as u64 + 1); + } + cursor = cursor + .checked_add(count as u64) + .context("transcript byte offset overflow")?; + } + bail!("checkpoint beginning at byte {start} has no complete first line") +} + +pub(super) fn read_range( + file: &mut File, + start: u64, + end: u64, + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result> { + let len = usize::try_from(end.saturating_sub(start)) + .context("transcript range does not fit in memory")?; + let mut bytes = vec![0_u8; len]; + read_exact_at(file, start, &mut bytes, instrumentation, label)?; + Ok(bytes) +} + +fn read_exact_at( + file: &mut File, + offset: u64, + buffer: &mut [u8], + instrumentation: &mut TranscriptInstrumentation, + label: &str, +) -> Result<()> { + file.seek(SeekFrom::Start(offset)) + .with_context(|| format!("seek transcript {label}"))?; + file.read_exact(buffer) + .with_context(|| format!("read transcript {label}"))?; + instrumentation.read_operations = instrumentation.read_operations.saturating_add(1); + instrumentation.bytes_read = instrumentation + .bytes_read + .saturating_add(buffer.len() as u64); + Ok(()) +} diff --git a/src/storage/message_log/transcript/tests.rs b/src/storage/message_log/transcript/tests.rs new file mode 100644 index 0000000..37201f5 --- /dev/null +++ b/src/storage/message_log/transcript/tests.rs @@ -0,0 +1,535 @@ +use std::{ + fs::OpenOptions as StdOpenOptions, + io::{Seek, SeekFrom, Write}, +}; + +use chrono::Utc; +use fmtview::view::{ + RecordLoadLimit, RecordTimeline, TimelineRead, TimelineRefresh, TimelineResetReason, +}; +use tempfile::TempDir; + +use super::*; +use crate::{ + artifact::ResultMetadata, + model::{Message, MessageContent, Role, ToolArguments}, + storage::message_log::{LocalState, MessageCheckpoint, StoredMessage}, + trajectory::message_ref, +}; + +mod review; + +async fn test_store(state: RunState) -> (TempDir, RunDirStore) { + let workspace = tempfile::tempdir().unwrap(); + let store = RunDirStore::new(workspace.path()); + let run = RunRecord::new( + "run-1", + "inspect", + "test-provider", + "test-model", + workspace.path().to_owned(), + None, + ); + store.create_run(&run).await.unwrap(); + store.update_state("run-1", state).await.unwrap(); + (workspace, store) +} + +fn raw_checkpoint(first_seq: u64, messages: Vec) -> Vec> { + let first_message_ref = message_ref(first_seq); + let count = messages.len() as u64; + messages + .into_iter() + .enumerate() + .map(|(index, message)| { + let mut raw = serde_json::to_vec(&StoredMessage { + message_ref: message_ref(first_seq + index as u64), + created_at: Utc::now(), + message, + local: LocalState { + checkpoint: Some(MessageCheckpoint { + first_message_ref: first_message_ref.clone(), + index: index as u64, + count, + }), + pending_input_id: None, + compaction: None, + }, + }) + .unwrap(); + raw.push(b'\n'); + raw + }) + .collect() +} + +fn append_raw(path: &Path, records: &[Vec]) { + let mut file = StdOpenOptions::new().append(true).open(path).unwrap(); + for record in records { + file.write_all(record).unwrap(); + } + file.flush().unwrap(); +} + +fn record_refs(read: TimelineRead) -> (Vec, TimelineReadNext) { + match read { + TimelineRead::Records { records, next } => ( + records + .into_iter() + .map(|record| { + serde_json::from_slice::(&record.raw).unwrap()["ref"] + .as_str() + .unwrap() + .to_owned() + }) + .collect(), + next, + ), + other => panic!("expected records, got {other:?}"), + } +} + +#[tokio::test] +async fn opens_at_tail_and_moves_by_whole_checkpoints_in_source_order() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "first")) + .await + .unwrap(); + store + .append_checkpoint( + "run-1", + &[ + Message::text(Role::Assistant, "call"), + Message::text(Role::User, "result one"), + Message::text(Role::User, "result two"), + ], + ) + .await + .unwrap(); + store + .append_message("run-1", &Message::text(Role::Assistant, "last")) + .await + .unwrap(); + + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + let (tail, next) = record_refs(timeline.load_older(RecordLoadLimit::new(1, 1)).unwrap()); + assert_eq!(tail, ["m5"]); + assert_eq!(next, TimelineReadNext::More); + + let (checkpoint, next) = record_refs(timeline.load_older(RecordLoadLimit::new(1, 1)).unwrap()); + assert_eq!(checkpoint, ["m2", "m3", "m4"]); + assert_eq!(next, TimelineReadNext::More); + + let (first, next) = record_refs( + timeline + .load_older(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + ); + assert_eq!(first, ["m1"]); + assert_eq!(next, TimelineReadNext::End); +} + +#[tokio::test] +async fn ignores_torn_lines_and_complete_but_incomplete_checkpoints() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "committed")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let incomplete = raw_checkpoint( + 2, + vec![ + Message::text(Role::Assistant, "call"), + Message::text(Role::User, "result"), + ], + ); + append_raw(&paths.messages, &incomplete[..1]); + let mut file = StdOpenOptions::new() + .append(true) + .open(&paths.messages) + .unwrap(); + file.write_all(b"{\"ref\":\"m3\"").unwrap(); + file.flush().unwrap(); + + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + assert!(timeline.snapshot().pending_bytes > 0); + let (records, next) = record_refs( + timeline + .load_older(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + ); + assert_eq!(records, ["m1"]); + assert_eq!(next, TimelineReadNext::End); +} + +#[tokio::test] +async fn pending_append_becomes_one_atomic_newer_batch() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "first")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let checkpoint = raw_checkpoint( + 2, + vec![ + Message::text(Role::Assistant, "call"), + Message::text(Role::User, "result"), + ], + ); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw(&paths.messages, &checkpoint[..1]); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Pending(_) + )); + assert!(matches!( + timeline + .load_newer(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + TimelineRead::Pending + )); + + append_raw(&paths.messages, &checkpoint[1..]); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Appended(_) + )); + let (records, next) = record_refs(timeline.load_newer(RecordLoadLimit::new(1, 1)).unwrap()); + assert_eq!(records, ["m2", "m3"]); + assert_eq!(next, TimelineReadNext::Pending); +} + +#[tokio::test] +async fn replacing_only_the_uncommitted_tail_does_not_reset_or_repeat() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "first")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let committed_end = std::fs::metadata(&paths.messages).unwrap().len(); + let discarded = raw_checkpoint( + 2, + vec![ + Message::text(Role::Assistant, "discarded"), + Message::text(Role::User, "missing"), + ], + ); + let replacement = raw_checkpoint(2, vec![Message::text(Role::Assistant, "replacement")]); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw(&paths.messages, &discarded[..1]); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Pending(_) + )); + + StdOpenOptions::new() + .write(true) + .open(&paths.messages) + .unwrap() + .set_len(committed_end) + .unwrap(); + append_raw(&paths.messages, &replacement); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Appended(_) + )); + assert_eq!(timeline.snapshot().epoch, 1); + let (records, _) = record_refs( + timeline + .load_newer(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + ); + assert_eq!(records, ["m2"]); + assert!(matches!( + timeline + .load_newer(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + TimelineRead::Pending + )); +} + +#[tokio::test] +async fn truncating_or_replacing_the_committed_prefix_resets_epoch() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "first")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let mut truncated = TranscriptTimeline::open(&store, "run-1").unwrap(); + StdOpenOptions::new() + .write(true) + .open(&paths.messages) + .unwrap() + .set_len(0) + .unwrap(); + assert!(matches!( + truncated.refresh().unwrap(), + TimelineRefresh::Reset { + reason: TimelineResetReason::Truncated, + .. + } + )); + assert_eq!(truncated.snapshot().epoch, 2); + + append_raw( + &paths.messages, + &raw_checkpoint(1, vec![Message::text(Role::User, "replacement")]), + ); + let mut replaced = TranscriptTimeline::open(&store, "run-1").unwrap(); + let temporary = paths.messages.with_extension("new"); + std::fs::write( + &temporary, + raw_checkpoint(1, vec![Message::text(Role::User, "new inode")]).concat(), + ) + .unwrap(); + std::fs::rename(temporary, &paths.messages).unwrap(); + assert!(matches!( + replaced.refresh().unwrap(), + TimelineRefresh::Reset { + reason: TimelineResetReason::IdentityChanged, + .. + } + )); +} + +#[tokio::test] +async fn same_inode_middle_rewrite_resets_as_replaced() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message( + "run-1", + &Message::text(Role::User, format!("before{}after", "a".repeat(2048))), + ) + .await + .unwrap(); + let paths = store.paths("run-1"); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + let bytes = std::fs::read(&paths.messages).unwrap(); + let middle = bytes.len() / 2; + assert_eq!(bytes[middle], b'a'); + let mut file = StdOpenOptions::new() + .write(true) + .open(&paths.messages) + .unwrap(); + file.seek(SeekFrom::Start(middle as u64)).unwrap(); + file.write_all(b"b").unwrap(); + file.flush().unwrap(); + + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Reset { + reason: TimelineResetReason::Replaced, + .. + } + )); +} + +#[tokio::test] +async fn terminal_state_turns_live_boundaries_into_end() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "first")) + .await + .unwrap(); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + store + .update_state("run-1", RunState::Completed) + .await + .unwrap(); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::End(_) + )); + assert!(matches!( + timeline + .load_newer(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + TimelineRead::End + )); +} + +#[tokio::test] +async fn raw_tool_arguments_and_lf_are_preserved() { + let (_workspace, store) = test_store(RunState::Completed).await; + let arguments = "{\n \"cmd\": \"cargo test\" \n}"; + store + .append_message( + "run-1", + &Message { + role: Role::Assistant, + content: vec![MessageContent::ToolCall { + id: "call-1".to_owned(), + name: "bash".to_owned(), + arguments: ToolArguments::from_raw(arguments), + }], + }, + ) + .await + .unwrap(); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + let read = timeline + .load_older(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(); + let TimelineRead::Records { records, .. } = read else { + panic!("expected records"); + }; + assert!(records[0].raw.ends_with(b"\n")); + let value: serde_json::Value = serde_json::from_slice(&records[0].raw).unwrap(); + assert_eq!(value["content"][0]["arguments"], arguments); +} + +#[tokio::test] +async fn prefix_probe_is_bounded_and_does_not_advance_tail_cursors() { + let (_workspace, store) = test_store(RunState::Running).await; + let paths = store.paths("run-1"); + let records = (1..=5000) + .flat_map(|seq| raw_checkpoint(seq, vec![Message::text(Role::User, "small")])) + .collect::>(); + append_raw(&paths.messages, &records); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + let before = timeline.instrumentation(); + let (prefix, _) = record_refs( + timeline + .probe_prefix(RecordLoadLimit::new(2, 1024)) + .unwrap(), + ); + let after = timeline.instrumentation(); + assert_eq!(prefix, ["m1", "m2"]); + assert!(after.bytes_read - before.bytes_read < 4096); + let (tail, _) = record_refs(timeline.load_older(RecordLoadLimit::new(1, 1)).unwrap()); + assert_eq!(tail, ["m5000"]); +} + +#[tokio::test] +async fn reverse_budget_does_not_decode_a_huge_checkpoint_before_a_small_tail() { + let (_workspace, store) = test_store(RunState::Running).await; + let paths = store.paths("run-1"); + let large = raw_checkpoint( + 1, + (0..1024) + .map(|index| { + Message::text(Role::User, format!("LARGE_{index:04}_{}", "x".repeat(4096))) + }) + .collect(), + ); + append_raw(&paths.messages, &large); + append_raw( + &paths.messages, + &raw_checkpoint(1025, vec![Message::text(Role::Assistant, "small tail")]), + ); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + + let before = timeline.instrumentation(); + let (tail, next) = record_refs( + timeline + .load_older(RecordLoadLimit::new(128, 4 * 1024 * 1024)) + .unwrap(), + ); + let after_tail = timeline.instrumentation(); + assert_eq!(tail, ["m1025"]); + assert_eq!(next, TimelineReadNext::More); + assert!(after_tail.bytes_read - before.bytes_read < 256 * 1024); + + let (large_group, next) = record_refs(timeline.load_older(RecordLoadLimit::new(1, 1)).unwrap()); + let after_large = timeline.instrumentation(); + assert_eq!(large_group.len(), 1024); + assert_eq!(large_group.first().unwrap(), "m1"); + assert_eq!(large_group.last().unwrap(), "m1024"); + assert_eq!(next, TimelineReadNext::End); + assert!(after_large.bytes_read - after_tail.bytes_read > 4 * 1024 * 1024); +} + +#[tokio::test] +async fn unchanged_large_incomplete_checkpoint_is_not_rescanned_each_refresh() { + let (_workspace, store) = test_store(RunState::Running).await; + let paths = store.paths("run-1"); + let checkpoint = raw_checkpoint( + 1, + (0..4000) + .map(|index| Message::text(Role::User, format!("pending-{index}"))) + .collect(), + ); + append_raw(&paths.messages, &checkpoint[..checkpoint.len() - 1]); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + let before = timeline.instrumentation(); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Pending(_) + )); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Pending(_) + )); + let unchanged = timeline.instrumentation(); + assert!(unchanged.bytes_read - before.bytes_read < 1024); + + append_raw(&paths.messages, &checkpoint[checkpoint.len() - 1..]); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Appended(_) + )); + let appended = timeline.instrumentation(); + assert!(appended.bytes_read - unchanged.bytes_read < 4096); + let (records, _) = record_refs(timeline.load_newer(RecordLoadLimit::new(1, 1)).unwrap()); + assert_eq!(records.len(), 4000); +} + +#[tokio::test] +async fn ndjson_writer_exposes_only_complete_checkpoints() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "committed")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let incomplete = raw_checkpoint( + 2, + vec![ + Message::text(Role::Assistant, "call"), + Message { + role: Role::Tool, + content: vec![MessageContent::ToolResult { + call_id: "call-1".to_owned(), + content: "result".to_owned(), + is_error: false, + metadata: ResultMetadata::empty(), + }], + }, + ], + ); + append_raw(&paths.messages, &incomplete[..1]); + let mut output = Vec::new(); + store + .write_committed_ndjson("run-1", &mut output) + .await + .unwrap(); + assert_eq!( + output, + std::fs::read(&paths.messages).unwrap()[..output.len()] + ); + assert_eq!(output.iter().filter(|byte| **byte == b'\n').count(), 1); +} + +#[tokio::test] +#[ignore = "million-record tail-open acceptance"] +async fn million_record_tail_open_reads_a_bounded_suffix() { + let (_workspace, store) = test_store(RunState::Running).await; + let paths = store.paths("run-1"); + let mut file = StdOpenOptions::new() + .append(true) + .open(&paths.messages) + .unwrap(); + for seq in 1..=1_000_000_u64 { + let record = raw_checkpoint(seq, vec![Message::text(Role::User, "x")]); + file.write_all(&record[0]).unwrap(); + } + file.flush().unwrap(); + let timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + assert_eq!(timeline.committed_next_seq, 1_000_001); + assert!(timeline.instrumentation().bytes_read < 512 * 1024); +} diff --git a/src/storage/message_log/transcript/tests/review.rs b/src/storage/message_log/transcript/tests/review.rs new file mode 100644 index 0000000..45a3864 --- /dev/null +++ b/src/storage/message_log/transcript/tests/review.rs @@ -0,0 +1,248 @@ +use super::*; + +#[tokio::test] +async fn concurrent_pending_truncate_retries_without_polluting_suffix_state() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "first")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let committed_end = std::fs::metadata(&paths.messages).unwrap().len(); + let discarded = raw_checkpoint( + 2, + vec![ + Message::text(Role::Assistant, format!("discarded-{}", "x".repeat(4096))), + Message::text(Role::User, "missing"), + ], + ); + let replacement = raw_checkpoint(2, vec![Message::text(Role::Assistant, "replacement")]); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw(&paths.messages, &discarded[..1]); + + let mut attempts = Vec::new(); + let refresh = timeline + .refresh_timeline_with_hook(|attempt| { + attempts.push(attempt); + if attempt == 1 { + StdOpenOptions::new() + .write(true) + .open(&paths.messages) + .unwrap() + .set_len(committed_end) + .unwrap(); + append_raw(&paths.messages, &replacement); + } + Ok(()) + }) + .unwrap(); + assert!(matches!(refresh, TimelineRefresh::Appended(_))); + assert_eq!(attempts, [1, 2]); + assert_eq!(timeline.snapshot().epoch, 1); + + let (records, next) = record_refs( + timeline + .load_newer(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + ); + assert_eq!(records, ["m2"]); + assert_eq!(next, TimelineReadNext::Pending); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::NoChange(_) + )); + assert!(matches!( + timeline + .load_newer(RecordLoadLimit::new(128, 1024 * 1024)) + .unwrap(), + TimelineRead::Pending + )); +} + +#[tokio::test] +async fn forward_budget_stops_before_a_huge_checkpoint_after_a_small_append() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "initial")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw( + &paths.messages, + &raw_checkpoint(2, vec![Message::text(Role::Assistant, "small append")]), + ); + append_raw( + &paths.messages, + &raw_checkpoint( + 3, + (0..256) + .map(|index| { + Message::text(Role::User, format!("large-{index}-{}", "x".repeat(4096))) + }) + .collect(), + ), + ); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Appended(_) + )); + + let before = timeline.instrumentation(); + let (small, next) = record_refs( + timeline + .load_newer(RecordLoadLimit::new(2, 1024 * 1024)) + .unwrap(), + ); + let after_small = timeline.instrumentation(); + assert_eq!(small, ["m2"]); + assert_eq!(next, TimelineReadNext::More); + assert!(after_small.bytes_read - before.bytes_read < 64 * 1024); + + let (large, next) = record_refs(timeline.load_newer(RecordLoadLimit::new(1, 1)).unwrap()); + assert_eq!(large.len(), 256); + assert_eq!(large.first().unwrap(), "m3"); + assert_eq!(large.last().unwrap(), "m258"); + assert_eq!(next, TimelineReadNext::Pending); +} + +#[tokio::test] +async fn forward_byte_preflight_restores_the_buffered_reader_position() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "initial")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw( + &paths.messages, + &raw_checkpoint(2, vec![Message::text(Role::Assistant, "small append")]), + ); + append_raw( + &paths.messages, + &raw_checkpoint( + 3, + vec![ + Message::text(Role::Assistant, "first in group"), + Message::text(Role::User, "x".repeat(32 * 1024)), + ], + ), + ); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Appended(_) + )); + + let (records, next) = record_refs( + timeline + .load_newer(RecordLoadLimit::new(4, 1024 * 1024)) + .unwrap(), + ); + assert_eq!(records, ["m2", "m3", "m4"]); + assert_eq!(next, TimelineReadNext::Pending); +} + +#[tokio::test] +async fn forward_byte_budget_scans_only_the_bounded_prefix_of_the_next_checkpoint() { + let (_workspace, store) = test_store(RunState::Running).await; + store + .append_message("run-1", &Message::text(Role::User, "initial")) + .await + .unwrap(); + let paths = store.paths("run-1"); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + append_raw( + &paths.messages, + &raw_checkpoint(2, vec![Message::text(Role::Assistant, "small append")]), + ); + append_raw( + &paths.messages, + &raw_checkpoint( + 3, + vec![ + Message::text(Role::Assistant, "small first line"), + Message::text(Role::User, "x".repeat(1024 * 1024)), + ], + ), + ); + assert!(matches!( + timeline.refresh().unwrap(), + TimelineRefresh::Appended(_) + )); + + let before = timeline.instrumentation(); + let (small, next) = record_refs( + timeline + .load_newer(RecordLoadLimit::new(10, 8 * 1024)) + .unwrap(), + ); + let after_small = timeline.instrumentation(); + assert_eq!(small, ["m2"]); + assert_eq!(next, TimelineReadNext::More); + assert!(after_small.bytes_read - before.bytes_read < 32 * 1024); + + let (large, next) = record_refs(timeline.load_newer(RecordLoadLimit::new(1, 1)).unwrap()); + assert_eq!(large, ["m3", "m4"]); + assert_eq!(next, TimelineReadNext::Pending); +} + +#[tokio::test] +async fn forward_preflight_rejects_malformed_checkpoint_start_before_budget_skip() { + let (_workspace, store) = test_store(RunState::Running).await; + let paths = store.paths("run-1"); + append_raw( + &paths.messages, + &raw_checkpoint(1, vec![Message::text(Role::Assistant, "valid prefix")]), + ); + let mut malformed = raw_checkpoint(2, vec![Message::text(Role::User, "malformed")]); + let mut value: serde_json::Value = serde_json::from_slice(&malformed[0]).unwrap(); + value["_fiasco"]["checkpoint"]["index"] = serde_json::json!(1_u64); + value["_fiasco"]["checkpoint"]["count"] = serde_json::json!(1_000_000_u64); + malformed[0] = serde_json::to_vec(&value).unwrap(); + malformed[0].push(b'\n'); + append_raw(&paths.messages, &malformed); + append_raw( + &paths.messages, + &raw_checkpoint(3, vec![Message::text(Role::Assistant, "valid tail")]), + ); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + + let error = timeline + .probe_prefix(RecordLoadLimit::new(2, 1024 * 1024)) + .unwrap_err(); + assert!(error.to_string().contains("starts at index 1 instead of 0")); +} + +#[tokio::test] +async fn skipped_malformed_older_checkpoint_makes_progress_then_errors() { + let (_workspace, store) = test_store(RunState::Running).await; + let paths = store.paths("run-1"); + let mut malformed = raw_checkpoint(1, vec![Message::text(Role::User, "malformed")]); + let mut value: serde_json::Value = serde_json::from_slice(&malformed[0]).unwrap(); + value["_fiasco"]["checkpoint"]["count"] = serde_json::json!(1_000_000_u64); + malformed[0] = serde_json::to_vec(&value).unwrap(); + malformed[0].push(b'\n'); + append_raw(&paths.messages, &malformed); + append_raw( + &paths.messages, + &raw_checkpoint(2, vec![Message::text(Role::Assistant, "tail")]), + ); + let mut timeline = TranscriptTimeline::open(&store, "run-1").unwrap(); + + let (tail, next) = record_refs( + timeline + .load_older(RecordLoadLimit::new(2, 1024 * 1024)) + .unwrap(), + ); + assert_eq!(tail, ["m2"]); + assert_eq!(next, TimelineReadNext::More); + let error = timeline + .load_older(RecordLoadLimit::new(2, 1024 * 1024)) + .unwrap_err(); + assert!( + error + .to_string() + .contains("incomplete checkpoint ends inside committed transcript") + ); +} diff --git a/src/storage/mod.rs b/src/storage/mod.rs index a457d6f..044a44b 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -20,6 +20,8 @@ mod input; mod message_log; mod trajectory; +pub use message_log::TranscriptTimeline; + pub const MESSAGE_FORMAT: &str = "fiasco-message"; const RUN_RECORD_VERSION: u32 = 10; @@ -258,21 +260,7 @@ impl RunDirStore { pub async fn load_run(&self, run_id: &str) -> Result { let run: RunRecord = read_json(&self.paths(run_id).metadata).await?; - ensure!( - run.version == RUN_RECORD_VERSION, - "unsupported run record version {}; expected {RUN_RECORD_VERSION}", - run.version - ); - ensure!( - run.message_format == MESSAGE_FORMAT, - "unsupported message format {}; expected {MESSAGE_FORMAT}", - run.message_format - ); - ensure!( - run.model_modalities.contains(&ModelModality::Text), - "run record model modalities must include text" - ); - validate_run_parentage(&run)?; + validate_loaded_run(&run)?; Ok(run) } @@ -312,6 +300,24 @@ impl RunDirStore { } } +pub(super) fn validate_loaded_run(run: &RunRecord) -> Result<()> { + ensure!( + run.version == RUN_RECORD_VERSION, + "unsupported run record version {}; expected {RUN_RECORD_VERSION}", + run.version + ); + ensure!( + run.message_format == MESSAGE_FORMAT, + "unsupported message format {}; expected {MESSAGE_FORMAT}", + run.message_format + ); + ensure!( + run.model_modalities.contains(&ModelModality::Text), + "run record model modalities must include text" + ); + validate_run_parentage(run) +} + fn validate_run_parentage(run: &RunRecord) -> Result<()> { match (run.profile.as_str(), run.parent_run_id.as_ref()) { ("root", None) => Ok(()), diff --git a/tests/inspect_cli.rs b/tests/inspect_cli.rs new file mode 100644 index 0000000..86f0d4c --- /dev/null +++ b/tests/inspect_cli.rs @@ -0,0 +1,132 @@ +use std::{fs::OpenOptions, io::Write, process::Command}; + +use fiasco::{ + model::{Message, Role}, + storage::{RunDirStore, RunRecord, RunState}, +}; +use tempfile::TempDir; + +async fn workspace_with_run() -> (TempDir, RunDirStore) { + let workspace = tempfile::tempdir().unwrap(); + let store = RunDirStore::new(workspace.path()); + let run = RunRecord::new( + "run-1", + "inspect me", + "test-provider", + "test-model", + workspace.path().to_owned(), + None, + ); + store.create_run(&run).await.unwrap(); + store + .append_message("run-1", &Message::text(Role::User, "hello")) + .await + .unwrap(); + store + .append_message("run-1", &Message::text(Role::Assistant, "goodbye")) + .await + .unwrap(); + store + .update_state("run-1", RunState::Completed) + .await + .unwrap(); + (workspace, store) +} + +fn fiasco(workspace: &TempDir) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_fiasco")); + command.args(["--workspace", workspace.path().to_str().unwrap()]); + command +} + +#[tokio::test] +async fn redirected_inspect_writes_exact_committed_ndjson_without_loading_config() { + let (workspace, store) = workspace_with_run().await; + let paths = store.paths("run-1"); + let committed = std::fs::read(&paths.messages).unwrap(); + let bad_config = workspace.path().join("bad.toml"); + std::fs::write(&bad_config, "this is not valid toml = [").unwrap(); + let mut file = OpenOptions::new() + .append(true) + .open(&paths.messages) + .unwrap(); + file.write_all(b"{\"ref\":\"m3\"").unwrap(); + file.flush().unwrap(); + + let output = fiasco(&workspace) + .args(["--config", bad_config.to_str().unwrap(), "inspect", "run-1"]) + .output() + .unwrap(); + + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + assert_eq!(output.stdout, committed); + for line in output + .stdout + .split(|byte| *byte == b'\n') + .filter(|line| !line.is_empty()) + { + serde_json::from_slice::(line).unwrap(); + } +} + +#[tokio::test] +async fn explicit_ndjson_matches_redirected_default() { + let (workspace, store) = workspace_with_run().await; + let expected = std::fs::read(store.paths("run-1").messages).unwrap(); + let output = fiasco(&workspace) + .args(["inspect", "run-1", "--output", "ndjson"]) + .output() + .unwrap(); + + assert!(output.status.success()); + assert_eq!(output.stdout, expected); +} + +#[tokio::test] +async fn summary_preserves_metadata_and_final_view() { + let (workspace, store) = workspace_with_run().await; + store.write_final("run-1", "FINAL_OK").await.unwrap(); + let output = fiasco(&workspace) + .args(["inspect", "run-1", "--summary"]) + .output() + .unwrap(); + + assert!(output.status.success()); + let stdout = String::from_utf8(output.stdout).unwrap(); + assert!(stdout.contains("\"id\": \"run-1\"")); + assert!(stdout.contains("\n--- final ---\nFINAL_OK\n")); +} + +#[tokio::test] +async fn conflicting_inspect_flags_are_rejected_by_clap() { + let (workspace, _store) = workspace_with_run().await; + for arguments in [ + &["inspect", "run-1", "--follow", "--summary"][..], + &["inspect", "run-1", "--follow", "--output", "ndjson"][..], + &["inspect", "run-1", "--summary", "--output", "ndjson"][..], + ] { + let output = fiasco(&workspace).args(arguments).output().unwrap(); + assert!(!output.status.success()); + assert!(String::from_utf8_lossy(&output.stderr).contains("cannot be used with")); + } +} + +#[tokio::test] +async fn redirected_follow_is_rejected_instead_of_degrading_to_snapshot_ndjson() { + let (workspace, _store) = workspace_with_run().await; + let output = fiasco(&workspace) + .args(["inspect", "run-1", "--follow"]) + .output() + .unwrap(); + + assert!(!output.status.success()); + assert!( + String::from_utf8_lossy(&output.stderr) + .contains("requires an interactive terminal on stdout") + ); + assert!(output.stdout.is_empty()); +}