From 9c0e237e0af11bc9b929d484cb49bd88f2183a00 Mon Sep 17 00:00:00 2001 From: A Tobey Date: Tue, 4 Aug 2026 18:26:34 -0400 Subject: [PATCH 1/3] =?UTF-8?q?feat(acp):=20kaibo=20acp=20=E2=80=94=20ACP?= =?UTF-8?q?=20v1=20scaffold=20and=20handshake=20(chunk=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a second protocol front door alongside MCP: `kaibo acp` runs kaibo as an Agent Client Protocol v1 agent on stdio, so Zed/Toad/JetBrains-style clients can drive kaibo directly instead of through an MCP client. This chunk is scaffold and handshake only — initialize, session/new, session/prompt, session/set_mode, and session/cancel all answer over the real wire, but session/prompt returns a canned notice instead of running the consult loop. Wiring run_kaish/the model team into the prompt turn is chunk 2. Decisions worth a second look: - `agent-client-protocol` is exact-pinned at `=1.3.0` deliberately, not a floating `"1"`. Read the real crate source (not memory) via a scratch probe crate and docs.rs: the 1.x line reshaped substantially even within minor versions (a builder + typed-dispatch API replacing an older fixed-trait/`AgentSideConnection` shape our starting assumptions expected from a possible older release) and crates.io's `1.x` moved fast (0.x -> 1.x -> 2.0.0 in one summer, 2.0.0 being the v2 DRAFT with its own `unstable_protocol_v2` feature gate). Verified pure-Rust and aws-lc/mimalloc/TLS-free with this dep present (`cargo tree -i` both empty) — its transport stack is async-io/blocking/futures, no TLS of its own. It carries `async-process` transitively for its CLIENT role (spawning a subprocess agent); dead weight on our agent-only surface. - Bumped `rust-version` 1.85 -> 1.88: 1.1.0+ of the crate requires it (async closures in its handler-registration API), so the old MSRV would have silently resolved the pin backward to 1.0.1's superseded API. The installed toolchain (1.96) already exceeds both floors; this only makes the manifest honest about the new one now that ACP is a hard dependency. - No dedicated thread or `LocalSet`, contrary to the working assumption going in. The crate's connection is executor-agnostic and `Send` throughout (`ConnectTo::connect_to` returns `impl Future<..> + Send`, every handler closure must itself be `Send`) — unlike the `!Send` kaish kernel, which stays on its own `KaishWorker` thread for an unrelated reason (rig tools need `Send` futures). `kaibo acp` runs on the existing multi-thread tokio runtime, same as `main`'s other front doors. Chunk 2 drives `KaishWorker` through its already-`Send` channel handle from inside a `Send` ACP handler, the same way `run_kaish` does today. - Session modes are wired to real cast names now, not stubbed: each configured cast becomes one advertised `SessionMode`, mode id == cast name, current mode starts on the config's default cast. `session/set_mode` validates against the roster and records the selection per session; chunk 2 is what actually reads it. - ACP sessions get their own tiny in-memory table (`session-N` ids, the `job-N` style already used in jobs.rs), not the existing consult `SessionStore`/durable store — an ACP session (cwd, MCP servers, mode) is a different shape than a consult session's question/answer replay history. Chunk 2 decides how much of this rides on the existing store. Tested with a real Client role from the same crate driving the Agent builder over a `tokio::io::duplex` (via `tokio-util`'s `compat` bridge to `futures::AsyncRead`/`AsyncWrite`) — actual JSON-RPC bytes on the wire, no mocks, no network: initialize negotiates v1 with no auth methods, session/new returns an id and advertises every configured cast as a mode, and session/prompt streams the canned `session/update` before a `StopReason::EndTurn` response. Full existing suite stays green (637 lib tests + all integration suites, including tests/no_write_path.rs and tests/sandbox.rs unchanged — this adds no filesystem writes). `cargo tree -i aws-lc-rs`/`-i mimalloc` both empty. Local prototype per instructions: no push, no PR. Co-authored-by: Claude Sonnet --- Cargo.lock | 416 ++++++++++++++++++++++++++++++++++++++++++++--- Cargo.toml | 37 ++++- src/acp.rs | 453 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/cli.rs | 43 +++++ src/lib.rs | 1 + src/main.rs | 1 + 6 files changed, 927 insertions(+), 24 deletions(-) create mode 100644 src/acp.rs diff --git a/Cargo.lock b/Cargo.lock index f5fbf58..ee3f1bd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -47,6 +47,56 @@ dependencies = [ "subtle", ] +[[package]] +name = "agent-client-protocol" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d386d6a58f4dbe0ebfa98e915caa54005dabdefd419de3d4678b28cd5752444" +dependencies = [ + "agent-client-protocol-derive", + "agent-client-protocol-schema", + "async-io", + "async-process", + "blocking", + "futures", + "futures-concurrency", + "rustc-hash 2.1.3", + "rustix 1.1.4", + "schemars 1.2.1", + "serde", + "serde_json", + "shell-words", + "tracing", + "uuid", + "windows-sys 0.61.2", +] + +[[package]] +name = "agent-client-protocol-derive" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97b94934d118a69e921d14e94d4af5b3076563318c52662c89a0ecb3f139e1da" +dependencies = [ + "quote", + "syn", +] + +[[package]] +name = "agent-client-protocol-schema" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06679e1542356341f4550ccfb16338b64f37f6af70de2105446ef6fbb078234c" +dependencies = [ + "anyhow", + "derive_more", + "schemars 1.2.1", + "serde", + "serde_json", + "serde_with", + "strum 0.28.0", + "tracing", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -209,6 +259,83 @@ version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bfdc70193dadb9d7287fa4b633f15f90c876915b31f6af17da307fc59c9859a8" +[[package]] +name = "async-channel" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2" +dependencies = [ + "concurrent-queue", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-io" +version = "2.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc" +dependencies = [ + "autocfg", + "cfg-if", + "concurrent-queue", + "futures-io", + "futures-lite", + "parking", + "polling", + "rustix 1.1.4", + "slab", + "windows-sys 0.61.2", +] + +[[package]] +name = "async-lock" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" +dependencies = [ + "event-listener", + "event-listener-strategy", + "pin-project-lite", +] + +[[package]] +name = "async-process" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc50921ec0055cdd8a16de48773bfeec5c972598674347252c0399676be7da75" +dependencies = [ + "async-channel", + "async-io", + "async-lock", + "async-signal", + "async-task", + "blocking", + "cfg-if", + "event-listener", + "futures-lite", + "rustix 1.1.4", +] + +[[package]] +name = "async-signal" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52b5aaafa020cf5053a01f2a60e8ff5dccf550f0f77ec54a4e47285ac2bab485" +dependencies = [ + "async-io", + "async-lock", + "atomic-waker", + "cfg-if", + "futures-core", + "futures-io", + "rustix 1.1.4", + "signal-hook-registry", + "slab", + "windows-sys 0.61.2", +] + [[package]] name = "async-stream" version = "0.3.6" @@ -231,6 +358,12 @@ dependencies = [ "syn", ] +[[package]] +name = "async-task" +version = "4.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" + [[package]] name = "async-trait" version = "0.1.89" @@ -335,6 +468,19 @@ dependencies = [ "generic-array", ] +[[package]] +name = "blocking" +version = "1.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21" +dependencies = [ + "async-channel", + "async-task", + "futures-io", + "futures-lite", + "piper", +] + [[package]] name = "branches" version = "0.4.4" @@ -344,6 +490,15 @@ dependencies = [ "rustc_version", ] +[[package]] +name = "bs58" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf88ba1141d185c399bee5288d850d63b8369520c1eafc32a0430b5b6c287bf4" +dependencies = [ + "tinyvec", +] + [[package]] name = "bstr" version = "1.12.1" @@ -554,6 +709,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "convert_case" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "633458d4ef8c78b72454de2d54fd6ab2e60f9e02be22f3c6104cdc8a4e0fceb9" +dependencies = [ + "unicode-segmentation", +] + [[package]] name = "convert_case" version = "0.11.0" @@ -732,6 +896,32 @@ name = "deranged" version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +dependencies = [ + "serde_core", +] + +[[package]] +name = "derive_more" +version = "2.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d751e9e49156b02b44f9c1815bcb94b984cdcc4396ecc32521c739452808b134" +dependencies = [ + "derive_more-impl", +] + +[[package]] +name = "derive_more-impl" +version = "2.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "799a97264921d8623a957f6c3b9011f3b5492f557bbb7a5a19b7fa6d06ba8dcb" +dependencies = [ + "convert_case 0.10.0", + "proc-macro2", + "quote", + "rustc_version", + "syn", + "unicode-xid", +] [[package]] name = "digest" @@ -820,6 +1010,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "event-listener" +version = "5.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", +] + [[package]] name = "eventsource-stream" version = "0.2.3" @@ -860,6 +1070,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "fixedbitset" +version = "0.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" + [[package]] name = "fnv" version = "1.0.7" @@ -918,6 +1134,19 @@ dependencies = [ "futures-sink", ] +[[package]] +name = "futures-concurrency" +version = "7.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "175cd8cca9e1d45b87f18ffa75088f2099e3c4fe5e2f83e42de112560bea8ea6" +dependencies = [ + "fixedbitset", + "futures-core", + "futures-lite", + "pin-project", + "smallvec", +] + [[package]] name = "futures-core" version = "0.3.32" @@ -941,6 +1170,19 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" +[[package]] +name = "futures-lite" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" +dependencies = [ + "fastrand", + "futures-core", + "futures-io", + "parking", + "pin-project-lite", +] + [[package]] name = "futures-macro" version = "0.3.32" @@ -1157,13 +1399,19 @@ dependencies = [ "futures-core", "futures-sink", "http", - "indexmap", + "indexmap 2.14.0", "slab", "tokio", "tokio-util", "tracing", ] +[[package]] +name = "hashbrown" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" + [[package]] name = "hashbrown" version = "0.15.5" @@ -1526,6 +1774,17 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "indexmap" +version = "1.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99" +dependencies = [ + "autocfg", + "hashbrown 0.12.3", + "serde", +] + [[package]] name = "indexmap" version = "2.14.0" @@ -1630,7 +1889,7 @@ dependencies = [ "bytes", "foldhash 0.1.5", "hifijson", - "indexmap", + "indexmap 2.14.0", "jaq-core", "jaq-std", "num-bigint", @@ -1763,6 +2022,7 @@ dependencies = [ name = "kaibo" version = "0.2.0" dependencies = [ + "agent-client-protocol", "anyhow", "async-trait", "base64 0.22.1", @@ -1787,6 +2047,7 @@ dependencies = [ "tempfile", "thiserror", "tokio", + "tokio-util", "toml", "tracing", "tracing-opentelemetry 0.33.0", @@ -2355,6 +2616,12 @@ dependencies = [ "bytemuck", ] +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -2434,6 +2701,17 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "piper" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" +dependencies = [ + "atomic-waker", + "fastrand", + "futures-io", +] + [[package]] name = "polling" version = "3.11.0" @@ -2807,10 +3085,10 @@ dependencies = [ "fastrand", "futures", "http", - "indexmap", + "indexmap 2.14.0", "rig-core", "rig-derive", - "schemars", + "schemars 1.2.1", "serde", "serde_json", "thiserror", @@ -2835,14 +3113,14 @@ dependencies = [ "futures-timer", "glob", "http", - "indexmap", + "indexmap 2.14.0", "mime", "mime_guess", "ordered-float", "pin-project-lite", "reqwest", "rig-derive", - "schemars", + "schemars 1.2.1", "serde", "serde_json", "thiserror", @@ -2858,7 +3136,7 @@ version = "0.41.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eb868fcebdf3ba425e3afad2e4926bb6d9e1188a856843b00bcee2e15c07424f" dependencies = [ - "convert_case", + "convert_case 0.11.0", "proc-macro-crate", "proc-macro2", "quote", @@ -2892,7 +3170,7 @@ dependencies = [ "pastey", "pin-project-lite", "rmcp-macros", - "schemars", + "schemars 1.2.1", "serde", "serde_json", "thiserror", @@ -3085,6 +3363,18 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "schemars" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cd191f9397d57d581cddd31014772520aa448f65ef991055d7f61582c65165f" +dependencies = [ + "dyn-clone", + "ref-cast", + "serde", + "serde_json", +] + [[package]] name = "schemars" version = "1.2.1" @@ -3211,7 +3501,7 @@ version = "1.0.150" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" dependencies = [ - "indexmap", + "indexmap 2.14.0", "itoa", "memchr", "serde", @@ -3240,6 +3530,38 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_with" +version = "3.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76a5c54c7310e7b8b9577c286d7e399ddd876c3e12b3ed917a8aabc4b96e9e8c" +dependencies = [ + "base64 0.22.1", + "bs58", + "chrono", + "hex", + "indexmap 1.9.3", + "indexmap 2.14.0", + "schemars 0.9.0", + "schemars 1.2.1", + "serde_core", + "serde_json", + "serde_with_macros", + "time", +] + +[[package]] +name = "serde_with_macros" +version = "3.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84d57bc0c8b9a17920c178daa6bb924850d54a9c97ab45194bb8c17ad66bb660" +dependencies = [ + "darling", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "sha1" version = "0.10.6" @@ -3277,6 +3599,12 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shell-words" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc6fe69c597f9c37bfeeeeeb33da3530379845f10be461a66d16d03eca2ded77" + [[package]] name = "shlex" version = "1.3.0" @@ -3309,6 +3637,16 @@ dependencies = [ "tracing", ] +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "simd_cesu8" version = "1.1.1" @@ -3405,7 +3743,16 @@ version = "0.26.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" dependencies = [ - "strum_macros", + "strum_macros 0.26.4", +] + +[[package]] +name = "strum" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +dependencies = [ + "strum_macros 0.28.0", ] [[package]] @@ -3421,6 +3768,18 @@ dependencies = [ "syn", ] +[[package]] +name = "strum_macros" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "subtle" version = "2.6.1" @@ -3574,6 +3933,21 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinyvec" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb4ebadaa0af04fab11ae01eb5f9fdb5f9c5b875506e210e71c07873528baa7f" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.52.3" @@ -3671,7 +4045,7 @@ version = "0.22.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" dependencies = [ - "indexmap", + "indexmap 2.14.0", "serde", "serde_spanned", "toml_datetime 0.6.11", @@ -3685,7 +4059,7 @@ version = "0.25.12+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2153edc6955a6c354fad8f5efd38b6a8769bdccf9fe50f8e1329f81b0baa5d7" dependencies = [ - "indexmap", + "indexmap 2.14.0", "toml_datetime 1.1.1+spec-1.1.0", "toml_parser", "winnow 1.0.3", @@ -3944,8 +4318,8 @@ dependencies = [ "shuttle", "simsimd", "smallvec", - "strum", - "strum_macros", + "strum 0.26.3", + "strum_macros 0.26.4", "tempfile", "thiserror", "tracing", @@ -3990,8 +4364,8 @@ dependencies = [ "bitflags 2.12.1", "memchr", "miette", - "strum", - "strum_macros", + "strum 0.26.3", + "strum_macros 0.26.4", "thiserror", "turso_macros", ] @@ -4323,7 +4697,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909" dependencies = [ "anyhow", - "indexmap", + "indexmap 2.14.0", "wasm-encoder", "wasmparser", ] @@ -4349,7 +4723,7 @@ checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" dependencies = [ "bitflags 2.12.1", "hashbrown 0.15.5", - "indexmap", + "indexmap 2.14.0", "semver", ] @@ -4607,7 +4981,7 @@ checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21" dependencies = [ "anyhow", "heck", - "indexmap", + "indexmap 2.14.0", "prettyplease", "syn", "wasm-metadata", @@ -4638,7 +5012,7 @@ checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" dependencies = [ "anyhow", "bitflags 2.12.1", - "indexmap", + "indexmap 2.14.0", "log", "serde", "serde_derive", @@ -4657,7 +5031,7 @@ checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736" dependencies = [ "anyhow", "id-arena", - "indexmap", + "indexmap 2.14.0", "log", "semver", "serde", diff --git a/Cargo.toml b/Cargo.toml index cfeb091..ed36856 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,12 @@ name = "kaibo" description = "kaibo — a two-phase MCP consult agent that explores a read-only filesystem through kaish builtins" version = "0.2.0" edition = "2021" -rust-version = "1.85" +# Bumped 1.85 -> 1.88 for `agent-client-protocol` (below): 1.1.0+ of that crate +# declares rust-version 1.88.0 (async closures — `async move |x| {..}` — used +# throughout its handler-registration API), and we pin the exact 1.3.0. The +# installed toolchain matters more than this field for actually building; this +# is the honest MSRV floor now that ACP is a hard (non-optional) dependency. +rust-version = "1.88" license = "MIT" [lib] @@ -102,6 +107,25 @@ toml = "0.8" # tree transitively via kaish-kernel, named here so the dep is ours, not implicit. regex = "1" +# The Agent Client Protocol (ACP) front door (`kaibo acp`, src/acp.rs) — an ACP v1 +# agent alongside the MCP server, so Zed/Toad/JetBrains-style clients can drive kaibo +# directly. EXACT-pinned (`=1.3.0`): this is a spec-tracking crate whose crates.io +# `1.x` line moved fast (0.x -> 1.x -> 2.0.0 in one summer) and whose API reshaped +# substantially between minor releases even within 1.x (a builder + typed-dispatch +# API replacing an older fixed-trait shape) — a bare `"1"` would silently ride a +# breaking rewrite in. `2.0.0` implements the ACP v2 DRAFT (an `unstable_protocol_v2` +# feature gate even within the crate itself); we speak v1, the shipping protocol every +# real client (Zed, Toad, JetBrains) negotiates today. Verified pure-Rust and +# aws-lc/mimalloc/TLS-free (`cargo tree -i aws-lc-rs`/`-i mimalloc` both empty with +# this dep present) — its own transport stack is `async-io`/`blocking`/`futures`, no +# TLS at all (stdio only, matching kaibo's own invariant). It carries `async-process` +# transitively (for its CLIENT role, spawning a subprocess agent) — dead code on our +# AGENT-only surface, never invoked, just weight in the dependency tree. Bumped our +# own `rust-version` above (1.85 -> 1.88) to match: 1.1.0+ requires it (async +# closures in its handler-registration API), so a bare `"1"` pin plus our old MSRV +# would have resolved backward to 1.0.1's now-superseded API anyway. +agent-client-protocol = "=1.3.0" + # MCP server (stdio only — never bind a socket; the FS is the boundary). rmcp = { version = "3.0.0-beta.5", features = ["server", "transport-io", "macros"] } clap = { version = "4", features = ["derive"] } @@ -174,8 +198,15 @@ libc = "0.2" [dev-dependencies] tempfile = "3" # `start_paused` tests (the deferred-generate poll loop) auto-advance tokio's clock -# instead of really sleeping the poll interval. -tokio = { version = "1", features = ["test-util"] } +# instead of really sleeping the poll interval. `io-util` adds `tokio::io::duplex` + +# `split`, feeding the ACP agent test (src/acp.rs) an in-memory transport pair. +tokio = { version = "1", features = ["test-util", "io-util"] } +# Bridges tokio's `AsyncRead`/`AsyncWrite` to the `futures` traits +# `agent_client_protocol::ByteStreams` requires, so the ACP agent test (src/acp.rs) +# can drive a real duplex-socket connection instead of calling handlers directly. +# `default-features = false` + `compat` only: we need just the trait-bridging +# wrappers, none of tokio-util's codec/net extras. +tokio-util = { version = "0.7", default-features = false, features = ["compat"] } # Request-body capture in `tests/effort_wire.rs` and `test_support::CaptureHttp`: # rig's `HttpClientExt` trait is typed in `bytes::Bytes`, so implementing a fake # transport needs the type by name. Already in the tree via reqwest/rig-core — zero diff --git a/src/acp.rs b/src/acp.rs new file mode 100644 index 0000000..a51216d --- /dev/null +++ b/src/acp.rs @@ -0,0 +1,453 @@ +//! `kaibo acp` — the Agent Client Protocol front door. +//! +//! kaibo's second protocol front door, alongside MCP (`src/server/`) and the CLI +//! (`src/cli.rs`): an ACP v1 agent, so an ACP client (Zed, Toad, JetBrains) drives +//! kaibo the same way an MCP client does, over stdio. This chunk is scaffold and +//! handshake only — `initialize`, `session/new`, `session/prompt`, `session/cancel`, +//! and `session/set_mode` all answer, but `session/prompt` returns a canned reply +//! instead of running the real consult loop. Wiring `run_kaish`/the model team into +//! the prompt turn is chunk 2. +//! +//! ACP's session concept doesn't map onto kaibo's `consult` session (a lean +//! question/answer history replayed as context, `src/session.rs`) or the durable +//! store (`src/store.rs`): an ACP session is a whole client conversation — a cwd, +//! an MCP server list, a mode — that a prompt turn runs inside. This chunk keeps +//! that mapping in a small in-memory table, keyed by a `session-N` id (the `job-N` +//! style already used in `src/jobs.rs`); chunk 2 decides how much of it, if any, +//! rides on the existing store. +//! +//! # Executor +//! +//! `agent-client-protocol` 1.3.0's connection is executor-agnostic and `Send` +//! throughout: `ConnectTo::connect_to` returns `impl Future<..> + Send`, and every +//! handler closure registered below must itself be `Send`. This is unlike the `!Send` +//! kaish kernel (`src/sandbox.rs`'s `KaishWorker`, run on its own thread because rig +//! tools require `Send` futures and the kernel's execution future is not one) — no +//! `LocalSet` or dedicated thread is needed here, and none is added. When chunk 2 +//! wires the real consult loop into `session/prompt`, it drives `KaishWorker` through +//! its already-`Send` channel handle, the same way `run_kaish` does today; the `!Send` +//! kernel stays on its own worker thread regardless of which front door called it. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use agent_client_protocol::schema::v1::{ + AgentCapabilities, CancelNotification, ContentBlock, ContentChunk, Implementation, + InitializeRequest, InitializeResponse, NewSessionRequest, NewSessionResponse, PromptRequest, + PromptResponse, SessionId, SessionMode, SessionModeState, SessionNotification, SessionUpdate, + SetSessionModeRequest, SetSessionModeResponse, StopReason, TextContent, +}; +use agent_client_protocol::schema::ProtocolVersion; +use agent_client_protocol::{Agent, Client, ConnectTo, Error}; + +use crate::config::Config; + +/// The canned `session/prompt` reply for this chunk. Plain declarative text — this +/// string rides over the wire to a real ACP client (Zed, Toad), not just a log line. +const SCAFFOLD_REPLY: &str = + "kaibo ACP scaffold. The real consult loop is not wired yet. It lands in chunk 2."; + +/// One ACP session this scaffold has minted: the client's working directory (read by +/// chunk 2's consult wiring, unused here beyond the debug log below) and which cast +/// the session is currently pinned to. +#[derive(Debug, Clone)] +struct SessionRecord { + cwd: PathBuf, + mode_id: String, +} + +/// Shared state behind every ACP connection this process serves: the cast roster +/// (each name becomes one advertised session mode) and the in-memory session table. +/// `Clone` is cheap — an `Arc` around the mutable inner state — so each registered +/// handler closure gets its own handle to the same table. +#[derive(Clone)] +pub struct AcpAgentState { + inner: Arc, +} + +struct Inner { + /// Configured cast names, in `Config::casts`'s `BTreeMap` order — deterministic, + /// so the advertised mode list doesn't reorder between calls. + cast_names: Vec, + default_cast: String, + sessions: Mutex>, + next_session: AtomicU64, +} + +impl AcpAgentState { + /// Build the shared state from a resolved [`Config`]: one session mode per + /// configured cast, starting on the config's default cast. + pub fn new(config: &Config) -> Self { + Self { + inner: Arc::new(Inner { + cast_names: config.casts.keys().cloned().collect(), + default_cast: config.default_cast.clone(), + sessions: Mutex::new(HashMap::new()), + next_session: AtomicU64::new(1), + }), + } + } + + /// The session modes to advertise: one per configured cast, mode id == cast name. + fn session_modes(&self) -> Vec { + self.inner + .cast_names + .iter() + .map(|name| SessionMode::new(name.clone(), name.clone())) + .collect() + } + + /// Whether `mode_id` names a configured cast. + fn known_mode(&self, mode_id: &str) -> bool { + self.inner.cast_names.iter().any(|name| name == mode_id) + } + + /// Mint the next `session-N` id — same style as the `job-N` ids in `src/jobs.rs`, + /// not a UUID: no new dependency for a value only this process's ACP connections + /// ever read back. + fn mint_session_id(&self) -> SessionId { + let n = self.inner.next_session.fetch_add(1, Ordering::Relaxed); + SessionId::new(format!("session-{n}")) + } + + /// Record a freshly created session, starting it on the default cast. + fn insert_session(&self, id: SessionId, cwd: PathBuf) { + let mut sessions = self + .inner + .sessions + .lock() + .expect("acp sessions mutex poisoned"); + sessions.insert( + id, + SessionRecord { + cwd, + mode_id: self.inner.default_cast.clone(), + }, + ); + } + + /// Snapshot a session's record, if it exists. + fn session(&self, id: &SessionId) -> Option { + self.inner + .sessions + .lock() + .expect("acp sessions mutex poisoned") + .get(id) + .cloned() + } + + /// Move a known session onto `mode_id`. Returns `false` if the session is unknown + /// (a caller-visible error, not this scaffold's problem to swallow). + fn set_mode(&self, id: &SessionId, mode_id: String) -> bool { + let mut sessions = self + .inner + .sessions + .lock() + .expect("acp sessions mutex poisoned"); + match sessions.get_mut(id) { + Some(record) => { + record.mode_id = mode_id; + true + } + None => false, + } + } +} + +/// Build the ACP agent connection: `initialize`, `session/new`, `session/prompt`, +/// `session/set_mode`, and `session/cancel`, wired to `state`. Returns an opaque +/// `ConnectTo` — the caller feeds it a transport with `.connect_to(transport)` +/// (`Stdio::new()` for the real CLI, an in-memory duplex in tests). +pub fn agent(state: AcpAgentState) -> impl ConnectTo { + Agent + .builder() + .name("kaibo") + .on_receive_request( + async move |req: InitializeRequest, responder, _cx| { + // No auth: kaibo has its own read-only sandbox (`src/sandbox.rs`), so + // there is nothing for an ACP auth method to gate. We only speak v1 — + // negotiating anything else here would be a lie the client can't act on. + let _ = req.protocol_version; + responder.respond( + InitializeResponse::new(ProtocolVersion::V1) + .agent_capabilities(AgentCapabilities::new()) + .auth_methods(Vec::new()) + .agent_info(Implementation::new("kaibo", env!("CARGO_PKG_VERSION"))), + ) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + { + let state = state.clone(); + async move |req: NewSessionRequest, responder, _cx| { + let session_id = state.mint_session_id(); + state.insert_session(session_id.clone(), req.cwd.clone()); + let modes = + SessionModeState::new(state.inner.default_cast.clone(), state.session_modes()); + tracing::debug!( + session_id = %session_id, + cwd = %req.cwd.display(), + "acp: session/new" + ); + responder.respond(NewSessionResponse::new(session_id).modes(modes)) + } + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + { + let state = state.clone(); + async move |req: PromptRequest, responder, cx| { + let Some(record) = state.session(&req.session_id) else { + return responder.respond_with_error( + Error::invalid_params() + .data(format!("unknown session {}", req.session_id)), + ); + }; + tracing::debug!( + session_id = %req.session_id, + cwd = %record.cwd.display(), + mode = %record.mode_id, + "acp: session/prompt (scaffold canned reply, chunk 2 wires the real loop)" + ); + let update = SessionNotification::new( + req.session_id.clone(), + SessionUpdate::AgentMessageChunk(ContentChunk::new(ContentBlock::Text( + TextContent::new(SCAFFOLD_REPLY), + ))), + ); + cx.send_notification(update)?; + responder.respond(PromptResponse::new(StopReason::EndTurn)) + } + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + { + let state = state.clone(); + async move |req: SetSessionModeRequest, responder, _cx| { + let mode_id = req.mode_id.to_string(); + if !state.known_mode(&mode_id) { + return responder.respond_with_error( + Error::invalid_params() + .data(format!("unknown session mode {mode_id}")), + ); + } + if !state.set_mode(&req.session_id, mode_id.clone()) { + return responder.respond_with_error( + Error::invalid_params() + .data(format!("unknown session {}", req.session_id)), + ); + } + tracing::debug!(session_id = %req.session_id, mode = %mode_id, "acp: session/set_mode"); + responder.respond(SetSessionModeResponse::new()) + } + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_notification( + { + let state = state.clone(); + async move |notif: CancelNotification, _cx| { + // Chunk 1 has no in-flight tool loop: `session/prompt` above already + // finished by the time a client could send this. Nothing to stop yet + // — chunk 2's real loop is the one that needs a cancellation flag to + // check. A notification carries no response, so an unknown session id + // is logged, not erred. + if state.session(¬if.session_id).is_none() { + tracing::debug!( + session_id = %notif.session_id, + "acp: session/cancel for unknown session" + ); + } + Ok(()) + } + }, + agent_client_protocol::on_receive_notification!(), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + use agent_client_protocol::schema::v1::{ + NewSessionRequest as ClientNewSessionRequest, PromptRequest as ClientPromptRequest, + }; + use agent_client_protocol::{ByteStreams, Client as ClientRole}; + use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; + + /// A resolved [`Config`] carrying only the fields this module reads (`casts`, + /// `default_cast`) via `Config::builtin()` — the built-in casts (`anthropic`, + /// `deepseek`, `gemini`, `openrouter`, `openai-local`) are what a fresh install + /// without a config.toml resolves to, so testing against them here tests the shape + /// a real first run advertises. + fn test_config() -> Config { + Config::builtin() + } + + /// One in-memory duplex, split and wrapped as two independent [`ByteStreams`] + /// transports — real JSON-RPC bytes over `tokio::io::duplex`, no network, no + /// stdio. `tokio_util`'s `compat` layer bridges tokio's `AsyncRead`/`AsyncWrite` to + /// the `futures` traits `ByteStreams` requires. + fn new_transport_pair() -> ( + impl ConnectTo + 'static, + impl ConnectTo + 'static, + ) { + let (agent_io, client_io) = tokio::io::duplex(64 * 1024); + let (agent_read, agent_write) = tokio::io::split(agent_io); + let (client_read, client_write) = tokio::io::split(client_io); + ( + ByteStreams::new(agent_write.compat_write(), agent_read.compat()), + ByteStreams::new(client_write.compat_write(), client_read.compat()), + ) + } + + #[tokio::test] + async fn initialize_negotiates_v1_with_no_auth_methods() { + let state = AcpAgentState::new(&test_config()); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + + let response = ClientRole + .builder() + .name("test-client") + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + assert_eq!(response.protocol_version, ProtocolVersion::V1); + assert!(response.auth_methods.is_empty()); + } + + #[tokio::test] + async fn session_new_returns_an_id_and_advertises_cast_modes() { + let state = AcpAgentState::new(&test_config()); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + + let response = ClientRole + .builder() + .name("test-client") + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + cx.send_request(ClientNewSessionRequest::new("/tmp/kaibo-acp-test")) + .block_task() + .await + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + assert!(!response.session_id.to_string().is_empty()); + let modes = response.modes.expect("session modes should be advertised"); + let mut mode_ids: Vec = modes + .available_modes + .iter() + .map(|m| m.id.to_string()) + .collect(); + mode_ids.sort(); + let mut expected: Vec = test_config().casts.keys().cloned().collect(); + expected.sort(); + assert_eq!(mode_ids, expected); + assert_eq!( + modes.current_mode_id.to_string(), + test_config().default_cast + ); + } + + #[tokio::test] + async fn prompt_yields_the_canned_update_then_a_completed_turn() { + let state = AcpAgentState::new(&test_config()); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + + // The client-side notification handler must be registered on the builder + // BEFORE `connect_with` — `ConnectionTo` (the `cx` handed to `connect_with`'s + // closure) has no way to add one mid-connection, only `Builder` does. + let updates: std::sync::Arc>> = + std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let recorded = updates.clone(); + + let prompt_response = ClientRole + .builder() + .name("test-client") + .on_receive_notification( + { + let updates = updates.clone(); + async move |notif: SessionNotification, _cx| { + updates + .lock() + .expect("updates mutex poisoned") + .push(notif.update); + Ok(()) + } + }, + agent_client_protocol::on_receive_notification!(), + ) + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + let session = cx + .send_request(ClientNewSessionRequest::new("/tmp/kaibo-acp-test")) + .block_task() + .await?; + cx.send_request(ClientPromptRequest::new( + session.session_id, + vec![ContentBlock::Text(TextContent::new("hello"))], + )) + .block_task() + .await + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + assert_eq!(prompt_response.stop_reason, StopReason::EndTurn); + let updates = recorded.lock().expect("updates mutex poisoned"); + assert_eq!(updates.len(), 1, "expected exactly one session/update"); + match &updates[0] { + SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content { + ContentBlock::Text(text) => assert_eq!(text.text, SCAFFOLD_REPLY), + other => panic!("expected a text content block, got {other:?}"), + }, + other => panic!("expected an AgentMessageChunk update, got {other:?}"), + } + } + + #[test] + fn known_mode_matches_a_configured_cast_name() { + let state = AcpAgentState::new(&test_config()); + assert!(state.known_mode(&test_config().default_cast)); + assert!(!state.known_mode("not-a-real-cast")); + } + + #[test] + fn set_mode_on_an_unknown_session_reports_failure() { + let state = AcpAgentState::new(&test_config()); + assert!(!state.set_mode(&SessionId::new("nope"), "anthropic".to_string())); + } +} diff --git a/src/cli.rs b/src/cli.rs index 99ce163..94ad91b 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -31,6 +31,7 @@ use std::collections::BTreeMap; use std::path::PathBuf; use std::sync::Arc; +use agent_client_protocol::ConnectTo; use clap::{Args, Parser, Subcommand}; use rig_core::completion::Usage; use rmcp::ErrorData as McpError; @@ -130,6 +131,10 @@ pub enum Command { /// List available models a backend's provider actually serves — read-only model /// discovery via the backend's real /models endpoint, no cast/model in the loop. Models(ModelsArgs), + /// Run kaibo as an Agent Client Protocol (ACP) v1 agent on stdio, for ACP clients + /// (Zed, Toad, JetBrains) instead of an MCP client. Scaffold: `session/prompt` + /// answers with a canned notice until the consult loop is wired (chunk 2). + Acp, /// Print the resolved runtime configuration (the `kaibo://config` document). Config, /// Print the guided "set up my models" walkthrough — the CLI equivalent of the @@ -1777,6 +1782,44 @@ pub async fn run_models(common: CommonArgs, args: ModelsArgs) -> i32 { } } +// --------------------------------------------------------------------------- +// acp +// --------------------------------------------------------------------------- + +/// Run `kaibo acp` — an ACP v1 agent on stdio, the same stdout-is-the-protocol / +/// stderr-is-everything-else split `serve` (MCP) uses. Runs until the client closes +/// the connection (clean EOF, exit 0) or the connection itself fails (exit 4 — a +/// transport/handler fault after startup, not a pre-flight rejection, so it doesn't +/// share `run_consult`'s usage/setup codes). A bad `--config`/cast-resolution error +/// is still a pre-flight usage/setup rejection, same as every other front door. +pub async fn run_acp(common: CommonArgs) -> i32 { + init_cli_logging(); + let config = match load_config(&common) { + Ok(c) => c, + Err(e) => { + eprintln!("kaibo: config error: {e:#}"); + return EXIT_USAGE; + } + }; + // The default cast must resolve now, same check `serve` does at startup — an ACP + // client has no equivalent of `kaibo config` to discover the failure later. + if let Err(e) = config.resolve_cast(&config.default_cast) { + eprintln!("kaibo: default cast: {e:#}"); + return EXIT_SETUP; + } + let state = crate::acp::AcpAgentState::new(&config); + match crate::acp::agent(state) + .connect_to(agent_client_protocol::Stdio::new()) + .await + { + Ok(()) => EXIT_OK, + Err(e) => { + eprintln!("kaibo: acp connection failed: {e}"); + EXIT_CONSULT_FAILURE + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/lib.rs b/src/lib.rs index 95449d4..d5b9e20 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17,6 +17,7 @@ //! store and the media CAS, refuse to resolve into any allowed tree) — and cannot //! shell out to external commands. +pub mod acp; pub mod attach; pub mod batch; pub mod cas; diff --git a/src/main.rs b/src/main.rs index 32a450a..a1dd9e0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -56,6 +56,7 @@ async fn main() -> Result<()> { Some(Command::Models(args)) => { std::process::exit(kaibo::cli::run_models(cli.common, args).await) } + Some(Command::Acp) => std::process::exit(kaibo::cli::run_acp(cli.common).await), Some(Command::Config) => std::process::exit(kaibo::cli::run_config(cli.common)), Some(Command::Configure(args)) => std::process::exit(kaibo::cli::run_configure(args.goal)), Some(Command::ExampleConfig) => std::process::exit(kaibo::cli::run_example_config()), From 1b7c823c4d195e2c1c6b5544a87172fd17b24a14 Mon Sep 17 00:00:00 2001 From: A Tobey Date: Tue, 4 Aug 2026 18:51:01 -0400 Subject: [PATCH 2/3] =?UTF-8?q?feat(acp):=20kaibo=20acp=20=E2=80=94=20the?= =?UTF-8?q?=20real=20consult=20loop=20(chunk=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit session/prompt now drives the real consult loop instead of chunk 1's canned reply. It resolves the session's current mode (cast) through the same shared Resolver every other front door uses (Arm::from_slot — the single live construction point), extracts the prompt's text content blocks, and runs consult() with the ACP session id doubling as the key into kaibo's own in-memory multi-turn Sessions store — so successive prompts in one ACP session replay through the real consult_session_turn history/record machinery, not a cut-down stand-in. Decisions worth a second look: - The event loop can't read a new message while a request handler's own future is still running (per the crate's own ConnectionTo::spawn doc), so session/prompt can't just .await the consult call inline — that would make session/cancel unreachable for the whole turn. The handler resolves cast/ arms/context synchronously, spawns the actual consult() on its own tokio::spawn (kept for its AbortHandle — cx.spawn doesn't hand one back), and returns immediately; the response is sent from a second task (via cx.spawn, matching the SDK's own documented cancellation pattern) that awaits the first and answers once it lands. - session/cancel aborts that AbortHandle directly — the same spawn-then- AbortHandle shape src/jobs.rs already uses for job_cancel. A cancelled task's JoinError::is_cancelled() becomes StopReason::Cancelled, per spec. This is ACP's own domain-level session/cancel notification, distinct from the SDK's low-level JSON-RPC $/cancel_request (a different mechanism for cancelling one outstanding request by id) — not touched here. - Progress renders as session/update through AcpProgressSink. kaibo's PhaseEvent granularity is per-milestone, not token-streamed: SweepStarted/ SweepFinished open and close one ACP ToolCall (the one pair the engine actually emits 1:1); Attached updates that same open call; KaishRun opens its own ToolCall left InProgress and never explicitly closed, because kaibo has no matching "finished" beat for it today — reporting only what kaibo actually knows rather than fabricating a completion. The bookending PhaseStarted/PhaseFinished/TurnCapReached beats are meta narration, not a tool call, so they render as AgentThoughtChunk text (kaibo's own PhaseEvent::message() one-liner — the same text the MCP progress notifications and the CLI's TerminalSink already show). - session/prompt accepts only ContentBlock::Text in this build; an image/ audio/resource block is refused with a clear error. consult's vision plumbing (view_image, ConsultAttachment::Image) expects a file path under the project root, not inline wire bytes, so accepting one honestly needs a real spooling hand-off this chunk doesn't build. Text-only is the honest baseline ACP itself requires every agent to support. - AcpAgentState::new is now fallible (Resolver::from_config can fail on a bad --root/--allow-path) and takes Arc directly rather than &Config — cli.rs's run_acp handles the error as a setup rejection, same as every other front door. - Test seam: AcpAgentState::new_scripted builds a real Resolver (so cast metadata, prompts, sandbox all come from genuine config) but resolves session/prompt's arms from a cast-name-keyed map of pre-built ScriptedClient arms instead of Resolver::arm's real backend construction, which can only ever build a networked client. Mirrors how consult's own offline tests hand a ScriptedClient through Arm::new; no parallel harness. Ten tests cover the wire end to end over tokio::io::duplex: the full prompt -> real consult loop -> session/update sequence -> EndTurn path (with the provenance footer asserted), multi-turn replay, a set_mode switch actually routing to the new cast's model, cancel mid-turn yielding StopReason::Cancelled, a no-op cancel on an idle session, and a non-text content block being refused. All ten pass; the full suite (642 lib tests + every integration suite, including tests/no_write_path.rs and tests/sandbox.rs) stays green. append_warnings moves from a private server::render import to the same pub(crate) re-export CLI-shared helpers already use (with_provenance, consultation_failure_text) — ACP is now a third consumer needing the same answer-rendering fold. cargo build, cargo clippy --all-targets (zero warnings), and cargo tree -i for both aws-lc-rs and mimalloc (both empty) all pass. rustfmt was run only on the three edited leaf files (src/acp.rs, src/cli.rs, src/server/mod.rs), never a crate-root file alongside others. Local prototype per instructions: no push, no PR. Co-authored-by: Claude Sonnet --- src/acp.rs | 1068 +++++++++++++++++++++++++++++++++++++++++---- src/cli.rs | 11 +- src/server/mod.rs | 16 +- 3 files changed, 1005 insertions(+), 90 deletions(-) diff --git a/src/acp.rs b/src/acp.rs index a51216d..f24f396 100644 --- a/src/acp.rs +++ b/src/acp.rs @@ -2,31 +2,61 @@ //! //! kaibo's second protocol front door, alongside MCP (`src/server/`) and the CLI //! (`src/cli.rs`): an ACP v1 agent, so an ACP client (Zed, Toad, JetBrains) drives -//! kaibo the same way an MCP client does, over stdio. This chunk is scaffold and -//! handshake only — `initialize`, `session/new`, `session/prompt`, `session/cancel`, -//! and `session/set_mode` all answer, but `session/prompt` returns a canned reply -//! instead of running the real consult loop. Wiring `run_kaish`/the model team into -//! the prompt turn is chunk 2. +//! kaibo the same way an MCP client does, over stdio. `session/prompt` now runs the +//! real consult loop: it resolves the session's current cast to arms the same way +//! the MCP `consult` tool does (`Arm::from_slot`, the single live construction +//! point), drives `consult` (`src/consult/`), and streams progress as ACP +//! `session/update` notifications while it works. //! //! ACP's session concept doesn't map onto kaibo's `consult` session (a lean //! question/answer history replayed as context, `src/session.rs`) or the durable //! store (`src/store.rs`): an ACP session is a whole client conversation — a cwd, //! an MCP server list, a mode — that a prompt turn runs inside. This chunk keeps //! that mapping in a small in-memory table, keyed by a `session-N` id (the `job-N` -//! style already used in `src/jobs.rs`); chunk 2 decides how much of it, if any, -//! rides on the existing store. +//! style already used in `src/jobs.rs`), and reuses the ACP session id itself as the +//! key into kaibo's own in-memory multi-turn `Sessions` store — so successive +//! prompts in one ACP session replay as one `consult` thread, through the real +//! `consult_session_turn` replay/record machinery, not a parallel one. //! -//! # Executor +//! # Progress → `session/update` //! -//! `agent-client-protocol` 1.3.0's connection is executor-agnostic and `Send` -//! throughout: `ConnectTo::connect_to` returns `impl Future<..> + Send`, and every -//! handler closure registered below must itself be `Send`. This is unlike the `!Send` -//! kaish kernel (`src/sandbox.rs`'s `KaishWorker`, run on its own thread because rig -//! tools require `Send` futures and the kernel's execution future is not one) — no -//! `LocalSet` or dedicated thread is needed here, and none is added. When chunk 2 -//! wires the real consult loop into `session/prompt`, it drives `KaishWorker` through -//! its already-`Send` channel handle, the same way `run_kaish` does today; the `!Send` -//! kernel stays on its own worker thread regardless of which front door called it. +//! [`AcpProgressSink`] renders each [`crate::progress::PhaseEvent`] as a +//! `session/update` notification: kaibo's granularity is per-milestone (a sweep +//! started, a kaish script ran), not token-streamed, so each beat becomes one +//! notification. `SweepStarted`/`SweepFinished` open and close one ACP `ToolCall` +//! (they are the one pair kaibo's engine already emits 1:1); `Attached` updates that +//! same open call; `KaishRun` opens its own `ToolCall` that kaibo has no matching +//! "finished" beat for today, so it is announced `InProgress` and never explicitly +//! closed — reporting only what kaibo actually knows, not a fabricated completion. +//! The bookending `PhaseStarted`/`PhaseFinished`/`TurnCapReached` beats are meta +//! narration about the whole turn, not a tool call, so they render as +//! `AgentThoughtChunk` text (kaibo's own `PhaseEvent::message()`, the same one-liner +//! the MCP progress notifications and the CLI's `TerminalSink` show). +//! +//! # Cancellation +//! +//! ACP's `session/cancel` is its own domain notification (`CancelNotification`), +//! distinct from the SDK's low-level JSON-RPC `$/cancel_request` (a different +//! mechanism, for cancelling one outstanding request by id — see +//! `agent_client_protocol::concepts::cancellation`). kaibo tracks the in-flight +//! turn's `tokio::task::AbortHandle` on the session record — the same +//! spawn-then-`AbortHandle` shape `src/jobs.rs` uses for `job_cancel` — so +//! `session/cancel` aborts it directly; a session with nothing running is a clean +//! no-op. `session/prompt` itself spawns the actual consult work and returns from +//! its handler immediately (mirroring the SDK's own documented cancellation +//! pattern): the connection's event loop can't process a new message while a +//! handler's future is still running, so awaiting the consult call *inline* would +//! make `session/cancel` unreachable until the turn already finished. +//! +//! # Non-text prompt content +//! +//! `session/prompt` accepts only `ContentBlock::Text` blocks in this build. An +//! image/audio/resource block is refused with a clear error naming the block kind: +//! consult's vision plumbing (`view_image`, `ConsultAttachment::Image`) expects a +//! file path under the project root, not inline bytes over the wire, so accepting +//! one honestly would need a real hand-off (spooling the bytes somewhere kaish can +//! read them) that this chunk does not build. Text-only is the honest baseline ACP +//! itself requires every agent to support. use std::collections::HashMap; use std::path::PathBuf; @@ -37,55 +67,108 @@ use agent_client_protocol::schema::v1::{ AgentCapabilities, CancelNotification, ContentBlock, ContentChunk, Implementation, InitializeRequest, InitializeResponse, NewSessionRequest, NewSessionResponse, PromptRequest, PromptResponse, SessionId, SessionMode, SessionModeState, SessionNotification, SessionUpdate, - SetSessionModeRequest, SetSessionModeResponse, StopReason, TextContent, + SetSessionModeRequest, SetSessionModeResponse, StopReason, TextContent, ToolCall, + ToolCallContent, ToolCallId, ToolCallStatus, ToolCallUpdate, ToolCallUpdateFields, ToolKind, }; use agent_client_protocol::schema::ProtocolVersion; -use agent_client_protocol::{Agent, Client, ConnectTo, Error}; - -use crate::config::Config; +use agent_client_protocol::{Agent, Client, ConnectTo, ConnectionTo, Error}; -/// The canned `session/prompt` reply for this chunk. Plain declarative text — this -/// string rides over the wire to a real ACP client (Zed, Toad), not just a log line. -const SCAFFOLD_REPLY: &str = - "kaibo ACP scaffold. The real consult loop is not wired yet. It lands in chunk 2."; +use crate::config::{Cast, Config, ModelRole}; +use crate::consult::{consult, Arm, ConsultConfig, ExploreConfig, PhaseContext}; +use crate::progress::{PhaseEvent, ProgressSink}; +use crate::server::{append_warnings, consultation_failure_text, with_provenance, Resolver}; +use crate::session::Sessions; -/// One ACP session this scaffold has minted: the client's working directory (read by -/// chunk 2's consult wiring, unused here beyond the debug log below) and which cast -/// the session is currently pinned to. +/// One ACP session this process has minted: the client's working directory (the +/// `consult` root every turn runs against), which cast the session is currently +/// pinned to, and the in-flight turn's abort handle, if one is running. #[derive(Debug, Clone)] struct SessionRecord { cwd: PathBuf, mode_id: String, + /// What `session/cancel` aborts. `None` between turns and while idle — a cancel + /// on an idle session is a clean no-op. + running: Option, } -/// Shared state behind every ACP connection this process serves: the cast roster -/// (each name becomes one advertised session mode) and the in-memory session table. -/// `Clone` is cheap — an `Arc` around the mutable inner state — so each registered -/// handler closure gets its own handle to the same table. +/// How `session/prompt` builds this turn's explorer/synth [`Arm`]s for a resolved +/// cast. The live path threads through the shared [`Resolver`] (`Arm::from_slot` — +/// real backends, keys, HTTP clients, the single live construction point every +/// other front door uses); tests substitute arms built directly over a +/// `ScriptedClient` via `Arm::new` — the exact injection seam `consult`'s own +/// offline tests use (`src/consult/engine.rs`'s test module), not a parallel +/// harness. Keyed by cast name, so a `session/set_mode` switch resolves a +/// different scripted pair. +enum ArmSource { + Live, + #[cfg(test)] + Scripted(HashMap), +} + +/// Shared state behind every ACP connection this process serves: the resolved +/// config (via the same [`Resolver`] the MCP handler and CLI use), the in-memory +/// session table, and the multi-turn `consult` session store the ACP session id +/// itself keys into. #[derive(Clone)] pub struct AcpAgentState { inner: Arc, } struct Inner { + resolver: Resolver, /// Configured cast names, in `Config::casts`'s `BTreeMap` order — deterministic, /// so the advertised mode list doesn't reorder between calls. cast_names: Vec, default_cast: String, + /// Multi-turn `consult` history, keyed by the ACP session id string — an + /// in-memory backend (today's only ACP option; the durable store is a later + /// chunk's call, same as it was for MCP before persistence shipped). + consult_sessions: Sessions, sessions: Mutex>, next_session: AtomicU64, + arms: ArmSource, } impl AcpAgentState { /// Build the shared state from a resolved [`Config`]: one session mode per - /// configured cast, starting on the config's default cast. - pub fn new(config: &Config) -> Self { + /// configured cast, starting on the config's default cast, arms resolved live + /// through a [`Resolver`] built the same way every other front door builds one. + /// Fallible for the same reason `Resolver::from_config` is everywhere else — a + /// nonexistent `--root`/`--allow-path` is a loud construction error, not a + /// silently-empty boundary. + pub fn new(config: Arc) -> anyhow::Result { + let resolver = Resolver::from_config(config)?; + Ok(Self::build(resolver, ArmSource::Live)) + } + + /// Test-only constructor: a real [`Resolver`] over `config` (so cast metadata, + /// prompts, sandbox, and session capacity are all the genuine article), but + /// `session/prompt`'s arms come from `arms` (cast name → (explorer, synth)) + /// instead of `Resolver::arm`'s real backend construction — which can only ever + /// build a networked client, never a `ScriptedClient`. Mirrors how `consult`'s + /// own tests hand a `ScriptedClient` through `Arm::new`; no parallel harness. + #[cfg(test)] + pub(crate) fn new_scripted(config: &Config, arms: HashMap) -> Self { + let resolver = Resolver::from_config(Arc::new(config.clone())) + .expect("test config resolves (no --root/--allow-path to canonicalize)"); + Self::build(resolver, ArmSource::Scripted(arms)) + } + + fn build(resolver: Resolver, arms: ArmSource) -> Self { + let cast_names = resolver.config.casts.keys().cloned().collect(); + let default_cast = resolver.config.default_cast.clone(); + let consult_sessions = Sessions::Memory(crate::session::SessionStore::new( + resolver.config.defaults.session_capacity, + )); Self { inner: Arc::new(Inner { - cast_names: config.casts.keys().cloned().collect(), - default_cast: config.default_cast.clone(), + resolver, + cast_names, + default_cast, + consult_sessions, sessions: Mutex::new(HashMap::new()), next_session: AtomicU64::new(1), + arms, }), } } @@ -124,6 +207,7 @@ impl AcpAgentState { SessionRecord { cwd, mode_id: self.inner.default_cast.clone(), + running: None, }, ); } @@ -154,6 +238,265 @@ impl AcpAgentState { None => false, } } + + /// Record the in-flight turn's abort handle, so a later `session/cancel` can + /// find it. A no-op if the session vanished between the prompt starting and + /// this call (can't happen today — sessions are never evicted — but harmless). + fn set_running(&self, id: &SessionId, handle: tokio::task::AbortHandle) { + let mut sessions = self + .inner + .sessions + .lock() + .expect("acp sessions mutex poisoned"); + if let Some(record) = sessions.get_mut(id) { + record.running = Some(handle); + } + } + + /// Clear a finished turn's abort handle so a stale one can't be aborted later. + fn clear_running(&self, id: &SessionId) { + let mut sessions = self + .inner + .sessions + .lock() + .expect("acp sessions mutex poisoned"); + if let Some(record) = sessions.get_mut(id) { + record.running = None; + } + } + + /// `session/cancel`: abort the session's in-flight prompt task, if any — + /// cooperative and best-effort, matching the spec's own framing. Returns + /// whether the session was known at all, so the notification handler can still + /// log a truly unknown id without treating "known but idle" as one. + fn cancel_running(&self, id: &SessionId) -> bool { + let sessions = self + .inner + .sessions + .lock() + .expect("acp sessions mutex poisoned"); + match sessions.get(id) { + Some(record) => { + if let Some(handle) = &record.running { + handle.abort(); + } + true + } + None => false, + } + } + + /// Resolve `mode_id` to a cast and refuse one whose synth runs on an offline + /// lane — the same gate the MCP `consult` tool applies (`reject_offline_cast`): + /// an ACP `session/prompt` turn is interactive by construction. + fn resolve_cast(&self, mode_id: &str) -> Result { + let cast = self + .inner + .resolver + .resolve_cast(Some(mode_id.to_string())) + .map_err(|e| e.message.to_string())?; + self.inner + .resolver + .reject_offline_cast(&cast, "session/prompt") + .map_err(|e| e.message.to_string())?; + Ok(cast) + } + + /// Resolve one of `cast`'s slots into a live [`Arm`] — through the real + /// [`Resolver`] in production, or the scripted map in tests. See [`ArmSource`]. + fn arm(&self, cast: &Cast, role: ModelRole) -> Result { + match &self.inner.arms { + ArmSource::Live => self + .inner + .resolver + .arm(cast, role) + .map_err(|e| e.message.to_string()), + #[cfg(test)] + ArmSource::Scripted(map) => { + let (explorer, synth) = map.get(&cast.name).ok_or_else(|| { + format!( + "no scripted arms registered for cast {:?} (registered: {:?})", + cast.name, + map.keys().collect::>() + ) + })?; + Ok(match role { + ModelRole::Explorer => explorer.clone(), + ModelRole::Synth => synth.clone(), + ModelRole::Image => { + return Err(format!( + "cast {:?}: session/prompt has no image role to resolve", + cast.name + )) + } + }) + } + } + } +} + +/// Extract the plain-text prompt from a `session/prompt`'s content blocks — text +/// blocks only in this build (see the module doc's "Non-text prompt content" +/// section). Multiple text blocks join with a blank line, matching how the MCP +/// `consult` tool folds multi-part input into one question string. +fn extract_prompt_text(blocks: &[ContentBlock]) -> Result { + let mut parts = Vec::with_capacity(blocks.len()); + for block in blocks { + match block { + ContentBlock::Text(t) => parts.push(t.text.clone()), + other => { + return Err(format!( + "session/prompt only accepts text content blocks in this build (got \ + {}) — image/audio/resource input isn't wired into consult's vision \ + plumbing yet. Send plain text.", + content_block_kind(other) + )) + } + } + } + if parts.iter().all(|p| p.trim().is_empty()) { + return Err( + "empty prompt — session/prompt needs at least one non-empty text block".to_string(), + ); + } + Ok(parts.join("\n\n")) +} + +/// A short name for a [`ContentBlock`] variant, for the non-text refusal message. +/// `ContentBlock` is `#[non_exhaustive]`, so this must carry a wildcard arm even +/// though every variant is named today. +fn content_block_kind(block: &ContentBlock) -> &'static str { + match block { + ContentBlock::Text(_) => "text", + ContentBlock::Image(_) => "image", + ContentBlock::Audio(_) => "audio", + ContentBlock::ResourceLink(_) => "resource_link", + ContentBlock::Resource(_) => "resource", + _ => "unknown", + } +} + +/// Renders kaibo's [`PhaseEvent`] beats as ACP `session/update` notifications. See +/// the module doc's "Progress → `session/update`" section for the mapping and why +/// each event lands where it does. Fire-and-forget (`ProgressSink`'s contract): a +/// send failure (client gone, connection tearing down) is dropped rather than +/// allowed to fail the turn — the final `PromptResponse` is the authoritative +/// outcome either way. +struct AcpProgressSink { + cx: ConnectionTo, + session_id: SessionId, + next_tool_id: AtomicU64, + /// The most recently opened sweep's tool-call id, so `SweepFinished`/`Attached` + /// update the SAME call instead of opening a new one. `consult` runs its + /// delegated sweeps one at a time today, so a single slot (not a stack) pairs + /// them correctly. + current_sweep: Mutex>, +} + +impl AcpProgressSink { + fn new(cx: ConnectionTo, session_id: SessionId) -> Self { + Self { + cx, + session_id, + next_tool_id: AtomicU64::new(1), + current_sweep: Mutex::new(None), + } + } + + fn mint_tool_id(&self) -> ToolCallId { + let n = self.next_tool_id.fetch_add(1, Ordering::Relaxed); + ToolCallId::new(format!("kaibo-{n}")) + } + + fn notify(&self, update: SessionUpdate) { + let notif = SessionNotification::new(self.session_id.clone(), update); + let _ = self.cx.send_notification(notif); + } + + /// Meta narration about the turn as a whole (not a specific tool call) — an + /// `AgentThoughtChunk` carrying kaibo's own `PhaseEvent::message()` one-liner, + /// the same text the MCP progress notifications and the CLI's `TerminalSink` + /// show. + fn thought(&self, text: String) { + self.notify(SessionUpdate::AgentThoughtChunk(ContentChunk::new( + ContentBlock::Text(TextContent::new(text)), + ))); + } +} + +impl std::fmt::Debug for AcpProgressSink { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AcpProgressSink") + .field("session_id", &self.session_id) + .finish_non_exhaustive() + } +} + +impl ProgressSink for AcpProgressSink { + fn emit(&self, event: PhaseEvent) { + let msg = event.message(); + match event { + PhaseEvent::PhaseStarted { .. } + | PhaseEvent::PhaseFinished { .. } + | PhaseEvent::TurnCapReached => self.thought(msg), + // Announced before the script runs (see `explorer.rs`) and kaibo has no + // matching "finished" beat for one today — reported honestly as + // in-progress rather than faking a completion kaibo never observed. + PhaseEvent::KaishRun { script } => { + let id = self.mint_tool_id(); + self.notify(SessionUpdate::ToolCall( + ToolCall::new(id, msg) + .kind(ToolKind::Execute) + .status(ToolCallStatus::InProgress) + .raw_input(serde_json::json!({ "script": script })), + )); + } + PhaseEvent::SweepStarted { .. } => { + let id = self.mint_tool_id(); + *self + .current_sweep + .lock() + .expect("acp progress sink mutex poisoned") = Some(id.clone()); + self.notify(SessionUpdate::ToolCall( + ToolCall::new(id, msg) + .kind(ToolKind::Search) + .status(ToolCallStatus::InProgress), + )); + } + PhaseEvent::SweepFinished => { + let current = self + .current_sweep + .lock() + .expect("acp progress sink mutex poisoned") + .take(); + match current { + Some(id) => self.notify(SessionUpdate::ToolCallUpdate(ToolCallUpdate::new( + id, + ToolCallUpdateFields::new().status(ToolCallStatus::Completed), + ))), + // No open sweep to close (shouldn't happen — the engine pairs + // these 1:1 — but narrate rather than silently drop the beat). + None => self.thought(msg), + } + } + PhaseEvent::Attached { .. } => { + let current = self + .current_sweep + .lock() + .expect("acp progress sink mutex poisoned") + .clone(); + match current { + Some(id) => self.notify(SessionUpdate::ToolCallUpdate(ToolCallUpdate::new( + id, + ToolCallUpdateFields::new().content(vec![ToolCallContent::from( + ContentBlock::Text(TextContent::new(msg)), + )]), + ))), + None => self.thought(msg), + } + } + } + } } /// Build the ACP agent connection: `initialize`, `session/new`, `session/prompt`, @@ -207,20 +550,157 @@ pub fn agent(state: AcpAgentState) -> impl ConnectTo { .data(format!("unknown session {}", req.session_id)), ); }; - tracing::debug!( - session_id = %req.session_id, - cwd = %record.cwd.display(), - mode = %record.mode_id, - "acp: session/prompt (scaffold canned reply, chunk 2 wires the real loop)" - ); - let update = SessionNotification::new( - req.session_id.clone(), - SessionUpdate::AgentMessageChunk(ContentChunk::new(ContentBlock::Text( - TextContent::new(SCAFFOLD_REPLY), - ))), - ); - cx.send_notification(update)?; - responder.respond(PromptResponse::new(StopReason::EndTurn)) + + let question = match extract_prompt_text(&req.prompt) { + Ok(q) => q, + Err(msg) => { + return responder + .respond_with_error(Error::invalid_params().data(msg)) + } + }; + + let cast = match state.resolve_cast(&record.mode_id) { + Ok(c) => c, + Err(msg) => { + return responder + .respond_with_error(Error::invalid_params().data(msg)) + } + }; + let explorer = match state.arm(&cast, ModelRole::Explorer) { + Ok(a) => a, + Err(msg) => { + return responder + .respond_with_error(Error::internal_error().data(msg)) + } + }; + let synth = match state.arm(&cast, ModelRole::Synth) { + Ok(a) => a, + Err(msg) => { + return responder + .respond_with_error(Error::internal_error().data(msg)) + } + }; + + let root = record.cwd.clone(); + let house_rules = match state.inner.resolver.house_rules(&root) { + Ok(h) => h, + Err(e) => { + return responder.respond_with_error( + Error::internal_error().data(e.message.to_string()), + ) + } + }; + let orientation = match state.inner.resolver.orientation(&root).await { + Ok(o) => o, + Err(e) => { + return responder.respond_with_error( + Error::internal_error().data(e.message.to_string()), + ) + } + }; + let prompts = state.inner.resolver.resolved_prompts(&cast); + let defaults = &state.inner.resolver.config.defaults; + + // Progress rides the whole turn: a clone feeds `ConsultConfig` (the + // spawned consult loop emits through it), and this one bookends the + // turn exactly as the MCP `consult` tool does — started here, + // finished once the spawned task actually lands an answer. + let progress: Arc = + Arc::new(AcpProgressSink::new(cx.clone(), req.session_id.clone())); + progress.emit(PhaseEvent::PhaseStarted { phase: "consult" }); + + let cfg = ConsultConfig { + explore: ExploreConfig { + phase: PhaseContext { + progress: progress.clone(), + house_rules, + prompts, + orientation, + call_deadline: defaults.call_deadline, + }, + explorer_max_turns: defaults.explorer_max_turns, + sandbox: state.inner.resolver.config.sandbox.clone(), + max_attachments: defaults.max_attachments, + }, + synth_max_turns: defaults.synth_max_turns, + attachments: Vec::new(), + }; + + let explorer_for_task = explorer.clone(); + let synth_for_task = synth.clone(); + let sessions_for_task = state.inner.consult_sessions.clone(); + let session_key = req.session_id.to_string(); + + // Spawn the actual model work on its own task so this handler can + // return immediately: the connection's event loop can't read the + // next message (a `session/cancel`, most importantly) while a + // handler's own future is still running. `tokio::spawn` (not + // `cx.spawn`) because we need its `AbortHandle` — `cx.spawn` runs the + // task but doesn't hand one back. + let consult_task = tokio::spawn(async move { + consult( + &question, + None, + root, + &explorer_for_task, + &synth_for_task, + &cfg, + Some((&sessions_for_task, &session_key)), + ) + .await + }); + state.set_running(&req.session_id, consult_task.abort_handle()); + + let explorer_model = explorer.model.clone(); + let synth_model = synth.model.clone(); + let cast_name = cast.name.clone(); + let session_id_for_task = req.session_id.clone(); + let state_for_task = state.clone(); + let cx_for_task = cx.clone(); + + cx.spawn(async move { + let outcome = consult_task.await; + state_for_task.clear_running(&session_id_for_task); + let response = match outcome { + Ok(Ok(out)) => { + progress.emit(PhaseEvent::PhaseFinished { phase: "consult" }); + let answer = with_provenance( + append_warnings(out.answer, &out.warnings), + &cast_name, + &[ + ("explorer", explorer_model.as_str()), + ("synth", synth_model.as_str()), + ], + &out.usage, + ); + let update = SessionNotification::new( + session_id_for_task.clone(), + SessionUpdate::AgentMessageChunk(ContentChunk::new( + ContentBlock::Text(TextContent::new(answer)), + )), + ); + cx_for_task.send_notification(update)?; + Ok(PromptResponse::new(StopReason::EndTurn)) + } + // A provider/model-loop failure — the turn ran and failed, + // not a bad request. Named and classified the same way the + // MCP `consult` tool renders a consultation failure. + Ok(Err(e)) => Err(Error::internal_error().data( + consultation_failure_text("session/prompt", &cast_name, e), + )), + // `session/cancel` aborted the spawned task: the spec's own + // stop reason for exactly this. + Err(join_err) if join_err.is_cancelled() => { + Ok(PromptResponse::new(StopReason::Cancelled)) + } + // The task panicked — a kaibo-side bug, not a provider + // failure or a cancellation. + Err(join_err) => Err(Error::internal_error().data(format!( + "the consult task ended unexpectedly: {join_err}" + ))), + }; + responder.respond_with_result(response) + }) } }, agent_client_protocol::on_receive_request!(), @@ -252,12 +732,12 @@ pub fn agent(state: AcpAgentState) -> impl ConnectTo { { let state = state.clone(); async move |notif: CancelNotification, _cx| { - // Chunk 1 has no in-flight tool loop: `session/prompt` above already - // finished by the time a client could send this. Nothing to stop yet - // — chunk 2's real loop is the one that needs a cancellation flag to - // check. A notification carries no response, so an unknown session id - // is logged, not erred. - if state.session(¬if.session_id).is_none() { + // Best-effort and cooperative, per the spec: aborting the in-flight + // task (if any) is all this needs to do — the prompt handler's own + // spawned tail observes the abort and answers `StopReason::Cancelled`. + // A notification carries no response, so an unknown session id is + // logged, not erred; a known-but-idle session is a silent no-op. + if !state.cancel_running(¬if.session_id) { tracing::debug!( session_id = %notif.session_id, "acp: session/cancel for unknown session" @@ -276,10 +756,14 @@ mod tests { use agent_client_protocol::schema::v1::{ NewSessionRequest as ClientNewSessionRequest, PromptRequest as ClientPromptRequest, + SetSessionModeRequest as ClientSetSessionModeRequest, }; use agent_client_protocol::{ByteStreams, Client as ClientRole}; use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; + use crate::consult::ModelCaps; + use crate::test_support::{text_response, transcript_text, ScriptedClient}; + /// A resolved [`Config`] carrying only the fields this module reads (`casts`, /// `default_cast`) via `Config::builtin()` — the built-in casts (`anthropic`, /// `deepseek`, `gemini`, `openrouter`, `openai-local`) are what a fresh install @@ -289,6 +773,22 @@ mod tests { Config::builtin() } + /// A scripted arm over `client`, addressing model `model` — same shape + /// `consult`'s own offline tests build (`src/consult/engine.rs`), vision off + /// (nothing here attaches an image). + fn scripted_arm(client: &ScriptedClient, model: &str) -> Arm { + Arm::new( + client.clone(), + model, + 16384, + None, + ModelCaps { + vision: false, + tool_result_images: true, + }, + ) + } + /// One in-memory duplex, split and wrapped as two independent [`ByteStreams`] /// transports — real JSON-RPC bytes over `tokio::io::duplex`, no network, no /// stdio. `tokio_util`'s `compat` layer bridges tokio's `AsyncRead`/`AsyncWrite` to @@ -297,7 +797,7 @@ mod tests { impl ConnectTo + 'static, impl ConnectTo + 'static, ) { - let (agent_io, client_io) = tokio::io::duplex(64 * 1024); + let (agent_io, client_io) = tokio::io::duplex(1 << 20); let (agent_read, agent_write) = tokio::io::split(agent_io); let (client_read, client_write) = tokio::io::split(client_io); ( @@ -306,9 +806,18 @@ mod tests { ) } + /// A project root with one real file, so orientation/house-rules assembly (which + /// spawns a real read-only kaish worker) has something to mount — a turn's cwd + /// must be a real directory even though the model side is fully scripted. + fn project_dir() -> tempfile::TempDir { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("README.md"), "kaibo acp test fixture\n").unwrap(); + dir + } + #[tokio::test] async fn initialize_negotiates_v1_with_no_auth_methods() { - let state = AcpAgentState::new(&test_config()); + let state = AcpAgentState::new(Arc::new(test_config())).expect("resolver builds"); let (agent_transport, client_transport) = new_transport_pair(); let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); @@ -334,7 +843,7 @@ mod tests { #[tokio::test] async fn session_new_returns_an_id_and_advertises_cast_modes() { - let state = AcpAgentState::new(&test_config()); + let state = AcpAgentState::new(Arc::new(test_config())).expect("resolver builds"); let (agent_transport, client_transport) = new_transport_pair(); let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); @@ -374,18 +883,47 @@ mod tests { ); } + #[test] + fn known_mode_matches_a_configured_cast_name() { + let state = AcpAgentState::new(Arc::new(test_config())).expect("resolver builds"); + assert!(state.known_mode(&test_config().default_cast)); + assert!(!state.known_mode("not-a-real-cast")); + } + + #[test] + fn set_mode_on_an_unknown_session_reports_failure() { + let state = AcpAgentState::new(Arc::new(test_config())).expect("resolver builds"); + assert!(!state.set_mode(&SessionId::new("nope"), "anthropic".to_string())); + } + + /// The full path, scripted: `session/prompt` drives the REAL consult loop (offline, + /// via `ScriptedClient`) and the caller sees at least one progress-derived + /// `session/update` before the final answer, which carries the provenance footer. #[tokio::test] - async fn prompt_yields_the_canned_update_then_a_completed_turn() { - let state = AcpAgentState::new(&test_config()); + async fn prompt_runs_the_real_consult_loop_and_streams_progress_before_the_answer() { + let client = ScriptedClient::builder() + .on_model("scripted-synth", |req| { + let shown = transcript_text(req); + Ok(text_response(format!("ANSWER seeing[{shown}]"))) + }) + .build(); + let default_cast = test_config().default_cast; + let mut arms = HashMap::new(); + arms.insert( + default_cast.clone(), + ( + scripted_arm(&client, "scripted-explorer"), + scripted_arm(&client, "scripted-synth"), + ), + ); + let state = AcpAgentState::new_scripted(&test_config(), arms); let (agent_transport, client_transport) = new_transport_pair(); let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); - // The client-side notification handler must be registered on the builder - // BEFORE `connect_with` — `ConnectionTo` (the `cx` handed to `connect_with`'s - // closure) has no way to add one mid-connection, only `Builder` does. - let updates: std::sync::Arc>> = - std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let updates: Arc>> = Arc::new(Mutex::new(Vec::new())); let recorded = updates.clone(); + let root = project_dir(); + let root_path = root.path().display().to_string(); let prompt_response = ClientRole .builder() @@ -408,12 +946,12 @@ mod tests { .block_task() .await?; let session = cx - .send_request(ClientNewSessionRequest::new("/tmp/kaibo-acp-test")) + .send_request(ClientNewSessionRequest::new(root_path)) .block_task() .await?; cx.send_request(ClientPromptRequest::new( session.session_id, - vec![ContentBlock::Text(TextContent::new("hello"))], + vec![ContentBlock::Text(TextContent::new("what does this do?"))], )) .block_task() .await @@ -427,27 +965,395 @@ mod tests { .expect("agent-side connection failed"); assert_eq!(prompt_response.stop_reason, StopReason::EndTurn); - let updates = recorded.lock().expect("updates mutex poisoned"); - assert_eq!(updates.len(), 1, "expected exactly one session/update"); - match &updates[0] { + let updates = recorded.lock().expect("updates mutex poisoned").clone(); + let answer_pos = updates + .iter() + .position(|u| matches!(u, SessionUpdate::AgentMessageChunk(_))) + .expect("a final AgentMessageChunk answer must be sent"); + assert!( + answer_pos > 0, + "at least one progress-derived update must precede the answer, got: {updates:?}" + ); + match &updates[answer_pos] { SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content { - ContentBlock::Text(text) => assert_eq!(text.text, SCAFFOLD_REPLY), + ContentBlock::Text(text) => { + assert!( + text.text.contains("ANSWER"), + "answer text present: {}", + text.text + ); + assert!( + text.text.contains(&format!("cast `{default_cast}`")), + "provenance names the cast: {}", + text.text + ); + assert!( + text.text.contains("scripted-synth"), + "provenance names the answering model: {}", + text.text + ); + } other => panic!("expected a text content block, got {other:?}"), }, other => panic!("expected an AgentMessageChunk update, got {other:?}"), } } - #[test] - fn known_mode_matches_a_configured_cast_name() { - let state = AcpAgentState::new(&test_config()); - assert!(state.known_mode(&test_config().default_cast)); - assert!(!state.known_mode("not-a-real-cast")); + /// A second prompt in the same ACP session replays the first turn's Q&A: the + /// scripted synth reads its own prior answer back in the transcript, proving + /// `session/prompt` threads the real `consult_session_turn` replay/record + /// machinery (keyed by the ACP session id), not a stateless call each time. + #[tokio::test] + async fn a_second_prompt_in_the_same_session_replays_prior_context() { + let client = ScriptedClient::builder() + .on_model("scripted-synth", |req| { + let shown = transcript_text(req); + if shown.contains("first turn marker") { + Ok(text_response("SECOND[saw the first turn]")) + } else { + Ok(text_response("FIRST[first turn marker]")) + } + }) + .build(); + let default_cast = test_config().default_cast; + let mut arms = HashMap::new(); + arms.insert( + default_cast, + ( + scripted_arm(&client, "scripted-explorer"), + scripted_arm(&client, "scripted-synth"), + ), + ); + let state = AcpAgentState::new_scripted(&test_config(), arms); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + + let root = project_dir(); + let root_path = root.path().display().to_string(); + + let (first, second) = ClientRole + .builder() + .name("test-client") + .on_receive_notification( + async move |_notif: SessionNotification, _cx| Ok(()), + agent_client_protocol::on_receive_notification!(), + ) + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + let session = cx + .send_request(ClientNewSessionRequest::new(root_path)) + .block_task() + .await?; + let first = cx + .send_request(ClientPromptRequest::new( + session.session_id.clone(), + vec![ContentBlock::Text(TextContent::new("first question"))], + )) + .block_task() + .await?; + let second = cx + .send_request(ClientPromptRequest::new( + session.session_id, + vec![ContentBlock::Text(TextContent::new("second question"))], + )) + .block_task() + .await?; + Ok((first, second)) + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + assert_eq!(first.stop_reason, StopReason::EndTurn); + assert_eq!(second.stop_reason, StopReason::EndTurn); } - #[test] - fn set_mode_on_an_unknown_session_reports_failure() { - let state = AcpAgentState::new(&test_config()); - assert!(!state.set_mode(&SessionId::new("nope"), "anthropic".to_string())); + /// `session/set_mode` switches which cast the NEXT prompt resolves to: two casts + /// in the built-in registry (`anthropic`, `deepseek`), each wired to a distinct + /// scripted model id, so the answering model's name in the provenance footer + /// proves the switch actually took. + #[tokio::test] + async fn set_mode_then_prompt_uses_the_newly_selected_casts_model() { + let client = ScriptedClient::builder() + .on_model("scripted-synth-a", |_req| Ok(text_response("from A"))) + .on_model("scripted-synth-b", |_req| Ok(text_response("from B"))) + .build(); + let mut arms = HashMap::new(); + arms.insert( + "anthropic".to_string(), + ( + scripted_arm(&client, "scripted-explorer-a"), + scripted_arm(&client, "scripted-synth-a"), + ), + ); + arms.insert( + "deepseek".to_string(), + ( + scripted_arm(&client, "scripted-explorer-b"), + scripted_arm(&client, "scripted-synth-b"), + ), + ); + let state = AcpAgentState::new_scripted(&test_config(), arms); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + + let updates: Arc>> = Arc::new(Mutex::new(Vec::new())); + let recorded = updates.clone(); + let root = project_dir(); + let root_path = root.path().display().to_string(); + + ClientRole + .builder() + .name("test-client") + .on_receive_notification( + { + let updates = updates.clone(); + async move |notif: SessionNotification, _cx| { + updates + .lock() + .expect("updates mutex poisoned") + .push(notif.update); + Ok(()) + } + }, + agent_client_protocol::on_receive_notification!(), + ) + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + let session = cx + .send_request(ClientNewSessionRequest::new(root_path)) + .block_task() + .await?; + // Switch onto "deepseek" (mode id == cast name) before the turn. + cx.send_request(ClientSetSessionModeRequest::new( + session.session_id.clone(), + "deepseek", + )) + .block_task() + .await?; + cx.send_request(ClientPromptRequest::new( + session.session_id, + vec![ContentBlock::Text(TextContent::new("which model answers?"))], + )) + .block_task() + .await + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + let updates = recorded.lock().expect("updates mutex poisoned"); + let answer = updates + .iter() + .find_map(|u| match u { + SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content { + ContentBlock::Text(text) => Some(text.text.clone()), + _ => None, + }, + _ => None, + }) + .expect("an answer chunk must have been sent"); + assert!( + answer.contains("from B") && answer.contains("scripted-synth-b"), + "switching to `deepseek` must route the turn to its scripted model, got: {answer}" + ); + assert!( + !answer.contains("from A"), + "the `anthropic` cast's model must not have answered: {answer}" + ); + } + + /// `session/cancel` mid-turn aborts the spawned consult task; the prompt's own + /// response lands with `StopReason::Cancelled`, per spec. `hang_model` makes the + /// synth's completion call park forever once invoked (see `test_support.rs`) — + /// the deterministic "still running" shape, the same tool `jobs.rs`'s own cancel + /// test uses a gated channel for. The request goes out (so `consult_task` is + /// genuinely in flight, not raced against a fast mock answer), then never + /// resolves on its own: only the abort ends it. + #[tokio::test] + async fn cancel_mid_turn_yields_the_cancelled_stop_reason() { + let client = ScriptedClient::builder() + .hang_model("scripted-synth") + .build(); + + let default_cast = test_config().default_cast; + let mut arms = HashMap::new(); + arms.insert( + default_cast, + ( + scripted_arm(&client, "scripted-explorer"), + scripted_arm(&client, "scripted-synth"), + ), + ); + let state = AcpAgentState::new_scripted(&test_config(), arms); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + + let root = project_dir(); + let root_path = root.path().display().to_string(); + + let response = ClientRole + .builder() + .name("test-client") + .on_receive_notification( + async move |_notif: SessionNotification, _cx| Ok(()), + agent_client_protocol::on_receive_notification!(), + ) + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + let session = cx + .send_request(ClientNewSessionRequest::new(root_path)) + .block_task() + .await?; + let prompt = cx.send_request(ClientPromptRequest::new( + session.session_id.clone(), + vec![ContentBlock::Text(TextContent::new("cancel me"))], + )); + cx.send_notification(CancelNotification::new(session.session_id))?; + prompt.block_task().await + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + assert_eq!( + response.stop_reason, + StopReason::Cancelled, + "a session/cancel mid-turn must yield the spec's Cancelled stop reason" + ); + } + + /// `session/cancel` with nothing running is a clean no-op: no panic, and a later + /// prompt on the same session still runs normally. + #[tokio::test] + async fn cancel_with_no_in_flight_turn_is_a_clean_no_op() { + let client = ScriptedClient::builder() + .on_model("scripted-synth", |_req| Ok(text_response("still works"))) + .build(); + let default_cast = test_config().default_cast; + let mut arms = HashMap::new(); + arms.insert( + default_cast, + ( + scripted_arm(&client, "scripted-explorer"), + scripted_arm(&client, "scripted-synth"), + ), + ); + let state = AcpAgentState::new_scripted(&test_config(), arms); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + + let root = project_dir(); + let root_path = root.path().display().to_string(); + + let response = ClientRole + .builder() + .name("test-client") + .on_receive_notification( + async move |_notif: SessionNotification, _cx| Ok(()), + agent_client_protocol::on_receive_notification!(), + ) + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + let session = cx + .send_request(ClientNewSessionRequest::new(root_path)) + .block_task() + .await?; + // Nothing running yet — must be a silent no-op. + cx.send_notification(CancelNotification::new(session.session_id.clone()))?; + cx.send_request(ClientPromptRequest::new( + session.session_id, + vec![ContentBlock::Text(TextContent::new("normal question"))], + )) + .block_task() + .await + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + assert_eq!( + response.stop_reason, + StopReason::EndTurn, + "an idle-session cancel must not disturb a subsequent normal turn" + ); + } + + /// A non-text content block is refused up front with a clear error, never + /// silently dropped or forwarded to the model. + #[tokio::test] + async fn a_non_text_content_block_is_refused() { + let state = AcpAgentState::new(Arc::new(test_config())).expect("resolver builds"); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + let root = project_dir(); + let root_path = root.path().display().to_string(); + + let result = ClientRole + .builder() + .name("test-client") + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + let session = cx + .send_request(ClientNewSessionRequest::new(root_path)) + .block_task() + .await?; + Ok(cx + .send_request(ClientPromptRequest::new( + session.session_id, + vec![ContentBlock::Image( + agent_client_protocol::schema::v1::ImageContent::new( + "aGk=", + "image/png", + ), + )], + )) + .block_task() + .await) + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + let err = result.expect_err("an image content block must be refused"); + assert!( + err.message.contains("text content blocks") + || err + .data + .as_ref() + .map(|d| d.to_string().contains("text content blocks")) + .unwrap_or(false), + "the refusal names the constraint: {err:?}" + ); } } diff --git a/src/cli.rs b/src/cli.rs index 94ad91b..40a39a9 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -1807,7 +1807,16 @@ pub async fn run_acp(common: CommonArgs) -> i32 { eprintln!("kaibo: default cast: {e:#}"); return EXIT_SETUP; } - let state = crate::acp::AcpAgentState::new(&config); + // Builds a `Resolver` under the hood (same containment/allowed-set setup every + // front door computes) — a bad `--root`/`--allow-path` is a setup rejection here + // too, not a silent empty boundary. + let state = match crate::acp::AcpAgentState::new(Arc::new(config)) { + Ok(s) => s, + Err(e) => { + eprintln!("kaibo: acp setup: {e:#}"); + return EXIT_SETUP; + } + }; match crate::acp::agent(state) .connect_to(agent_client_protocol::Stdio::new()) .await diff --git a/src/server/mod.rs b/src/server/mod.rs index 310db33..4eb5bfc 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -59,19 +59,19 @@ mod resolver; pub use resolver::Resolver; -// Re-exported for the CLI front door (`crate::cli`), which renders the same answer -// footer, failure text, `kaibo://config` document, and `batch list` recency window the -// MCP handler does. +// Re-exported for the other front doors (`crate::cli`, `crate::acp`), which render +// the same answer footer, non-fatal warnings, failure text, `kaibo://config` +// document, and `batch list` recency window the MCP handler does. pub(crate) use config_resource::render_config_resource; pub(crate) use render::{ - batch_within_window, consultation_failure_text, now_epoch_secs, with_provenance, - BATCH_RECENCY_WINDOW_SECS, + append_warnings, batch_within_window, consultation_failure_text, now_epoch_secs, + with_provenance, BATCH_RECENCY_WINDOW_SECS, }; use render::{ - append_warnings, batch_poll_brief, consult_result, consultation_failed, fmt_usage, - is_batch_handle, parse_batch_handle, render_job, render_jobs_section, render_wait, - wait_level_floor, wait_level_label, + batch_poll_brief, consult_result, consultation_failed, fmt_usage, is_batch_handle, + parse_batch_handle, render_job, render_jobs_section, render_wait, wait_level_floor, + wait_level_label, }; /// kaibo's resource URI namespace. Everything kaish-related hangs off `kaibo://kaish/`. From 7163f06278de66e7c6bc29563f6648089c627bd2 Mon Sep 17 00:00:00 2001 From: A Tobey Date: Tue, 4 Aug 2026 19:02:52 -0400 Subject: [PATCH 3/3] fix(acp): route session/new cwd through resolve_root containment chunk 2's real consult loop threaded an ACP session's cwd straight into ConsultConfig/the sandbox root without ever running it through the same containment boundary the MCP path argument and the CLI --root/--allow-path enforce (Resolver::resolve_root, src/server/containment.rs). The gap: an ACP client's session/new could aim kaibo's read-only shell at any directory the process could read, not just the configured allowed set. Read-scope is supposed to be bounded for every front door alike, so this was a structural hole in an otherwise-enforced invariant, flagged in the chunk-2 report. Fix: AcpAgentState::resolve_session_cwd wraps the same Resolver::resolve_root the MCP consult tool calls, so a session/new cwd is canonicalized (symlinks, `..` resolved) and checked against the allowed set built from --root/ --allow-path/launch cwd -- no parallel canonicalizer. session/new now validates before minting a session id: a cwd outside the allowed set or one that can't canonicalize (nonexistent, not a directory) is refused with a clear invalid_params JSON-RPC error, matching resolve_root's own message text, never silently substituted or waved through. The canonicalized path becomes the SessionRecord's cwd, which is what every later session/prompt turn already threads into house_rules/orientation/ConsultConfig -- chunk 2's plumbing was honest, just unvalidated at the door. Tests (src/acp.rs, over the existing duplex-wire harness -- tests/ containment.rs is MCP-shaped around KaiboHandler/run_kaish and has no ACP transport harness to extend, so the wire tests carry this): - session_new_refuses_a_cwd_that_does_not_exist: chunk 1's original session_new_returns_an_id_and_advertises_cast_modes test predates containment and passed a deliberately nonexistent cwd expecting success; renamed and updated to expect refusal, since accepting that cwd was exactly the bug. - session_new_refuses_a_cwd_outside_the_allowed_set: a real, existing directory outside every allowed tree is refused the same way. - session_new_with_an_in_bounds_cwd_returns_an_id_and_advertises_cast_modes: a cwd equal to the configured root is accepted, and the session record holds the canonicalized path. - Every existing turn-driving test (prompt/cancel/set_mode/second-turn/ non-text-block) now builds its Config with root pinned to its project_dir() fixture (test_config_rooted), since those fixtures live outside the crate's inferred cwd and would otherwise be refused by the same check. Verified failing-first: temporarily bypassed resolve_session_cwd (return the raw cwd unchecked) and confirmed both refusal tests fail without the fix, then restored it. Gates: cargo build, clippy --all-targets (zero warnings), full cargo test (644+ passed, 0 failed), cargo tree -i aws-lc-rs / -i mimalloc both empty. Co-authored-by: Claude Sonnet --- src/acp.rs | 200 ++++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 182 insertions(+), 18 deletions(-) diff --git a/src/acp.rs b/src/acp.rs index f24f396..34674b1 100644 --- a/src/acp.rs +++ b/src/acp.rs @@ -84,6 +84,10 @@ use crate::session::Sessions; /// pinned to, and the in-flight turn's abort handle, if one is running. #[derive(Debug, Clone)] struct SessionRecord { + /// The session's consult root — already resolved through + /// [`AcpAgentState::resolve_session_cwd`] at `session/new`, so this is always + /// the CANONICALIZED path, contained in the allowed set. Never the client's raw + /// `cwd` string. cwd: PathBuf, mode_id: String, /// What `session/cancel` aborts. `None` between turns and while idle — a cancel @@ -195,7 +199,10 @@ impl AcpAgentState { SessionId::new(format!("session-{n}")) } - /// Record a freshly created session, starting it on the default cast. + /// Record a freshly created session, starting it on the default cast. `cwd` + /// must already be the canonicalized, contained root from + /// [`Self::resolve_session_cwd`] — this method trusts its caller and does not + /// re-check containment. fn insert_session(&self, id: SessionId, cwd: PathBuf) { let mut sessions = self .inner @@ -302,6 +309,22 @@ impl AcpAgentState { Ok(cast) } + /// Resolve and canonicalize a `session/new` `cwd` through the same containment + /// boundary every other front door enforces (`Resolver::resolve_root` — + /// symlinks and `..` resolved, then required at-or-under an allowed tree; + /// `--root`/`--allow-path`, the launch cwd when unset). Returns the + /// canonicalized path, which becomes this session's consult root. A `cwd` + /// outside the allowed set, or one that doesn't canonicalize at all + /// (nonexistent, not a directory), is refused here with the same clear error + /// `resolve_root` gives the MCP `path` argument — a loud refusal, never a + /// silent fallback to some other root. + fn resolve_session_cwd(&self, cwd: &std::path::Path) -> Result { + self.inner + .resolver + .resolve_root(Some(cwd.to_string_lossy().into_owned())) + .map_err(|e| e.message.to_string()) + } + /// Resolve one of `cast`'s slots into a live [`Arm`] — through the real /// [`Resolver`] in production, or the scripted map in tests. See [`ArmSource`]. fn arm(&self, cast: &Cast, role: ModelRole) -> Result { @@ -526,13 +549,28 @@ pub fn agent(state: AcpAgentState) -> impl ConnectTo { { let state = state.clone(); async move |req: NewSessionRequest, responder, _cx| { + // Containment first, same boundary as every other front door + // (`Resolver::resolve_root`): an ACP client names its own `cwd`, + // so it gets no more trust than an MCP caller's `path` argument. + // Refuse loudly — outside the allowed set, or not even a real + // directory — rather than minting a session pinned to whatever + // the client asked for. + let root = match state.resolve_session_cwd(&req.cwd) { + Ok(r) => r, + Err(msg) => { + return responder.respond_with_error( + Error::invalid_params().data(msg), + ) + } + }; let session_id = state.mint_session_id(); - state.insert_session(session_id.clone(), req.cwd.clone()); + state.insert_session(session_id.clone(), root.clone()); let modes = SessionModeState::new(state.inner.default_cast.clone(), state.session_modes()); tracing::debug!( session_id = %session_id, cwd = %req.cwd.display(), + resolved_root = %root.display(), "acp: session/new" ); responder.respond(NewSessionResponse::new(session_id).modes(modes)) @@ -773,6 +811,20 @@ mod tests { Config::builtin() } + /// `test_config()`, with `root` fixed to `root_dir` — mirrors + /// `handler_with_allowed` in `tests/containment.rs`: naming a root makes it + /// both the allowed tree and the inferred default root, so a + /// `session/new` whose `cwd` is `root_dir` (or a path under it) passes + /// containment. Every test that drives a real turn against a `project_dir()` + /// fixture needs this — `session/new` now enforces the same boundary the MCP + /// `path` argument always has, so an unrelated tempdir is no longer waved + /// through. + fn test_config_rooted(root_dir: &std::path::Path) -> Config { + let mut config = Config::builtin(); + config.root = Some(root_dir.to_path_buf()); + config + } + /// A scripted arm over `client`, addressing model `model` — same shape /// `consult`'s own offline tests build (`src/consult/engine.rs`), vision off /// (nothing here attaches an image). @@ -841,12 +893,22 @@ mod tests { assert!(response.auth_methods.is_empty()); } + /// `session/new` with a `cwd` inside the allowed set (here, exactly the + /// resolver's configured `root`) is accepted: it returns a nonempty session id, + /// advertises one mode per configured cast, and — the point of this fix — the + /// session record now holds the CANONICALIZED root (what + /// `resolve_session_cwd`/`resolve_root` returned), not the client's raw `cwd` + /// string, since that canonicalized path is what every later `session/prompt` + /// turn runs `consult` against. #[tokio::test] - async fn session_new_returns_an_id_and_advertises_cast_modes() { - let state = AcpAgentState::new(Arc::new(test_config())).expect("resolver builds"); + async fn session_new_with_an_in_bounds_cwd_returns_an_id_and_advertises_cast_modes() { + let root = project_dir(); + let config = test_config_rooted(root.path()); + let state = AcpAgentState::new(Arc::new(config)).expect("resolver builds"); let (agent_transport, client_transport) = new_transport_pair(); - let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + let agent_task = tokio::spawn(agent(state.clone()).connect_to(agent_transport)); + let root_path = root.path().display().to_string(); let response = ClientRole .builder() .name("test-client") @@ -854,7 +916,7 @@ mod tests { cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) .block_task() .await?; - cx.send_request(ClientNewSessionRequest::new("/tmp/kaibo-acp-test")) + cx.send_request(ClientNewSessionRequest::new(root_path)) .block_task() .await }) @@ -881,6 +943,107 @@ mod tests { modes.current_mode_id.to_string(), test_config().default_cast ); + + // The point of this fix: the session's stored root is the canonicalized + // path `resolve_root` computed, so a later turn's `consult` call runs + // against exactly what containment vetted — not the client's raw string. + let canonicalized = std::fs::canonicalize(root.path()).expect("root canonicalizes"); + let record = state + .session(&response.session_id) + .expect("session was recorded"); + assert_eq!( + record.cwd, canonicalized, + "the session's consult root must be the canonicalized, contained cwd" + ); + } + + /// A `session/new` whose `cwd` cannot canonicalize at all (it doesn't exist) is + /// refused with a clear JSON-RPC error — never silently accepted and never + /// silently substituted with some other root. This behavior is new: chunk 2's + /// `session/new` accepted any `cwd` string, including this deliberately + /// nonexistent one, without ever driving it through containment. + #[tokio::test] + async fn session_new_refuses_a_cwd_that_does_not_exist() { + let state = AcpAgentState::new(Arc::new(test_config())).expect("resolver builds"); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + + let result = ClientRole + .builder() + .name("test-client") + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + Ok(cx + .send_request(ClientNewSessionRequest::new( + "/tmp/kaibo-acp-test-cwd-does-not-exist", + )) + .block_task() + .await) + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + let err = result.expect_err( + "a cwd that cannot canonicalize (nonexistent) must be refused, never silently \ + accepted", + ); + let data = err.data.as_ref().map(|d| d.to_string()).unwrap_or_default(); + assert!( + err.message.contains("could not be resolved") || data.contains("could not be resolved"), + "the refusal names the canonicalization failure: {err:?}" + ); + } + + /// A `session/new` whose `cwd` is a real, existing directory — just one outside + /// every allowed tree — is refused the same as a nonexistent one. Containment + /// bounds *where* a session can point, not merely whether the path resolves. + #[tokio::test] + async fn session_new_refuses_a_cwd_outside_the_allowed_set() { + let allowed_root = project_dir(); + let outside_root = project_dir(); + let config = test_config_rooted(allowed_root.path()); + let state = AcpAgentState::new(Arc::new(config)).expect("resolver builds"); + let (agent_transport, client_transport) = new_transport_pair(); + let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); + + let outside_path = outside_root.path().display().to_string(); + let result = ClientRole + .builder() + .name("test-client") + .connect_with(client_transport, async move |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + Ok(cx + .send_request(ClientNewSessionRequest::new(outside_path)) + .block_task() + .await) + }) + .await + .expect("client-side connection failed"); + + agent_task + .await + .expect("agent task panicked") + .expect("agent-side connection failed"); + + let err = result.expect_err( + "a cwd outside every allowed tree must be refused — an ACP client cannot aim \ + kaibo's read-only shell at an arbitrary directory", + ); + let data = err.data.as_ref().map(|d| d.to_string()).unwrap_or_default(); + assert!( + err.message.contains("outside the allowed set") + || data.contains("outside the allowed set"), + "the refusal names the containment boundary: {err:?}" + ); } #[test] @@ -916,13 +1079,13 @@ mod tests { scripted_arm(&client, "scripted-synth"), ), ); - let state = AcpAgentState::new_scripted(&test_config(), arms); + let root = project_dir(); + let state = AcpAgentState::new_scripted(&test_config_rooted(root.path()), arms); let (agent_transport, client_transport) = new_transport_pair(); let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); let updates: Arc>> = Arc::new(Mutex::new(Vec::new())); let recorded = updates.clone(); - let root = project_dir(); let root_path = root.path().display().to_string(); let prompt_response = ClientRole @@ -1024,11 +1187,11 @@ mod tests { scripted_arm(&client, "scripted-synth"), ), ); - let state = AcpAgentState::new_scripted(&test_config(), arms); + let root = project_dir(); + let state = AcpAgentState::new_scripted(&test_config_rooted(root.path()), arms); let (agent_transport, client_transport) = new_transport_pair(); let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); - let root = project_dir(); let root_path = root.path().display().to_string(); let (first, second) = ClientRole @@ -1099,13 +1262,13 @@ mod tests { scripted_arm(&client, "scripted-synth-b"), ), ); - let state = AcpAgentState::new_scripted(&test_config(), arms); + let root = project_dir(); + let state = AcpAgentState::new_scripted(&test_config_rooted(root.path()), arms); let (agent_transport, client_transport) = new_transport_pair(); let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); let updates: Arc>> = Arc::new(Mutex::new(Vec::new())); let recorded = updates.clone(); - let root = project_dir(); let root_path = root.path().display().to_string(); ClientRole @@ -1197,11 +1360,11 @@ mod tests { scripted_arm(&client, "scripted-synth"), ), ); - let state = AcpAgentState::new_scripted(&test_config(), arms); + let root = project_dir(); + let state = AcpAgentState::new_scripted(&test_config_rooted(root.path()), arms); let (agent_transport, client_transport) = new_transport_pair(); let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); - let root = project_dir(); let root_path = root.path().display().to_string(); let response = ClientRole @@ -1257,11 +1420,11 @@ mod tests { scripted_arm(&client, "scripted-synth"), ), ); - let state = AcpAgentState::new_scripted(&test_config(), arms); + let root = project_dir(); + let state = AcpAgentState::new_scripted(&test_config_rooted(root.path()), arms); let (agent_transport, client_transport) = new_transport_pair(); let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); - let root = project_dir(); let root_path = root.path().display().to_string(); let response = ClientRole @@ -1307,10 +1470,11 @@ mod tests { /// silently dropped or forwarded to the model. #[tokio::test] async fn a_non_text_content_block_is_refused() { - let state = AcpAgentState::new(Arc::new(test_config())).expect("resolver builds"); + let root = project_dir(); + let state = + AcpAgentState::new(Arc::new(test_config_rooted(root.path()))).expect("resolver builds"); let (agent_transport, client_transport) = new_transport_pair(); let agent_task = tokio::spawn(agent(state).connect_to(agent_transport)); - let root = project_dir(); let root_path = root.path().display().to_string(); let result = ClientRole