fix: port shared review and ACP lifecycle fixes

src/crates/interfaces/acp/AGENTS.md

@@ -41,4 +41,11 @@ Keep these role features additive and do not replace either closure with
 cargo check -p openbitfun-acp --no-default-features --features client
 cargo check -p openbitfun-acp --no-default-features --features server
 cargo test -p openbitfun-acp
+cargo test -p openbitfun-acp --no-default-features --features client,openbitfun-core/git --lib client::prompt::tests
 ```
+
+The focused client prompt tests cover protocol errors, retry, cancellation,
+partial output, and transport termination with in-memory agent streams. These
+fixtures do not require a live provider or a device connection. The explicit
+Core `git` feature satisfies the worktree tool dependency in the current client
+closure without enabling the server role or `product-full`.

src/crates/interfaces/acp/src/client/manager.rs

@@ -44,6 +44,7 @@ use super::config::{
     AcpClientRequirementProbe, AcpClientStatus, RemoteAcpClientRequirementSnapshot,
 };
 use super::dsh_profile::{ensure_bundled_profile, ensure_bundled_profile_remote};
+use super::prompt::AcpPrompt;
 use super::remote_capability_store::RemoteAcpCapabilityStore;
 use super::remote_session::{preferred_resume_strategies, AcpRemoteSessionStrategy};
 use super::remote_shell::{remote_user_shell_command, render_remote_env_assignments, shell_escape};
@@ -63,6 +64,7 @@ use super::stream::{
     AcpToolCallTracker,
 };
 use super::tool::AcpAgentTool;
+use super::transport::{handle_transport_closed, AcpTransport};
 
 const CONFIG_PATH: &str = "acp_clients";
 const CLIENT_STARTUP_TIMEOUT_SECS: u64 = 60;
@@ -641,6 +643,7 @@ impl AcpClientService {
                 return Err(error);
             }
         };
+        let transport = AcpTransport::new(transport);
         *connection.child.lock().await = child;
         let service = self.clone();
         let connection_for_task = connection.clone();
@@ -652,6 +655,10 @@ impl AcpClientService {
             let result = Client
                 .builder()
                 .name("openbitfun-acp-client")
+                .on_receive_notification(
+                    handle_transport_closed,
+                    agent_client_protocol::on_receive_notification!(),
+                )
                 .on_receive_request(
                     {
                         let service = service.clone();
@@ -1184,8 +1191,8 @@ impl AcpClientService {
                 .active
                 .as_mut()
                 .ok_or_else(|| OpenBitFunError::service("ACP session was not initialized"))?;
-            active.send_prompt(prompt).map_err(protocol_error)?;
-            read_turn_to_string(&mut session).await
+            let mut prompt = AcpPrompt::start(active, prompt);
+            read_turn_to_string(&mut session, &mut prompt).await
         };
 
         if let Some(seconds) = timeout_seconds.filter(|seconds| *seconds > 0) {
@@ -1235,13 +1242,13 @@ impl AcpClientService {
             .await?;
 
             discard_pending_session_updates_if_needed(&mut session).await;
-            {
+            let mut prompt = {
                 let active = session
                     .active
                     .as_mut()
                     .ok_or_else(|| OpenBitFunError::service("ACP session was not initialized"))?;
-                active.send_prompt(prompt).map_err(protocol_error)?;
-            }
+                AcpPrompt::start(active, prompt)
+            };
             let mut round_tracker = AcpStreamRoundTracker::new();
             let mut tool_call_tracker = AcpToolCallTracker::new();
 
@@ -1250,7 +1257,7 @@ impl AcpClientService {
                     let active = session.active.as_mut().ok_or_else(|| {
                         OpenBitFunError::service("ACP session was not initialized")
                     })?;
-                    active.read_update().await.map_err(protocol_error)?
+                    prompt.read_update(active).await.map_err(protocol_error)?
                 };
 
                 match message {
@@ -2343,7 +2350,10 @@ where
     Ok(())
 }
 
-async fn read_turn_to_string(session: &mut AcpRemoteSession) -> OpenBitFunResult<String> {
+async fn read_turn_to_string(
+    session: &mut AcpRemoteSession,
+    prompt: &mut AcpPrompt,
+) -> OpenBitFunResult<String> {
     let mut output = String::new();
     let mut tool_call_tracker = AcpToolCallTracker::new();
     loop {
@@ -2352,7 +2362,7 @@ async fn read_turn_to_string(session: &mut AcpRemoteSession) -> OpenBitFunResult
                 .active
                 .as_mut()
                 .ok_or_else(|| OpenBitFunError::service("ACP session was not initialized"))?;
-            active.read_update().await.map_err(protocol_error)?
+            prompt.read_update(active).await.map_err(protocol_error)?
         };
 
         match message {

src/crates/interfaces/acp/src/client/mod.rs

@@ -2,6 +2,7 @@ mod builtin_clients;
 mod config;
 mod dsh_profile;
 mod manager;
+mod prompt;
 mod remote_capability_store;
 mod remote_session;
 mod remote_shell;
@@ -11,6 +12,7 @@ mod session_persistence;
 mod stream;
 mod tool;
 mod tool_card_bridge;
+mod transport;
 
 pub use config::{
     AcpClientConfig, AcpClientConfigFile, AcpClientInfo, AcpClientPermissionMode,

src/crates/interfaces/acp/src/client/prompt.rs

@@ -0,0 +1,42 @@
+use std::future::Future;
+use std::pin::Pin;
+
+use agent_client_protocol::schema::{PromptRequest, PromptResponse};
+use agent_client_protocol::{ActiveSession, Agent, Error, SessionMessage};
+
+/// Own the prompt response separately from the session notification queue.
+/// ACP 0.12's ActiveSession::send_prompt only queues successful stop reasons;
+/// its error callback terminates the connection without waking read_update.
+pub(super) struct AcpPrompt {
+    response: Pin<Box<dyn Future<Output = Result<PromptResponse, Error>> + Send>>,
+}
+
+impl AcpPrompt {
+    pub(super) fn start(active: &ActiveSession<'_, Agent>, prompt: String) -> Self {
+        let request = PromptRequest::new(active.session_id().clone(), vec![prompt.into()]);
+        let response = active.connection().send_request(request).block_task();
+        Self {
+            response: Box::pin(response),
+        }
+    }
+
+    pub(super) async fn read_update(
+        &mut self,
+        active: &mut ActiveSession<'_, Agent>,
+    ) -> Result<SessionMessage, Error> {
+        tokio::select! {
+            // Preserve notifications queued before the response, including any
+            // partial output preceding an error. The response also wakes us
+            // when the connection drops its pending request sender.
+            biased;
+            update = active.read_update() => update,
+            response = &mut self.response => {
+                response.map(|response| SessionMessage::StopReason(response.stop_reason))
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+#[path = "prompt/tests.rs"]
+mod tests;