diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 20ba888f4..45877e78a 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -32,6 +32,7 @@ jobs: fmt: ${{ steps.filter.outputs.fmt }} fmt-comments: ${{ steps.filter.outputs.fmt-comments }} fmt-markdown: ${{ steps.filter.outputs.fmt-markdown }} + typecheck-js: ${{ steps.filter.outputs.typecheck-js }} test: ${{ steps.filter.outputs.test }} docs: ${{ steps.filter.outputs.docs }} coverage: ${{ steps.filter.outputs.coverage }} @@ -76,6 +77,7 @@ jobs: cargo='(^|/)Cargo\.toml$|^Cargo\.lock$|^rust-toolchain\.toml$' rust='\.rs$' markdown='\.md$|\.markdown$' + javascript='\.js$|(^|/)tsconfig\.json$' snapshots='\.snap$|^\.config/insta\.yaml$' source="$rust|$cargo|$workflow|$justfile" @@ -83,6 +85,7 @@ jobs: set_task fmt "$rust|^\.rustfmt\.toml$|$workflow|$justfile" set_task fmt-comments "$rust|$workflow|$justfile" set_task fmt-markdown "$markdown|$workflow|$justfile" + set_task typecheck-js "$javascript|$workflow|$justfile" set_task test "$source|$snapshots" set_task docs "$source" set_task coverage "$source|$snapshots" @@ -96,7 +99,7 @@ jobs: strategy: matrix: os: [ubuntu-latest] - task: [lint, fmt, fmt-comments, fmt-markdown, test, docs, coverage, deny, insta, shear, vet] + task: [lint, fmt, fmt-comments, fmt-markdown, typecheck-js, test, docs, coverage, deny, insta, shear, vet] include: - task: test os: windows-latest @@ -125,6 +128,7 @@ jobs: FMT: ${{ needs.changes.outputs.fmt }} FMT_COMMENTS: ${{ needs.changes.outputs.fmt-comments }} FMT_MARKDOWN: ${{ needs.changes.outputs.fmt-markdown }} + TYPECHECK_JS: ${{ needs.changes.outputs.typecheck-js }} TEST: ${{ needs.changes.outputs.test }} DOCS: ${{ needs.changes.outputs.docs }} COVERAGE: ${{ needs.changes.outputs.coverage }} @@ -140,6 +144,7 @@ jobs: fmt) run="$FMT" ;; fmt-comments) run="$FMT_COMMENTS" ;; fmt-markdown) run="$FMT_MARKDOWN" ;; + typecheck-js) run="$TYPECHECK_JS" ;; test) run="$TEST" ;; docs) run="$DOCS" ;; coverage) run="$COVERAGE" ;; @@ -197,6 +202,8 @@ jobs: if: ${{ steps.task.outputs.run == 'true' && matrix.task == 'test' }} shell: bash run: cargo metadata --locked --format-version 1 > /dev/null + - if: ${{ steps.task.outputs.run == 'true' && matrix.task == 'typecheck-js' }} + uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 - if: ${{ steps.task.outputs.run == 'true' }} run: just ${{ matrix.task }}-ci ${{ matrix.scope }} - if: ${{ matrix.task == 'coverage' && steps.task.outputs.run == 'true' }} diff --git a/Cargo.lock b/Cargo.lock index cb88f3395..95a73a8d3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -56,6 +56,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -270,6 +276,7 @@ checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", "bytes", + "form_urlencoded", "futures-util", "http", "http-body", @@ -283,6 +290,9 @@ dependencies = [ "percent-encoding", "pin-project-lite", "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", "sync_wrapper", "tokio", "tower", @@ -370,11 +380,11 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" [[package]] name = "bitflags" -version = "2.9.4" +version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2261d10cca569e4643e526d8dc2e62e433cc8aba21ab764233731f8d369bf394" +checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" dependencies = [ - "serde", + "serde_core", ] [[package]] @@ -443,6 +453,12 @@ version = "3.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43" +[[package]] +name = "bytecount" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "175812e0be2bccb6abe50bb8d566126198344f707e304f45c648fd8f2cc0365e" + [[package]] name = "byteorder" version = "1.5.0" @@ -519,6 +535,15 @@ dependencies = [ "unicode-normalization", ] +[[package]] +name = "castaway" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dec551ab6e7578819132c713a93c022a05d60159dc86e7a7050223577484c55a" +dependencies = [ + "rustversion", +] + [[package]] name = "cc" version = "1.2.56" @@ -657,6 +682,19 @@ dependencies = [ "unicode-width", ] +[[package]] +name = "compact_str" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79fcda08c33bb58b97008b2cdada6622500e949e060f5913361763121abd2416" +dependencies = [ + "castaway", + "cfg-if", + "itoa", + "static_assertions", + "zmij", +] + [[package]] name = "comrak" version = "0.52.0" @@ -667,7 +705,7 @@ dependencies = [ "entities", "finl_unicode", "jetscii", - "phf", + "phf 0.13.1", "phf_codegen", "rustc-hash 2.1.1", "smallvec", @@ -748,6 +786,12 @@ version = "3.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7704b5fdd17b18ae31c4c1da5a2e0305a2bf17b5249300a9ee9ed7b72114c636" +[[package]] +name = "cow-utils" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "417bef24afe1460300965a25ff4a24b8b45ad011948302ec221e8a0a81eb2c79" + [[package]] name = "cpp_demangle" version = "0.5.1" @@ -815,7 +859,7 @@ version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d8b9f2e4c67f833b660cdb0a3523065869fb35570177239812ed4c905aeff87b" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "crossterm_winapi", "derive_more", "document-features", @@ -856,7 +900,7 @@ dependencies = [ "cssparser-macros", "dtoa-short", "itoa", - "phf", + "phf 0.13.1", "smallvec", ] @@ -1079,6 +1123,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" +[[package]] +name = "dragonbox_ecma" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd8e701084c37e7ef62d3f9e453b618130cbc0ef3573847785952a3ac3f746bf" + [[package]] name = "dtoa" version = "1.0.11" @@ -1621,6 +1671,7 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" dependencies = [ + "allocator-api2", "foldhash", ] @@ -1647,7 +1698,7 @@ checksum = "de550515ae03ff01fb033658945ba393c8db391297978a1f988ecb436e072f87" dependencies = [ "html5ever", "markup5ever_rcdom", - "phf", + "phf 0.13.1", ] [[package]] @@ -1998,7 +2049,7 @@ name = "inquire" version = "0.9.1" source = "git+https://github.com/JeanMertz/inquire?branch=merged#93ecb2750bc244b2af69b4294d6809957ec7adf0" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "crossterm", "dyn-clone", "unicode-segmentation", @@ -2032,7 +2083,7 @@ version = "0.7.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "046fa2d4d00aea763528b4950358d0ead425372445dc8ff86312b3c69ff7727b" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "cfg-if", "libc", ] @@ -2068,11 +2119,20 @@ dependencies = [ "either", ] +[[package]] +name = "itertools" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b4baf93f58d4425749ca49a51c50ebab072c5df6994d08fed93541c331481dc" +dependencies = [ + "either", +] + [[package]] name = "itoa" -version = "1.0.15" +version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "jetscii" @@ -2104,9 +2164,15 @@ dependencies = [ "axum", "chrono", "comrak", + "form_urlencoded", "jp_plugin", "maud", + "oxc_allocator", + "oxc_parser", + "oxc_semantic", + "oxc_span", "pretty_assertions", + "serde", "serde_json", "sha2", "tokio", @@ -2793,7 +2859,7 @@ version = "0.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "416f7e718bdb06000964960ffa43b4335ad4012ae8b99060261aa4a8088d5ccb" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "libc", ] @@ -2974,9 +3040,9 @@ dependencies = [ [[package]] name = "memchr" -version = "2.8.0" +version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "memmap2" @@ -3051,7 +3117,7 @@ version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab2156c4fce2f8df6c499cc1c763e4394b7482525bf2a9701c9d79d215f519e4" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "cfg-if", "cfg_aliases 0.1.1", "libc", @@ -3063,7 +3129,7 @@ version = "0.30.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "cfg-if", "cfg_aliases 0.2.1", "libc", @@ -3079,6 +3145,12 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "nonmax" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "610a5acd306ec67f907abe5567859a3c693fb9886eb1f012ab8f2a47bef3db51" + [[package]] name = "nu-ansi-term" version = "0.50.1" @@ -3088,6 +3160,25 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "num-bigint" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0" +dependencies = [ + "num-integer", + "num-traits", +] + +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -3194,6 +3285,231 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "owo-colors" +version = "4.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d211803b9b6b570f68772237e415a029d5a50c65d382910b879fb19d3271f94d" + +[[package]] +name = "oxc_allocator" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "064fce871c5cb07e557049ed27c8ca2aa530db797f73424c67e4bb8cfe61175e" +dependencies = [ + "allocator-api2", + "hashbrown 0.17.1", + "oxc_data_structures", + "rustc-hash 2.1.1", +] + +[[package]] +name = "oxc_ast" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a999fd4494b604fc0328fc43443bc1298722500587d87d7e993118bd82c65d7b" +dependencies = [ + "bitflags 2.13.1", + "oxc_allocator", + "oxc_ast_macros", + "oxc_data_structures", + "oxc_diagnostics", + "oxc_estree", + "oxc_regular_expression", + "oxc_span", + "oxc_str", + "oxc_syntax", +] + +[[package]] +name = "oxc_ast_macros" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d715e3c300c95b1d797b05526567161fd86fcac2ac68235101c61b736f216cde" +dependencies = [ + "phf 0.14.0", + "proc-macro2", + "quote", + "syn 3.0.4", +] + +[[package]] +name = "oxc_ast_visit" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81b9d77b5c27575ef7d624de1f226caeca3c0d3fcc2347bdaa788d5c4a85512d" +dependencies = [ + "oxc_allocator", + "oxc_ast", + "oxc_span", + "oxc_syntax", +] + +[[package]] +name = "oxc_data_structures" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d15a23b57a931fda6bc9a4fdc3fbabcac6a2edfd0b7216cb7c81432f0e9f54d9" + +[[package]] +name = "oxc_diagnostics" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2d0ce1ec51b07b6501eecedd5ab50835b7ebe5c8463cada3b2e6d984d0943af" +dependencies = [ + "bytecount", + "cow-utils", + "itoa", + "memchr", + "owo-colors", + "oxc_span", + "percent-encoding", + "smallvec", + "textwrap", + "unicode-segmentation", + "unicode-width", +] + +[[package]] +name = "oxc_ecmascript" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b5d3d07165b7aadcd021f62cd95fb7d62cb6977b8bbfb42fa12adbfb7e75257" +dependencies = [ + "dragonbox_ecma", + "itoa", + "num-bigint", + "num-traits", + "oxc_ast", + "oxc_data_structures", + "oxc_span", + "oxc_syntax", +] + +[[package]] +name = "oxc_estree" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc7af86a59b7aeb2845ffea2cc21ae259053865b175e9ac936a7d72f2744abe5" + +[[package]] +name = "oxc_index" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "191884bee6c3744909a51acc7d78d4ae370d817b25875b10642f632327b6296e" +dependencies = [ + "nonmax", + "serde", +] + +[[package]] +name = "oxc_parser" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "311c29dfdf55ea8bf065cf300b3d0dca5fe1c0fb8059c9ba70a6c98cce75306e" +dependencies = [ + "bitflags 2.13.1", + "cow-utils", + "memchr", + "num-bigint", + "num-traits", + "oxc_allocator", + "oxc_ast", + "oxc_data_structures", + "oxc_diagnostics", + "oxc_ecmascript", + "oxc_span", + "oxc_str", + "oxc_syntax", + "rustc-hash 2.1.1", + "seq-macro", +] + +[[package]] +name = "oxc_regular_expression" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31839a373ad02a37fad54244a3eb710f9a94b5e2e16ca0e1d28a37b4c75b8ebc" +dependencies = [ + "bitflags 2.13.1", + "oxc_allocator", + "oxc_ast_macros", + "oxc_diagnostics", + "oxc_span", + "oxc_str", + "phf 0.14.0", + "rustc-hash 2.1.1", + "unicode-id-start", +] + +[[package]] +name = "oxc_semantic" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "366191309f606847936aade62707b06683ed5e0b0c8cb66c27710a61f64a50f1" +dependencies = [ + "itertools 0.15.0", + "memchr", + "oxc_allocator", + "oxc_ast", + "oxc_ast_visit", + "oxc_data_structures", + "oxc_diagnostics", + "oxc_ecmascript", + "oxc_index", + "oxc_span", + "oxc_str", + "oxc_syntax", + "rustc-hash 2.1.1", + "self_cell", + "smallvec", +] + +[[package]] +name = "oxc_span" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c355ae1fa3e865c28cb8a85735ecafd1bb6340a5b880c638d127c1f2d61a98a5" +dependencies = [ + "compact_str", + "oxc_allocator", + "oxc_ast_macros", + "oxc_estree", + "oxc_str", +] + +[[package]] +name = "oxc_str" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ba45ef9efa8296e647aa37b2b21936cb520953031edc1ef6352281bbea22faa" +dependencies = [ + "compact_str", + "hashbrown 0.17.1", + "oxc_allocator", + "oxc_estree", +] + +[[package]] +name = "oxc_syntax" +version = "0.146.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0970d9be099a082711b574d58add258549580a70fe6546b99a067bbd7ad4a264" +dependencies = [ + "bitflags 2.13.1", + "cow-utils", + "dragonbox_ecma", + "nonmax", + "oxc_allocator", + "oxc_ast_macros", + "oxc_estree", + "oxc_index", + "oxc_span", + "oxc_str", + "phf 0.14.0", + "unicode-id-start", +] + [[package]] name = "parking" version = "2.2.1" @@ -3300,8 +3616,19 @@ version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c1562dc717473dbaa4c1f85a36410e03c047b2e7df7f45ee938fbef64ae7fadf" dependencies = [ - "phf_macros", - "phf_shared", + "phf_macros 0.13.1", + "phf_shared 0.13.1", + "serde", +] + +[[package]] +name = "phf" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "010378780309880b08997fae13be7834dba947d36393bd372f2b1556deb2a2f6" +dependencies = [ + "phf_macros 0.14.0", + "phf_shared 0.14.0", "serde", ] @@ -3311,8 +3638,8 @@ version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "49aa7f9d80421bca176ca8dbfebe668cc7a2684708594ec9f3c0db0805d5d6e1" dependencies = [ - "phf_generator", - "phf_shared", + "phf_generator 0.13.1", + "phf_shared 0.13.1", ] [[package]] @@ -3322,7 +3649,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "135ace3a761e564ec88c03a77317a7c6b80bb7f7135ef2544dbe054243b89737" dependencies = [ "fastrand", - "phf_shared", + "phf_shared 0.13.1", +] + +[[package]] +name = "phf_generator" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aeb62e0959d5a1bebc965f4d15d9e2b7cea002b6b0f5ba8cde6cc26738467100" +dependencies = [ + "fastrand", + "phf_shared 0.14.0", ] [[package]] @@ -3331,8 +3668,21 @@ version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "812f032b54b1e759ccd5f8b6677695d5268c588701effba24601f6932f8269ef" dependencies = [ - "phf_generator", - "phf_shared", + "phf_generator 0.13.1", + "phf_shared 0.13.1", + "proc-macro2", + "quote", + "syn 2.0.114", +] + +[[package]] +name = "phf_macros" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fa8d0ca26d424d27630da600c6624696e7dec8bf7b3b492b383c5dc49e5e085" +dependencies = [ + "phf_generator 0.14.0", + "phf_shared 0.14.0", "proc-macro2", "quote", "syn 2.0.114", @@ -3347,6 +3697,15 @@ dependencies = [ "siphasher", ] +[[package]] +name = "phf_shared" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6fd9027e2d9319be6349febd1db4e8d02aa544921200c9b777720ac34a3aa89" +dependencies = [ + "siphasher", +] + [[package]] name = "pin-project-lite" version = "0.2.16" @@ -3461,7 +3820,7 @@ version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bee689443a2bd0a16ab0348b52ee43e3b2d1b1f931c8aa5c9f8de4c86fbe8c40" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "num-traits", "rand", "rand_chacha", @@ -3626,7 +3985,7 @@ version = "0.5.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5407465600fb0548f1442edf71dd20683c6ed326200ace4b1ef0763521bb3b77" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", ] [[package]] @@ -3648,7 +4007,7 @@ dependencies = [ "chrono", "crossterm", "fd-lock", - "itertools", + "itertools 0.13.0", "nu-ansi-term", "serde", "strip-ansi-escapes", @@ -3867,7 +4226,7 @@ version = "0.38.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1c93dd1c9683b438c392c492109cb702b8090b2bfc8fed6f6e4eb4523f17af3" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "fallible-iterator", "fallible-streaming-iterator", "libsqlite3-sys", @@ -3899,7 +4258,7 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd15f8a2c5551a84d56efdc1cd049089e409ac19a3072d5037a17fd70719ff3e" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "errno", "libc", "linux-raw-sys", @@ -3908,9 +4267,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.35" +version = "0.23.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "533f54bc6a7d4f647e46ad909549eda97bf5afc1585190ef692b4286b198bd8f" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" dependencies = [ "log", "once_cell", @@ -3945,9 +4304,9 @@ dependencies = [ [[package]] name = "rustls-webpki" -version = "0.103.13" +version = "0.103.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" dependencies = [ "ring", "rustls-pki-types", @@ -4137,7 +4496,7 @@ version = "3.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cc198e42d9b7510827939c9a15f5062a0c913f3371d765977e586d2fe6c16f4a" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "core-foundation", "core-foundation-sys", "libc", @@ -4160,12 +4519,12 @@ version = "0.33.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "feef350c36147532e1b79ea5c1f3791373e61cbd9a6a2615413b3807bb164fb7" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "cssparser", "derive_more", "log", "new_debug_unreachable", - "phf", + "phf 0.13.1", "phf_codegen", "precomputed-hash", "rustc-hash 2.1.1", @@ -4173,6 +4532,12 @@ dependencies = [ "smallvec", ] +[[package]] +name = "self_cell" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ab42ca02749e120097e328d91d415325bdf43b1c72c4c8badf37375fe40a813" + [[package]] name = "semver" version = "1.0.28" @@ -4183,6 +4548,12 @@ dependencies = [ "serde_core", ] +[[package]] +name = "seq-macro" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1bc711410fbe7399f390ca1c3b60ad0f53f80e95c5eb935e52268a0e2cd49acc" + [[package]] name = "serde" version = "1.0.228" @@ -4491,9 +4862,18 @@ checksum = "7a2ae44ef20feb57a68b23d846850f861394c2e02dc425a50098ae8c90267589" [[package]] name = "smallvec" -version = "1.15.1" +version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +dependencies = [ + "serde", +] + +[[package]] +name = "smawk" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8e2fb0f499abb4d162f2bedad68f5ef91a1682b5a03596ddb67efd37768d100" [[package]] name = "socket2" @@ -4537,7 +4917,7 @@ checksum = "a18596f8c785a729f2819c0f6a7eae6ebeebdfffbfe4214ae6b087f690e31901" dependencies = [ "new_debug_unreachable", "parking_lot", - "phf_shared", + "phf_shared 0.13.1", "precomputed-hash", ] @@ -4547,8 +4927,8 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "585635e46db231059f76c5849798146164652513eb9e8ab2685939dd90f29b69" dependencies = [ - "phf_generator", - "phf_shared", + "phf_generator 0.13.1", + "phf_shared 0.13.1", "proc-macro2", "quote", ] @@ -4745,6 +5125,11 @@ name = "textwrap" version = "0.16.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c13547615a44dc9c452a8a534638acdf07120d4b6847c8178705da06306a3057" +dependencies = [ + "smawk", + "unicode-linebreak", + "unicode-width", +] [[package]] name = "thiserror" @@ -5032,7 +5417,7 @@ version = "0.6.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8" dependencies = [ - "bitflags 2.9.4", + "bitflags 2.13.1", "bytes", "futures-util", "http", @@ -5224,12 +5609,24 @@ version = "2.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "75b844d17643ee918803943289730bec8aac480150456169e647ed0b576ba539" +[[package]] +name = "unicode-id-start" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81b79ad29b5e19de4260020f8919b443b2ef0277d242ce532ec7b7a2cc8b6007" + [[package]] name = "unicode-ident" version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-linebreak" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b09c83c3c29d37506a3e260c08c03743a6bb66a9cd432c6934ab501a190571f" + [[package]] name = "unicode-normalization" version = "0.1.24" @@ -5247,9 +5644,9 @@ checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" [[package]] name = "unicode-segmentation" -version = "1.12.0" +version = "1.13.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6ccf251212114b54433ec949fd6a7841275f9ada20dddd2f29e9ceea4501493" +checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" [[package]] name = "unicode-width" @@ -5501,7 +5898,7 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "57a9779e9f04d2ac1ce317aee707aa2f6b773afba7b931222bff6983843b1576" dependencies = [ - "phf", + "phf 0.13.1", "phf_codegen", "string_cache", "string_cache_codegen", diff --git a/Cargo.toml b/Cargo.toml index 49fe5c828..2dc2fc4e1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,6 +76,7 @@ dyn-clone = { version = "1", default-features = false } dyn-hash = { version = "1", default-features = false } eventsource-stream = { version = "0.2", default-features = false } fancy-regex = { version = "0.19", default-features = false } +form_urlencoded = { version = "1", default-features = false, features = ["alloc"] } futures = { version = "0.3", default-features = false } gemini_client_rs = { git = "https://github.com/JeanMertz/gemini-client", default-features = false } # gimli = { version = "0.33" } @@ -104,6 +105,10 @@ object = { version = "0.39" } # which is the last release on reqwest 0.12. ollama-rs = { git = "https://github.com/JeanMertz/ollama-rs", rev = "6270148236d2d8e199580ee8f5a529868134c0a5", default-features = false } openai_responses = { version = "0.1", default-features = false } +oxc_allocator = { version = "0.146", default-features = false } +oxc_parser = { version = "0.146", default-features = false } +oxc_semantic = { version = "0.146", default-features = false } +oxc_span = { version = "0.146", default-features = false } parking_lot = { version = "0.12", default-features = false, features = ["arc_lock"] } paste = { version = "1", default-features = false } percent-encoding = { version = "2", default-features = false } diff --git a/crates/jp_cli/src/cmd/plugin/dispatch.rs b/crates/jp_cli/src/cmd/plugin/dispatch.rs index 02612e36f..7bd9bb1ad 100644 --- a/crates/jp_cli/src/cmd/plugin/dispatch.rs +++ b/crates/jp_cli/src/cmd/plugin/dispatch.rs @@ -4,7 +4,7 @@ //! the plugin sends `exit` or the process terminates. use std::{ - collections::{BTreeSet, HashSet}, + collections::{BTreeSet, HashMap, HashSet}, fmt::Write as _, fs, io::{self, BufRead, BufReader, Write}, @@ -40,25 +40,33 @@ use jp_plugin::{ ComposeMode, ComposeOption, ComposeRequest, ComposeResponse, ConfigEntry, ConfigResponse, ConfigsResponse, ConversationSummary, ConversationsResponse, CreatedResponse, DescribeResponse, DoneResponse, DraftResponse, ErrorResponse, EventsResponse, HostToPlugin, - InitMessage, LogMessage, PathsInfo, PluginToHost, QueryCompleteResponse, QueryRequest, - SetTitleRequest, WorkspaceInfo, WriteDraftRequest, + InitMessage, InterruptAction as WireAction, InterruptRequest, LockState, LogMessage, + OutputFormat as PluginOutputFormat, PathsInfo, PluginToHost, QueryCompleteResponse, + QueryRequest, SetTitleRequest, WorkspaceInfo, WriteDraftRequest, }, }; -use jp_printer::Printer; +use jp_printer::{OutputFormat, Printer}; use jp_storage::backend::{FsStorageBackend, Projection}; +use jp_task::task::TitleGeneratorTask; use jp_workspace::{ConversationLock, LockResult, Workspace, session::Session}; use serde_json::Value; -use tokio::{sync::mpsc, task::JoinSet}; +use tokio::{ + sync::mpsc::{self, error::TrySendError}, + task::JoinSet, +}; use tracing::{debug, error, info, trace, warn}; use super::registry; use crate::{ Ctx, KeyValueOrPath, cmd, - cmd::query::{PendingStreamTrim, TurnInputs, interrupt::reply_edit_mode}, + cmd::query::{ + NewTitle, PendingStreamTrim, TurnInputs, + interrupt::{InterruptAction, TurnInterruptSender, TurnInterrupts, reply_edit_mode}, + resolve_new_title, + }, config_pipeline::{build_partial_over, config_search_roots}, ctx::McpServerScope, editor::{draft_query_text, draft_revision, report_editor_failure}, - signals::SignalRouter, }; /// Runs the prompts a plugin asks for. @@ -249,6 +257,7 @@ fn init_message( paths: PluginPaths<'_>, config: &Arc, log_level: u8, + format: OutputFormat, ) -> Result<(HostToPlugin, Value), cmd::Error> { let config_json = serde_json::to_value(config.as_ref().to_partial()) .map_err(|e| cmd::Error::from(format!("failed to serialize config: {e}")))?; @@ -276,6 +285,7 @@ fn init_message( options, args: args.to_vec(), log_level, + output_format: output_format(format), }); Ok((init, config_json)) @@ -313,6 +323,7 @@ pub(crate) async fn run_plugin( paths, &config, ctx.term.args.verbose, + ctx.printer.format(), )?; let composer = Composer { @@ -394,6 +405,7 @@ pub(crate) async fn run_plugin( &shutdown_sent, &composer, &mut turns, + &RunningTurns::default(), ) .await; @@ -530,6 +542,19 @@ fn stop_plugin(stdin: &Mutex, sent: &AtomicBool, child_id: u32, grac kill_child(child_id); } +/// The host's output format, in the protocol's vocabulary. +/// +/// Two enums rather than one shared type: the protocol should not depend on a +/// particular renderer, so it carries its own. +fn output_format(format: OutputFormat) -> PluginOutputFormat { + match format { + OutputFormat::Text => PluginOutputFormat::Text, + OutputFormat::TextPretty => PluginOutputFormat::TextPretty, + OutputFormat::Json => PluginOutputFormat::Json, + OutputFormat::JsonPretty => PluginOutputFormat::JsonPretty, + } +} + /// The JP directories a plugin is told about, so it needs no platform logic of /// its own. fn well_known_paths(user_storage_path: Option<&Utf8Path>) -> PathsInfo { @@ -587,6 +612,7 @@ async fn message_loop( shutdown_sent: &AtomicBool, composer: &Composer, turns: &mut JoinSet<()>, + running: &RunningTurns, ) -> Result<(), cmd::Error> { while let Some(line) = requests.recv().await { if line.trim().is_empty() { @@ -612,7 +638,7 @@ async fn message_loop( PluginToHost::Query(request) => { // `None` means the turn is running and will answer for itself. - if let Some(response) = run_query(ctx, request, stdin, turns).await { + if let Some(response) = run_query(ctx, request, stdin, turns, running).await { let mut writer = stdin.lock().expect("stdin lock poisoned"); write_message(&mut *writer, &response) .map_err(|e| cmd::Error::from(format!("failed to answer a query: {e}")))?; @@ -623,7 +649,6 @@ async fn message_loop( let config = ctx.config(); let fs_backend = ctx.fs_backend.clone(); let session = ctx.session.clone(); - let signals = ctx.signals.clone(); let mut writer = stdin.lock().expect("stdin lock poisoned"); if handle_request( @@ -634,7 +659,7 @@ async fn message_loop( session.as_ref(), fs_backend.as_deref(), &config, - &signals, + running, )? == Flow::Stop { return Ok(()); @@ -674,6 +699,7 @@ async fn run_query( request: QueryRequest, stdin: &Arc>, turns: &mut JoinSet<()>, + running: &RunningTurns, ) -> Option { let reply_id = request.id.clone(); let failed = |message: String| Some(query_error(reply_id.clone(), message)); @@ -693,7 +719,7 @@ async fn run_query( // dispatched while the task may not have started. A handler registered by // the task itself would not exist yet, and the stop request would be // reported as reaching nothing. - let turn_interrupt = ctx.signals.turn_interrupt(lock.id()); + let turn_interrupt = ctx.signals.turn_interrupt(); // Read from the lock, not the request: a new conversation was named by the // host, and the plugin has no other way to learn its id. @@ -737,24 +763,54 @@ async fn run_query( "Running a delegated query.", ); + let chat_request = ChatRequest { + content: request.content, + author: config.user.name.clone(), + ..ChatRequest::default() + }; + + // Registered before the turn starts, so an interrupt that arrives while it + // is still being prepared is delivered rather than refused as "no turn is + // running". + let interrupts = running.register(lock.id()); + // Swapped around collecting only, because that is the part that reads the // context. The turn itself carries the config it was given. let host_config = ctx.swap_config(Arc::clone(&config)); - let prepared = prepare_turn(ctx, config, &lock, request.content).await; + let prepared = prepare_turn( + ctx, + Arc::clone(&config), + &lock, + chat_request.clone(), + interrupts, + ) + .await; ctx.swap_config(host_config); let (inputs, stream) = match prepared { Ok(prepared) => prepared, - Err(error) => return failed(error.to_string()), + Err(error) => { + running.finished(lock.id()); + return failed(error.to_string()); + } }; + let title_task = resolve_title(&config, &lock, &stream, &chat_request); + // Hand the turn to its own task. It owns everything it needs and the lock owns // itself, so nothing here is borrowed for the minutes a turn can take, which // is what keeps the message loop answering reads while it runs. let stdin = Arc::clone(stdin); + let running = running.clone(); turns.spawn(async move { - let outcome = inputs.run(&lock, stream, turn_interrupt).await; + // Alongside the turn rather than after it: the two are independent + // requests, and whoever is looking at a list of conversations wants a + // name for this one long before the answer arrives. + let (outcome, ()) = tokio::join!( + inputs.run(&lock, stream, turn_interrupt), + write_generated_title(title_task, &lock), + ); // Reported through tracing rather than to the terminal. These are facts // about the host, not content: the turn's output belongs to the @@ -776,6 +832,14 @@ async fn run_query( } }; + // Both before the answer goes out, because the answer is what a client + // waits on before sending its next request. Reporting the turn as + // finished while it still held the conversation would have that request + // refused as already-locked, and an interrupt aimed at a turn that has + // stopped reading would be accepted and then dropped. + running.finished(lock.id()); + drop(lock); + let mut writer = stdin.lock().expect("stdin lock poisoned"); drop(write_message(&mut *writer, &reply)); }); @@ -914,7 +978,8 @@ async fn prepare_turn( ctx: &mut Ctx, config: Arc, lock: &ConversationLock, - content: String, + chat_request: ChatRequest, + interrupts: TurnInterrupts, ) -> Result<(TurnInputs, ConversationStream), cmd::Error> { // The client was built from the config this host read at startup. A provider // added to the workspace since then is otherwise unknown to it, and starting @@ -931,12 +996,6 @@ async fn prepare_turn( .configure_active_mcp_servers(forced_tool, McpServerScope::Shared) .await?; - let chat_request = ChatRequest { - content, - author: config.user.name.clone(), - ..ChatRequest::default() - }; - // The message has moved from draft to request, so the draft is done. Clearing // it here rather than from the caller gives it one owner: a client that // cleared its own draft would be racing its debounced save, and losing. @@ -1001,12 +1060,90 @@ async fn prepare_turn( // tool at `ask` runs rather than stopping for approval, matching what // `jp query --no-interactive` does. false, + interrupts, ) .await?; Ok((inputs, stream)) } +/// Decide how a conversation nobody has named gets a title from its first +/// message. +/// +/// A leading markdown heading is written straight to the conversation. +/// Anything else needs the model, and comes back as a task for the caller to +/// run. +/// Returns `None` when the conversation already has a title, already has +/// events, or the configuration asks for neither route. +fn resolve_title( + config: &AppConfig, + lock: &ConversationLock, + stream: &ConversationStream, + chat_request: &ChatRequest, +) -> Option { + if lock.metadata().title.is_some() || !stream.is_empty() { + return None; + } + + match resolve_new_title( + config.conversation.title.from_heading, + config.conversation.title.generate.auto, + &chat_request.content, + ) { + NewTitle::FromHeading(title) => { + debug!(conversation = %lock.id(), "Titling from the prompt's leading heading."); + lock.as_mut() + .update_metadata(|meta| meta.title = Some(title)); + None + } + NewTitle::Generate => { + // The title model is configured separately from the assistant's, so + // a broken one must not take the turn down with it. + let mut events = stream.clone(); + events.start_turn(chat_request.clone()); + + match TitleGeneratorTask::new(lock.id(), events, config, false) { + Ok(task) => Some(task), + Err(error) => { + warn!(%error, "Skipping title generation."); + None + } + } + } + NewTitle::Skip => None, + } +} + +/// Run a title task and record what it produced. +/// +/// Writes through the turn's own lock, so the name is on disk as soon as the +/// model answers rather than when the turn ends. +async fn write_generated_title(task: Option, lock: &ConversationLock) { + let Some(task) = task else { + return; + }; + + let title = match task.generate().await { + Ok(Some(title)) => title, + Ok(None) => { + warn!(conversation = %lock.id(), "The title model answered without a title."); + return; + } + Err(error) => { + warn!(%error, conversation = %lock.id(), "Failed to generate a title."); + return; + } + }; + + debug!(conversation = %lock.id(), %title, "Generated a conversation title."); + + let mut conv = lock.as_mut(); + conv.update_metadata(|meta| meta.title = Some(title)); + if let Err(error) = conv.flush() { + warn!(%error, "Failed to persist the generated title."); + } +} + /// Flatten an error and its sources into one line. /// /// JP's error types label a category and carry the cause underneath, so the @@ -1046,6 +1183,135 @@ enum Flow { Stop, } +/// The turns this host is running, and how to reach each one. +/// +/// An interrupt names the conversation it means, because a host runs several +/// turns at once and stopping the wrong one is worse than stopping nothing. +/// +/// An entry lives for exactly as long as its turn: registered before the turn +/// starts, removed once it ends. +/// A request that arrives a moment too late therefore finds nothing, and is +/// told so rather than being dropped — which is what lets a client tell an +/// interrupted turn from one that had already finished. +#[derive(Clone, Default)] +struct RunningTurns(Arc>>); + +impl RunningTurns { + /// Open the interrupt channel for a turn about to start on `conversation`. + /// + /// Returns the end the turn keeps. + /// Any earlier entry for the same conversation is replaced, which cannot + /// happen while the conversation lock is doing its job. + fn register(&self, conversation: ConversationId) -> TurnInterrupts { + let (tx, interrupts) = TurnInterrupts::channel(); + self.0 + .lock() + .expect("running turns lock poisoned") + .insert(conversation, tx); + + interrupts + } + + /// Forget the turn on `conversation`. + fn finished(&self, conversation: ConversationId) { + self.0 + .lock() + .expect("running turns lock poisoned") + .remove(&conversation); + } + + /// Deliver `action` to the turn running on `conversation`. + /// + /// The error is the message to report, phrased for whoever sent the + /// interrupt. + fn interrupt( + &self, + conversation: ConversationId, + action: InterruptAction, + ) -> Result<(), String> { + let sender = self + .0 + .lock() + .expect("running turns lock poisoned") + .get(&conversation) + .cloned() + .ok_or_else(|| format!("no turn is running on conversation {conversation}"))?; + + sender.try_send(action).map_err(|error| match error { + TrySendError::Closed(_) => { + format!("the turn on conversation {conversation} has ended") + } + TrySendError::Full(_) => format!( + "the turn on conversation {conversation} has not caught up with the interrupts \ + already sent" + ), + }) + } +} + +/// Deliver an interrupt to the turn it names. +/// +/// Returns the answer to send, or `None` when the request carried no id: a +/// plugin that is not waiting for one has nowhere to put an uncorrelated +/// response. +fn handle_interrupt(req: InterruptRequest, turns: &RunningTurns) -> Option { + let delivered = parse_conversation_id(&req.conversation) + .and_then(|id| Ok((id, requested_action(&req)?))) + .and_then(|(id, action)| turns.interrupt(id, action)); + + match &delivered { + Ok(()) => debug!( + conversation = %req.conversation, + action = ?req.action, + "Interrupted a turn on a plugin's behalf." + ), + Err(error) => debug!( + conversation = %req.conversation, + action = ?req.action, + error, + "Could not interrupt a turn on a plugin's behalf." + ), + } + + let id = req.id?; + + Some(match delivered { + Ok(()) => HostToPlugin::Done(DoneResponse { id: Some(id) }), + Err(message) => HostToPlugin::Error(ErrorResponse { + id: Some(id), + request: Some("interrupt".to_owned()), + message, + }), + }) +} + +/// Read the interrupt a request is asking for. +/// +/// The error is the message to report: a `reply` without anything to say is the +/// one combination the wire format allows and a turn cannot act on. +fn requested_action(request: &InterruptRequest) -> Result { + match request.action { + WireAction::Stop => Ok(InterruptAction::Stop), + WireAction::Abort => Ok(InterruptAction::Abort), + WireAction::Reply => { + let content = request + .content + .as_deref() + .map(str::trim) + .filter(|content| !content.is_empty()) + .ok_or("a reply needs `content`")?; + + Ok(InterruptAction::Reply { + content: content.to_owned(), + + // Typed somewhere else entirely, so the terminal the turn runs + // in has never seen it. + echo: true, + }) + } + } +} + /// Answer one request from the plugin. /// /// Runs with the writer lock held, so everything here has to be quick: anything @@ -1058,7 +1324,7 @@ fn handle_request( session: Option<&Session>, fs_backend: Option<&FsStorageBackend>, config: &AppConfig, - signals: &SignalRouter, + turns: &RunningTurns, ) -> Result { match msg { PluginToHost::Ready(ready) => { @@ -1113,39 +1379,8 @@ fn handle_request( } PluginToHost::Interrupt(req) => { - // Aimed at the named conversation, not at whatever is topmost. - // - // Several turns can be running at once, and the request already said - // which one it means. Falling back to the untargeted path would stop - // an arbitrary other turn, which is worse than stopping nothing. - // - // Nothing to answer: what the interrupt did lands in the - // conversation, and the turn's own outcome is still the reply to its - // `query`. - // - // A scope with no handler is benign, so it stays at debug: the turn - // finished before the request arrived. An id that does not parse is - // the plugin's bug, and silence would leave its author unable to - // tell the two apart. - // - // `interrupt_scope` is called before the macro, not inside it: a - // tracing field expression only runs when the callsite is enabled, - // and a run whose log file could not be created installs no - // subscriber at all. The interrupt has to happen either way. - match parse_conversation_id(&req.conversation) { - Ok(id) => { - let reached = signals.interrupt_scope(id); - debug!( - conversation = %req.conversation, - reached, - "Interrupting on a plugin's behalf." - ); - } - Err(error) => warn!( - conversation = %req.conversation, - %error, - "Ignoring an interrupt that names an unparseable conversation." - ), + if let Some(response) = handle_interrupt(req, turns) { + write_message(writer, &response)?; } } @@ -1655,13 +1890,44 @@ fn handle_read_events( jp_conversation::decode_event_value(value); } + // Carried here so labelling one conversation doesn't cost a plugin the whole + // conversation list, which reads every conversation's metadata. + let title = workspace + .metadata(&handle) + .ok() + .and_then(|meta| meta.title.clone()); + HostToPlugin::Events(EventsResponse { id: req_id, conversation: conversation_id.to_owned(), + lock: lock_state(workspace, &conv_id), + title, data: event_values, }) } +/// Whether a turn is running on a conversation, and whose it is. +/// +/// Read from the lock rather than from the transcript: a stream ending in a +/// request looks identical whether a turn is running, was interrupted, or +/// failed outright. +/// +/// A lock file outlives the process that wrote it when that process is killed, +/// so a recorded holder that is no longer alive counts as no holder at all. +/// Otherwise a crashed run would leave a conversation looking busy forever. +fn lock_state(workspace: &Workspace, id: &ConversationId) -> LockState { + workspace + .conversation_lock_info(id) + .filter(|info| is_process_alive(info.pid)) + .map_or(LockState::Free, |info| { + if info.pid == std::process::id() { + LockState::Here + } else { + LockState::Elsewhere + } + }) +} + fn handle_read_config( config_json: &Value, path: Option, diff --git a/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs b/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs index 4d3fbbf11..abe7165aa 100644 --- a/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs +++ b/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs @@ -16,14 +16,6 @@ fn bare_workspace() -> Workspace { Workspace::in_memory("/tmp/jp-test-plugin") } -/// A router with no signal source, for requests that never reach one. -/// -/// Must be called inside a tokio runtime, which is why the tests using it are -/// `#[tokio::test]` despite `handle_request` being synchronous. -fn router() -> SignalRouter { - crate::signals::testing::detached_router() -} - /// How a conversation is spelled on the wire, matching `list_conversations`. fn wire_id(id: ConversationId) -> String { id.to_string() @@ -42,63 +34,61 @@ fn conversation_id(secs: u64) -> ConversationId { /// A run whose `--log-file` names a directory that does not exist installs no /// tracing subscriber at all, and a `tracing` field expression does not run /// when its callsite is disabled. -/// These tests install no subscriber either, so an `interrupt_scope` call -/// written inside the macro is never made. -#[tokio::test] -async fn an_interrupt_is_issued_without_a_tracing_subscriber() { +/// These tests install no subscriber either, so a delivery written inside the +/// macro is never made. +#[test] +fn an_interrupt_is_issued_without_a_tracing_subscriber() { let mut ws = bare_workspace(); let mut sink: Vec = Vec::new(); - let signals = router(); + let turns = RunningTurns::default(); let id = conversation_id(1_700_000_000); - let (_guard, mut interrupted) = signals.push_handler_for(id); + let mut interrupted = turns.register(id); handle_request( - PluginToHost::Interrupt(InterruptRequest { - conversation: wire_id(id), - }), + PluginToHost::Interrupt(InterruptRequest::stop(wire_id(id))), &mut sink, &mut ws, &json!({}), None, None, &AppConfig::new_test(), - &signals, + &turns, ) .unwrap(); - assert!( - interrupted.try_recv().is_ok(), + assert_eq!( + interrupted.try_next(), + Some(InterruptAction::Stop), "the turn was never told to stop" ); } /// An interrupt naming something that is not a conversation id is the plugin's /// bug, and must not be mistaken for a turn that already finished. -#[tokio::test] -async fn an_unparseable_interrupt_reaches_no_handler() { +#[test] +fn an_unparseable_interrupt_reaches_no_handler() { let mut ws = bare_workspace(); let mut sink: Vec = Vec::new(); - let signals = router(); + let turns = RunningTurns::default(); - let (_guard, mut interrupted) = signals.push_handler_for(conversation_id(1_700_000_000)); + let mut interrupted = turns.register(conversation_id(1_700_000_000)); handle_request( - PluginToHost::Interrupt(InterruptRequest { - conversation: "not-an-id".to_owned(), - }), + PluginToHost::Interrupt(InterruptRequest::stop("not-an-id".to_owned())), &mut sink, &mut ws, &json!({}), None, None, &AppConfig::new_test(), - &signals, + &turns, ) .unwrap(); - assert!( - interrupted.try_recv().is_err(), + assert_eq!( + interrupted.try_next(), + None, "a malformed id must not stop an unrelated turn" ); } @@ -600,7 +590,7 @@ async fn a_conversation_written_after_startup_is_listed() { None, None, &AppConfig::new_test(), - &router(), + &RunningTurns::default(), ) .unwrap(); assert_eq!(response, Flow::Continue); @@ -650,7 +640,7 @@ async fn an_event_written_after_a_read_is_served_by_the_next_read() { None, None, &AppConfig::new_test(), - &router(), + &RunningTurns::default(), ) .unwrap(); @@ -672,7 +662,7 @@ async fn an_event_written_after_a_read_is_served_by_the_next_read() { None, None, &AppConfig::new_test(), - &router(), + &RunningTurns::default(), ) .unwrap(); @@ -738,7 +728,7 @@ async fn a_ready_carries_on_and_a_clean_exit_stops() { None, None, &AppConfig::new_test(), - &router(), + &RunningTurns::default(), ) .unwrap(), Flow::Continue @@ -756,7 +746,7 @@ async fn a_ready_carries_on_and_a_clean_exit_stops() { None, None, &AppConfig::new_test(), - &router(), + &RunningTurns::default(), ) .unwrap(), Flow::Stop @@ -781,7 +771,7 @@ async fn a_failing_exit_carries_its_code_and_reason() { None, None, &AppConfig::new_test(), - &router(), + &RunningTurns::default(), ) .expect_err("a non-zero exit is an error"); @@ -961,7 +951,7 @@ async fn a_plugin_needing_a_newer_protocol_is_refused() { None, None, &AppConfig::new_test(), - &router(), + &RunningTurns::default(), ) .expect_err("a plugin needing a newer protocol must be refused"); @@ -971,6 +961,149 @@ async fn a_plugin_needing_a_newer_protocol_is_refused() { ); } +/// An interrupt reaches the turn it names and no other. +/// +/// The failure this guards against is stopping the wrong turn: a host runs +/// several at once, and the request is the only thing that says which. +#[test] +fn an_interrupt_reaches_only_the_turn_it_names() { + let turns = RunningTurns::default(); + let wanted = conversation_id(1_700_000_000); + let other = conversation_id(1_700_000_001); + + let mut wanted_rx = turns.register(wanted); + let mut other_rx = turns.register(other); + + turns.interrupt(wanted, InterruptAction::Stop).unwrap(); + + assert_eq!(wanted_rx.try_next(), Some(InterruptAction::Stop)); + assert_eq!(other_rx.try_next(), None); +} + +/// A turn that has finished is reported as gone rather than silently accepting +/// an interrupt nothing will read. +/// +/// This is what lets a client tell "interrupted" from "it had already +/// finished", and send its message as a new turn instead of losing it. +#[test] +fn interrupting_a_finished_turn_says_so() { + let turns = RunningTurns::default(); + let id = conversation_id(1_700_000_000); + + let _rx = turns.register(id); + turns.interrupt(id, InterruptAction::Stop).unwrap(); + + turns.finished(id); + + let error = turns + .interrupt(id, InterruptAction::Stop) + .expect_err("the turn is gone"); + + assert_eq!( + error, + format!("no turn is running on conversation {id}"), + "the message names the conversation, because a client can be watching several" + ); +} + +/// A turn whose receiver is gone is reported as ended, not as never having +/// existed: the entry is still registered, so the two are distinguishable. +#[test] +fn interrupting_a_turn_that_stopped_reading_says_it_ended() { + let turns = RunningTurns::default(); + let id = conversation_id(1_700_000_000); + + drop(turns.register(id)); + + let error = turns + .interrupt(id, InterruptAction::Stop) + .expect_err("nothing is reading"); + + assert_eq!(error, format!("the turn on conversation {id} has ended")); +} + +/// The wire's `reply` carries the text the turn answers with. +#[test] +fn a_reply_request_carries_its_content_to_the_turn() { + let action = requested_action(&InterruptRequest::reply( + "jp-c17000000000".to_owned(), + " use Rust instead ".to_owned(), + )) + .expect("a reply with content is deliverable"); + + assert_eq!(action, InterruptAction::Reply { + // Trimmed, because a browser's textarea keeps the newline a user + // pressed Enter on before deciding to send. + content: "use Rust instead".to_owned(), + + // Nobody watched this arrive at the terminal the turn runs in. + echo: true, + }); +} + +/// A `reply` with nothing to say is refused rather than delivered as an empty +/// message the assistant has to answer. +#[test] +fn a_reply_request_without_content_is_refused() { + let blank = InterruptRequest { + content: Some(" \n ".to_owned()), + ..InterruptRequest::reply("jp-c17000000000".to_owned(), String::new()) + }; + + assert_eq!( + requested_action(&blank).expect_err("a blank reply is not deliverable"), + "a reply needs `content`" + ); + + let missing = InterruptRequest { + action: WireAction::Reply, + ..InterruptRequest::stop("jp-c17000000000".to_owned()) + }; + + assert_eq!( + requested_action(&missing).expect_err("a reply with no content field is not deliverable"), + "a reply needs `content`" + ); +} + +/// An `interrupt` from a protocol 8 plugin carries no action, and means the +/// stop it meant then. +#[test] +fn an_interrupt_without_an_action_stops_the_turn() { + let request: InterruptRequest = + serde_json::from_value(json!({ "conversation": "jp-c17000000000" })).unwrap(); + + assert_eq!(requested_action(&request).unwrap(), InterruptAction::Stop); +} + +/// A request that asked for an answer gets one; one that did not is left alone. +/// +/// An uncorrelated response is a message the plugin has nowhere to put, and the +/// dispatcher pairs replies to requests by id. +#[test] +fn only_an_interrupt_with_an_id_is_answered() { + let turns = RunningTurns::default(); + let id = conversation_id(1_700_000_000); + let _rx = turns.register(id); + + assert!( + handle_interrupt(InterruptRequest::stop(id.to_string()), &turns).is_none(), + "a fire-and-forget interrupt is not answered" + ); + + let answered = handle_interrupt( + InterruptRequest::stop(id.to_string()).with_id("req-1".to_owned()), + &turns, + ); + + assert_eq!( + answered, + Some(HostToPlugin::Done(DoneResponse { + id: Some("req-1".to_owned()) + })) + ); +} + /// The user-global root is `/config/`, the directory `--cfg` itself /// searches, so a listing built from its parent reports nothing. /// diff --git a/crates/jp_cli/src/cmd/query.rs b/crates/jp_cli/src/cmd/query.rs index 1c59dca42..894d0fe91 100644 --- a/crates/jp_cli/src/cmd/query.rs +++ b/crates/jp_cli/src/cmd/query.rs @@ -68,6 +68,7 @@ use chrono::{DateTime, Utc}; use clap::{ArgAction, builder::TypedValueParser as _}; use crossterm::style::Stylize as _; use indexmap::IndexMap; +use interrupt::TurnInterrupts; use jp_attachment::Attachment; use jp_config::{ AppConfig, PartialAppConfig, PartialConfig as _, PartialConfigDelta as _, @@ -705,11 +706,14 @@ impl Query { mcp_servers_handle, ctx.printer.clone(), ctx.term.interactive, + // Ctrl-C is how this turn is interrupted, and that arrives through + // the signal router. + TurnInterrupts::none(), ) .await?; let mut turn_result = inputs - .run(lock, stream, ctx.signals.turn_interrupt(lock.id())) + .run(lock, stream, ctx.signals.turn_interrupt()) .await .map_err(|error| cmd::Error::from(error).with_persistence(true)); @@ -1054,6 +1058,7 @@ impl Query { invocation: InvocationContext, pending_trim: PendingStreamTrim, mut turn_interrupt: TurnInterrupt, + interrupts: TurnInterrupts, ) -> Result<()> { let model_id = cfg.assistant.model.id.resolved(); let provider: Arc = Arc::from(provider::get_provider( @@ -1113,6 +1118,7 @@ impl Query { invocation, pending_trim, turn_interrupt, + interrupts, ) .await } @@ -1376,6 +1382,13 @@ pub(crate) struct TurnInputs { /// MCP servers starting in the background, awaited by [`TurnInputs::run`]. mcp_servers: StartupSet, + + /// Interrupts aimed at this turn by whoever asked for it. + /// + /// A turn typed at a terminal is interrupted through the signal router and + /// takes [`TurnInterrupts::none`]; a turn asked for from elsewhere is + /// reached through here. + interrupts: TurnInterrupts, } impl TurnInputs { @@ -1407,6 +1420,7 @@ impl TurnInputs { mcp_servers: StartupSet, printer: Arc, interactive: bool, + interrupts: TurnInterrupts, ) -> Result { let urls: Vec = config .conversation @@ -1456,6 +1470,7 @@ impl TurnInputs { chat_request, pending_trim, config, + interrupts, }) } @@ -1561,6 +1576,7 @@ impl TurnInputs { }, self.pending_trim, turn_interrupt, + self.interrupts, ) .await } @@ -2242,7 +2258,7 @@ enum QuerySource { /// How a new conversation's title is set from its first prompt, before the turn /// runs. #[derive(Debug, PartialEq)] -enum NewTitle { +pub(crate) enum NewTitle { /// Use this text, taken verbatim from a leading markdown heading. FromHeading(String), @@ -2259,7 +2275,11 @@ enum NewTitle { /// background generation is chosen when `generate_auto` is enabled. /// The two flags are independent: disabling generation does not disable /// heading-derived titles. -fn resolve_new_title(from_heading: bool, generate_auto: bool, content: &str) -> NewTitle { +pub(crate) fn resolve_new_title( + from_heading: bool, + generate_auto: bool, + content: &str, +) -> NewTitle { if from_heading && let Some(title) = jp_md::heading::leading_heading(content) { return NewTitle::FromHeading(title); } diff --git a/crates/jp_cli/src/cmd/query/interrupt.rs b/crates/jp_cli/src/cmd/query/interrupt.rs index 3657b08c5..853e5ee29 100644 --- a/crates/jp_cli/src/cmd/query/interrupt.rs +++ b/crates/jp_cli/src/cmd/query/interrupt.rs @@ -3,10 +3,13 @@ //! Provides context-aware interrupt menus (streaming vs tool execution) and //! routes OS signals to the appropriate handlers. +pub(crate) mod channel; pub(crate) mod handler; pub(crate) mod signals; +pub(crate) use channel::{TurnInterruptSender, TurnInterrupts}; pub(crate) use handler::{InterruptAction, reply_edit_mode}; pub(crate) use signals::{ - LoopAction, StreamingInterruptResult, handle_llm_event, handle_streaming_interrupt, + LoopAction, StreamingInterruptResult, apply_streaming_interrupt, handle_llm_event, + handle_streaming_interrupt, }; diff --git a/crates/jp_cli/src/cmd/query/interrupt/channel.rs b/crates/jp_cli/src/cmd/query/interrupt/channel.rs new file mode 100644 index 000000000..d1778fda0 --- /dev/null +++ b/crates/jp_cli/src/cmd/query/interrupt/channel.rs @@ -0,0 +1,79 @@ +//! Interrupts aimed at a running turn from outside the process. +//! +//! A turn driven from a terminal learns what the user wants through its own +//! prompt: Ctrl-C notifies the loop, the loop shows the interrupt menu, and the +//! menu returns an [`InterruptAction`]. +//! A turn nobody is sitting in front of has no menu to show, so the action +//! arrives already decided, over the channel in this module. +//! +//! [`TurnInterrupts`] is the receiving end, owned by the turn. +//! [`TurnInterruptSender`] is the sending end, held by whoever is driving that +//! turn from elsewhere. + +use std::task::{Context, Poll}; + +use tokio::sync::mpsc; + +use super::InterruptAction; + +/// How many undelivered interrupts a turn will hold. +/// +/// A turn consumes these between provider events, so the queue only builds up +/// while one is in flight. +/// Small on purpose: a client that has sent eight unanswered interrupts is not +/// going to be helped by a ninth being accepted. +const CAPACITY: usize = 8; + +/// Sends interrupts to one running turn. +/// +/// Cloneable, and every clone reaches the same turn. +/// Sending fails once the turn has ended, which is how a client learns there +/// was nothing left to interrupt. +pub(crate) type TurnInterruptSender = mpsc::Sender; + +/// Receives interrupts aimed at this turn. +pub(crate) struct TurnInterrupts(mpsc::Receiver); + +impl TurnInterrupts { + /// Create both ends of a turn's interrupt channel. + pub(crate) fn channel() -> (TurnInterruptSender, Self) { + let (tx, rx) = mpsc::channel(CAPACITY); + (tx, Self(rx)) + } + + /// An end nothing can send to. + /// + /// For a turn whose only user is at the terminal it is running in: the + /// keyboard reaches it through the signal router, and no client holds a + /// sender. + /// Reads as an ended stream, so a caller polling it alongside other sources + /// drops it on the first poll. + pub(crate) fn none() -> Self { + let (_, rx) = mpsc::channel(1); + Self(rx) + } + + /// Take the next interrupt if one has already arrived. + /// + /// Never waits. + pub(crate) fn try_next(&mut self) -> Option { + self.0.try_recv().ok() + } + + /// Wait for the next interrupt. + /// + /// Yields `None` once every sender is gone, which for + /// [`TurnInterrupts::none`] is immediately. + /// Cancel-safe, so it can be raced against other work. + pub(crate) async fn next(&mut self) -> Option { + self.0.recv().await + } + + /// Poll for the next interrupt. + /// + /// Yields `Ready(None)` once every sender is gone, which for + /// [`TurnInterrupts::none`] is immediately. + pub(crate) fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_recv(cx) + } +} diff --git a/crates/jp_cli/src/cmd/query/interrupt/handler.rs b/crates/jp_cli/src/cmd/query/interrupt/handler.rs index b2f89f86f..3b0857a37 100644 --- a/crates/jp_cli/src/cmd/query/interrupt/handler.rs +++ b/crates/jp_cli/src/cmd/query/interrupt/handler.rs @@ -98,13 +98,14 @@ pub enum InterruptAction { /// The reply text. content: String, - /// Whether the reply was composed in the external editor. + /// Whether the caller has to render the reply to the terminal. /// - /// An editor-composed reply never appeared on the terminal, so the - /// caller should echo it back. - /// An inline-composed reply is already visible in scrollback on the - /// widget's own line. - from_editor: bool, + /// `false` when the text is already on the terminal because that is + /// where it was typed — an inline-composed reply sits in scrollback on + /// the widget's own line, and rendering it again would double it. + /// `true` for a reply the terminal never saw: composed in the external + /// editor, or sent by a client somewhere else entirely. + echo: bool, }, /// Resume generation (if stream is alive) or wait (if tool is running). @@ -283,7 +284,10 @@ impl InterruptHandler

{ ReplyResult::Reply { text, from_editor } => { return InterruptAction::Reply { content: text, - from_editor, + // The editor took over the screen and gave it back: + // whatever was composed there was never rendered + // here. + echo: from_editor, }; } // Empty submit or `Ctrl+C` in a menu-driven reply re-shows diff --git a/crates/jp_cli/src/cmd/query/interrupt/handler_tests.rs b/crates/jp_cli/src/cmd/query/interrupt/handler_tests.rs index 23bd69704..a26aa26f1 100644 --- a/crates/jp_cli/src/cmd/query/interrupt/handler_tests.rs +++ b/crates/jp_cli/src/cmd/query/interrupt/handler_tests.rs @@ -172,7 +172,7 @@ fn streaming_interrupt_reply_submits() { ); assert_eq!(action, InterruptAction::Reply { content: "my reply message".into(), - from_editor: false, + echo: false, }); } @@ -206,7 +206,7 @@ fn streaming_interrupt_reply_empty_returns_to_menu_then_submits() { ); assert_eq!(action, InterruptAction::Reply { content: "second try".into(), - from_editor: false, + echo: false, }); } @@ -227,11 +227,11 @@ fn streaming_interrupt_open_editor_re_seeds_then_submits() { &make_printer(), true, ); - // The `Ctrl+X` round-trip ends with a widget submission, so the reply - // counts as inline-composed (visible in scrollback). + // The `Ctrl+X` round-trip ends with a widget submission, so the text is on + // the terminal and echoing it would double it. assert_eq!(action, InterruptAction::Reply { content: "from the editor, edited inline".into(), - from_editor: false, + echo: false, }); } @@ -424,7 +424,7 @@ fn configured_streaming_reply_uses_inline_prompt() { ); assert_eq!(action, InterruptAction::Reply { content: "changed my mind".into(), - from_editor: false, + echo: false, }); } @@ -521,7 +521,7 @@ fn compose_in_editor_opens_editor_directly() { ); assert_eq!(action, InterruptAction::Reply { content: "written in the editor".into(), - from_editor: true, + echo: true, }); } @@ -579,7 +579,7 @@ fn compose_never_uses_inline_widget() { let action = handler(backend).handle_streaming_interrupt(&config, &make_printer(), true); assert_eq!(action, InterruptAction::Reply { content: "inline only".into(), - from_editor: false, + echo: false, }); } @@ -597,7 +597,7 @@ fn compose_in_editor_without_editor_falls_back_to_inline() { let action = handler(backend).handle_streaming_interrupt(&config, &make_printer(), true); assert_eq!(action, InterruptAction::Reply { content: "typed inline".into(), - from_editor: false, + echo: false, }); } @@ -621,7 +621,7 @@ fn compose_in_editor_spawn_failure_falls_back_to_inline() { ); assert_eq!(action, InterruptAction::Reply { content: "typed inline".into(), - from_editor: false, + echo: false, }); } @@ -646,6 +646,6 @@ fn inline_editor_escape_spawn_failure_keeps_buffer() { ); assert_eq!(action, InterruptAction::Reply { content: "draft, then more".into(), - from_editor: false, + echo: false, }); } diff --git a/crates/jp_cli/src/cmd/query/interrupt/signals.rs b/crates/jp_cli/src/cmd/query/interrupt/signals.rs index 867751873..7829b54ff 100644 --- a/crates/jp_cli/src/cmd/query/interrupt/signals.rs +++ b/crates/jp_cli/src/cmd/query/interrupt/signals.rs @@ -82,6 +82,37 @@ pub fn handle_streaming_interrupt( config: &StreamingInterruptConfig, llm_stream_finished: bool, ) -> StreamingInterruptResult { + let action = decide_streaming_interrupt( + turn_coordinator, + printer, + backend, + editor, + edit_mode, + config, + llm_stream_finished, + ); + + apply_streaming_interrupt(action, turn_coordinator, conversation_stream) +} + +/// Ask the user what an interrupt during LLM streaming should do. +/// +/// Shows the menu when `config.action` is `prompt`, and otherwise resolves the +/// configured action without asking. +/// Decides only: nothing is committed to the conversation and the turn's phase +/// does not move until the action reaches [`apply_streaming_interrupt`]. +/// +/// `llm_stream_finished` narrows the menu to what a dead stream can still +/// offer. +pub fn decide_streaming_interrupt( + turn_coordinator: &mut TurnCoordinator, + printer: &Printer, + backend: &dyn PromptBackend, + editor: Option>, + edit_mode: ReplyEditMode, + config: &StreamingInterruptConfig, + llm_stream_finished: bool, +) -> InterruptAction { info!("Interrupt received during streaming."); // Flush the renderer's markdown buffer to the printer queue, then drain @@ -93,6 +124,26 @@ pub fn handle_streaming_interrupt( let action = InterruptHandler::with_backend(backend, editor, edit_mode) .handle_streaming_interrupt(config, printer, !llm_stream_finished); + debug!( + ?action, + llm_stream_finished, "Streaming interrupt resolved." + ); + + action +} + +/// Apply an already-decided interrupt action to a streaming turn. +/// +/// Commits partial assistant content, appends a reply, or moves the phase, +/// depending on the action, and reports what the streaming loop should do next. +/// +/// The action's source does not matter here: a menu choice and a command that +/// arrived from a client are applied identically. +pub fn apply_streaming_interrupt( + action: InterruptAction, + turn_coordinator: &mut TurnCoordinator, + conversation_stream: &mut ConversationStream, +) -> StreamingInterruptResult { // `Resume` means "keep waiting for the current stream." The state // machine is a no-op for it, and we must NOT break the inner loop: // breaking drops the live `SelectAll` and forces a redundant new @@ -100,10 +151,6 @@ pub fn handle_streaming_interrupt( // polling instead. let is_resume = matches!(action, InterruptAction::Resume); let is_escalate = matches!(action, InterruptAction::Escalate); - debug!( - ?action, - llm_stream_finished, "Streaming interrupt resolved." - ); // A menu that never ran decided nothing, so the state machine is left // untouched: no partial commit, no phase change. @@ -277,6 +324,46 @@ pub fn handle_tool_interrupt( .handle_tool_interrupt(config, printer); debug!(?action, "Tool interrupt resolved."); + apply_tool_interrupt(action, cancellation_token, turn_coordinator) +} + +/// Restate an interrupt decided elsewhere in the terms tool execution acts on. +/// +/// The two menus offer different verbs for the same intents, and only these +/// ones mean anything while tools are running: a reply becomes the answer each +/// cancelled tool gives back, which is what `[r] Stop & respond` does at the +/// terminal, and stopping cancels them and ends the turn. +#[must_use] +pub fn as_tool_interrupt(action: InterruptAction) -> InterruptAction { + match action { + InterruptAction::Reply { content, .. } => InterruptAction::ToolCancelled { + response: Some(content), + exit: false, + }, + + // Abort discards the turn's uncommitted work, and cancelling the + // running tools is as much of that as this phase owns. + InterruptAction::Stop | InterruptAction::Abort => InterruptAction::ToolCancelled { + response: None, + exit: true, + }, + + other => other, + } +} + +/// Apply an already-decided interrupt action to running tools. +/// +/// Cancels the running tools where the action calls for it, and reports what +/// the execution loop should do with the results. +/// +/// The action's source does not matter here: a menu choice and a command that +/// arrived from a client are applied identically. +pub fn apply_tool_interrupt( + action: InterruptAction, + cancellation_token: &CancellationToken, + turn_coordinator: &mut TurnCoordinator, +) -> ToolInterruptResult { // A menu that never ran decided nothing: the running tools are left alone // and the state machine is not notified. if matches!(action, InterruptAction::PromptFailed) { diff --git a/crates/jp_cli/src/cmd/query/stream/retry.rs b/crates/jp_cli/src/cmd/query/stream/retry.rs index 2d5db0f80..e5f0e8504 100644 --- a/crates/jp_cli/src/cmd/query/stream/retry.rs +++ b/crates/jp_cli/src/cmd/query/stream/retry.rs @@ -390,13 +390,8 @@ pub async fn handle_stream_error( // scope (stacked above the streaming handler for the duration of the // wait) catches the press, so the caller can show the interrupt menu // immediately instead of after the wait. - // - // Scoped to the conversation, because this is the handler being polled for - // the length of the backoff. An unscoped one would leave a targeted - // interrupt queued on the streaming handler instead, and a `Retry` outcome - // drops that receiver on its way out of the streaming loop. let delay = retry_state.backoff_duration(&error); - let (interrupt_guard, mut interrupt_rx) = signals.push_handler_for(conv.id()); + let (interrupt_guard, mut interrupt_rx) = signals.push_handler(); let notice = tokio::select! { biased; notice = interrupt_rx.recv() => notice, diff --git a/crates/jp_cli/src/cmd/query/tool/coordinator.rs b/crates/jp_cli/src/cmd/query/tool/coordinator.rs index 93f7ea56b..db303fab6 100644 --- a/crates/jp_cli/src/cmd/query/tool/coordinator.rs +++ b/crates/jp_cli/src/cmd/query/tool/coordinator.rs @@ -122,7 +122,12 @@ use super::{ use crate::{ Error, cmd::query::{ - interrupt::signals::{ToolInterruptResult, handle_tool_interrupt}, + interrupt::{ + InterruptAction, TurnInterrupts, + signals::{ + ToolInterruptResult, apply_tool_interrupt, as_tool_interrupt, handle_tool_interrupt, + }, + }, turn::{ TurnCoordinator, state::{PermissionCacheKey, ToolAnswerCacheKey, TurnState}, @@ -132,11 +137,59 @@ use crate::{ signals::{InterruptNotice, SignalRouter}, }; +/// Fold a resolved tool interrupt into the execution loop's state. +/// +/// The tools with no result yet are the ones a cancellation answers for, so +/// they are collected as the interrupt lands rather than after the loop, where +/// a result that arrived in between would have filled one in. +fn record_tool_interrupt( + result: &ToolInterruptResult, + results: &[Option], + outcome: &mut ExecutionOutcome, + tools_cancelled: &mut bool, + cancellation_message: &mut Option, + cancelled_indices: &mut Vec, +) { + match result { + // Either the user chose to keep waiting, or the menu could not be shown + // and nothing happened. A declined press was already handed down the + // stack. + ToolInterruptResult::Continue + | ToolInterruptResult::PromptFailed + | ToolInterruptResult::Declined => {} + + ToolInterruptResult::Restart => outcome.upgrade(ExecutionOutcome::Restart), + + ToolInterruptResult::Cancelled { response, exit } => { + *cancelled_indices = results + .iter() + .enumerate() + .filter(|(_, r)| r.is_none()) + .map(|(i, _)| i) + .collect(); + *tools_cancelled = true; + *cancellation_message = response.clone(); + if *exit { + outcome.upgrade(ExecutionOutcome::Stopped); + } + } + + // The menu itself was cancelled with Ctrl-C: the tools are already + // cancelled; surface the escalation so the turn loop begins a graceful + // shutdown. + ToolInterruptResult::Escalate => outcome.upgrade(ExecutionOutcome::Escalated), + } +} + #[derive(Debug)] enum ExecutionEvent { /// A Ctrl-C press delivered to this execution phase's interrupt handler. Interrupt(InterruptNotice), + /// An interrupt from a client driving this turn from outside the process, + /// which arrives already decided because there was no menu to show. + ClientInterrupt(InterruptAction), + ToolResult { index: usize, result: ExecutorResult, @@ -953,6 +1006,7 @@ impl ToolCoordinator { root: &Utf8Path, tool_renderer: &mut ToolRenderer, interactive: bool, + interrupts: &mut TurnInterrupts, ) -> ExecutionResult { if executors.is_empty() { return ExecutionResult { @@ -966,12 +1020,7 @@ impl ToolCoordinator { // Register the tool interrupt handler for this execution phase. While // registered, the first Ctrl-C press is delivered to this event loop; // the guard deregisters the handler when execution completes. - // - // Scoped to the conversation, so an interrupt that names one reaches the - // handler this loop is polling. Tool execution has no timeout of its - // own, so an unscoped handler here would leave a targeted interrupt - // waiting for the longest tool to finish. - let (interrupt_guard, mut interrupt_rx) = signals.push_handler_for(conv.id()); + let (interrupt_guard, mut interrupt_rx) = signals.push_handler(); // The caller's `index` values come from the execution plan and may // be sparse (e.g. when some tools in the same plan are @@ -1060,7 +1109,18 @@ impl ToolCoordinator { let mut cancellation_message: Option = None; let mut cancelled_indices: Vec = Vec::new(); - while let Some(event) = event_rx.recv().await { + loop { + // A client's interrupt arrives on its own channel rather than + // through the router, so it is polled alongside the tools rather + // than forwarded by a task: the receiver belongs to the turn, and + // the turn outlives this phase. + let event = tokio::select! { + event = event_rx.recv() => event, + Some(action) = interrupts.next() => Some(ExecutionEvent::ClientInterrupt(action)), + }; + + let Some(event) = event else { break }; + match event { ExecutionEvent::ToolResult { index, result } => { let Some(tool) = executing_tools.get_mut(&index) else { @@ -1265,39 +1325,37 @@ impl ToolCoordinator { ToolInterruptResult::Escalate | ToolInterruptResult::PromptFailed => {} } - match result { - // Either the user chose to keep waiting, or the menu - // could not be shown and nothing happened. A - // declined press was already handed down the stack. - ToolInterruptResult::Continue - | ToolInterruptResult::PromptFailed - | ToolInterruptResult::Declined => {} - ToolInterruptResult::Restart => { - outcome.upgrade(ExecutionOutcome::Restart); - } - ToolInterruptResult::Cancelled { response, exit } => { - cancelled_indices = results - .iter() - .enumerate() - .filter(|(_, r)| r.is_none()) - .map(|(i, _)| i) - .collect(); - tools_cancelled = true; - cancellation_message = response; - if exit { - outcome.upgrade(ExecutionOutcome::Stopped); - } - } - // The menu itself was cancelled with Ctrl-C: the - // tools are already cancelled; surface the - // escalation so the turn loop begins a graceful - // shutdown. - ToolInterruptResult::Escalate => { - outcome.upgrade(ExecutionOutcome::Escalated); - } - } + record_tool_interrupt( + &result, + &results, + &mut outcome, + &mut tools_cancelled, + &mut cancellation_message, + &mut cancelled_indices, + ); } } + + ExecutionEvent::ClientInterrupt(action) => { + // Applied even while a tool prompt is active, unlike a + // Ctrl-C: the press competes with the prompt for the + // terminal, and this does not. The prompt is cancelled + // along with the tool that asked. + let result = apply_tool_interrupt( + as_tool_interrupt(action), + &cancellation_token, + turn_coordinator, + ); + + record_tool_interrupt( + &result, + &results, + &mut outcome, + &mut tools_cancelled, + &mut cancellation_message, + &mut cancelled_indices, + ); + } } if results.iter().all(Option::is_some) { diff --git a/crates/jp_cli/src/cmd/query/turn/coordinator.rs b/crates/jp_cli/src/cmd/query/turn/coordinator.rs index 6981d1227..474b26925 100644 --- a/crates/jp_cli/src/cmd/query/turn/coordinator.rs +++ b/crates/jp_cli/src/cmd/query/turn/coordinator.rs @@ -564,10 +564,7 @@ impl TurnCoordinator { self.prepare_continuation(); } - InterruptAction::Reply { - content, - from_editor, - } => { + InterruptAction::Reply { content, echo } => { // Inject partial reasoning + message as assistant events first, // before the user's reply, so the resumed model sees its own // interrupted reasoning as context. @@ -581,14 +578,12 @@ impl TurnCoordinator { author: self.author.clone(), }; - // An editor-composed reply never appeared on the terminal, so - // echo it through the view (labeled user header + body), same - // as replay would emit for this `ChatRequest`. An - // inline-composed reply is already visible in scrollback on - // the widget's own line, so skip the echo — but still reset - // the assistant-header gate so the next assistant chunk - // prints a fresh `── jp …` header. - if from_editor { + // A reply the terminal never saw goes through the view + // (labeled user header + body), the same way replay would emit + // this `ChatRequest`. One that is already in scrollback is + // skipped — but the assistant-header gate still resets, so the + // next assistant chunk prints a fresh `── jp …` header. + if echo { self.view.render_user_request(&request); } else { self.view.begin_turn(); diff --git a/crates/jp_cli/src/cmd/query/turn/coordinator_tests.rs b/crates/jp_cli/src/cmd/query/turn/coordinator_tests.rs index 515c5a76e..9e518b6f3 100644 --- a/crates/jp_cli/src/cmd/query/turn/coordinator_tests.rs +++ b/crates/jp_cli/src/cmd/query/turn/coordinator_tests.rs @@ -800,12 +800,12 @@ fn interrupt_continue_before_first_chunk_emits_assistant_header_on_resume() { assert_eq!(*out.lock(), "hi there\n\n"); } -/// Regression: an editor-composed Reply interrupt inserts a new `ChatRequest` -/// boundary whose text never appeared on the terminal, so live mode must echo -/// it: a labeled user header AND a fresh assistant header for the following -/// content, matching what replay renders for this `ChatRequest`. +/// Regression: a Reply interrupt the terminal never saw inserts a new +/// `ChatRequest` boundary that live mode must echo: a labeled user header AND a +/// fresh assistant header for the following content, matching what replay +/// renders for this `ChatRequest`. #[test] -fn interrupt_reply_from_editor_renders_user_header_for_new_request() { +fn an_unseen_interrupt_reply_renders_a_user_header_for_the_new_request() { let mut stream = ConversationStream::new_test(); let (printer, _out, err) = Printer::memory(OutputFormat::Text); let printer = Arc::new(printer); @@ -822,11 +822,12 @@ fn interrupt_reply_from_editor_renders_user_header_for_new_request() { // Some assistant content arrives so the assistant header is emitted. coordinator.handle_event(&mut stream, Event::message(0, "partial answer")); - // User interrupts with a follow-up reply composed in the external editor. + // User interrupts with a follow-up reply composed in the external editor, + // which took over the screen and gave it back with nothing rendered here. coordinator.handle_streaming_interrupt( InterruptAction::Reply { content: "actually, ignore that".into(), - from_editor: true, + echo: true, }, &mut stream, ); @@ -878,7 +879,7 @@ fn interrupt_reply_inline_skips_user_header_but_resets_assistant_header() { coordinator.handle_streaming_interrupt( InterruptAction::Reply { content: "actually, ignore that".into(), - from_editor: false, + echo: false, }, &mut stream, ); @@ -942,7 +943,7 @@ fn interrupt_reply_during_reasoning_preserves_partial_reasoning() { coordinator.handle_streaming_interrupt( InterruptAction::Reply { content: "actually, do X instead".into(), - from_editor: false, + echo: false, }, &mut stream, ); diff --git a/crates/jp_cli/src/cmd/query/turn_loop.rs b/crates/jp_cli/src/cmd/query/turn_loop.rs index 739ff2657..09b44ff62 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop.rs @@ -47,8 +47,8 @@ use tracing::{debug, info, warn}; use super::{ PendingStreamTrim, build_sections, build_thread, interrupt::{ - LoopAction, StreamingInterruptResult, handle_llm_event, handle_streaming_interrupt, - reply_edit_mode, + InterruptAction, LoopAction, StreamingInterruptResult, TurnInterrupts, + apply_streaming_interrupt, handle_llm_event, handle_streaming_interrupt, reply_edit_mode, }, stream::{ ResponseBoundary, StreamErrorOutcome, StreamRetryState, commit_partial_response, @@ -77,6 +77,9 @@ enum StreamingLoopEvent { /// A Ctrl-C press delivered by the signal router, carried as the notice the /// loop resolves once it has decided what the press did. Interrupt(InterruptNotice), + /// An interrupt from a client driving this turn from outside the process, + /// which arrives already decided because there was no menu to show. + ClientInterrupt(InterruptAction), /// An event from the LLM provider stream. Llm(Box>), } @@ -87,14 +90,16 @@ enum StreamingLoopEvent { /// [`StreamingLoopEvent`]. /// This avoids boxing while allowing `select_all` to poll them as a single /// merged stream. -enum StreamSource { +enum StreamSource { Interrupt(S), + Client(C), Llm(L), } -impl Stream for StreamSource +impl Stream for StreamSource where S: Stream + Unpin, + C: Stream + Unpin, L: Stream + Unpin, { type Item = StreamingLoopEvent; @@ -102,6 +107,7 @@ where fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { match self.get_mut() { Self::Interrupt(s) => Pin::new(s).poll_next(cx), + Self::Client(s) => Pin::new(s).poll_next(cx), Self::Llm(s) => Pin::new(s).poll_next(cx), } } @@ -146,7 +152,7 @@ fn event_keeps_waiting_indicator(event: &StreamingLoopEvent) -> bool { result.as_ref(), Ok(Event::KeepAlive | Event::Patch(_) | Event::Flush { .. }) ), - StreamingLoopEvent::Interrupt(_) => false, + StreamingLoopEvent::Interrupt(_) | StreamingLoopEvent::ClientInterrupt(_) => false, } } @@ -188,6 +194,7 @@ pub(super) async fn run_turn_loop( invocation: InvocationContext, pending_trim: PendingStreamTrim, mut turn_interrupt: TurnInterrupt, + mut interrupts: TurnInterrupts, ) -> Result<(), Error> { // The turn-level interrupt handler (RFD 045) is the outermost handler scope // within the turn: it owns the gaps between phases (persistence, thread @@ -270,6 +277,30 @@ pub(super) async fn run_turn_loop( notice.handled(); } + // A client's interrupt that landed between phases is applied the same + // way the streaming loop applies one: a stop completes the turn, a + // reply becomes the next request. + // + // One per iteration, because applying one can end the turn, and + // whatever followed it was aimed at a turn that no longer exists. + // + // Not before the turn has started: a client can reach this turn from + // the moment it is registered, which is a little before its own request + // is appended, and a reply applied in that window would sit above the + // message it answers. + if turn_coordinator.current_phase() != TurnPhase::Idle + && let Some(action) = interrupts.try_next() + { + info!(?action, "Client interrupt received between turn phases."); + let result = lock.as_mut().update_events(|stream| { + apply_streaming_interrupt(action, &mut turn_coordinator, stream) + }); + + if result == StreamingInterruptResult::Abort { + return Ok(()); + } + } + match turn_coordinator.current_phase() { TurnPhase::Idle => { // The turn-start commit point: any replay trim deferred while @@ -325,15 +356,19 @@ pub(super) async fn run_turn_loop( // set up is delivered as soon as the loop starts polling. The // guard deregisters the handler when the cycle ends. // - // Scoped like the turn-level handler above, and for the same - // reason: this is the handler being polled while a response - // streams, so an interrupt naming this conversation has to be - // able to reach it rather than wait for the phase to end. - let (interrupt_guard, interrupt_rx) = signals.push_handler_for(lock.id()); + let (interrupt_guard, interrupt_rx) = signals.push_handler(); let interrupt_stream = StreamSource::Interrupt( ReceiverStream::new(interrupt_rx).map(StreamingLoopEvent::Interrupt), ); + // Polled alongside the provider stream so a client's interrupt + // lands while the turn is streaming, rather than waiting for + // the phase to end on its own. + let client_stream = StreamSource::Client( + stream::poll_fn(|cx| interrupts.poll_next(cx)) + .map(StreamingLoopEvent::ClientInterrupt), + ); + let raw_stream = provider .chat_completion_stream(model, query) .await @@ -381,7 +416,7 @@ pub(super) async fn run_turn_loop( let mut received_provider_event = false; let mut streams: SelectAll<_> = - SelectAll::from_iter([interrupt_stream, llm_stream]); + SelectAll::from_iter([interrupt_stream, client_stream, llm_stream]); let mut conv = lock.as_mut(); @@ -445,6 +480,32 @@ pub(super) async fn run_turn_loop( } } + StreamingLoopEvent::ClientInterrupt(action) => { + // Nothing is about to prompt, but the partial + // content this commits has to reach the terminal in + // the order it was produced, which is what the menu + // path flushes for too. + turn_coordinator.flush_renderer(); + printer.flush_instant(); + + info!(?action, "Client interrupt received during streaming."); + + let result = conv.update_events(|stream| { + apply_streaming_interrupt(action, &mut turn_coordinator, stream) + }); + + match result { + StreamingInterruptResult::Continue + | StreamingInterruptResult::PromptFailed => {} + StreamingInterruptResult::Break => break, + StreamingInterruptResult::Abort => return Ok(()), + StreamingInterruptResult::Escalate => { + signals.shutdown_token().cancel(); + return Err(cmd::Error::interrupted().into()); + } + } + } + StreamingLoopEvent::Llm(event) => { let event = *event; @@ -861,6 +922,7 @@ pub(super) async fn run_turn_loop( root, &mut tool_renderer, interactive, + &mut interrupts, ) .await; diff --git a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs index f3dd7dae3..bc7ba3f52 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs @@ -321,6 +321,308 @@ impl Provider for StallingMockProvider { } } +/// A provider that stalls on its first request and answers the next. +/// +/// The stall is the window a mid-turn interrupt has to land in; the answer is +/// how the test can tell the turn carried on rather than ending there. +#[derive(Debug)] +struct StallThenAnswerProvider { + /// Streamed before the first request parks. + partial: String, + + /// The whole of the second request's response. + answer: String, + + calls: AtomicUsize, + model: ModelDetails, + + /// Notified on the first poll of the first request's pending tail. + stalled: Arc, +} + +impl StallThenAnswerProvider { + fn new(partial: &str, answer: &str, stalled: &Arc) -> Self { + Self { + partial: partial.to_owned(), + answer: answer.to_owned(), + calls: AtomicUsize::new(0), + model: ModelDetails::empty(id::ModelIdConfig { + provider: ProviderId::Test, + name: "stall-then-answer".parse().expect("valid name"), + }), + stalled: Arc::clone(stalled), + } + } + + /// How many requests the turn has sent. + fn calls(&self) -> usize { + self.calls.load(Ordering::SeqCst) + } +} + +#[async_trait] +impl Provider for StallThenAnswerProvider { + async fn model_details(&self, name: &id::Name) -> Result { + let mut model = self.model.clone(); + model.id.name = name.clone(); + Ok(model) + } + + async fn models(&self) -> Result, LlmError> { + Ok(vec![self.model.clone()]) + } + + async fn chat_completion_stream( + &self, + _model: &ModelDetails, + _query: ChatQuery, + ) -> Result { + if self.calls.fetch_add(1, Ordering::SeqCst) > 0 { + let events = vec![ + Event::message(0, &self.answer), + Event::flush(0), + Event::Finished(FinishReason::Completed), + ]; + + return Ok(Box::pin(stream::iter(events.into_iter().map(Ok)))); + } + + let events: Vec> = + vec![Ok(Event::message(0, &self.partial)), Ok(Event::flush(0))]; + + let stalled = Arc::clone(&self.stalled); + let mut notified = false; + let tail = stream::poll_fn(move |_| { + if !notified { + notified = true; + stalled.notify_one(); + } + std::task::Poll::Pending + }); + + Ok(Box::pin(stream::iter(events).chain(tail))) + } +} + +/// A reply from a client lands in the running turn and the turn carries on. +/// +/// This is the Ctrl-C `[r] Reply` path reached without a keyboard: the partial +/// answer is kept, the reply follows it, and the assistant answers it in the +/// same turn rather than a second one. +#[tokio::test(flavor = "multi_thread")] +async fn a_client_reply_continues_the_running_turn() { + let test_result = Box::pin(timeout(Duration::from_secs(10), async { + let tmp = tempdir().unwrap(); + let root = tmp.path(); + let storage = root.join(".jp"); + + let config = AppConfig::new_test(); + let fs = Arc::new(FsStorageBackend::new(&storage).expect("failed to create backend")); + let mut workspace = Workspace::in_memory(root).with_backend(fs.clone()); + + let lock = workspace + .create_and_lock_conversation(Conversation::default(), config.clone().into(), None) + .unwrap(); + let conv_id = lock.id(); + + let stalled = Arc::new(Notify::new()); + let provider = Arc::new(StallThenAnswerProvider::new( + "Let me start with Python.", + "Rust it is.", + &stalled, + )); + let model = provider + .model_details(&"test-model".parse().unwrap()) + .await + .unwrap(); + + let (printer, _out, _err) = Printer::memory(OutputFormat::TextPretty); + let printer = Arc::new(printer); + let mcp_client = jp_mcp::Client::default(); + let router = detached_router(); + + let (interrupt_tx, interrupts) = TurnInterrupts::channel(); + + // Sent once the first request has parked, which is the only window in + // which this is an interrupt rather than the next turn's question. + let reply_handle = tokio::spawn({ + let stalled = Arc::clone(&stalled); + async move { + stalled.notified().await; + interrupt_tx + .send(InterruptAction::Reply { + content: "No, use Rust.".to_owned(), + echo: true, + }) + .await + .expect("the turn is still running"); + } + }); + + let result = run_turn_loop( + provider.clone() as Arc, + &model, + &config, + &router, + &mcp_client, + root, + false, // is_tty + &[], // attachments + &lock, + ToolChoice::Auto, + &[], // tools + printer.clone(), + Arc::new(MockPromptBackend::new()), + ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), + ChatRequest::from("Which language?"), + InvocationContext::default(), + PendingStreamTrim::default(), + router.turn_interrupt(), + interrupts, + ) + .await; + + reply_handle.await.unwrap(); + assert!(result.is_ok(), "the turn should complete: {result:?}"); + + assert_eq!( + provider.calls(), + 2, + "the reply belongs to the turn that was interrupted, so it is asked again rather than \ + left for a second turn" + ); + + let content = fs + .read_test_events_raw(&conv_id) + .expect("events should be persisted"); + + let question = content.find("Which language?").expect("the first request"); + let partial = content + .find("Let me start with Python.") + .expect("the interrupted answer, kept"); + let reply = content.find("No, use Rust.").expect("the reply"); + let answer = content + .find("Rust it is.") + .expect("the answer to the reply"); + + assert!( + question < partial && partial < reply && reply < answer, + "the assistant's interrupted answer must precede the reply, so the model reads its \ + own partial output as context.\nFile contents:\n{content}" + ); + + assert_eq!( + content.matches("turn_start").count(), + 1, + "the reply belongs to the turn that was interrupted; a second `turn_start` would mean \ + it started a new one.\nFile contents:\n{content}" + ); + })) + .await; + + assert!(test_result.is_ok(), "Test timed out after 10 seconds"); +} + +/// A stop from a client ends the turn where it is, mid-stream. +/// +/// The provider parks after its first response and would never finish on its +/// own, so a turn that ends at all is one the stop reached while it was +/// streaming. +#[tokio::test(flavor = "multi_thread")] +async fn a_client_stop_ends_the_turn_mid_stream() { + let test_result = Box::pin(timeout(Duration::from_secs(10), async { + let tmp = tempdir().unwrap(); + let root = tmp.path(); + let storage = root.join(".jp"); + + let config = AppConfig::new_test(); + let fs = Arc::new(FsStorageBackend::new(&storage).expect("failed to create backend")); + let mut workspace = Workspace::in_memory(root).with_backend(fs.clone()); + + let lock = workspace + .create_and_lock_conversation(Conversation::default(), config.clone().into(), None) + .unwrap(); + let conv_id = lock.id(); + + let stalled = Arc::new(Notify::new()); + let provider = Arc::new(StallThenAnswerProvider::new( + "Thinking about it.", + "never asked", + &stalled, + )); + let model = provider + .model_details(&"test-model".parse().unwrap()) + .await + .unwrap(); + + let (printer, _out, _err) = Printer::memory(OutputFormat::TextPretty); + let printer = Arc::new(printer); + let mcp_client = jp_mcp::Client::default(); + let router = detached_router(); + + let (interrupt_tx, interrupts) = TurnInterrupts::channel(); + + let stop_handle = tokio::spawn({ + let stalled = Arc::clone(&stalled); + async move { + stalled.notified().await; + interrupt_tx + .send(InterruptAction::Stop) + .await + .expect("the turn is still running"); + } + }); + + let result = run_turn_loop( + provider.clone() as Arc, + &model, + &config, + &router, + &mcp_client, + root, + false, // is_tty + &[], // attachments + &lock, + ToolChoice::Auto, + &[], // tools + printer.clone(), + Arc::new(MockPromptBackend::new()), + ToolCoordinator::new(config.conversation.tools.clone(), empty_executor_source()), + ChatRequest::from("What is 2+2?"), + InvocationContext::default(), + PendingStreamTrim::default(), + router.turn_interrupt(), + interrupts, + ) + .await; + + stop_handle.await.unwrap(); + assert!(result.is_ok(), "the turn should complete: {result:?}"); + + assert_eq!( + provider.calls(), + 1, + "a stop ends the turn, so nothing is asked again" + ); + + let content = fs + .read_test_events_raw(&conv_id) + .expect("events should be persisted"); + + assert!( + content.contains("Thinking about it."), + "what the assistant had produced is kept.\nFile contents:\n{content}" + ); + assert!( + !content.contains("never asked"), + "the turn ended, so the provider was never asked again.\nFile contents:\n{content}" + ); + })) + .await; + + assert!(test_result.is_ok(), "Test timed out after 10 seconds"); +} + #[tokio::test] async fn test_interrupt_stop_during_streaming_persists_content() { // A Ctrl-C press is routed to the streaming loop's registered interrupt @@ -395,7 +697,8 @@ async fn test_interrupt_stop_during_streaming_persists_content() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -495,7 +798,8 @@ async fn a_completed_block_is_persisted_before_the_turn_ends() { ChatRequest::from("What is 2+2?"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -579,7 +883,8 @@ async fn a_refusal_takes_back_content_it_had_persisted() { ChatRequest::from("something declined"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -666,7 +971,8 @@ async fn test_streaming_interrupt_menu_cancel_escalates() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -750,7 +1056,8 @@ async fn test_normal_completion_persists_content() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -836,7 +1143,8 @@ async fn premature_stream_end_without_finished_returns_error() { ChatRequest::from("hi"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ), ) .await @@ -900,7 +1208,8 @@ async fn premature_stream_end_exhausts_retry_budget() { ChatRequest::from("hi"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ), ) .await @@ -980,7 +1289,8 @@ async fn output_ceiling_ends_turn_without_re_requesting() { ChatRequest::from("hi"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ), ) .await @@ -1085,7 +1395,8 @@ async fn orphan_tool_call_is_sanitized_before_provider_request() { ChatRequest::from("new query"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -1170,7 +1481,8 @@ async fn test_tool_call_cycle_completes_with_followup() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -1468,7 +1780,8 @@ async fn test_tool_interrupt_menu_cancel_escalates() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -1618,7 +1931,8 @@ async fn test_tool_stop_on_interrupt_commits_responses_without_follow_up() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -1761,7 +2075,8 @@ async fn test_interrupt_during_tool_prompt_completes_turn_early() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -1873,7 +2188,8 @@ async fn test_multiple_tool_calls_in_sequence() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -1964,7 +2280,8 @@ async fn test_empty_tool_response_continues_cycle() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -2110,7 +2427,8 @@ async fn test_tool_restart_on_interrupt() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -2232,7 +2550,8 @@ async fn test_merged_stream_exits_after_tool_response() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -2360,7 +2679,8 @@ async fn test_tool_call_with_run_mode_ask_approves() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -2503,7 +2823,8 @@ async fn test_tool_call_with_run_mode_ask_skips() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -2657,7 +2978,8 @@ async fn test_permission_prompt_follows_interactive_not_is_tty() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -2781,7 +3103,8 @@ async fn test_tool_call_with_run_mode_unattended() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -2929,7 +3252,8 @@ async fn test_tool_call_with_run_mode_skip() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -3133,7 +3457,8 @@ async fn test_multiple_tools_with_different_run_modes() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -3282,7 +3607,8 @@ async fn test_tool_call_returns_error() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -3518,7 +3844,8 @@ async fn test_waiting_indicator_shows_during_delay() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -3619,7 +3946,8 @@ async fn test_waiting_indicator_survives_keep_alive_and_shows_status() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -3734,7 +4062,8 @@ async fn test_waiting_indicator_cleared_before_retry_notice() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -3822,7 +4151,8 @@ async fn test_waiting_indicator_not_shown_when_disabled() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -3902,7 +4232,8 @@ async fn test_waiting_indicator_not_shown_for_non_tty() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -3983,7 +4314,8 @@ async fn test_waiting_indicator_follows_stderr_not_stdout() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -4169,7 +4501,8 @@ async fn test_multi_part_tool_call_shows_preparing_spinner() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -4257,7 +4590,8 @@ async fn test_turn_start_event_is_emitted() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -4321,7 +4655,8 @@ async fn test_turn_start_index_increments_across_turns() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -4357,7 +4692,8 @@ async fn test_turn_start_index_increments_across_turns() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -4453,7 +4789,8 @@ async fn test_markdown_flushed_before_tool_header() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -4639,7 +4976,8 @@ async fn test_parallel_tool_calls_rendered_atomically() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -4799,7 +5137,8 @@ async fn test_single_tool_call_rendered_with_args() { chat_request.clone(), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -5046,7 +5385,8 @@ async fn a_running_tools_stderr_reaches_the_progress_window() { ChatRequest::from("Build it"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -5144,7 +5484,8 @@ async fn parallel_tools_label_their_window_rows() { ChatRequest::from("Run both"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -5266,7 +5607,8 @@ async fn a_tool_result_survives_a_live_window() { ChatRequest::from("Run both"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -5382,7 +5724,8 @@ async fn a_sink_survives_the_re_spawn_an_answer_triggers() { ChatRequest::from("Ask then work"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -5516,7 +5859,8 @@ async fn a_tool_can_opt_out_of_the_progress_window() { ChatRequest::from("Run both"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -5741,7 +6085,8 @@ async fn a_tool_prompt_hides_the_window_and_restores_it() { ChatRequest::from("Ask me"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -6237,7 +6582,8 @@ async fn test_tool_with_single_inquiry() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -6366,7 +6712,8 @@ async fn test_secret_question_without_tty_fails_tool() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -6476,7 +6823,8 @@ async fn test_secret_question_with_assistant_target_fails_tool() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -6579,7 +6927,8 @@ async fn test_secret_prompter_answer_is_redacted() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -6686,7 +7035,8 @@ async fn test_secret_static_answer_is_redacted() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -6794,7 +7144,8 @@ async fn test_static_answer_records_answered_inquiry() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -6910,7 +7261,8 @@ async fn test_remembered_answer_cache_hit_records_new_inquiry_pair() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -7033,7 +7385,8 @@ async fn test_tool_with_multiple_inquiries() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -7186,7 +7539,8 @@ async fn test_parallel_tools_one_with_inquiry() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -7325,7 +7679,8 @@ async fn test_parallel_tools_both_with_inquiries() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -7476,7 +7831,8 @@ async fn test_retry_counter_resets_on_successful_event() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -7616,7 +7972,8 @@ async fn test_unavailable_tool_before_approved_does_not_panic() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -7727,7 +8084,8 @@ async fn test_inquiry_failure_marks_tool_as_error() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -7922,7 +8280,8 @@ async fn test_live_header_uses_configured_model_id_not_provider_returned() { chat_request, InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -8038,7 +8397,8 @@ async fn reasoning_before_a_tool_call_shades_the_tool_chrome() { ChatRequest::from("use the tool"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); @@ -8167,7 +8527,8 @@ async fn test_rebuild_cap_stops_a_provider_that_keeps_requesting_rebuilds() { ChatRequest::from("repair this"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -8259,7 +8620,8 @@ async fn test_refused_rebuild_clears_the_retry_line() { ChatRequest::from("answer this"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; @@ -8344,7 +8706,8 @@ async fn test_refused_rebuild_persists_streamed_content() { ChatRequest::from("answer this"), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await; diff --git a/crates/jp_cli/src/cmd/query_tests.rs b/crates/jp_cli/src/cmd/query_tests.rs index 71cee3596..9a2f71876 100644 --- a/crates/jp_cli/src/cmd/query_tests.rs +++ b/crates/jp_cli/src/cmd/query_tests.rs @@ -43,7 +43,7 @@ use crate::{ cmd::target::{ConversationTarget, PickerFilter}, config_pipeline::ConfigPipeline, resolve_config, - signals::testing::detached_router, + signals::testing::{detached_router, test_router}, }; fn make_partial_with_tools() -> PartialAppConfig { @@ -331,7 +331,7 @@ async fn an_interrupt_during_mcp_startup_stops_the_turn_before_it_runs() { }); let (mcp_servers, _lines) = startup_set(joins, vec![McpServerId::new("bookworm")]); - let router = detached_router(); + let (router, signals) = test_router(); let (printer, _out, _err) = Printer::memory(OutputFormat::TextPretty); let inputs = TurnInputs { @@ -347,16 +347,17 @@ async fn an_interrupt_during_mcp_startup_stops_the_turn_before_it_runs() { workspace_id: workspace.id().clone(), pending_trim: PendingStreamTrim::default(), mcp_servers, + // Ctrl-C is how this turn is interrupted, and this test drives that + // through the router rather than over a channel. + interrupts: TurnInterrupts::none(), }; - // Registered where the lock is taken, so the request reaches it even though - // the turn has not started. The notice waits on the channel until `run` - // polls it. - let interrupt = router.turn_interrupt(conversation_id); - assert!( - router.interrupt_scope(conversation_id), - "the handler exists as soon as the conversation is locked" - ); + // Registered where the lock is taken, so the press reaches it even though + // the turn has not started. It is the only handler on the stack at this + // point, so it is the one a Ctrl-C finds, and the notice waits on the + // channel until `run` polls it. + let interrupt = router.turn_interrupt(); + signals.interrupt().await; let stream = lock.events().clone(); tokio::time::timeout( @@ -408,7 +409,8 @@ async fn run_mock_turn( ChatRequest::from(prompt), InvocationContext::default(), PendingStreamTrim::default(), - router.turn_interrupt(lock.id()), + router.turn_interrupt(), + TurnInterrupts::none(), ) .await .unwrap(); diff --git a/crates/jp_cli/src/signals.rs b/crates/jp_cli/src/signals.rs index b76877fc8..82206216b 100644 --- a/crates/jp_cli/src/signals.rs +++ b/crates/jp_cli/src/signals.rs @@ -46,7 +46,6 @@ use std::{ }; use futures::{Stream, StreamExt as _}; -use jp_conversation::ConversationId; use tokio::{ runtime::{Handle, Runtime}, sync::mpsc::{self, error::TrySendError}, @@ -185,30 +184,17 @@ impl SignalRouter { /// fires, and resolves the notice with the outcome. #[must_use] pub fn push_handler(&self) -> (InterruptGuard, mpsc::Receiver) { - self.inner.push_handler(None) - } - - /// Register an interrupt handler scope that can also be interrupted by - /// name. - /// - /// Behaves as [`push_handler`] for a Ctrl-C, which still goes to whichever - /// handler is topmost. - /// The scope only matters to [`interrupt_scope`], for interrupts that - /// arrive from somewhere with no notion of "topmost". - /// - /// [`interrupt_scope`]: Self::interrupt_scope - /// [`push_handler`]: Self::push_handler - #[must_use] - pub fn push_handler_for( - &self, - conversation: ConversationId, - ) -> (InterruptGuard, mpsc::Receiver) { - self.inner.push_handler(Some(conversation)) + self.inner.push_handler() } /// Register the handler a turn is driven through, for as long as its /// conversation is locked. /// + /// Reached by a Ctrl-C the same way any other handler is, by being topmost. + /// An interrupt that names a conversation does not arrive here at all: it + /// reaches the turn over its own channel, carrying the action it has + /// already decided. + /// /// Registered by whoever takes the lock, so every span between taking it /// and releasing it reads the same receiver: waiting on MCP servers, /// looking up the model, and the turn itself. @@ -216,30 +202,10 @@ impl SignalRouter { /// the rest of it — a press delivered to a receiver nobody polls again is /// reported as delivered and then dropped. #[must_use] - pub fn turn_interrupt(&self, conversation: ConversationId) -> TurnInterrupt { - let (guard, rx) = self.inner.push_handler(Some(conversation)); + pub fn turn_interrupt(&self) -> TurnInterrupt { + let (guard, rx) = self.inner.push_handler(); TurnInterrupt { _guard: guard, rx } } - - /// Interrupt one named scope, leaving every other handler alone. - /// - /// For interrupts that arrive with a target rather than from a keyboard. - /// A Ctrl-C means "whatever I am looking at", and the topmost handler is - /// the right guess; a request naming a conversation means that - /// conversation, and with several turns running the topmost handler is very - /// likely the wrong one. - /// - /// Outside the escalation ladder: a repeat re-asks the same turn to stop. - /// A Ctrl-C escalates to cancelling the shutdown token and then to exiting - /// the process, neither of which a request naming one conversation should - /// reach. - /// - /// Returns whether the interrupt reached a handler for that scope. - /// `false` means nothing is registered for it, which usually means the turn - /// has already finished. - pub fn interrupt_scope(&self, conversation: ConversationId) -> bool { - self.inner.notify_scope(conversation) - } } /// The interrupt handler a turn is driven through. @@ -288,13 +254,6 @@ pub struct InterruptNotice { /// with two turns in flight the stack interleaves, so "second from the top" /// is very likely a handler belonging to the other turn. from: HandlerId, - - /// The conversation this press named, when it named one. - /// - /// A decline stays inside it. - /// `None` is a keypress, which belongs to whatever the user is looking at - /// and so declines down the whole stack. - scope: Option, } impl fmt::Debug for InterruptNotice { @@ -314,16 +273,13 @@ impl InterruptNotice { /// The handler declined this press. /// - /// Notifies the next handler below the one this press reached, skipping any - /// that belong to another conversation. + /// Notifies the next handler below the one this press reached. /// The press keeps its place on the escalation ladder: nothing has acted on /// it yet. /// - /// A keypress with nothing left below it requests a graceful shutdown. - /// A press that named a conversation does not: a request about one turn - /// must not be able to shut the process down. + /// With nothing left below it, the press requests a graceful shutdown. pub fn decline(self) { - self.inner.notify_below(self.from, self.scope); + self.inner.notify_below(self.from); } } @@ -347,21 +303,6 @@ struct HandlerId(u64); struct RegisteredHandler { id: HandlerId, - /// The conversation this handler is a scope for, when something other than - /// a keypress might want to interrupt it specifically. - /// - /// A Ctrl-C is aimed at whatever the user is looking at, which is the - /// topmost handler, so the terminal path ignores this. - /// An interrupt arriving over a protocol names its target instead: several - /// turns can be running at once, and stopping the wrong one is worse than - /// stopping nothing. - /// - /// The id itself, not a rendering of it. - /// An id has more than one spelling, and comparing one spelling to another - /// matches nothing while looking exactly like a stop button that does not - /// work. - scope: Option, - /// Notifies the handler's event loop that SIGINT arrived. /// The event loop runs the interrupt logic; the router never does. notify_tx: mpsc::Sender, @@ -487,7 +428,7 @@ impl RouterInner { } if let Some((id, notify_tx)) = self.topmost() { - return match notify_tx.try_send(self.notice(id, None)) { + return match notify_tx.try_send(self.notice(id)) { // A full channel means the handler already has a pending // interrupt notification; nothing to add. The undelivered // notice is dropped unresolved, leaving this press on the @@ -518,14 +459,10 @@ impl RouterInner { } /// Build a notice for a press about to be delivered to `id`. - /// - /// `scope` is the conversation the press named, and bounds where a decline - /// may go next. - fn notice(self: &Arc, id: HandlerId, scope: Option) -> InterruptNotice { + fn notice(self: &Arc, id: HandlerId) -> InterruptNotice { InterruptNotice { inner: Arc::clone(self), from: id, - scope, } } @@ -540,20 +477,13 @@ impl RouterInner { /// Register a handler scope: push a fresh notification channel onto the /// stack and return the deregistration guard plus the receiver. - fn push_handler( - self: &Arc, - scope: Option, - ) -> (InterruptGuard, mpsc::Receiver) { + fn push_handler(self: &Arc) -> (InterruptGuard, mpsc::Receiver) { let (notify_tx, notify_rx) = mpsc::channel(1); let id = HandlerId(self.next_handler_id.fetch_add(1, Ordering::Relaxed)); self.stack .lock() .expect("handler stack lock poisoned") - .push(RegisteredHandler { - id, - scope, - notify_tx, - }); + .push(RegisteredHandler { id, notify_tx }); ( InterruptGuard { @@ -564,38 +494,6 @@ impl RouterInner { ) } - /// Notify the handler registered for `scope`, if one still is. - /// - /// Searched from the top down, so the innermost handler for a conversation - /// is the one reached, matching how a Ctrl-C finds the innermost handler - /// overall. - /// - /// Returns whether anything was notified. - /// `false` means the scope has no handler, usually because its work already - /// finished, and is not an error: there was nothing left to interrupt. - fn notify_scope(self: &Arc, scope: ConversationId) -> bool { - let found = self - .stack - .lock() - .expect("handler stack lock poisoned") - .iter() - .rev() - .find(|handler| handler.scope == Some(scope)) - .map(|handler| (handler.id, handler.notify_tx.clone())); - - match found { - // A full channel means the handler already has a notice it has not - // picked up yet, so this one has nothing to add and is dropped - // unresolved. The handler has still been told, which is what the - // caller is asking about. - Some((id, tx)) => !matches!( - tx.try_send(self.notice(id, Some(scope))), - Err(TrySendError::Closed(_)) - ), - None => false, - } - } - /// Remove a handler by id. /// /// Id-based rather than positional so guards can drop in any order: early @@ -607,18 +505,16 @@ impl RouterInner { } } - /// Notify the first handler below `from`, skipping any outside `scope`. + /// Notify the first handler below `from`. /// /// Walking from `from` rather than from the top of the stack is what keeps /// a decline inside the turn that declined: two turns in flight interleave /// their handlers, so the entry below the *topmost* one often belongs to /// the other turn. /// - /// With nothing left below, a keypress (`scope` of `None`) requests a - /// graceful shutdown, which is the terminal's escalation ladder. - /// A press that named a conversation stops instead: it asked about one - /// turn, and shutting the process down is not among the answers. - fn notify_below(self: &Arc, from: HandlerId, scope: Option) { + /// With nothing left below, the press requests a graceful shutdown, which + /// is the last rung of the terminal's escalation ladder. + fn notify_below(self: &Arc, from: HandlerId) { let next = { let stack = self.stack.lock().expect("handler stack lock poisoned"); stack @@ -626,25 +522,22 @@ impl RouterInner { .rev() .skip_while(|handler| handler.id != from) .skip(1) - .find(|handler| scope.is_none() || handler.scope == scope) .map(|handler| (handler.id, handler.notify_tx.clone())) + .next() }; let Some((id, notify_tx)) = next else { - if scope.is_none() { - self.shutdown_token.cancel(); - } + self.shutdown_token.cancel(); return; }; - // A closed channel means that handler's event loop is already gone. For - // a keypress that leaves the ladder with nothing further to try, so it - // falls through to shutdown; a scoped press has nowhere else to go. - let closed = matches!( - notify_tx.try_send(self.notice(id, scope)), + // A closed channel means that handler's event loop is already gone, + // which leaves the ladder with nothing further to try, so it falls + // through to shutdown. + if matches!( + notify_tx.try_send(self.notice(id)), Err(TrySendError::Closed(_)) - ); - if closed && scope.is_none() { + ) { self.shutdown_token.cancel(); } } diff --git a/crates/jp_cli/src/signals_tests.rs b/crates/jp_cli/src/signals_tests.rs index 37990c850..1a6ba75de 100644 --- a/crates/jp_cli/src/signals_tests.rs +++ b/crates/jp_cli/src/signals_tests.rs @@ -6,144 +6,7 @@ use super::*; /// Push a handler scope onto the router state. fn push_handler(inner: &Arc) -> (InterruptGuard, mpsc::Receiver) { - inner.push_handler(None) -} - -/// A fixed conversation id, distinct per `secs`. -fn conversation(secs: u64) -> ConversationId { - ConversationId::try_from( - chrono::DateTime::::UNIX_EPOCH + Duration::from_secs(secs), - ) - .unwrap() -} - -/// A targeted interrupt reaches the named scope and nothing else. -/// -/// The failure this guards against is stopping the wrong turn: with several -/// running, the topmost handler is very likely not the one that was asked for. -#[test] -fn a_scoped_interrupt_notifies_only_its_own_scope() { - let inner = RouterInner::new(Duration::from_secs(2)); - let wanted = conversation(1_700_000_000); - let other = conversation(1_700_000_001); - - let (_guard_wanted, mut rx_wanted) = inner.push_handler(Some(wanted)); - // Pushed after, so it is topmost and would be the one a Ctrl-C reached. - let (_guard_other, mut rx_other) = inner.push_handler(Some(other)); - let (_guard_plain, mut rx_plain) = push_handler(&inner); - - assert!(inner.notify_scope(wanted)); - - assert!(took_notice(&mut rx_wanted)); - assert_eq!(recv_error(&mut rx_other), Some(TryRecvError::Empty)); - assert_eq!(recv_error(&mut rx_plain), Some(TryRecvError::Empty)); -} - -/// A turn registers a handler per phase under the same conversation, and the -/// innermost one is the one being polled. -/// -/// The outer turn-level handler is read only between phases, so reaching it -/// while a response streams or a tool runs would leave a targeted interrupt -/// waiting for that phase to finish. -#[test] -fn a_scoped_interrupt_reaches_the_innermost_handler_for_its_scope() { - let inner = RouterInner::new(Duration::from_secs(2)); - let id = conversation(1_700_000_000); - - // The turn-level handler, then the one a streaming or executing phase adds. - let (_turn_guard, mut rx_turn) = inner.push_handler(Some(id)); - let (phase_guard, mut rx_phase) = inner.push_handler(Some(id)); - - assert!(inner.notify_scope(id)); - - assert!(took_notice(&mut rx_phase)); - assert_eq!(recv_error(&mut rx_turn), Some(TryRecvError::Empty)); - - // The phase ends, and the turn-level handler is innermost again. - drop(phase_guard); - assert!(inner.notify_scope(id)); - assert!(took_notice(&mut rx_turn)); -} - -/// A repeat while the handler has not picked up the first one still counts as -/// reached: the turn has been told, so reporting otherwise would read as "that -/// conversation is not running". -#[test] -fn a_repeat_interrupt_with_one_still_pending_counts_as_reached() { - let inner = RouterInner::new(Duration::from_secs(2)); - let id = conversation(1_700_000_000); - let (_guard, mut rx) = inner.push_handler(Some(id)); - - // The channel holds one notice, so the second send finds it full. - assert!(inner.notify_scope(id)); - assert!(inner.notify_scope(id)); - - assert!(took_notice(&mut rx)); - assert_eq!( - recv_error(&mut rx), - Some(TryRecvError::Empty), - "the repeat added nothing to a handler that had not looked yet" - ); -} - -/// A scope whose handler's event loop is gone is not reached, even though its -/// guard has not dropped yet. -#[test] -fn a_scope_whose_receiver_is_gone_is_not_reached() { - let inner = RouterInner::new(Duration::from_secs(2)); - let id = conversation(1_700_000_000); - let (_guard, rx) = inner.push_handler(Some(id)); - - drop(rx); - - assert!(!inner.notify_scope(id)); -} - -/// A scope with no handler is not an error: its work already finished, so there -/// was nothing left to interrupt. -#[test] -fn interrupting_an_unknown_scope_reports_that_nothing_was_reached() { - let inner = RouterInner::new(Duration::from_secs(2)); - let (_guard, mut rx) = push_handler(&inner); - - assert!(!inner.notify_scope(conversation(1_700_000_000))); - - assert_eq!( - recv_error(&mut rx), - Some(TryRecvError::Empty), - "an unscoped handler is not a fallback target" - ); -} - -/// A dropped guard takes its scope with it, so a later interrupt finds nothing -/// rather than a stale channel. -#[test] -fn a_finished_scope_is_no_longer_reachable() { - let inner = RouterInner::new(Duration::from_secs(2)); - let id = conversation(1_700_000_000); - - let (guard, _rx) = inner.push_handler(Some(id)); - assert!(inner.notify_scope(id)); - - drop(guard); - assert!(!inner.notify_scope(id)); -} - -/// A Ctrl-C still goes to whichever handler is topmost, scoped or not. -/// -/// The scope is extra information for targeted interrupts, not a change to how -/// the keyboard path chooses. -#[test] -fn a_scope_does_not_change_where_a_keypress_lands() { - let inner = RouterInner::new(Duration::from_secs(2)); - - let (_guard_bottom, mut rx_bottom) = inner.push_handler(Some(conversation(1_700_000_000))); - let (_guard_top, mut rx_top) = inner.push_handler(Some(conversation(1_700_000_001))); - - assert_eq!(inner.route_interrupt(Instant::now()), Routed::Handler); - - assert!(took_notice(&mut rx_top)); - assert_eq!(recv_error(&mut rx_bottom), Some(TryRecvError::Empty)); + inner.push_handler() } /// Take a delivered press off the channel and drop it unresolved. @@ -359,7 +222,7 @@ fn decline_notifies_next_handler_down() { let (_guard_bottom, mut rx_bottom) = push_handler(&inner); let (guard_top, mut rx_top) = push_handler(&inner); - inner.notify_below(guard_top.id, None); + inner.notify_below(guard_top.id); assert!(took_notice(&mut rx_bottom)); assert_eq!(recv_error(&mut rx_top), Some(TryRecvError::Empty)); @@ -371,66 +234,12 @@ fn decline_with_single_handler_requests_shutdown() { let inner = RouterInner::new(Duration::from_secs(2)); let (guard, mut rx) = push_handler(&inner); - inner.notify_below(guard.id, None); + inner.notify_below(guard.id); assert_eq!(recv_error(&mut rx), Some(TryRecvError::Empty)); assert!(inner.shutdown_token.is_cancelled()); } -/// A declined press walks down from the handler that declined it, not from the -/// top of the stack. -/// -/// Two turns in flight interleave their handlers, so the entry below the -/// topmost one routinely belongs to the other turn. -/// Declining A's press has to reach A's next handler, not whichever turn -/// happens to be on top. -#[test] -fn a_decline_stays_inside_its_own_conversation() { - let inner = RouterInner::new(Duration::from_secs(2)); - let a = conversation(1_700_000_000); - let b = conversation(1_700_000_001); - - // Turn A registers its turn-level and tool handlers, then turn B starts and - // pushes its own on top. - let (_a_turn, mut rx_a_turn) = inner.push_handler(Some(a)); - let (a_tool_guard, mut rx_a_tool) = inner.push_handler(Some(a)); - let (_b_turn, mut rx_b_turn) = inner.push_handler(Some(b)); - let (_b_stream, mut rx_b_stream) = inner.push_handler(Some(b)); - - // A's tool handler declines the press it was given. - inner.notify_below(a_tool_guard.id, Some(a)); - - assert!( - took_notice(&mut rx_a_turn), - "the press belongs to A, so A's next handler answers it" - ); - assert_eq!(recv_error(&mut rx_a_tool), Some(TryRecvError::Empty)); - assert_eq!( - recv_error(&mut rx_b_turn), - Some(TryRecvError::Empty), - "B is topmost but the press was never about B" - ); - assert_eq!(recv_error(&mut rx_b_stream), Some(TryRecvError::Empty)); - assert!(!inner.shutdown_token.is_cancelled()); -} - -/// A scoped press that nothing below it can answer stops there. -/// -/// A keypress escalates to shutdown at this point, which is right for someone -/// at a terminal. -/// A request naming one conversation must not be able to shut the process down. -#[test] -fn a_declined_scoped_press_does_not_request_shutdown() { - let inner = RouterInner::new(Duration::from_secs(2)); - let id = conversation(1_700_000_000); - let (guard, mut rx) = inner.push_handler(Some(id)); - - inner.notify_below(guard.id, Some(id)); - - assert_eq!(recv_error(&mut rx), Some(TryRecvError::Empty)); - assert!(!inner.shutdown_token.is_cancelled()); -} - #[test] fn escalation_counter_bumps_and_resets() { let mut state = EscalationState::new(Duration::from_secs(2)); diff --git a/crates/jp_plugin/src/message.rs b/crates/jp_plugin/src/message.rs index 22fdf8ada..e2e29d8e3 100644 --- a/crates/jp_plugin/src/message.rs +++ b/crates/jp_plugin/src/message.rs @@ -137,7 +137,7 @@ pub enum PluginToHost { /// Ask the host to run a turn on a conversation. Query(QueryRequest), - /// Ask the host to interrupt the turn running on a conversation. + /// Ask the host to change what a running turn is doing. Interrupt(InterruptRequest), /// Print user-facing output through JP's printer. @@ -175,20 +175,102 @@ impl PluginToHost { Self::ArchiveConversation(m) | Self::ReadDraft(m) => m.id.as_deref(), Self::SetTitle(m) => m.id.as_deref(), Self::WriteDraft(m) => m.id.as_deref(), + Self::Interrupt(m) => m.id.as_deref(), // Not requests: nothing is waiting on an answer to any of these. - Self::Ready(_) - | Self::Interrupt(_) - | Self::Print(_) - | Self::Log(_) - | Self::Describe(_) - | Self::Exit(_) => None, + Self::Ready(_) | Self::Print(_) | Self::Log(_) | Self::Describe(_) | Self::Exit(_) => { + None + } } } } // --- Host-to-Plugin messages --- +/// How the host renders what it prints. +/// +/// A plugin reads this to decide the shape of its own output, so `jp --format +/// json` reaches a plugin's listings the way it reaches the host's own commands +/// and a caller does not have to learn a separate flag per plugin. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum OutputFormat { + /// Plain text, with no ANSI colors and no unicode decoration. + #[default] + Text, + + /// Text with ANSI colors and unicode decoration. + TextPretty, + + /// Compact JSON, one line per print. + Json, + + /// Indented JSON. + JsonPretty, +} + +impl OutputFormat { + /// Whether output should be machine-readable. + #[must_use] + pub const fn is_json(self) -> bool { + matches!(self, Self::Json | Self::JsonPretty) + } + + /// Whether JSON output should be indented. + #[must_use] + pub const fn is_json_pretty(self) -> bool { + matches!(self, Self::JsonPretty) + } + + /// Whether text output can carry ANSI colors and unicode decoration. + #[must_use] + pub const fn is_pretty(self) -> bool { + matches!(self, Self::TextPretty) + } +} + +/// Who holds a conversation. +/// +/// A conversation is locked for the length of a turn, so this says whether one +/// is running, and whether it is the reader's to interrupt. +/// A turn in another process can be waited for but not signalled from here. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum LockState { + /// Nobody. + /// No turn is running. + #[default] + Free, + + /// A turn in another process. + Elsewhere, + + /// A turn in the host answering this request. + Here, +} + +impl LockState { + /// Whether no turn is running. + /// + /// Takes a reference because `skip_serializing_if` calls it with one. + #[must_use] + pub const fn is_free(&self) -> bool { + matches!(self, Self::Free) + } + + /// Whether a turn is running, wherever it is. + #[must_use] + pub const fn is_held(&self) -> bool { + !self.is_free() + } + + /// Whether the running turn can be interrupted through this connection. + #[must_use] + pub const fn is_here(&self) -> bool { + matches!(self, Self::Here) + } +} + /// The `init` message sent to the plugin on startup. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct InitMessage { @@ -227,6 +309,18 @@ pub struct InitMessage { /// that stderr output matches the host's `-v` flags. #[serde(default)] pub log_level: u8, + + /// The shape the host's own output takes, resolved from `--format`. + /// + /// A plugin that prints listings or records should match it, so one flag + /// governs the whole invocation. + /// + /// Reads as [`OutputFormat::Text`] when the host is old enough not to send + /// it, which is the shape plugins printed before they could ask. + /// That fallback is why this needs no protocol version of its own: there is + /// nothing a plugin has to refuse to run without. + #[serde(default)] + pub output_format: OutputFormat, } /// Workspace metadata included in the `init` message. @@ -283,6 +377,21 @@ pub struct EventsResponse { /// The conversation ID. pub conversation: String, + /// Who holds this conversation, if anyone. + /// + /// Read from the conversation lock, which is the only authoritative answer: + /// a transcript ending in a request looks identical whether a turn is + /// running, was interrupted, or failed outright. + #[serde(default, skip_serializing_if = "LockState::is_free")] + pub lock: LockState, + + /// The conversation's title, if it has one. + /// + /// Saves a plugin from asking for the whole conversation list to label one + /// conversation, which reads every conversation's metadata. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, + /// Serialized conversation events. pub data: Vec, } @@ -426,27 +535,105 @@ pub struct QueryRequest { pub cfg: Vec, } -/// Ask the host to interrupt the turn running on a conversation. +/// Ask the host to change what a running turn is doing. /// -/// Reaches the turn the way a Ctrl-C does, so it asks the turn to wrap up: -/// partial output is kept and the conversation is left in a state a later turn -/// can continue from. +/// Reaches the turn the way the terminal's Ctrl-C menu does, with the choice +/// already made: `stop` keeps what the assistant has produced and ends the +/// turn, `abort` discards it, and `reply` interrupts the assistant and answers +/// it without leaving the turn. /// /// A repeat re-asks the same turn. /// It does not escalate the way a terminal's second and third Ctrl-C do, since /// cancelling the host's shutdown token or exiting the process is not something /// a request naming one conversation should reach. /// -/// Fire-and-forget: the host sends no acknowledgement, because what the -/// interrupt did shows up in the conversation itself. -/// The outcome of the turn still arrives as the reply to the original `query`. +/// Answered with [`HostToPlugin::Done`] when `id` is set, and with +/// [`HostToPlugin::Error`] when there is no turn left to reach — which is how +/// a client tells "interrupted" from "it had already finished" without guessing +/// from a timeout. +/// Without an `id` it stays fire-and-forget. +/// The outcome of the turn itself still arrives as the reply to the original +/// `query`. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct InterruptRequest { - /// The conversation whose turn should stop. + /// Optional request correlation ID. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub id: Option, + + /// The conversation whose turn to reach. /// /// Required, and not a convenience: a host can be running several turns at /// once, so there is no "the" turn to infer. pub conversation: String, + + /// What the turn should do. + /// + /// Defaults to `stop`. + #[serde(default)] + pub action: InterruptAction, + + /// What to say to the assistant, for `reply`. + /// + /// Required by `reply` and ignored by every other action. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub content: Option, +} + +impl InterruptRequest { + /// Stop the turn running on `conversation`. + #[must_use] + pub fn stop(conversation: String) -> Self { + Self { + id: None, + conversation, + action: InterruptAction::Stop, + content: None, + } + } + + /// Interrupt the turn running on `conversation` and answer it with + /// `content`. + #[must_use] + pub fn reply(conversation: String, content: String) -> Self { + Self { + id: None, + conversation, + action: InterruptAction::Reply, + content: Some(content), + } + } + + /// Correlate the answer to this request with `id`. + #[must_use] + pub fn with_id(mut self, id: String) -> Self { + self.id = Some(id); + self + } +} + +/// What an interrupt asks a running turn to do. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum InterruptAction { + /// Keep what the assistant has produced so far and end the turn. + /// + /// What a terminal user gets from `[s] Stop`. + #[default] + Stop, + + /// End the turn, discarding what the assistant produced since the last + /// persisted boundary. + /// + /// What a terminal user gets from `[a] Abort`. + Abort, + + /// Interrupt the assistant and answer it, staying in the same turn. + /// + /// What the assistant had produced so far is kept as context, `content` + /// follows it as the user's next message, and the turn continues from + /// there. + /// What a terminal user gets from `[r] Reply`. + Reply, } /// Response to `query`, sent once the turn has finished. diff --git a/crates/jp_plugin/src/message_tests.rs b/crates/jp_plugin/src/message_tests.rs index 1e213f52a..065eb9d3a 100644 --- a/crates/jp_plugin/src/message_tests.rs +++ b/crates/jp_plugin/src/message_tests.rs @@ -21,6 +21,7 @@ fn host_init_roundtrip() { options: Map::from_iter([("port".to_owned(), json!(8080))]), args: vec!["--web".to_owned()], log_level: 0, + output_format: OutputFormat::JsonPretty, }); let json = serde_json::to_string(&msg).unwrap(); diff --git a/crates/jp_plugin/src/protocol.rs b/crates/jp_plugin/src/protocol.rs index 9d4df5292..63b45baa5 100644 --- a/crates/jp_plugin/src/protocol.rs +++ b/crates/jp_plugin/src/protocol.rs @@ -16,7 +16,9 @@ use crate::message::{ExitMessage, ReadyMessage}; /// | 5 | `list_configs`, naming the configurations a query can select. | /// | 6 | `query`, with `created` and `query_complete` in reply. | /// | 7 | `interrupt`, for stopping a turn the host is running. | -pub const PROTOCOL_VERSION: u32 = 7; +/// | 8 | `lock` on `events`, saying whether a turn is running. | +/// | 9 | `action` on `interrupt`, and an answer when it carries an `id`. | +pub const PROTOCOL_VERSION: u32 = 9; /// Answer a host's `init`, refusing it when it is too old to serve this plugin. /// diff --git a/crates/jp_task/src/task/title_generator.rs b/crates/jp_task/src/task/title_generator.rs index e54c97d1f..c263c841a 100644 --- a/crates/jp_task/src/task/title_generator.rs +++ b/crates/jp_task/src/task/title_generator.rs @@ -56,6 +56,16 @@ impl TitleGeneratorTask { }) } + /// Ask the model for a title and return it. + /// + /// For a caller that already holds the conversation lock and can write the + /// title itself. + /// Returns `None` when the model answered without a usable title. + pub async fn generate(mut self) -> Result, Box> { + self.update_title().await?; + Ok(self.title) + } + async fn update_title(&mut self) -> Result<(), Box> { trace!(conversation_id = %self.conversation_id, "Updating conversation title."); diff --git a/crates/plugins/command/gui/src/main_tests.rs b/crates/plugins/command/gui/src/main_tests.rs index e8cc34395..b2bfe26a7 100644 --- a/crates/plugins/command/gui/src/main_tests.rs +++ b/crates/plugins/command/gui/src/main_tests.rs @@ -1,6 +1,6 @@ use std::{cell::RefCell, env, io::Cursor}; -use jp_plugin::message::{PathsInfo, ReadyMessage, WorkspaceInfo}; +use jp_plugin::message::{OutputFormat, PathsInfo, ReadyMessage, WorkspaceInfo}; use pretty_assertions::assert_eq; use super::*; @@ -48,6 +48,7 @@ fn init_message(root: &str, args: &[&str]) -> String { options: serde_json::Map::new(), args: args.iter().map(|a| (*a).to_owned()).collect(), log_level: 0, + output_format: OutputFormat::default(), }; format!( diff --git a/crates/plugins/command/serve-web/Cargo.toml b/crates/plugins/command/serve-web/Cargo.toml index bc959e27a..810e11516 100644 --- a/crates/plugins/command/serve-web/Cargo.toml +++ b/crates/plugins/command/serve-web/Cargo.toml @@ -2,7 +2,7 @@ name = "jp-serve-web" authors.workspace = true -description = "Read-only web UI for browsing JP conversations." +description = "Web UI for browsing JP conversations and continuing them." documentation.workspace = true edition.workspace = true homepage.workspace = true @@ -15,16 +15,24 @@ version.workspace = true [dependencies] jp_plugin = { workspace = true } -axum = { workspace = true, features = ["http1", "tokio"] } +axum = { workspace = true, features = ["form", "http1", "json", "query", "tokio"] } chrono = { workspace = true } comrak = { workspace = true } +form_urlencoded = { workspace = true } maud = { workspace = true, features = ["axum"] } +serde = { workspace = true, features = ["derive", "std"] } serde_json = { workspace = true, features = ["std"] } sha2 = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true, features = ["ansi", "env-filter", "fmt", "std"] } +[build-dependencies] +oxc_allocator = { workspace = true } +oxc_parser = { workspace = true } +oxc_semantic = { workspace = true } +oxc_span = { workspace = true } + [dev-dependencies] pretty_assertions = { workspace = true, features = ["std"] } tracing = { workspace = true, features = ["std"] } @@ -35,7 +43,7 @@ workspace = true [package.metadata.jp-registry] id = "serve-web" command = ["serve", "web"] -description = "Read-only web UI for browsing conversations" +description = "Web UI for browsing conversations and continuing them" official = true requires = ["serve"] repository = "https://github.com/dcdpr/jp" diff --git a/crates/plugins/command/serve-web/README.md b/crates/plugins/command/serve-web/README.md new file mode 100644 index 000000000..f98f770e5 --- /dev/null +++ b/crates/plugins/command/serve-web/README.md @@ -0,0 +1,150 @@ +# jp-serve-web + +A command plugin that serves JP conversations over HTTP, and lets you continue +them from a browser. + +Run it with `jp serve-web`. +The server is read-write: it renders the transcript, takes a message from a +composer, and asks the host to run the turn. + +```sh +jp serve-web --bind 127.0.0.1 --port 3000 +``` + +## What it does and does not own + +The plugin is a presentation layer. +It never talks to a model, holds a credential, executes a tool, or writes to a +conversation. +Everything it shows it asked the host for, and every turn it starts the host +runs. + +That split is the reason the protocol exists. +A plugin that ran its own agent loop would need the user's API keys, the tool +registry, the MCP servers, and a second copy of the turn loop to keep in step +with the first. + +| Concern | Owner | +| --------------------------- | ------ | +| Rendering, routing, styling | Plugin | +| Conversation storage | Host | +| Config resolution | Host | +| Model calls and tool runs | Host | +| Interrupting a turn | Host | + +## Protocol + +Needs protocol 9 (`REQUIRED_PROTOCOL`). +The host refuses an older pairing at the handshake rather than failing later, so +a stale `jp` alongside a fresh plugin is an error message and not a mystery. + +| Message | Direction | Used for | +| ---------------------- | --------- | -------------------------------------------------- | +| `list_conversations` | → host | The conversation index | +| `read_events` | → host | One conversation's transcript, title and lock | +| `list_configs` | → host | The configurations a message can name | +| `query` | → host | Start a turn, or start a conversation | +| `created` | ← host | The id of a conversation just created | +| `query_complete` | ← host | That turn finished | +| `interrupt` | → host | Stop a named turn, or answer one without ending it | +| `read_draft` | → host | The message being composed, as the CLI stores it | +| `write_draft` | → host | Save it back, conditional on a revision | +| `archive_conversation` | → host | Move one conversation to the archive | +| `set_title` | → host | Rename one conversation | + +Starting a conversation is answered twice: `created` as soon as there is +somewhere to send the reader, and `query_complete` when the first turn ends. +The client registers both waiters before sending, because a turn that finishes +quickly would otherwise arrive before anything was listening for it. + +## How the page stays current + +There is no push channel yet, so the page polls `/conversations/{id}/messages` +every second while a turn is running and every three when it isn't. +The page says how much of the transcript it holds and how far back its copy is +provisional; the answer carries only the part it does not have, or drew in a +form that has since changed, so a tick that brings nothing new costs one small +response and no re-render. + +That second half is what a tool call needs. +A call is drawn when it is requested and gains its result later, so the entry on +the page is not final when it first appears — and by the time the result lands, +the server's own boundary has moved on to the next call still waiting. +Only the page knows how far back to ask from, so it is the page that says. + +A boundary is not enough for the newest entry, though, so while a turn is +running that one is re-sent on every tick if it is one that can still change. +A run of assistant text is rendered as one block that the next flush adds to, +and it sits at the end — nothing after it is provisional and the count does not +move, so neither the boundary nor the page's own floor reaches back for it. +An entry that is finished the moment it appears, such as the request itself, is +not re-sent; waiting for the first token is the longest stretch of a turn, and +nothing changes on the page during it. + +The host re-reads the conversation from disk on each request, which means a turn +you started in a terminal shows up in the browser too, without a restart. + +Sending while a turn is running interrupts the assistant and answers it, inside +the turn that was already going: what it had produced is kept, the message +follows it, and it carries on from there. +This is Ctrl-C then `[r] Reply` at a terminal, and the conversation is never +unlocked in between, so there is no window in which the message could be refused. + +Events arrive in batches rather than token by token: the turn loop persists at +each streaming boundary, so a page sees a complete assistant response or tool +call at a time. +Per-token updates need the host to push, which is future work. + +Everything on the page works without JavaScript except the polling. +The composer and the stop button are plain form posts, and the transcript is +server-rendered. + +## Endpoints + +| Path | Method | Purpose | +| ------------------------------- | ------ | -------------------------------- | +| `/conversations` | GET | Index | +| `/conversations/{id}` | GET | Transcript and composer | +| `/conversations/{id}/turn` | POST | Send a message | +| `/conversations/{id}/messages` | GET | Transcript as JSON, for the poll | +| `/conversations/{id}/interrupt` | POST | Stop the running turn | +| `/configs` | GET | The configuration chooser | +| `/status` | GET | Whether a turn is in flight | + +The chooser offers the configurations on the load paths by name, and rows for +setting a value directly. +Both reach the host as `--cfg` arguments, the typed values last, so one of those +overrides the same key set by a configuration chosen beside it. + +`/configs` answers with markup rather than data. +The new-conversation form offers the same chooser, and rendering it in one place +is what keeps the two grouping, labelling and posting the choices alike. + +`/status` exists for whoever supervises the process: restarting to pick up a new +build aborts a turn in flight, so a supervisor polls it and waits for `busy` to +go false. + +## Security + +No authentication, and every conversation in the workspace is readable. +Anyone who can reach the port can also start a turn, which spends tokens and +runs whatever tools the conversation allows. + +Binding to a non-loopback address hands that to the network. +The plugin warns on startup when you do. + +Writing requests are refused when they come from a page on another origin, which +is checked from `Sec-Fetch-Site` and `Origin`. +Loopback is no defence on its own here: a form post is not subject to a +preflight, so any site a browser visits can submit one to `127.0.0.1` and start +a turn, and the same-origin policy only stops it reading the answer. +A request that carries neither header — `curl`, a script, another tool — is +allowed, since no browser can be made to omit both. + +## Development + +A file watcher that restarts on save can't be used here: a turn started from the +browser runs inside the host process the plugin is attached to, so restarting on +save aborts whatever the assistant was in the middle of — including the +assistant editing these files. +Poll `/status` and restart only once `busy` is false. diff --git a/crates/plugins/command/serve-web/build.rs b/crates/plugins/command/serve-web/build.rs new file mode 100644 index 000000000..265ca93b5 --- /dev/null +++ b/crates/plugins/command/serve-web/build.rs @@ -0,0 +1,75 @@ +//! Fails the build when an embedded script cannot run. +//! +//! The scripts are served verbatim inside `