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..34674b1
--- /dev/null
+++ b/src/acp.rs
@@ -0,0 +1,1523 @@
+//! `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. `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`), 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.
+//!
+//! # Progress → `session/update`
+//!
+//! [`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;
+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, ToolCall,
+    ToolCallContent, ToolCallId, ToolCallStatus, ToolCallUpdate, ToolCallUpdateFields, ToolKind,
+};
+use agent_client_protocol::schema::ProtocolVersion;
+use agent_client_protocol::{Agent, Client, ConnectTo, ConnectionTo, Error};
+
+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 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 {
+    /// 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
+    /// on an idle session is a clean no-op.
+    running: Option<tokio::task::AbortHandle>,
+}
+
+/// 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<String, (Arm, Arm)>),
+}
+
+/// 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<Inner>,
+}
+
+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<String>,
+    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<HashMap<SessionId, SessionRecord>>,
+    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, 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<Config>) -> anyhow::Result<Self> {
+        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<String, (Arm, Arm)>) -> 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 {
+                resolver,
+                cast_names,
+                default_cast,
+                consult_sessions,
+                sessions: Mutex::new(HashMap::new()),
+                next_session: AtomicU64::new(1),
+                arms,
+            }),
+        }
+    }
+
+    /// The session modes to advertise: one per configured cast, mode id == cast name.
+    fn session_modes(&self) -> Vec<SessionMode> {
+        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. `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
+            .sessions
+            .lock()
+            .expect("acp sessions mutex poisoned");
+        sessions.insert(
+            id,
+            SessionRecord {
+                cwd,
+                mode_id: self.inner.default_cast.clone(),
+                running: None,
+            },
+        );
+    }
+
+    /// Snapshot a session's record, if it exists.
+    fn session(&self, id: &SessionId) -> Option<SessionRecord> {
+        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,
+        }
+    }
+
+    /// 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<Cast, String> {
+        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 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<PathBuf, String> {
+        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<Arm, String> {
+        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::<Vec<_>>()
+                    )
+                })?;
+                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<String, String> {
+    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<Client>,
+    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<Option<ToolCallId>>,
+}
+
+impl AcpProgressSink {
+    fn new(cx: ConnectionTo<Client>, 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`,
+/// `session/set_mode`, and `session/cancel`, wired to `state`. Returns an opaque
+/// `ConnectTo<Client>` — 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<Client> {
+    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| {
+                    // 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(), 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))
+                }
+            },
+            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)),
+                        );
+                    };
+
+                    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<dyn ProgressSink> =
+                        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!(),
+        )
+        .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| {
+                    // 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(&notif.session_id) {
+                        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,
+        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
+    /// 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()
+    }
+
+    /// `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).
+    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
+    /// the `futures` traits `ByteStreams` requires.
+    fn new_transport_pair() -> (
+        impl ConnectTo<Agent> + 'static,
+        impl ConnectTo<Client> + 'static,
+    ) {
+        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);
+        (
+            ByteStreams::new(agent_write.compat_write(), agent_read.compat()),
+            ByteStreams::new(client_write.compat_write(), client_read.compat()),
+        )
+    }
+
+    /// 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(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 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());
+    }
+
+    /// `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_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.clone()).connect_to(agent_transport));
+
+        let root_path = root.path().display().to_string();
+        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(root_path))
+                    .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<String> = modes
+            .available_modes
+            .iter()
+            .map(|m| m.id.to_string())
+            .collect();
+        mode_ids.sort();
+        let mut expected: Vec<String> = 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
+        );
+
+        // 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]
+    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_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 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<Mutex<Vec<SessionUpdate>>> = Arc::new(Mutex::new(Vec::new()));
+        let recorded = updates.clone();
+        let root_path = root.path().display().to_string();
+
+        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(root_path))
+                    .block_task()
+                    .await?;
+                cx.send_request(ClientPromptRequest::new(
+                    session.session_id,
+                    vec![ContentBlock::Text(TextContent::new("what does this do?"))],
+                ))
+                .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").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!(
+                        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:?}"),
+        }
+    }
+
+    /// 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 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_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);
+    }
+
+    /// `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 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<Mutex<Vec<SessionUpdate>>> = Arc::new(Mutex::new(Vec::new()));
+        let recorded = updates.clone();
+        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 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_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 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_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 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_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 99ce163..40a39a9 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,53 @@ 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;
+    }
+    // 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
+    {
+        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()),
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/`.