diff --git a/harness/Cargo.lock b/harness/Cargo.lock index 653291eab..279d67e45 100644 --- a/harness/Cargo.lock +++ b/harness/Cargo.lock @@ -502,7 +502,7 @@ dependencies = [ [[package]] name = "harness" -version = "1.0.2" +version = "1.0.3" dependencies = [ "anyhow", "async-trait", diff --git a/workers-dev/Cargo.lock b/workers-dev/Cargo.lock new file mode 100644 index 000000000..52181dcee --- /dev/null +++ b/workers-dev/Cargo.lock @@ -0,0 +1,948 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys 0.61.2", +] + +[[package]] +name = "anyhow" +version = "1.0.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" + +[[package]] +name = "bitflags" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" + +[[package]] +name = "bytes" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ae3f5d315924270530207e2a68396c3cc547f6dca3fbdca317cfb1a51edb593" + +[[package]] +name = "cassowary" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df8670b8c7b9dae1793364eafadf7239c40d669904660c5960d74cfd80b46a53" + +[[package]] +name = "castaway" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dec551ab6e7578819132c713a93c022a05d60159dc86e7a7050223577484c55a" +dependencies = [ + "rustversion", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "clap" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "clap_lex" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" + +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + +[[package]] +name = "compact_str" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7fd622ebbb56a5b2ccb651b32b911cdeb2a9b4b11776b2473bf26a26a286244e" +dependencies = [ + "castaway", + "cfg-if", + "itoa", + "rustversion", + "ryu", + "static_assertions", +] + +[[package]] +name = "crossterm" +version = "0.28.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "829d955a0bb380ef178a640b91779e3987da38c9aea133b20614cfed8cdea9c6" +dependencies = [ + "bitflags", + "crossterm_winapi", + "mio", + "parking_lot", + "rustix 0.38.44", + "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 = "darling" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" +dependencies = [ + "darling_core", + "darling_macro", +] + +[[package]] +name = "darling_core" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0" +dependencies = [ + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn", +] + +[[package]] +name = "darling_macro" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" +dependencies = [ + "darling_core", + "quote", + "syn", +] + +[[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" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "fastrand" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" + +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + +[[package]] +name = "futures-core" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" + +[[package]] +name = "futures-sink" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", +] + +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] + +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "ident_case" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" + +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "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]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "libc" +version = "0.2.186" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" + +[[package]] +name = "linux-raw-sys" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab" + +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + +[[package]] +name = "log" +version = "0.4.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" + +[[package]] +name = "lru" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" +dependencies = [ + "hashbrown 0.15.5", +] + +[[package]] +name = "memchr" +version = "2.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" + +[[package]] +name = "mio" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" +dependencies = [ + "libc", + "log", + "wasi", + "windows-sys 0.61.2", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + +[[package]] +name = "paste" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "ratatui" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eabd94c2f37801c20583fc49dd5cd6b0ba68c716787c2dd6ed18571e1e63117b" +dependencies = [ + "bitflags", + "cassowary", + "compact_str", + "crossterm", + "indoc", + "instability", + "itertools", + "lru", + "paste", + "strum", + "unicode-segmentation", + "unicode-truncate", + "unicode-width 0.2.0", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "rustix" +version = "0.38.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys 0.4.15", + "windows-sys 0.59.0", +] + +[[package]] +name = "rustix" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys 0.12.1", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustversion" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" + +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "serde_yaml" +version = "0.9.34+deprecated" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" +dependencies = [ + "indexmap", + "itoa", + "ryu", + "serde", + "unsafe-libyaml", +] + +[[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" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[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.26.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" +dependencies = [ + "strum_macros", +] + +[[package]] +name = "strum_macros" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "rustversion", + "syn", +] + +[[package]] +name = "syn" +version = "2.0.118" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom", + "once_cell", + "rustix 1.1.4", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio" +version = "1.52.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "signal-hook-registry", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio-stream" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", + "tokio-util", +] + +[[package]] +name = "tokio-util" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "unicode-ident" +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 = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3644627a5af5fa321c95b9b235a72fd24cd29c648c2c379431e6628655627bf" +dependencies = [ + "itertools", + "unicode-segmentation", + "unicode-width 0.1.14", +] + +[[package]] +name = "unicode-width" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" + +[[package]] +name = "unicode-width" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fc81956842c57dac11422a97c3b8195a1ff727f06e85c84ed2e8aa277c9a0fd" + +[[package]] +name = "unsafe-libyaml" +version = "0.2.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" + +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[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-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "workers-dev" +version = "0.1.0" +dependencies = [ + "anyhow", + "clap", + "crossterm", + "ratatui", + "serde", + "serde_json", + "serde_yaml", + "tempfile", + "tokio", + "tokio-stream", +] + +[[package]] +name = "zmij" +version = "1.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" diff --git a/workers-dev/Cargo.toml b/workers-dev/Cargo.toml new file mode 100644 index 000000000..8b3274f89 --- /dev/null +++ b/workers-dev/Cargo.toml @@ -0,0 +1,26 @@ +[workspace] + +[package] +name = "workers-dev" +version = "0.1.0" +edition = "2021" +publish = false +description = "Local dev orchestrator for iii workers in this repo" + +[[bin]] +name = "workers-dev" +path = "src/main.rs" + +[dependencies] +anyhow = "1" +clap = { version = "4", features = ["derive"] } +crossterm = "0.28" +ratatui = "0.29" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +serde_yaml = "0.9" +tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal", "time", "io-util", "process"] } +tokio-stream = { version = "0.1", features = ["sync"] } + +[dev-dependencies] +tempfile = "3" diff --git a/workers-dev/README.md b/workers-dev/README.md new file mode 100644 index 000000000..c4d0628a2 --- /dev/null +++ b/workers-dev/README.md @@ -0,0 +1,107 @@ +# workers-dev + +Local dev orchestrator for all iii workers in this repo. Starts Rust binary workers via `cargo run`, shows a TUI dashboard with live engine connection status, and supports dependency-aware restarts. + +## Prerequisites + +1. **`iii` CLI** on `PATH`. +2. **Running iii engine** (this tool does not start the engine): + +```bash +iii -c harness/engine.config.yaml +``` + +3. **Provider API keys** for the harness stack: + +```bash +export ANTHROPIC_API_KEY=sk-ant-... +export OPENAI_API_KEY=sk-... +``` + +## Install + +```bash +cargo install --path workers-dev +``` + +## Worker groups + +Workers are **discovered automatically** from top-level `*/iii.worker.yaml` in this repo. + +| Group | Workers | Started by | +|-------|---------|------------| +| **harness stack** | `session-manager`, `llm-router`, `context-manager`, `provider-anthropic`, `provider-openai`, `approval-gate`, `harness` | `workers-dev up`, `Ctrl+u` in TUI, `workers-dev start` | +| **other** | All remaining repo workers (e.g. `telegram-bot`, `shell`, `console`, …) | `workers-dev start `, `workers-dev start --all`, `Ctrl+a` in TUI | + +Only **Rust `deploy: binary`** workers can be started with `cargo run`. Node/bundle workers show as `(iii worker add)` — install them via the iii registry instead. + +## Usage + +```bash +workers-dev up # start harness stack + TUI +workers-dev # TUI only +workers-dev start # harness stack (CLI, waits for connect) +workers-dev start --all # every discovered Rust worker +workers-dev start telegram-bot # one worker (+ deps) +workers-dev restart llm-router # rebuild + restart dependents +workers-dev logs harness -f +workers-dev status +``` + +Global flags: `--repo`, `--url`, `--port`, `--release`, `--config workers-dev.yaml`, `--stop-on-exit`, `--color auto|always|never`. + +Environment: `WORKERS_DEV_REPO` overrides repo auto-detection. Set `NO_COLOR` to disable colors (also respected when `--color auto`). + +## Colors + +Semantic colors are applied at render time (logs stay plain text in the ring buffer): + +- **Worker table**: green/yellow/red/gray for status, process, and engine columns +- **Log pane / `workers-dev logs -f`**: cargo progress (yellow), build done (cyan), errors (red), tracing levels (INFO/WARN/DEBUG) + +Use `--color never` or `NO_COLOR=1` to force plain output. Default `--color auto` enables colors on a TTY and disables them when stdout is piped. + +## TUI keys + +| Key | Action | +|-----|--------| +| `↑`/`↓` | Select worker (skips group headers) | +| `s` | Start selected worker | +| `x` | Stop selected worker | +| `r` | Restart selected worker + dependents | +| `l` | Follow logs | +| `Ctrl+u` | Start harness stack | +| `Ctrl+a` | Start all managed Rust workers | +| `q` | Quit | + +The log pane shows the last 20 lines for the **selected worker only** (cargo build output + worker logs). Lines are sanitized (no ANSI, no `\r` overwrite garbage). + +## Config (`workers-dev.yaml`) + +```yaml +repo: /path/to/workers +engine_url: ws://127.0.0.1:49134 +release: false +workers: # optional override; default = all discovered + - session-manager + - harness +harness_stack: # optional override; default = harness stack subset + - session-manager + - llm-router + - harness +color: auto # auto | always | never (respects NO_COLOR) +``` + +## Troubleshooting + +**Garbled log lines in the dashboard** + +Usually caused by cargo `\r` progress lines or ANSI color codes. Current versions normalize both. Reinstall: `cargo install --path workers-dev`. + +**Engine not reachable** + +Start the engine: `iii -c harness/engine.config.yaml` + +**Non-Rust worker won't start** + +Expected — use `iii worker add ` for JavaScript/bundle workers. diff --git a/workers-dev/src/color.rs b/workers-dev/src/color.rs new file mode 100644 index 000000000..15bcaebb8 --- /dev/null +++ b/workers-dev/src/color.rs @@ -0,0 +1,36 @@ +use std::io::IsTerminal; + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum ColorMode { + #[default] + Auto, + Always, + Never, +} + +impl ColorMode { + pub fn parse(s: &str) -> Option { + match s { + "auto" => Some(Self::Auto), + "always" => Some(Self::Always), + "never" => Some(Self::Never), + _ => None, + } + } + + pub fn enabled_for_stdout(&self) -> bool { + match self { + Self::Never => false, + Self::Always => true, + Self::Auto => std::env::var("NO_COLOR").is_err() && std::io::stdout().is_terminal(), + } + } + + /// TUI runs on an alternate screen; treat as a terminal unless explicitly disabled. + pub fn enabled_for_tui(&self) -> bool { + match self { + Self::Never => false, + Self::Always | Self::Auto => std::env::var("NO_COLOR").is_err(), + } + } +} diff --git a/workers-dev/src/commands/mod.rs b/workers-dev/src/commands/mod.rs new file mode 100644 index 000000000..39ba4b7df --- /dev/null +++ b/workers-dev/src/commands/mod.rs @@ -0,0 +1,93 @@ +use std::io::{self, Write}; +use std::sync::Arc; + +use anyhow::Result; + +use crate::logs; +use crate::orchestrator::Orchestrator; +use crate::status; + +pub async fn run_status(orchestrator: &Orchestrator) -> Result<()> { + let views = orchestrator.worker_views().await?; + status::print_status_table(&views); + Ok(()) +} + +pub async fn run_start(orchestrator: &Orchestrator, workers: Vec, all: bool) -> Result<()> { + if workers.is_empty() { + if all { + orchestrator.start_all_managed(true).await?; + println!("started all managed workers"); + } else { + orchestrator.start_harness_stack(true).await?; + println!("started harness stack"); + } + } else { + orchestrator.start_workers(&workers, true).await?; + for worker in &workers { + println!("started {worker}"); + } + } + Ok(()) +} + +pub async fn run_stop(orchestrator: &Orchestrator, workers: Vec) -> Result<()> { + orchestrator.stop_workers(&workers).await?; + Ok(()) +} + +pub async fn run_restart(orchestrator: &Orchestrator, worker: &str) -> Result<()> { + println!("restarting {worker} and dependents…"); + orchestrator.restart_worker(worker).await?; + println!("done"); + Ok(()) +} + +pub async fn run_logs( + orchestrator: Arc, + worker: String, + follow: bool, + lines: usize, +) -> Result<()> { + if !follow { + let tail = orchestrator.logs_tail(&worker, lines).await?; + let color_enabled = orchestrator.config.color_mode.enabled_for_stdout(); + for line in tail { + logs::print_colored_line(&line, color_enabled, &mut io::stdout())?; + } + return Ok(()); + } + + let tail = orchestrator.logs_tail(&worker, lines).await?; + let color_enabled = orchestrator.config.color_mode.enabled_for_stdout(); + for line in tail { + logs::print_colored_line(&line, color_enabled, &mut io::stdout())?; + } + io::stdout().flush()?; + + let mut rx = orchestrator.subscribe_logs(&worker).await?; + loop { + match rx.recv().await { + Ok(line) => { + logs::print_colored_line(&line, color_enabled, &mut io::stdout())?; + io::stdout().flush()?; + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + let proc = orchestrator.worker_views().await?; + match proc.iter().find(|v| v.name == worker) { + Some(v) if v.process_status == "running" || v.process_status == "compiling" => {} + _ => break, + } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + rx = orchestrator.subscribe_logs(&worker).await?; + } + } + } + Ok(()) +} + +pub async fn run_up(orchestrator: Arc) -> Result<()> { + orchestrator.start_harness_stack(false).await?; + crate::tui::run(orchestrator).await +} diff --git a/workers-dev/src/config.rs b/workers-dev/src/config.rs new file mode 100644 index 000000000..9b09fd7b2 --- /dev/null +++ b/workers-dev/src/config.rs @@ -0,0 +1,219 @@ +use std::path::{Path, PathBuf}; + +use anyhow::{bail, Context, Result}; +use serde::Deserialize; + +use crate::color::ColorMode; +use crate::discover::{discover_repo_workers, harness_stack_names, order_worker_names, WorkerSpec}; + +pub const DEFAULT_ENGINE_URL: &str = "ws://127.0.0.1:49134"; +pub const DEFAULT_POLL_INTERVAL_MS: u64 = 2000; +pub const DEFAULT_CONNECT_TIMEOUT_MS: u64 = 120_000; +pub const LOG_RING_CAPACITY: usize = 500; +pub const DEFAULT_LOG_TAIL: usize = 100; +pub const DASHBOARD_LOG_LINES: usize = 20; + +#[derive(Debug, Clone)] +pub struct Config { + pub repo_root: PathBuf, + pub engine_url: String, + pub engine_host: String, + pub engine_port: u16, + pub release: bool, + pub poll_interval_ms: u64, + pub connect_timeout_ms: u64, + pub workers: Vec, + pub harness_stack: Vec, + pub worker_specs: Vec, + pub stop_on_exit: bool, + pub color_mode: ColorMode, +} + +#[derive(Debug, Deserialize, Default)] +struct FileConfig { + repo: Option, + engine_url: Option, + release: Option, + poll_interval_ms: Option, + connect_timeout_ms: Option, + workers: Option>, + harness_stack: Option>, + stop_on_exit: Option, + color: Option, +} + +impl Config { + pub fn load( + repo: Option, + engine_url: Option, + port: Option, + release: bool, + config_path: Option, + stop_on_exit: bool, + color: Option, + ) -> Result { + let file_cfg = if let Some(path) = config_path { + let raw = std::fs::read_to_string(&path) + .with_context(|| format!("read config file {}", path.display()))?; + serde_yaml::from_str(&raw) + .with_context(|| format!("parse config file {}", path.display()))? + } else { + FileConfig::default() + }; + + let repo_root = resolve_repo_root(repo.or(file_cfg.repo))?; + let worker_specs = discover_repo_workers(&repo_root)?; + if worker_specs.is_empty() { + bail!("no workers discovered under {}", repo_root.display()); + } + + let engine_url = engine_url + .or(file_cfg.engine_url) + .unwrap_or_else(|| DEFAULT_ENGINE_URL.to_string()); + let (engine_host, engine_port) = parse_engine_url(&engine_url, port)?; + + let discovered_names = order_worker_names(&worker_specs); + let workers = file_cfg.workers.unwrap_or(discovered_names); + let harness_stack = file_cfg + .harness_stack + .unwrap_or_else(|| harness_stack_names(&worker_specs)); + + let color_mode = color + .or(file_cfg.color) + .as_deref() + .and_then(|s| { + ColorMode::parse(s).or_else(|| { + eprintln!("warning: invalid color mode {s:?}; using auto"); + None + }) + }) + .unwrap_or_default(); + + Ok(Self { + repo_root, + engine_url, + engine_host, + engine_port, + release: release || file_cfg.release.unwrap_or(false), + poll_interval_ms: file_cfg + .poll_interval_ms + .unwrap_or(DEFAULT_POLL_INTERVAL_MS), + connect_timeout_ms: file_cfg + .connect_timeout_ms + .unwrap_or(DEFAULT_CONNECT_TIMEOUT_MS), + workers, + harness_stack, + worker_specs, + stop_on_exit: stop_on_exit || file_cfg.stop_on_exit.unwrap_or(false), + color_mode, + }) + } + + pub fn worker_spec(&self, name: &str) -> Option<&WorkerSpec> { + self.worker_specs.iter().find(|s| s.name == name) + } +} + +pub fn resolve_repo_root(explicit: Option) -> Result { + if let Some(path) = explicit { + let canonical = path + .canonicalize() + .with_context(|| format!("repo path {}", path.display()))?; + validate_repo_root(&canonical)?; + return Ok(canonical); + } + + if let Ok(env) = std::env::var("WORKERS_DEV_REPO") { + let path = PathBuf::from(env); + let canonical = path + .canonicalize() + .with_context(|| format!("WORKERS_DEV_REPO={}", path.display()))?; + validate_repo_root(&canonical)?; + return Ok(canonical); + } + + let mut candidates = Vec::new(); + if let Ok(cwd) = std::env::current_dir() { + candidates.push(cwd); + } + candidates.push(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("..")); + + for start in candidates { + if let Some(root) = find_repo_root(&start) { + return Ok(root); + } + } + + bail!( + "could not find workers repo root (expected top-level */iii.worker.yaml); \ + pass --repo or set WORKERS_DEV_REPO" + ) +} + +fn find_repo_root(start: &Path) -> Option { + for ancestor in start.ancestors() { + if validate_repo_root(ancestor).is_ok() { + return ancestor.canonicalize().ok(); + } + } + None +} + +fn validate_repo_root(path: &Path) -> Result<()> { + let count = std::fs::read_dir(path) + .with_context(|| format!("read {}", path.display()))? + .filter_map(Result::ok) + .filter(|entry| entry.path().join("iii.worker.yaml").is_file()) + .count(); + if count > 0 { + Ok(()) + } else { + bail!("no top-level iii.worker.yaml workers under {}", path.display()) + } +} + +pub fn parse_engine_url(url: &str, port_override: Option) -> Result<(String, u16)> { + let stripped = url + .strip_prefix("ws://") + .or_else(|| url.strip_prefix("wss://")) + .unwrap_or(url); + + let authority = stripped.split('/').next().unwrap_or(stripped); + + let (host, port) = if let Some((host, port_str)) = authority.rsplit_once(':') { + let port: u16 = port_str + .parse() + .with_context(|| format!("invalid port in engine url {url}"))?; + (host.to_string(), port) + } else { + (authority.to_string(), 49134) + }; + + Ok((host, port_override.unwrap_or(port))) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_default_url() { + let (host, port) = parse_engine_url(DEFAULT_ENGINE_URL, None).unwrap(); + assert_eq!(host, "127.0.0.1"); + assert_eq!(port, 49134); + } + + #[test] + fn parse_url_with_trailing_slash() { + let (host, port) = parse_engine_url("ws://127.0.0.1:49134/", None).unwrap(); + assert_eq!(host, "127.0.0.1"); + assert_eq!(port, 49134); + } + + #[test] + fn parse_url_with_path() { + let (host, port) = parse_engine_url("ws://127.0.0.1:49134/ws", None).unwrap(); + assert_eq!(host, "127.0.0.1"); + assert_eq!(port, 49134); + } +} diff --git a/workers-dev/src/discover.rs b/workers-dev/src/discover.rs new file mode 100644 index 000000000..709bea640 --- /dev/null +++ b/workers-dev/src/discover.rs @@ -0,0 +1,182 @@ +use std::fs; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use serde::Deserialize; + +/// Core harness loop workers — shown first in the dashboard and started by `workers-dev up`. +pub const HARNESS_STACK: &[&str] = &[ + "session-manager", + "llm-router", + "context-manager", + "provider-anthropic", + "provider-openai", + "approval-gate", + "harness", +]; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkerGroup { + HarnessStack, + Other, +} + +impl WorkerGroup { + pub fn label(self) -> &'static str { + match self { + Self::HarnessStack => "harness stack", + Self::Other => "other", + } + } +} + +#[derive(Debug, Clone)] +pub enum SpawnKind { + CargoRun, + Unsupported { reason: String }, +} + +#[derive(Debug, Clone)] +pub struct WorkerSpec { + pub name: String, + pub dir: PathBuf, + pub group: WorkerGroup, + pub spawn: SpawnKind, +} + +#[derive(Debug, Deserialize)] +struct WorkerYaml { + name: Option, + language: Option, + deploy: Option, +} + +pub fn discover_repo_workers(repo_root: &Path) -> Result> { + let mut specs = Vec::new(); + for entry in fs::read_dir(repo_root) + .with_context(|| format!("read repo root {}", repo_root.display()))? + { + let entry = entry?; + if !entry.file_type()?.is_dir() { + continue; + } + let dir = entry.path(); + let folder = entry.file_name().to_string_lossy().into_owned(); + if folder.starts_with('.') { + continue; + } + let yaml_path = dir.join("iii.worker.yaml"); + if !yaml_path.is_file() { + continue; + } + + let raw = fs::read_to_string(&yaml_path) + .with_context(|| format!("read {}", yaml_path.display()))?; + let parsed: WorkerYaml = serde_yaml::from_str(&raw) + .with_context(|| format!("parse {}", yaml_path.display()))?; + let name = parsed.name.clone().unwrap_or(folder.clone()); + if name != folder { + anyhow::bail!( + "worker folder {folder} has iii.worker.yaml name={name} (mismatch)" + ); + } + + let group = if HARNESS_STACK.contains(&name.as_str()) { + WorkerGroup::HarnessStack + } else { + WorkerGroup::Other + }; + + let spawn = classify_spawn(&dir, &parsed); + specs.push(WorkerSpec { + name, + dir, + group, + spawn, + }); + } + + specs.sort_by(|a, b| { + group_rank(a.group) + .cmp(&group_rank(b.group)) + .then_with(|| a.name.cmp(&b.name)) + }); + Ok(specs) +} + +fn group_rank(group: WorkerGroup) -> u8 { + match group { + WorkerGroup::HarnessStack => 0, + WorkerGroup::Other => 1, + } +} + +fn classify_spawn(dir: &Path, yaml: &WorkerYaml) -> SpawnKind { + let language = yaml.language.as_deref().unwrap_or(""); + let deploy = yaml.deploy.as_deref().unwrap_or(""); + if language == "rust" && deploy == "binary" && dir.join("Cargo.toml").is_file() { + SpawnKind::CargoRun + } else { + SpawnKind::Unsupported { + reason: format!("{language}/{deploy} (use iii worker add for non-Rust workers)"), + } + } +} + +pub fn order_worker_names(specs: &[WorkerSpec]) -> Vec { + specs.iter().map(|s| s.name.clone()).collect() +} + +pub fn harness_stack_names(specs: &[WorkerSpec]) -> Vec { + HARNESS_STACK + .iter() + .filter_map(|name| { + specs + .iter() + .find(|s| s.name == *name) + .map(|s| s.name.clone()) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + use tempfile::TempDir; + + fn write_worker(tmp: &TempDir, name: &str, language: &str, deploy: &str, with_cargo: bool) { + let dir = tmp.path().join(name); + fs::create_dir_all(&dir).unwrap(); + fs::write( + dir.join("iii.worker.yaml"), + format!( + "iii: v1\nname: {name}\nlanguage: {language}\ndeploy: {deploy}\nmanifest: Cargo.toml\nbin: {name}\ndescription: test\n" + ), + ) + .unwrap(); + if with_cargo { + fs::write(dir.join("Cargo.toml"), "[workspace]\n").unwrap(); + } + } + + #[test] + fn discovers_and_groups_workers() { + let tmp = TempDir::new().unwrap(); + write_worker(&tmp, "harness", "rust", "binary", true); + write_worker(&tmp, "telegram-bot", "rust", "binary", true); + write_worker(&tmp, "claude-code", "javascript", "bundle", false); + + let specs = discover_repo_workers(tmp.path()).unwrap(); + assert_eq!(specs.len(), 3); + assert_eq!(specs[0].name, "harness"); + assert_eq!(specs[0].group, WorkerGroup::HarnessStack); + assert!(matches!(specs[0].spawn, SpawnKind::CargoRun)); + assert_eq!(specs[2].name, "telegram-bot"); + assert_eq!(specs[2].group, WorkerGroup::Other); + assert!(matches!( + specs.iter().find(|s| s.name == "claude-code").unwrap().spawn, + SpawnKind::Unsupported { .. } + )); + } +} diff --git a/workers-dev/src/graph.rs b/workers-dev/src/graph.rs new file mode 100644 index 000000000..8eb4fdecd --- /dev/null +++ b/workers-dev/src/graph.rs @@ -0,0 +1,333 @@ +use std::collections::{HashMap, HashSet, VecDeque}; +use std::path::Path; + +use anyhow::{bail, Context, Result}; +use serde::Deserialize; + +#[derive(Debug, Clone)] +pub struct WorkerGraph { + workers: Vec, + deps: HashMap>, +} + +#[derive(Debug, Deserialize)] +struct WorkerYaml { + name: Option, + dependencies: Option>, +} + +impl WorkerGraph { + pub fn load(repo_root: &Path, workers: &[String]) -> Result { + let managed: HashSet<&str> = workers.iter().map(String::as_str).collect(); + let mut deps = HashMap::new(); + + for worker in workers { + let yaml_path = repo_root.join(worker).join("iii.worker.yaml"); + let raw = std::fs::read_to_string(&yaml_path) + .with_context(|| format!("read {}", yaml_path.display()))?; + let parsed: WorkerYaml = serde_yaml::from_str(&raw) + .with_context(|| format!("parse {}", yaml_path.display()))?; + + if let Some(name) = &parsed.name { + if name != worker { + bail!( + "worker folder {worker} has iii.worker.yaml name={name} (mismatch)" + ); + } + } + + let worker_deps: Vec = parsed + .dependencies + .unwrap_or_default() + .into_keys() + .filter(|dep| managed.contains(dep.as_str())) + .collect(); + + deps.insert(worker.clone(), worker_deps); + } + + Ok(Self { + workers: workers.to_vec(), + deps, + }) + } + + pub fn workers(&self) -> &[String] { + &self.workers + } + + pub fn dependencies(&self, worker: &str) -> &[String] { + self.deps + .get(worker) + .map(Vec::as_slice) + .unwrap_or(&[]) + } + + /// Topological start order (dependencies before dependents). + pub fn topo_start_order(&self, subset: &[String]) -> Result> { + let mut deduped: Vec<&String> = Vec::new(); + let mut seen = HashSet::new(); + for worker in subset { + if !self.workers.iter().any(|w| w == worker) { + bail!("unknown worker {worker}"); + } + if seen.insert(worker.as_str()) { + deduped.push(worker); + } + } + let subset_set: HashSet<&str> = deduped.iter().map(|w| w.as_str()).collect(); + let mut in_degree: HashMap<&str, usize> = HashMap::new(); + let mut adj: HashMap<&str, Vec<&str>> = HashMap::new(); + + for worker in &deduped { + in_degree.entry(worker.as_str()).or_insert(0); + for dep in self.dependencies(worker) { + if !subset_set.contains(dep.as_str()) { + continue; + } + *in_degree.entry(worker.as_str()).or_insert(0) += 1; + adj.entry(dep.as_str()) + .or_default() + .push(worker.as_str()); + } + } + + let mut queue: VecDeque<&str> = in_degree + .iter() + .filter(|(_, deg)| **deg == 0) + .map(|(name, _)| *name) + .collect(); + queue.make_contiguous().sort_unstable(); + + let mut order = Vec::with_capacity(deduped.len()); + while let Some(node) = queue.pop_front() { + order.push(node.to_string()); + if let Some(children) = adj.get(node) { + let mut next = Vec::new(); + for child in children { + let entry = in_degree.get_mut(child).expect("child in subset"); + *entry -= 1; + if *entry == 0 { + next.push(*child); + } + } + next.sort_unstable(); + for child in next { + queue.push_back(child); + } + } + } + + if order.len() != deduped.len() { + bail!("dependency cycle detected among workers"); + } + + Ok(order) + } + + /// Reverse topological stop order (dependents before dependencies). + pub fn topo_stop_order(&self, subset: &[String]) -> Result> { + let mut order = self.topo_start_order(subset)?; + order.reverse(); + Ok(order) + } + + /// All managed workers that transitively depend on `worker` (excluding `worker`). + pub fn reverse_dependents(&self, worker: &str) -> Result> { + let mut seen = HashSet::new(); + let mut queue = VecDeque::from([worker.to_string()]); + + while let Some(current) = queue.pop_front() { + for candidate in &self.workers { + if self.dependencies(candidate).contains(¤t) + && seen.insert(candidate.clone()) + { + queue.push_back(candidate.clone()); + } + } + } + + seen.remove(worker); + let mut dependents: Vec = seen.into_iter().collect(); + dependents.sort_unstable(); + Ok(dependents) + } + + /// Closure for restart: worker + all transitive dependents, in start order. + pub fn restart_closure(&self, worker: &str) -> Result> { + if !self.workers.iter().any(|w| w == worker) { + bail!("unknown worker {worker}"); + } + let mut affected: Vec = vec![worker.to_string()]; + affected.extend(self.reverse_dependents(worker)?); + self.topo_start_order(&affected) + } + + /// Expand requested workers with missing dependencies (within managed set). + pub fn closure_with_deps(&self, requested: &[String]) -> Result> { + let mut seen = HashSet::new(); + let mut queue: VecDeque = requested.iter().cloned().collect(); + + while let Some(worker) = queue.pop_front() { + if !self.workers.iter().any(|w| w == &worker) { + bail!("unknown worker {worker}"); + } + if !seen.insert(worker.clone()) { + continue; + } + for dep in self.dependencies(&worker) { + if !seen.contains(dep) { + queue.push_back(dep.clone()); + } + } + } + + let subset: Vec = seen.into_iter().collect(); + self.topo_start_order(&subset) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + use tempfile::TempDir; + + fn write_worker_yaml(dir: &Path, name: &str, deps: &[&str]) { + let worker_dir = dir.join(name); + fs::create_dir_all(&worker_dir).unwrap(); + let mut yaml = format!( + "iii: v1\nname: {name}\nlanguage: rust\ndeploy: binary\nmanifest: Cargo.toml\nbin: {name}\ndescription: test\n" + ); + if !deps.is_empty() { + yaml.push_str("dependencies:\n"); + for dep in deps { + yaml.push_str(&format!(" {dep}: \"^1.0.0\"\n")); + } + } + fs::write(worker_dir.join("iii.worker.yaml"), yaml).unwrap(); + } + + fn fixture_repo() -> TempDir { + let tmp = TempDir::new().unwrap(); + write_worker_yaml(tmp.path(), "session-manager", &[]); + write_worker_yaml(tmp.path(), "llm-router", &[]); + write_worker_yaml(tmp.path(), "context-manager", &["llm-router"]); + write_worker_yaml(tmp.path(), "provider-anthropic", &["llm-router"]); + write_worker_yaml(tmp.path(), "provider-openai", &["llm-router"]); + write_worker_yaml(tmp.path(), "approval-gate", &["session-manager"]); + write_worker_yaml(tmp.path(), "harness", &[ + "session-manager", + "context-manager", + "approval-gate", + "provider-anthropic", + "provider-openai", + ]); + tmp + } + + #[test] + fn topo_start_order_respects_dependencies() { + let tmp = fixture_repo(); + let workers: Vec = [ + "session-manager", + "llm-router", + "context-manager", + "provider-anthropic", + "provider-openai", + "approval-gate", + "harness", + ] + .into_iter() + .map(str::to_string) + .collect(); + let graph = WorkerGraph::load(tmp.path(), &workers).unwrap(); + let order = graph.topo_start_order(&workers).unwrap(); + + fn idx(order: &[String], name: &str) -> usize { + order.iter().position(|w| w == name).unwrap() + } + + assert!(idx(&order, "llm-router") < idx(&order, "context-manager")); + assert!(idx(&order, "session-manager") < idx(&order, "approval-gate")); + assert!(idx(&order, "context-manager") < idx(&order, "harness")); + } + + #[test] + fn reverse_dependents_of_llm_router() { + let tmp = fixture_repo(); + let workers: Vec = [ + "session-manager", + "llm-router", + "context-manager", + "provider-anthropic", + "provider-openai", + "approval-gate", + "harness", + ] + .into_iter() + .map(str::to_string) + .collect(); + let graph = WorkerGraph::load(tmp.path(), &workers).unwrap(); + let deps = graph.reverse_dependents("llm-router").unwrap(); + assert!(deps.contains(&"context-manager".to_string())); + assert!(deps.contains(&"provider-anthropic".to_string())); + assert!(deps.contains(&"provider-openai".to_string())); + assert!(deps.contains(&"harness".to_string())); + } + + #[test] + fn restart_closure_includes_self_and_dependents() { + let tmp = fixture_repo(); + let workers: Vec = [ + "session-manager", + "llm-router", + "context-manager", + "provider-anthropic", + "provider-openai", + "approval-gate", + "harness", + ] + .into_iter() + .map(str::to_string) + .collect(); + let graph = WorkerGraph::load(tmp.path(), &workers).unwrap(); + let closure = graph.restart_closure("llm-router").unwrap(); + assert_eq!(closure.first().map(String::as_str), Some("llm-router")); + assert!(closure.contains(&"harness".to_string())); + } + + #[test] + fn topo_start_order_rejects_unknown_worker() { + let tmp = fixture_repo(); + let workers: Vec = ["session-manager", "llm-router"] + .into_iter() + .map(str::to_string) + .collect(); + let graph = WorkerGraph::load(tmp.path(), &workers).unwrap(); + let err = graph + .topo_start_order(&["session-manager".to_string(), "typo".to_string()]) + .unwrap_err(); + assert!(err.to_string().contains("unknown worker typo")); + } + + #[test] + fn topo_start_order_deduplicates_subset() { + let tmp = fixture_repo(); + let workers: Vec = ["session-manager", "llm-router"] + .into_iter() + .map(str::to_string) + .collect(); + let graph = WorkerGraph::load(tmp.path(), &workers).unwrap(); + let order = graph + .topo_start_order(&[ + "session-manager".to_string(), + "session-manager".to_string(), + "llm-router".to_string(), + ]) + .unwrap(); + assert_eq!(order.len(), 2); + assert!(order.contains(&"session-manager".to_string())); + assert!(order.contains(&"llm-router".to_string())); + } +} diff --git a/workers-dev/src/logs.rs b/workers-dev/src/logs.rs new file mode 100644 index 000000000..6a84e2320 --- /dev/null +++ b/workers-dev/src/logs.rs @@ -0,0 +1,272 @@ +use std::io::Write; + +use crossterm::style::{Color as CrosstermColor, ResetColor, SetForegroundColor}; +use ratatui::style::{Color, Modifier, Style}; +use ratatui::text::{Line, Span}; + +/// Normalize process output for TUI display and ring-buffer storage. +pub fn normalize_log_line(raw: &str) -> String { + let segment = raw + .rsplit('\r') + .next() + .unwrap_or(raw) + .trim_end(); + let stripped = strip_ansi(segment); + stripped.trim_end().to_string() +} + +pub fn format_for_display(line: &str, max_width: usize) -> String { + if max_width == 0 { + return String::new(); + } + truncate_chars(line, max_width) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LogKind { + CargoProgress, + CargoDone, + Error, + Warn, + Info, + Debug, + Plain, +} + +pub fn classify_log_line(line: &str) -> LogKind { + if line.contains("process exited") + || line.contains("could not compile") + || line.contains("error[E") + || line.contains("error:") + { + return LogKind::Error; + } + if line.contains("Compiling") + || line.contains("Building") + || line.contains("Downloading") + || line.contains("Updating") + || line.contains("Blocking waiting") + { + return LogKind::CargoProgress; + } + if line.contains("Finished `") && line.contains("profile") || line.contains("Running `target/") { + return LogKind::CargoDone; + } + if contains_level_token(line, "WARN") { + return LogKind::Warn; + } + if contains_level_token(line, "ERROR") { + return LogKind::Error; + } + if contains_level_token(line, "INFO") { + return LogKind::Info; + } + if contains_level_token(line, "DEBUG") || contains_level_token(line, "TRACE") { + return LogKind::Debug; + } + LogKind::Plain +} + +fn contains_level_token(line: &str, level: &str) -> bool { + line.contains(&format!(" {level} ")) + || line.contains(&format!(" {level}:")) + || line.contains(&format!(" {level}\t")) +} + +pub fn log_line_to_ratatui(line: &str, max_width: usize, color_enabled: bool) -> Line<'static> { + if max_width == 0 { + return Line::from(""); + } + + if !color_enabled { + return Line::from(format_for_display(line, max_width)); + } + + if let Some((ts, rest)) = split_tracing_timestamp(line) { + let kind = classify_log_line(line); + let mut spans = vec![ + Span::styled( + format_for_display(ts, max_width), + log_timestamp_style(true), + ), + Span::raw(" "), + ]; + let rest_width = max_width.saturating_sub(ts.chars().count().min(max_width) + 1); + spans.push(Span::styled( + format_for_display(rest, rest_width), + log_kind_style(kind, true), + )); + return Line::from(spans); + } + + let kind = classify_log_line(line); + Line::from(Span::styled( + format_for_display(line, max_width), + log_kind_style(kind, true), + )) +} + +fn log_kind_style(kind: LogKind, color_enabled: bool) -> Style { + if !color_enabled { + return Style::default(); + } + match kind { + LogKind::CargoProgress | LogKind::Warn => Style::default().fg(Color::Yellow), + LogKind::CargoDone => Style::default().fg(Color::Cyan), + LogKind::Error => Style::default().fg(Color::Red).add_modifier(Modifier::BOLD), + LogKind::Info => Style::default().fg(Color::Green), + LogKind::Debug => Style::default().fg(Color::DarkGray), + LogKind::Plain => Style::default(), + } +} + +fn log_timestamp_style(color_enabled: bool) -> Style { + if color_enabled { + Style::default().fg(Color::DarkGray) + } else { + Style::default() + } +} + +fn log_crossterm_color(kind: LogKind) -> Option { + use crossterm::style::Color as CrosstermColor; + match kind { + LogKind::CargoProgress | LogKind::Warn => Some(CrosstermColor::Yellow), + LogKind::CargoDone => Some(CrosstermColor::Cyan), + LogKind::Error => Some(CrosstermColor::Red), + LogKind::Info => Some(CrosstermColor::Green), + LogKind::Debug => Some(CrosstermColor::DarkGrey), + LogKind::Plain => None, + } +} + +fn write_crossterm_colored_line( + line: &str, + kind: LogKind, + out: &mut impl Write, +) -> std::io::Result<()> { + if let Some(color) = log_crossterm_color(kind) { + crossterm::execute!(out, SetForegroundColor(color))?; + write!(out, "{line}")?; + crossterm::execute!(out, ResetColor)?; + } else { + write!(out, "{line}")?; + } + Ok(()) +} + +pub fn print_colored_line( + line: &str, + color_enabled: bool, + out: &mut impl Write, +) -> std::io::Result<()> { + if !color_enabled { + writeln!(out, "{line}")?; + return Ok(()); + } + let kind = classify_log_line(line); + write_crossterm_colored_line(line, kind, out)?; + writeln!(out)?; + Ok(()) +} + +fn split_tracing_timestamp(line: &str) -> Option<(&str, &str)> { + if !line.starts_with("20") { + return None; + } + let sep = line.find(" ")?; + if !(20..=40).contains(&sep) { + return None; + } + let ts = &line[..sep]; + if !ts.contains('T') { + return None; + } + Some((ts, line[sep..].trim_start())) +} + +fn strip_ansi(input: &str) -> String { + let mut out = String::with_capacity(input.len()); + let mut chars = input.chars().peekable(); + while let Some(ch) = chars.next() { + if ch == '\x1b' { + for next in chars.by_ref() { + if next.is_ascii_alphabetic() { + break; + } + } + continue; + } + out.push(ch); + } + out +} + +fn truncate_chars(input: &str, max_width: usize) -> String { + let mut width = 0usize; + let mut out = String::new(); + for ch in input.chars() { + let w = unicode_width(ch); + if width + w > max_width { + break; + } + width += w; + out.push(ch); + } + out +} + +fn unicode_width(ch: char) -> usize { + if ch.is_ascii() { 1 } else { 2 } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn strips_ansi_and_carriage_return_overwrites() { + let raw = " Blocking waiting\r Blocking waiting for file lock\r Compiling harness v1.0.0"; + assert_eq!( + normalize_log_line(raw), + " Compiling harness v1.0.0" + ); + } + + #[test] + fn strips_color_codes() { + let raw = "\x1b[2m2026-06-22T15:47:56 INFO\x1b[0m harness: ready"; + assert_eq!(normalize_log_line(raw), "2026-06-22T15:47:56 INFO harness: ready"); + } + + #[test] + fn classify_cargo_and_tracing() { + assert_eq!( + classify_log_line(" Compiling harness v1.0.0"), + LogKind::CargoProgress + ); + assert_eq!( + classify_log_line(" Finished `dev` profile [unoptimized + debuginfo] target(s) in 9.21s"), + LogKind::CargoDone + ); + assert_eq!( + classify_log_line("2026-06-22T15:47:56.851170Z INFO harness: ready"), + LogKind::Info + ); + assert_eq!( + classify_log_line("2026-06-22T15:47:56.851170Z WARN harness::config: bad"), + LogKind::Warn + ); + assert_eq!( + classify_log_line("error[E0425]: cannot find value `foo`"), + LogKind::Error + ); + } + + #[test] + fn log_line_to_ratatui_respects_width() { + let line = log_line_to_ratatui(" Compiling harness v1.0.0", 10, true); + assert_eq!(line.spans.len(), 1); + assert_eq!(line.spans[0].content, " Compili"); + } +} diff --git a/workers-dev/src/main.rs b/workers-dev/src/main.rs new file mode 100644 index 000000000..7fb48cb0e --- /dev/null +++ b/workers-dev/src/main.rs @@ -0,0 +1,135 @@ +mod color; +mod commands; +mod config; +mod discover; +mod graph; +mod logs; +mod orchestrator; +mod runtime; +mod status; +mod tui; + +use std::path::PathBuf; +use std::sync::Arc; + +use anyhow::Result; +use clap::{Parser, Subcommand}; + +use crate::config::Config; +use crate::orchestrator::Orchestrator; + +#[derive(Parser, Debug)] +#[command( + name = "workers-dev", + about = "Local dev orchestrator for iii workers in this repo", + version +)] +struct Cli { + /// Workers repo root (auto-detected when omitted) + #[arg(long, global = true)] + repo: Option, + + /// Engine WebSocket URL + #[arg(long, global = true)] + url: Option, + + /// Engine WebSocket port (overrides port in --url) + #[arg(long, global = true)] + port: Option, + + /// Run workers with cargo run --release + #[arg(long, global = true)] + release: bool, + + /// Optional workers-dev.yaml config file + #[arg(long, global = true)] + config: Option, + + /// Stop all managed workers when the process exits + #[arg(long, global = true)] + stop_on_exit: bool, + + /// Color output: auto, always, or never (respects NO_COLOR) + #[arg(long, global = true, default_value = "auto")] + color: String, + + #[command(subcommand)] + command: Option, +} + +#[derive(Subcommand, Debug)] +enum Command { + /// Start the harness stack and open the TUI dashboard + Up, + /// Start workers (default: harness stack; use --all for every worker) + Start { + #[arg(value_name = "WORKER")] + workers: Vec, + /// Start every discovered worker instead of the harness stack + #[arg(long)] + all: bool, + }, + /// Stop workers (default: all managed) + Stop { + #[arg(value_name = "WORKER")] + workers: Vec, + }, + /// Rebuild and restart a worker, then restart its dependents + Restart { + worker: String, + }, + /// Print worker logs from the local ring buffer + Logs { + worker: String, + /// Follow log output + #[arg(short = 'f', long)] + follow: bool, + /// Number of lines to show + #[arg(short = 'n', long, default_value_t = crate::config::DEFAULT_LOG_TAIL)] + lines: usize, + }, + /// Print one-shot status table (no TUI) + Status, +} + +#[tokio::main] +async fn main() -> Result<()> { + let cli = Cli::parse(); + let config = Config::load( + cli.repo, + cli.url, + cli.port, + cli.release, + cli.config, + cli.stop_on_exit, + Some(cli.color), + )?; + + let orchestrator = Arc::new(Orchestrator::new(config)?); + + let result = match cli.command { + None => crate::tui::run(orchestrator.clone()).await, + Some(Command::Up) => commands::run_up(orchestrator.clone()).await, + Some(Command::Start { workers, all }) => { + commands::run_start(&orchestrator, workers, all).await + } + Some(Command::Stop { workers }) => commands::run_stop(&orchestrator, workers).await, + Some(Command::Restart { worker }) => { + commands::run_restart(&orchestrator, &worker).await + } + Some(Command::Logs { + worker, + follow, + lines, + }) => commands::run_logs(orchestrator.clone(), worker, follow, lines).await, + Some(Command::Status) => commands::run_status(&orchestrator).await, + }; + + if orchestrator.config.stop_on_exit { + if let Err(err) = orchestrator.stop_workers(&[]).await { + eprintln!("warning: {err:#}"); + } + } + + result +} diff --git a/workers-dev/src/orchestrator.rs b/workers-dev/src/orchestrator.rs new file mode 100644 index 000000000..5d4045354 --- /dev/null +++ b/workers-dev/src/orchestrator.rs @@ -0,0 +1,414 @@ +use std::collections::HashMap; +use std::process::Stdio; +use std::time::{Duration, Instant}; + +use anyhow::{bail, Context, Result}; +use tokio::io::{AsyncBufReadExt, BufReader}; +use tokio::process::Command; +use tokio::time; + +use crate::config::{Config, DASHBOARD_LOG_LINES}; +use crate::discover::SpawnKind; +use crate::graph::WorkerGraph; +use crate::logs; +use crate::runtime::{ProcState, SharedRuntimes, WorkerRuntime}; +use crate::status::{self, EngineWorker, WorkerView}; + +pub struct Orchestrator { + pub config: Config, + pub graph: WorkerGraph, + pub runtimes: SharedRuntimes, +} + +impl Orchestrator { + pub fn new(config: Config) -> Result { + let graph = WorkerGraph::load(&config.repo_root, &config.workers)?; + let runtimes = crate::runtime::new_runtimes(&config.workers); + Ok(Self { + config, + graph, + runtimes, + }) + } + + pub async fn engine_preflight(&self) -> Result<()> { + status::fetch_engine_workers(&self.config) + .await + .with_context(|| { + format!( + "engine not reachable at {} (start with: iii -c harness/engine.config.yaml)", + self.config.engine_url + ) + })?; + Ok(()) + } + + pub async fn start_harness_stack(&self, wait_connected: bool) -> Result<()> { + self.start_workers(&self.config.harness_stack, wait_connected) + .await + } + + pub async fn start_all_managed(&self, wait_connected: bool) -> Result<()> { + self.start_workers(&self.config.workers, wait_connected) + .await + } + + pub async fn start_workers(&self, names: &[String], wait_connected: bool) -> Result<()> { + self.engine_preflight().await?; + let order = if names.is_empty() { + self.graph.topo_start_order(self.graph.workers())? + } else { + self.graph.closure_with_deps(names)? + }; + + for worker in order { + self.start_one(&worker, wait_connected).await?; + } + Ok(()) + } + + pub async fn stop_workers(&self, names: &[String]) -> Result<()> { + let order = if names.is_empty() { + self.graph.topo_stop_order(self.graph.workers())? + } else { + self.graph.topo_stop_order(names)? + }; + + for worker in order { + self.stop_one(&worker).await; + } + Ok(()) + } + + pub async fn restart_worker(&self, name: &str) -> Result<()> { + let closure = self.graph.restart_closure(name)?; + let stop_order = self.graph.topo_stop_order(&closure)?; + for worker in &stop_order { + self.stop_one(worker).await; + } + for worker in &closure { + self.start_one(worker, true).await?; + } + Ok(()) + } + + async fn start_one(&self, name: &str, wait_connected: bool) -> Result<()> { + self.stop_one(name).await; + + let spec = self + .config + .worker_spec(name) + .with_context(|| format!("unknown worker {name}"))?; + match &spec.spawn { + SpawnKind::CargoRun => {} + SpawnKind::Unsupported { reason } => { + bail!("worker {name} cannot be started from workers-dev: {reason}"); + } + } + + let worker_dir = &spec.dir; + if !worker_dir.is_dir() { + bail!("worker directory not found: {}", worker_dir.display()); + } + + let mut cmd = Command::new("cargo"); + cmd.arg("run"); + if self.config.release { + cmd.arg("--release"); + } + cmd.args(["--", "--url", &self.config.engine_url]); + cmd.current_dir(worker_dir); + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + cmd.kill_on_drop(true); + cmd.env("CARGO_TERM_COLOR", "never"); + cmd.env("CARGO_TERM_PROGRESS", "never"); + cmd.env("CLICOLOR_FORCE", "0"); + + let mut child = cmd + .spawn() + .with_context(|| format!("spawn cargo run for {name}"))?; + + let stdout = child.stdout.take().context("stdout pipe")?; + let stderr = child.stderr.take().context("stderr pipe")?; + + let expected_pid = child.id(); + + { + let mut runtimes = self.runtimes.write().await; + let rt = runtimes.get_mut(name).context("unknown worker runtime")?; + rt.proc_state = ProcState::Compiling; + rt.exit_code = None; + rt.started_at = Some(Instant::now()); + rt.set_child(child); + } + + let name_stdout = name.to_string(); + let name_stderr = name.to_string(); + let runtimes_stdout = self.runtimes.clone(); + let runtimes_stderr = self.runtimes.clone(); + tokio::spawn(async move { + read_stream(name_stdout, stdout, runtimes_stdout).await; + }); + tokio::spawn(async move { + read_stream(name_stderr, stderr, runtimes_stderr).await; + }); + + let name_wait = name.to_string(); + let runtimes_wait = self.runtimes.clone(); + tokio::spawn(async move { + wait_for_exit(&name_wait, expected_pid, runtimes_wait).await; + }); + + if wait_connected { + self.wait_connected(name).await?; + } + + Ok(()) + } + + async fn stop_one(&self, name: &str) { + let child = { + let mut runtimes = self.runtimes.write().await; + let Some(rt) = runtimes.get_mut(name) else { + return; + }; + rt.take_child() + }; + + if let Some(mut child) = child { + let _ = child.start_kill(); + let _ = child.wait().await; + } + + let mut runtimes = self.runtimes.write().await; + if let Some(rt) = runtimes.get_mut(name) { + if rt.proc_state != ProcState::Crashed { + rt.proc_state = ProcState::Stopped; + } + rt.clear_child(); + } + } + + pub async fn wait_connected(&self, name: &str) -> Result<()> { + let deadline = + Instant::now() + Duration::from_millis(self.config.connect_timeout_ms); + loop { + if Instant::now() >= deadline { + bail!("timed out waiting for {name} to connect to engine"); + } + let engine = status::fetch_engine_workers(&self.config).await.unwrap_or_default(); + if engine.iter().any(|w| w.name.as_deref() == Some(name) && w.status == "connected") { + return Ok(()); + } + let proc = self.proc_state(name).await; + if proc == ProcState::Crashed || proc == ProcState::Stopped { + bail!("worker {name} exited before connecting to engine"); + } + time::sleep(Duration::from_millis(500)).await; + } + } + + async fn proc_state(&self, name: &str) -> ProcState { + let runtimes = self.runtimes.read().await; + runtimes + .get(name) + .map(|rt| rt.proc_state) + .unwrap_or(ProcState::Stopped) + } + + pub async fn worker_views(&self) -> Result> { + let engine = status::fetch_engine_workers(&self.config) + .await + .unwrap_or_default(); + let engine_by_name: HashMap = engine + .into_iter() + .filter_map(|w| w.name.clone().map(|name| (name, w))) + .collect(); + + let runtimes = self.runtimes.read().await; + let mut views = Vec::new(); + for worker in self.graph.workers() { + let rt = runtimes.get(worker).expect("runtime"); + let spec = self.config.worker_spec(worker).expect("spec"); + views.push(build_view(spec, rt, engine_by_name.get(worker))); + } + Ok(views) + } + + pub async fn logs_tail(&self, name: &str, n: usize) -> Result> { + let runtimes = self.runtimes.read().await; + let rt = runtimes.get(name).context("unknown worker")?; + Ok(rt.logs.tail(n)) + } + + pub async fn subscribe_logs(&self, name: &str) -> Result> { + let runtimes = self.runtimes.read().await; + let rt = runtimes.get(name).context("unknown worker")?; + Ok(rt.subscribe_logs()) + } +} + +fn build_view( + spec: &crate::discover::WorkerSpec, + rt: &WorkerRuntime, + engine: Option<&EngineWorker>, +) -> WorkerView { + let process = rt.proc_state.label().to_string(); + let (engine_status, uptime) = if let Some(w) = engine { + ( + w.status.clone(), + format_uptime(w.connected_at_ms), + ) + } else { + ("—".to_string(), "—".to_string()) + }; + + let display = if engine.map(|w| w.status.as_str()) == Some("connected") { + "connected" + } else if rt.proc_state == ProcState::Crashed { + "crashed" + } else if rt.proc_state == ProcState::Compiling { + "compiling" + } else if rt.proc_state == ProcState::Running { + "disconnected" + } else { + "stopped" + }; + + WorkerView { + name: spec.name.clone(), + group: spec.group, + spawnable: matches!(spec.spawn, SpawnKind::CargoRun), + display_status: display.to_string(), + process_status: process, + engine_status, + local_pid: rt.pid(), + engine_pid: None, + uptime, + last_logs: rt.logs.tail(DASHBOARD_LOG_LINES), + } +} + +fn format_uptime(connected_at_ms: u64) -> String { + if connected_at_ms == 0 { + return "—".to_string(); + } + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let secs = now_ms.saturating_sub(connected_at_ms) / 1000; + if secs < 60 { + format!("{secs}s") + } else if secs < 3600 { + format!("{}m", secs / 60) + } else { + format!("{}h", secs / 3600) + } +} + +async fn read_stream( + worker: String, + stream: impl tokio::io::AsyncRead + Unpin + Send + 'static, + runtimes: SharedRuntimes, +) { + let mut lines = BufReader::new(stream).lines(); + loop { + match lines.next_line().await { + Ok(Some(line)) => { + let normalized = logs::normalize_log_line(&line); + if normalized.is_empty() { + continue; + } + let mut runtimes = runtimes.write().await; + if let Some(rt) = runtimes.get_mut(&worker) { + rt.logs.push(normalized.clone()); + update_proc_state_from_line(rt, &normalized); + let _ = rt.log_tx.send(normalized); + } + } + Ok(None) => break, + Err(err) => { + let mut runtimes = runtimes.write().await; + if let Some(rt) = runtimes.get_mut(&worker) { + let msg = format!("log read error: {err}"); + rt.logs.push(msg.clone()); + let _ = rt.log_tx.send(msg); + } + break; + } + } + } +} + +fn update_proc_state_from_line(rt: &mut WorkerRuntime, line: &str) { + if line.contains("Compiling") + || line.contains("Building") + || line.contains("Downloading") + || line.contains("Updating") + { + rt.proc_state = ProcState::Compiling; + } else if line.contains("Finished") + || line.contains("Running") + || line.contains("INFO") + || line.contains("registered") + { + if rt.proc_state != ProcState::Crashed { + rt.proc_state = ProcState::Running; + } + } else if line.contains("error:") + || line.contains("error[E") + || line.contains("could not compile") + { + rt.proc_state = ProcState::Compiling; + } +} + +async fn wait_for_exit(worker: &str, expected_pid: Option, runtimes: SharedRuntimes) { + loop { + time::sleep(Duration::from_millis(250)).await; + + let status = { + let mut guard = runtimes.write().await; + let Some(rt) = guard.get_mut(worker) else { + return; + }; + if rt.pid() != expected_pid { + return; + } + let Some(child) = rt.child_mut() else { + return; + }; + match child.try_wait() { + Ok(Some(status)) => Some(status), + Ok(None) => None, + Err(err) => { + rt.logs.push(format!("process wait error: {err}")); + rt.proc_state = ProcState::Crashed; + rt.clear_child(); + return; + } + } + }; + + let Some(status) = status else { + continue; + }; + + let mut guard = runtimes.write().await; + if let Some(rt) = guard.get_mut(worker) { + rt.clear_child(); + rt.exit_code = status.code(); + rt.proc_state = if status.success() { + ProcState::Stopped + } else { + let code = status.code().map(|c| c.to_string()).unwrap_or_else(|| "?".into()); + rt.logs + .push(format!("process exited with status {code}")); + ProcState::Crashed + }; + } + return; + } +} diff --git a/workers-dev/src/runtime.rs b/workers-dev/src/runtime.rs new file mode 100644 index 000000000..c6469a43b --- /dev/null +++ b/workers-dev/src/runtime.rs @@ -0,0 +1,112 @@ +use std::collections::{HashMap, VecDeque}; +use std::sync::Arc; +use std::time::Instant; + +use tokio::sync::{broadcast, RwLock}; + +use crate::config::LOG_RING_CAPACITY; + +#[derive(Debug, Clone)] +pub struct RingBuffer { + lines: VecDeque, + capacity: usize, +} + +impl RingBuffer { + pub fn new(capacity: usize) -> Self { + Self { + lines: VecDeque::with_capacity(capacity.min(64)), + capacity, + } + } + + pub fn push(&mut self, line: String) { + if self.lines.len() >= self.capacity { + self.lines.pop_front(); + } + self.lines.push_back(line); + } + + pub fn tail(&self, n: usize) -> Vec { + let start = self.lines.len().saturating_sub(n); + self.lines.iter().skip(start).cloned().collect() + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ProcState { + Stopped, + Compiling, + Running, + Crashed, +} + +impl ProcState { + pub fn label(self) -> &'static str { + match self { + Self::Stopped => "stopped", + Self::Compiling => "compiling", + Self::Running => "running", + Self::Crashed => "crashed", + } + } +} + +#[derive(Debug)] +pub struct WorkerRuntime { + pub proc_state: ProcState, + pub exit_code: Option, + pub started_at: Option, + pub logs: RingBuffer, + pub log_tx: broadcast::Sender, + child: Option, +} + +impl WorkerRuntime { + fn new() -> Self { + let (log_tx, _) = broadcast::channel(256); + Self { + proc_state: ProcState::Stopped, + exit_code: None, + started_at: None, + logs: RingBuffer::new(LOG_RING_CAPACITY), + log_tx, + child: None, + } + } + + pub fn pid(&self) -> Option { + self.child.as_ref().and_then(|c| c.id()) + } + + pub fn subscribe_logs(&self) -> broadcast::Receiver { + self.log_tx.subscribe() + } + + pub(crate) fn set_child(&mut self, child: tokio::process::Child) { + self.child = Some(child); + } + + pub(crate) fn take_child(&mut self) -> Option { + self.child.take() + } + + pub(crate) fn child_mut(&mut self) -> Option<&mut tokio::process::Child> { + self.child.as_mut() + } + + pub(crate) fn clear_child(&mut self) { + self.child = None; + } +} + +pub type SharedRuntimes = Arc>>; + +pub fn new_runtimes(workers: &[String]) -> SharedRuntimes { + let mut map = HashMap::new(); + for worker in workers { + map.insert(worker.clone(), WorkerRuntime::new()); + } + Arc::new(RwLock::new(map)) +} + diff --git a/workers-dev/src/status.rs b/workers-dev/src/status.rs new file mode 100644 index 000000000..2758948b8 --- /dev/null +++ b/workers-dev/src/status.rs @@ -0,0 +1,107 @@ +use std::process::Stdio; +use std::time::Duration; + +use anyhow::{Context, Result}; +use serde::Deserialize; +use tokio::process::Command; +use tokio::time::timeout; + +use crate::discover::WorkerGroup; +use crate::config::Config; + +#[derive(Debug, Clone, Deserialize)] +pub struct EngineWorkersResponse { + pub workers: Vec, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct EngineWorker { + #[serde(default)] + pub name: Option, + pub status: String, + #[serde(default)] + pub connected_at_ms: u64, +} + +#[derive(Debug, Clone)] +pub struct WorkerView { + pub name: String, + pub group: WorkerGroup, + pub spawnable: bool, + pub display_status: String, + pub process_status: String, + pub engine_status: String, + pub local_pid: Option, + pub engine_pid: Option, + pub uptime: String, + pub last_logs: Vec, +} + +pub async fn fetch_engine_workers(config: &Config) -> Result> { + let mut cmd = Command::new("iii"); + cmd.args([ + "trigger", + "engine::workers::list", + "--json", + "{}", + "--address", + &config.engine_host, + "--port", + &config.engine_port.to_string(), + ]); + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + + let output = timeout(Duration::from_secs(10), cmd.output()) + .await + .context("timed out waiting for iii trigger engine::workers::list")? + .context("spawn iii trigger (is iii on PATH?)")?; + + let out = String::from_utf8_lossy(&output.stdout).to_string(); + let err = String::from_utf8_lossy(&output.stderr).to_string(); + + if !output.status.success() { + anyhow::bail!( + "iii trigger engine::workers::list failed: {}", + err.trim() + ); + } + + let parsed: EngineWorkersResponse = serde_json::from_str(out.trim()) + .with_context(|| format!("parse engine workers response: {}", out.trim()))?; + Ok(parsed.workers) +} + +pub fn print_status_table(views: &[WorkerView]) { + if views.is_empty() { + return; + } + let mut last_group: Option = None; + for v in views { + if last_group != Some(v.group) { + if last_group.is_some() { + println!(); + } + println!("── {} ──", v.group.label()); + println!( + "{:<28} {:<12} {:<12} {:<8} {:<8}", + "WORKER", "PROCESS", "ENGINE", "PID", "UPTIME" + ); + last_group = Some(v.group); + } + let pid = v + .engine_pid + .or(v.local_pid) + .map(|p| p.to_string()) + .unwrap_or_else(|| "—".to_string()); + let name = if v.spawnable { + v.name.clone() + } else { + format!("{} (iii worker add)", v.name) + }; + println!( + "{:<28} {:<12} {:<12} {:<8} {:<8}", + name, v.process_status, v.engine_status, pid, v.uptime + ); + } +} diff --git a/workers-dev/src/tui/mod.rs b/workers-dev/src/tui/mod.rs new file mode 100644 index 000000000..2417f6720 --- /dev/null +++ b/workers-dev/src/tui/mod.rs @@ -0,0 +1,539 @@ +mod theme; + +use std::io; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use anyhow::Result; +use crossterm::event::{self, Event, KeyCode, KeyEvent, KeyModifiers}; +use crossterm::execute; +use crossterm::terminal::{ + disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen, +}; +use ratatui::layout::{Constraint, Direction, Layout, Rect}; +use ratatui::style::{Modifier, Style}; +use ratatui::text::{Line, Span}; +use ratatui::widgets::{Block, Borders, Cell, Clear, Paragraph, Row, Table, TableState}; +use ratatui::Frame; +use ratatui::Terminal; + +use crate::discover::WorkerGroup; +use crate::logs; +use crate::orchestrator::Orchestrator; +use crate::status::WorkerView; +use theme::{ + confirm_prompt_style, engine_style, engine_url_style, footer_style, group_header_style, + header_accent_style, hint_style, log_title_style, muted_cell_style, non_spawnable_style, + overlay_bg_style, process_style, selection_row_style, status_style, +}; + +enum UiMode { + Dashboard, + ConfirmRestart(String), + Busy(String), +} + +enum DisplayRowKind { + Header(WorkerGroup), + Worker(usize), +} + +struct DisplayRow { + kind: DisplayRowKind, +} + +pub async fn run(orchestrator: Arc) -> Result<()> { + enable_raw_mode()?; + execute!(io::stdout(), EnterAlternateScreen)?; + + let result: Result<()> = async { + let stdout = io::stdout(); + let backend = ratatui::backend::CrosstermBackend::new(stdout); + let mut terminal = Terminal::new(backend)?; + + let mut views: Vec = orchestrator.worker_views().await?; + let mut display_rows = build_display_rows(&views); + let mut table_state = TableState::default(); + table_state.select(Some(first_worker_row(&display_rows).unwrap_or(0))); + + let mut mode = UiMode::Dashboard; + let poll = Duration::from_millis(orchestrator.config.poll_interval_ms); + let mut last_poll = Instant::now(); + let mut running = true; + + let color_enabled = orchestrator.config.color_mode.enabled_for_tui(); + + while running { + display_rows = build_display_rows(&views); + terminal.draw(|f| { + draw_ui( + f, + &orchestrator.config.engine_url, + &views, + &display_rows, + &mut table_state, + &mode, + color_enabled, + ); + })?; + + if last_poll.elapsed() >= poll { + if let Ok(v) = orchestrator.worker_views().await { + views = v; + } + last_poll = Instant::now(); + if matches!(mode, UiMode::Busy(_)) { + mode = UiMode::Dashboard; + } + } + + if event::poll(Duration::from_millis(100))? { + if let Event::Key(key) = event::read()? { + match &mode { + UiMode::ConfirmRestart(worker) => { + let worker = worker.clone(); + match key.code { + KeyCode::Char('y') | KeyCode::Enter => { + spawn_restart(orchestrator.clone(), worker.clone()); + mode = UiMode::Busy(format!("restarting {worker} + dependents…")); + } + KeyCode::Char('n') | KeyCode::Esc => { + mode = UiMode::Dashboard; + } + _ => {} + } + } + UiMode::Busy(_) => { + if key.code == KeyCode::Esc { + mode = UiMode::Dashboard; + } + } + UiMode::Dashboard => { + running = handle_dashboard_key( + orchestrator.clone(), + key, + &mut table_state, + &display_rows, + &views, + &mut mode, + ) + .await?; + } + } + } + } + } + + Ok(()) + } + .await; + + let _ = disable_raw_mode(); + let _ = execute!(io::stdout(), LeaveAlternateScreen); + result +} + +#[allow(clippy::too_many_arguments)] +async fn handle_dashboard_key( + orchestrator: Arc, + key: KeyEvent, + table_state: &mut TableState, + display_rows: &[DisplayRow], + views: &[WorkerView], + mode: &mut UiMode, +) -> Result { + let worker_name = selected_worker(display_rows, views, table_state.selected().unwrap_or(0)); + + match key.code { + KeyCode::Char('q') | KeyCode::Esc => return Ok(false), + KeyCode::Up | KeyCode::Char('k') => { + if let Some(i) = move_selection(display_rows, table_state.selected().unwrap_or(0), false) + { + table_state.select(Some(i)); + } + } + KeyCode::Down | KeyCode::Char('j') => { + if let Some(i) = move_selection(display_rows, table_state.selected().unwrap_or(0), true) { + table_state.select(Some(i)); + } + } + KeyCode::Char('r') => { + if let Some(name) = worker_name { + if views.iter().any(|v| v.name == name && v.spawnable) { + *mode = UiMode::ConfirmRestart(name); + } else { + *mode = UiMode::Busy( + "worker not startable from workers-dev (use iii worker add)".into(), + ); + } + } + } + KeyCode::Char('s') => { + if let Some(name) = worker_name { + if views.iter().any(|v| v.name == name && v.spawnable) { + spawn_start(orchestrator.clone(), vec![name.clone()]); + *mode = UiMode::Busy(format!("starting {name}…")); + } else { + *mode = UiMode::Busy( + "worker not startable from workers-dev (use iii worker add)".into(), + ); + } + } + } + KeyCode::Char('x') => { + if let Some(name) = worker_name { + spawn_stop(orchestrator.clone(), vec![name.clone()]); + *mode = UiMode::Busy(format!("stopping {name}…")); + } + } + KeyCode::Char('l') => { + if let Some(name) = worker_name { + disable_raw_mode()?; + execute!(io::stdout(), LeaveAlternateScreen)?; + println!("Following logs for {name} (Ctrl+C to return)…\n"); + let _ = crate::commands::run_logs(orchestrator.clone(), name, true, 20).await; + enable_raw_mode()?; + execute!(io::stdout(), EnterAlternateScreen)?; + } + } + KeyCode::Char('u') if key.modifiers.contains(KeyModifiers::CONTROL) => { + spawn_start_harness_stack(orchestrator.clone()); + *mode = UiMode::Busy("starting harness stack…".to_string()); + } + KeyCode::Char('a') if key.modifiers.contains(KeyModifiers::CONTROL) => { + spawn_start(orchestrator.clone(), orchestrator.config.workers.clone()); + *mode = UiMode::Busy("starting all managed workers…".to_string()); + } + _ => {} + } + Ok(true) +} + +fn spawn_start(orchestrator: Arc, names: Vec) { + tokio::spawn(async move { + if let Err(err) = orchestrator.start_workers(&names, false).await { + eprintln!("workers-dev: start failed: {err:#}"); + } + }); +} + +fn spawn_start_harness_stack(orchestrator: Arc) { + tokio::spawn(async move { + if let Err(err) = orchestrator.start_harness_stack(false).await { + eprintln!("workers-dev: start failed: {err:#}"); + } + }); +} + +fn spawn_stop(orchestrator: Arc, names: Vec) { + tokio::spawn(async move { + if let Err(err) = orchestrator.stop_workers(&names).await { + eprintln!("workers-dev: stop failed: {err:#}"); + } + }); +} + +fn spawn_restart(orchestrator: Arc, worker: String) { + tokio::spawn(async move { + if let Err(err) = orchestrator.restart_worker(&worker).await { + eprintln!("workers-dev: restart failed: {err:#}"); + } + }); +} + +fn build_display_rows(views: &[WorkerView]) -> Vec { + let mut rows = Vec::new(); + let mut last_group = None; + for (idx, view) in views.iter().enumerate() { + if last_group != Some(view.group) { + rows.push(DisplayRow { + kind: DisplayRowKind::Header(view.group), + }); + last_group = Some(view.group); + } + rows.push(DisplayRow { + kind: DisplayRowKind::Worker(idx), + }); + } + rows +} + +fn first_worker_row(display_rows: &[DisplayRow]) -> Option { + display_rows.iter().position(|row| matches!(row.kind, DisplayRowKind::Worker(_))) +} + +fn move_selection(display_rows: &[DisplayRow], current: usize, down: bool) -> Option { + let mut i = current; + loop { + if down { + if i + 1 >= display_rows.len() { + return None; + } + i += 1; + } else if i == 0 { + return None; + } else { + i -= 1; + } + if matches!(display_rows[i].kind, DisplayRowKind::Worker(_)) { + return Some(i); + } + } +} + +fn selected_worker( + display_rows: &[DisplayRow], + views: &[WorkerView], + row: usize, +) -> Option { + match display_rows.get(row)?.kind { + DisplayRowKind::Worker(idx) => views.get(idx).map(|v| v.name.clone()), + DisplayRowKind::Header(_) => None, + } +} + +fn selected_worker_view<'a>( + display_rows: &[DisplayRow], + views: &'a [WorkerView], + row: usize, +) -> Option<&'a WorkerView> { + match display_rows.get(row)?.kind { + DisplayRowKind::Worker(idx) => views.get(idx), + DisplayRowKind::Header(_) => None, + } +} + +fn styled_if(enabled: bool, style: Style) -> Style { + if enabled { + style + } else { + Style::default() + } +} + +fn draw_ui( + f: &mut Frame, + engine_url: &str, + views: &[WorkerView], + display_rows: &[DisplayRow], + table_state: &mut TableState, + mode: &UiMode, + color_enabled: bool, +) { + let chunks = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Length(3), + Constraint::Min(12), + Constraint::Length(22), + Constraint::Length(1), + ]) + .split(f.area()); + + let header = Paragraph::new(Line::from(vec![ + Span::styled( + "workers-dev", + styled_if(color_enabled, header_accent_style()) + .add_modifier(Modifier::BOLD), + ), + Span::raw(" engine "), + Span::styled(engine_url, styled_if(color_enabled, engine_url_style())), + ])) + .block(Block::default().borders(Borders::ALL).title(" workers-dev ")); + f.render_widget(header, chunks[0]); + + let rows: Vec = display_rows + .iter() + .map(|row| match row.kind { + DisplayRowKind::Header(group) => Row::new(vec![Cell::from(format!( + "── {} ──", + group.label() + )) + .style(styled_if(color_enabled, group_header_style(group)))]), + DisplayRowKind::Worker(idx) => { + let v = &views[idx]; + let icon = status_icon(&v.display_status); + let name_cell = if v.spawnable { + Cell::from(Span::styled( + format!("{icon} {}", v.name), + styled_if(color_enabled, status_style(&v.display_status)), + )) + } else { + Cell::from(Line::from(vec![ + Span::styled( + format!("{icon} {} ", v.name), + styled_if(color_enabled, status_style(&v.display_status)), + ), + Span::styled( + "(iii worker add)", + styled_if(color_enabled, non_spawnable_style()), + ), + ])) + }; + let pid = v + .local_pid + .map(|p| p.to_string()) + .unwrap_or_else(|| "—".to_string()); + Row::new(vec![ + name_cell, + Cell::from(Span::styled( + v.process_status.clone(), + styled_if(color_enabled, process_style(&v.process_status)), + )), + Cell::from(Span::styled( + v.engine_status.clone(), + styled_if(color_enabled, engine_style(&v.engine_status)), + )), + Cell::from(Span::styled( + pid.clone(), + styled_if(color_enabled, muted_cell_style_for(&pid)), + )), + Cell::from(Span::styled( + v.uptime.clone(), + styled_if(color_enabled, muted_cell_style_for(&v.uptime)), + )), + ]) + } + }) + .collect(); + + let table = Table::new( + rows, + [ + Constraint::Percentage(38), + Constraint::Percentage(14), + Constraint::Percentage(14), + Constraint::Percentage(12), + Constraint::Percentage(22), + ], + ) + .header( + Row::new(vec!["Worker", "Process", "Engine", "PID", "Uptime"]).style( + Style::default().add_modifier(Modifier::BOLD), + ), + ) + .block(Block::default().borders(Borders::ALL).title(" Workers ")) + .row_highlight_style(styled_if(color_enabled, selection_row_style())); + f.render_stateful_widget(table, chunks[1], table_state); + + let selected_row = table_state.selected().unwrap_or(0); + render_log_pane( + f, + chunks[2], + selected_worker_view(display_rows, views, selected_row), + color_enabled, + ); + + let (help, overlay) = match mode { + UiMode::Dashboard => ( + " ↑↓ select s start x stop r restart l logs Ctrl+u harness stack Ctrl+a all q quit ", + None, + ), + UiMode::ConfirmRestart(name) => ( + " ↑↓ select s start x stop r restart l logs Ctrl+u harness stack Ctrl+a all q quit ", + Some(format!("Restart {name} and dependents? y/Enter yes n/Esc no")), + ), + UiMode::Busy(msg) => (msg.as_str(), None), + }; + let footer = Paragraph::new(help).style(styled_if(color_enabled, footer_style())); + f.render_widget(footer, chunks[3]); + + if let Some(text) = overlay { + draw_overlay(f, chunks[1], &text, color_enabled); + } +} + +fn muted_cell_style_for(value: &str) -> Style { + if value == "—" { + muted_cell_style() + } else { + Style::default() + } +} + +fn render_log_pane(f: &mut Frame, area: Rect, worker: Option<&WorkerView>, color_enabled: bool) { + let inner_width = area.width.saturating_sub(2) as usize; + let inner_height = area.height.saturating_sub(2) as usize; + + let title = worker.map(|v| { + Line::from(vec![ + Span::raw(" logs: "), + Span::styled(v.name.clone(), styled_if(color_enabled, log_title_style())), + Span::raw(" "), + ]) + }); + + let mut lines: Vec = if let Some(v) = worker { + if v.last_logs.is_empty() { + vec![Line::from(Span::styled( + "(no output yet — press s to start, or `workers-dev logs -f`)", + styled_if(color_enabled, hint_style()), + ))] + } else { + v.last_logs + .iter() + .map(|line| logs::log_line_to_ratatui(line, inner_width, color_enabled)) + .collect() + } + } else { + vec![Line::from("")] + }; + + while lines.len() < inner_height { + lines.push(Line::from(" ".repeat(inner_width.max(1)))); + } + lines.truncate(inner_height); + + let mut block = Block::default().borders(Borders::ALL); + if let Some(title_line) = title { + block = block.title(title_line); + } else { + block = block.title(" logs "); + } + let logs = Paragraph::new(lines).block(block); + f.render_widget(logs, area); +} + +fn draw_overlay(f: &mut Frame, area: Rect, message: &str, color_enabled: bool) { + let popup = centered_rect(60, 20, area); + f.render_widget(Clear, popup); + f.render_widget( + Paragraph::new(message) + .style(styled_if(color_enabled, confirm_prompt_style())) + .block( + Block::default() + .borders(Borders::ALL) + .title(" confirm ") + .style(styled_if(color_enabled, overlay_bg_style())), + ), + popup, + ); +} + +fn centered_rect(percent_x: u16, percent_y: u16, r: Rect) -> Rect { + let popup_layout = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Percentage((100 - percent_y) / 2), + Constraint::Percentage(percent_y), + Constraint::Percentage((100 - percent_y) / 2), + ]) + .split(r); + Layout::default() + .direction(Direction::Horizontal) + .constraints([ + Constraint::Percentage((100 - percent_x) / 2), + Constraint::Percentage(percent_x), + Constraint::Percentage((100 - percent_x) / 2), + ]) + .split(popup_layout[1])[1] +} + +fn status_icon(status: &str) -> &'static str { + match status { + "connected" => "●", + "disconnected" | "compiling" => "◐", + "crashed" => "✗", + _ => "○", + } +} diff --git a/workers-dev/src/tui/theme.rs b/workers-dev/src/tui/theme.rs new file mode 100644 index 000000000..33415220f --- /dev/null +++ b/workers-dev/src/tui/theme.rs @@ -0,0 +1,79 @@ +use ratatui::style::{Color, Modifier, Style}; + +use crate::discover::WorkerGroup; + +pub fn header_accent_style() -> Style { + Style::default() + .fg(Color::Cyan) + .add_modifier(Modifier::BOLD) +} + +pub fn engine_url_style() -> Style { + Style::default().fg(Color::Gray) +} + +pub fn footer_style() -> Style { + Style::default().fg(Color::DarkGray) +} + +/// Selected row background: ~10% blue tint on a dark terminal (no fg override). +pub fn selection_row_style() -> Style { + Style::default().bg(Color::Rgb(18, 28, 45)) +} + +pub fn group_header_style(group: WorkerGroup) -> Style { + match group { + WorkerGroup::HarnessStack => Style::default().fg(Color::Cyan), + WorkerGroup::Other => Style::default().fg(Color::DarkGray), + } +} + +pub fn status_style(display_status: &str) -> Style { + match display_status { + "connected" => Style::default().fg(Color::Green), + "compiling" | "disconnected" => Style::default().fg(Color::Yellow), + "crashed" => Style::default().fg(Color::Red), + _ => Style::default().fg(Color::DarkGray), + } +} + +pub fn process_style(process_status: &str) -> Style { + match process_status { + "running" => Style::default().fg(Color::Green), + "compiling" => Style::default().fg(Color::Yellow), + "crashed" => Style::default().fg(Color::Red), + _ => Style::default().fg(Color::DarkGray), + } +} + +pub fn engine_style(engine_status: &str) -> Style { + match engine_status { + "connected" => Style::default().fg(Color::Green), + "—" => Style::default().fg(Color::DarkGray), + _ => Style::default().fg(Color::Yellow), + } +} + +pub fn muted_cell_style() -> Style { + Style::default().fg(Color::DarkGray) +} + +pub fn non_spawnable_style() -> Style { + Style::default().fg(Color::DarkGray) +} + +pub fn hint_style() -> Style { + Style::default().fg(Color::DarkGray) +} + +pub fn confirm_prompt_style() -> Style { + Style::default().fg(Color::Yellow) +} + +pub fn overlay_bg_style() -> Style { + Style::default().bg(Color::Black) +} + +pub fn log_title_style() -> Style { + Style::default().fg(Color::Cyan) +}