fix(acp)!: v1 session stream reports prompt errors and full responses

md/migration-v5.md

@@ -41,7 +41,7 @@ versions (`AgentProtocolRouter`, `ProxyProtocolRouter`,
 | `SessionSetup` | a session and the response of the request that set it up | a session and the response of the request that set it up |
 | `ActiveSession` | the session handle with its update stream | the command-only session handle |
 | `ForkSessionBuilder` | - | `fork_session*` (`unstable_session_fork`) |
-| `LoadSessionBuilder`, `RestoreSessionBuilder`, `SessionMessage`, `Blocking`, `NonBlocking`, `SessionBlockState` | v1 session helpers | - |
+| `LoadSessionBuilder`, `SessionMessage`, `Blocking`, `NonBlocking`, `SessionBlockState` | v1 session helpers | - |
 | `InitializeOutcome`, `InitializationVersionMismatch`, `SelectedInitializeResponse` | - | the result of `send_initialize` |
 | `schema` | `schema::v1` | `schema::v2` |
 
@@ -110,8 +110,21 @@ Every setup method returns one, in both versions: v1 `start_session` for
 and `session/resume`; v2 `session/new`, `session/resume` and
 `session/fork`. The v1 `run_until` and `on_session_start` callbacks receive
 one too, as do the v2 `on_proxy_session_start` callbacks. The v1
-`on_proxy_session_start` callback still receives the `SessionId`, because the
-session is proxied rather than handed to the caller.
+`on_proxy_session_start` callback receives a clone of the complete
+`NewSessionResponse` forwarded to the client, and `start_session_proxy`
+returns it. Both used to give only the `SessionId`. There is no
+`SessionSetup` here, because the session is proxied rather than handed to the
+caller.
+
+```rust,ignore
+// Before
+.on_proxy_session_start(responder, async |session_id| { ... })
+let session_id = builder.block_task().start_session_proxy(responder).await?;
+
+// After
+.on_proxy_session_start(responder, async |response| { let session_id = response.session_id; ... })
+let session_id = builder.block_task().start_session_proxy(responder).await?.session_id;
+```
 
 The accessors are removed in favor of the fields: `session()`,
 `session_mut()`, `response()`, `into_parts()` and `into_session()`. The v1
@@ -135,6 +148,49 @@ let SessionSetup { session, response, .. } = restored;
 let session = opened.session;
 ```
 
+### The v1 session stream reports prompt responses and errors
+
+`v1::ActiveSession::send_prompt` used to close the whole connection, and every
+session on it, when the agent answered `session/prompt` with an error. The
+caller never saw the error. The response, or the error, now arrives on the
+session's update stream, and `v1::SessionMessage` changed to carry it:
+
+- `SessionMessage::StopReason(StopReason)` is now
+  `SessionMessage::PromptResponse(Result<PromptResponse, Error>)`. It keeps
+  the full `PromptResponse`, including `usage` and `_meta`, which used to be
+  dropped.
+- `SessionMessage::SessionMessage(Dispatch)` is now
+  `SessionMessage::Message(v1::Dispatch)`, whose request handles are typed
+  for v1. Call `.into_raw()` before handing it to `MatchDispatch`.
+- `read_to_string` returns the prompt's error instead of the connection
+  closing.
+- `send_prompt_blocks(Vec<ContentBlock>)` sends prompts that are not plain
+  text.
+
+`v1::SessionBlockState` is sealed. `v1::Blocking` and `v1::NonBlocking` are
+its only implementations.
+
+```rust,ignore
+// Before
+match session.read_update().await? {
+    SessionMessage::SessionMessage(dispatch) => MatchDispatch::new(dispatch)
+        /* .if_notification(..) */,
+    SessionMessage::StopReason(stop_reason) => { /* .. */ }
+    _ => {}
+}
+
+// After
+match session.read_update().await? {
+    SessionMessage::Message(dispatch) => MatchDispatch::new(dispatch.into_raw())
+        /* .if_notification(..) */,
+    SessionMessage::PromptResponse(response) => {
+        let stop_reason = response?.stop_reason;
+        /* .. */
+    }
+    _ => {}
+}
+```
+
 ### Renamed and moved items
 
 Every path is relative to `agent_client_protocol`. Items not listed keep
@@ -162,8 +218,10 @@ their path.
 | `SessionBuilder` | `v1::SessionBuilder` |
 | `ActiveSession` | `v1::ActiveSession` |
 | `RestoredSession` | `v1::SessionSetup` |
-| `RestoreSessionBuilder` | `v1::RestoreSessionBuilder` (also `v1::LoadSessionBuilder` / `v1::ResumeSessionBuilder` for each request) |
+| `RestoreSessionBuilder` | `v1::LoadSessionBuilder` / `v1::ResumeSessionBuilder` (the generic builder is no longer exported) |
 | `SessionMessage` | `v1::SessionMessage` |
+| `SessionMessage::SessionMessage(Dispatch)` | `v1::SessionMessage::Message(v1::Dispatch)` |
+| `SessionMessage::StopReason(StopReason)` | `v1::SessionMessage::PromptResponse(Result<PromptResponse, Error>)` |
 | `Blocking`, `NonBlocking`, `SessionBlockState` | `v1::Blocking`, `v1::NonBlocking`, `v1::SessionBlockState` |
 | `schema::InitializeProxyRequest` | `schema::v1::InitializeProxyRequest` |
 | `OpenedSession` (`v1` and `v2`) | `v1::SessionSetup`, `v2::SessionSetup` (public `session` and `response` fields) |

src/agent-client-protocol-conductor/tests/request_cancellation.rs

@@ -849,7 +849,7 @@ async fn proxy_session_helper_cancellation_propagates_to_agent() -> Result<(), E
                responder: Responder<NewSessionResponse, V1>,
                cx: ConnectionTo<Conductor>| {
             cx.build_session_from(request)
-                .on_proxy_session_start(responder, async |_session_id| Ok::<(), Error>(()))
+                .on_proxy_session_start(responder, async |_response| Ok::<(), Error>(()))
         },
         agent_client_protocol::on_receive_request!(),
     );
@@ -1056,7 +1056,7 @@ async fn proxy_session_helper_cleans_up_mcp_handlers_after_cancelled_session() -
             );
             cx.build_session_from(request)
                 .with_mcp_server(mcp_server)?
-                .on_proxy_session_start(responder, async |_session_id| Ok::<(), Error>(()))
+                .on_proxy_session_start(responder, async |_response| Ok::<(), Error>(()))
         },
         agent_client_protocol::on_receive_request!(),
     );

src/agent-client-protocol-conductor/tests/v1_chain_equivalence.rs

@@ -11,12 +11,15 @@ use std::{
 };
 
 use agent_client_protocol::V1;
+use agent_client_protocol::mcp_server::McpServer;
 use agent_client_protocol::v1::ConnectionTo;
 use agent_client_protocol::{
-    Agent, Client, ConnectTo, DynConnectTo, Error, Lines, Proxy, Responder, UntypedMessage,
+    Agent, Client, Conductor, ConnectTo, DynConnectTo, Error, Lines, Proxy, Responder,
+    UntypedMessage,
     schema::{ProtocolVersion, v1},
 };
 use agent_client_protocol_conductor::{ConductorImpl, SelectedInitializeRequest};
+use agent_client_protocol_rmcp::rmcp_3::McpServerExt as _;
 use futures::{SinkExt as _, StreamExt as _, channel::mpsc};
 use serde_json::{Value, json};
 
@@ -78,6 +81,21 @@ struct ChainRun {
 
 /// Initialize a v1 chain with `initialize`, then create a session.
 async fn run_v1_chain(initialize: &Value) -> ChainRun {
+    run_v1_chain_with(
+        initialize,
+        json!({ "cwd": "/v1-equivalence", "mcpServers": [], "_unknownRequestField": true }),
+        || DynConnectTo::new(Proxy.v1()),
+    )
+    .await
+}
+
+/// Initialize a chain of `proxy` and a v1 agent with `initialize`, then send
+/// `session/new` with `session_params`.
+async fn run_v1_chain_with(
+    initialize: &Value,
+    session_params: Value,
+    proxy: fn() -> DynConnectTo<Conductor>,
+) -> ChainRun {
     let received = Received::default();
     let instantiated_version = Arc::new(Mutex::new(None));
     let conductor = ConductorImpl::new_agent("v1-equivalence", {
@@ -86,7 +104,7 @@ async fn run_v1_chain(initialize: &Value) -> ChainRun {
         move |request: SelectedInitializeRequest| async move {
             *instantiated_version.lock().expect("version lock") = Some(request.protocol_version());
             let request = request.into_v1()?;
-            let proxies = vec![DynConnectTo::new(Proxy.v1())];
+            let proxies = vec![proxy()];
             let agent = DynConnectTo::new(recording_v1_agent(received));
             Ok((request.into(), proxies, agent))
         }
@@ -122,7 +140,7 @@ async fn run_v1_chain(initialize: &Value) -> ChainRun {
         "jsonrpc": "2.0",
         "id": 2,
         "method": "session/new",
-        "params": { "cwd": "/v1-equivalence", "mcpServers": [], "_unknownRequestField": true },
+        "params": session_params,
     }));
     let session_response = recv().await;
 
@@ -239,3 +257,45 @@ async fn v1_chain_negotiates_newer_requests_to_v1() {
     assert!(received.get("_unknownRequestField").is_none(), "{received}");
     assert_session_forwarded_unchanged(&run);
 }
+
+/// A v1 proxy built with `with_mcp_server` appends its declaration to the raw
+/// `session/new` params. Fields the v1 schema types do not model reach the
+/// agent unchanged.
+#[tokio::test(flavor = "current_thread")]
+async fn v1_mcp_proxy_keeps_unmodeled_session_fields() {
+    let run = run_v1_chain_with(
+        &initialize_params(1),
+        json!({
+            "cwd": "/v1-equivalence",
+            "mcpServers": [],
+            "futureField": { "nested": ["keep me"] },
+            "_unknownRequestField": true,
+        }),
+        || {
+            DynConnectTo::new(
+                Proxy
+                    .v1()
+                    .with_mcp_server(McpServer::builder("raw-fields").build()),
+            )
+        },
+    )
+    .await;
+
+    assert_eq!(run.responses[1]["id"], 2, "{}", run.responses[1]);
+    assert_eq!(
+        run.responses[1]["result"]["sessionId"], "v1-session",
+        "{}",
+        run.responses[1]
+    );
+    let (method, params) = &run.received[1];
+    assert_eq!(method, "session/new");
+    assert_eq!(
+        params["futureField"],
+        json!({ "nested": ["keep me"] }),
+        "{params}"
+    );
+    assert_eq!(params["_unknownRequestField"], true, "{params}");
+    let servers = params["mcpServers"].as_array().expect("mcpServers array");
+    assert_eq!(servers.len(), 1, "{params}");
+    assert_eq!(servers[0]["name"], "raw-fields", "{params}");
+}

src/agent-client-protocol-cookbook/src/lib.rs

@@ -346,7 +346,9 @@ pub mod connecting_as_client {
     //!
     //! - [`send_prompt`] - Send a text prompt to the agent
     //! - [`read_update`] - Read the next update (text chunk, tool call, etc.)
-    //! - [`read_to_string`] - Read all text until the turn ends
+    //!   or the prompt's response
+    //! - [`read_to_string`] - Read all text until the turn ends. If the agent
+    //!   answers the prompt with an error, this returns that error.
     //!
     //! With the core SDK's `unstable_mcp_over_acp` feature, the session builder
     //! also supports adding MCP servers with [`with_mcp_server`].
@@ -826,14 +828,14 @@ pub mod per_session_mcp_server {
     //!             // Build the session and run code after it starts
     //!             connection.build_session_from(request)
     //!                 .with_mcp_server(mcp_server)?
-    //!                 .on_proxy_session_start(responder, async move |session_id| {
+    //!                 .on_proxy_session_start(responder, async move |response| {
     //!                     // Session proxying is installed before this callback is spawned.
     //!                     //
     //!                     // Use this for follow-up work that may wait for later connection
     //!                     // traffic. Register ID-independent state in the request handler.
     //!                     // For ID-keyed state, preinstall a gate that later handlers await,
     //!                     // then populate it here.
-    //!                     tracing::info!(%session_id, "Session started");
+    //!                     tracing::info!(session_id = %response.session_id, "Session started");
     //!                     Ok(())
     //!                 })
     //!         }, agent_client_protocol::on_receive_request!())
@@ -849,7 +851,7 @@ pub mod per_session_mcp_server {
     //! 1. Sends `NewSessionRequest` to the agent
     //! 2. When the response arrives, responds to the client automatically
     //! 3. Sets up message proxying for the session
-    //! 4. Runs your callback with the `SessionId`
+    //! 4. Runs your callback with a clone of the `NewSessionResponse`
     //!
     //! The callback runs after the session is established but doesn't block
     //! the message handler. It is suitable for follow-up work and eventual
@@ -866,8 +868,8 @@ pub mod per_session_mcp_server {
     //! `v2::ResumeSessionBuilder` handles `session/resume`. With
     //! `unstable_session_fork`, `v2::ForkSessionBuilder` handles `session/fork`.
     //! Their `on_proxy_session_start` callbacks receive a `v2::SessionSetup`,
-    //! not just a session ID, so they retain both the command-only handle and
-    //! the complete operation-specific response:
+    //! which holds the command-only handle as well as the complete
+    //! operation-specific response:
     //!
     //! ```rust,ignore
     //! use agent_client_protocol::schema::v2;
@@ -953,13 +955,13 @@ pub mod per_session_mcp_server {
     //!
     //!             let task_connection = connection.clone();
     //!             connection.spawn(async move {
-    //!                 let session_id = task_connection.build_session_from(request)
+    //!                 let response = task_connection.build_session_from(request)
     //!                     .with_mcp_server(mcp_server)?
     //!                     .block_task()
     //!                     .start_session_proxy(responder)
     //!                     .await?;
     //!
-    //!                 tracing::info!(%session_id, "Session started");
+    //!                 tracing::info!(session_id = %response.session_id, "Session started");
     //!                 Ok(())
     //!             })?;
     //!             Ok(())

src/agent-client-protocol/src/concepts/proxies.rs

@@ -119,8 +119,8 @@
 //!
 //! The v1 builder injects the declaration into new, load, resume, and
 //! feature-gated fork requests. The v2 builder injects it into new, resume,
-//! and feature-gated fork requests while preserving unrelated setup fields.
-//! Both reuse one connection-scoped server ID.
+//! and feature-gated fork requests. Both preserve unrelated setup fields and
+//! reuse one connection-scoped server ID.
 //!
 //! ## Per-Session MCP Server
 //!
@@ -135,8 +135,8 @@
 //!         let my_mcp_server = McpServer::<Conductor, _>::builder("tools").build();
 //!         cx.build_session_from(req)
 //!             .with_mcp_server(my_mcp_server)?
-//!             .on_proxy_session_start(responder, async |session_id| {
-//!                 // Session started with MCP server attached
+//!             .on_proxy_session_start(responder, async |response| {
+//!                 // Session `response.session_id` started with MCP server attached
 //!                 Ok(())
 //!             })
 //!     }, agent_client_protocol::on_receive_request!())

src/agent-client-protocol/src/concepts/sessions.rs

@@ -129,6 +129,13 @@
 //! # }
 //! ```
 //!
+//! The `session/prompt` response arrives on the session's update stream as
+//! [`SessionMessage::PromptResponse`](crate::v1::SessionMessage::PromptResponse),
+//! after the updates the agent sent for that turn. An error response arrives
+//! there too, and `read_to_string` returns it. It does not close the
+//! connection. Use `read_update` to see each
+//! [`SessionMessage`](crate::v1::SessionMessage), including the stop reason.
+//!
 //! # Adding MCP Servers
 //!
 //! You can attach MCP (Model Context Protocol) servers to a session to provide

src/agent-client-protocol/src/jsonrpc.rs

@@ -2050,6 +2050,7 @@ impl<
     ///
     /// The same native MCP server declaration is added to new, load, and resume
     /// requests, plus fork requests when `unstable_session_fork` is enabled.
+    /// Unrelated request fields are preserved exactly.
     ///
     /// Only applicable to proxies. Use the same method on `v2::Builder` to
     /// attach the server to protocol v2 setup requests.