From c0caa310d72f7317f26b539fc38d2938ecbcbb6f Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Wed, 14 Oct 2026 10:30:29 +0200 Subject: [PATCH 1/4] fix(mcp, cli): Keep tool sessions alive through long turns A turn could lose every tool for the rest of its run once the model spent more than five minutes generating between tool calls. The in-process JP MCP Server used rmcp's default session config, whose 300-second idle timeout ended the Host's session while no traffic flowed. Every later call then failed with `Session expired (HTTP 404)`, and the assistant was handed those failures as tool results to retry against a connection that could not recover. The endpoint's sessions no longer expire when idle. They last until their client closes them or the endpoint shuts down at the end of the turn, which already bounds a session an external client abandons. When JP's own connection to the server is lost anyway, the turn now ends with "Lost the connection to the JP MCP Server" instead of continuing. Every unsettled call is recorded as not executed, or as possibly run if it had been released, so each request stays paired with a response; no follow-up request is sent, and nothing more is prompted for. A single `WARN` names the error and the endpoint. A tool that fails on its own still fails only its own call. Signed-off-by: Jean Mertz --- crates/jp_cli/src/cmd.rs | 13 ++ .../jp_cli/src/cmd/query/tool/coordinator.rs | 119 +++++++++++++-- .../src/cmd/query/tool/coordinator_tests.rs | 139 ++++++++++++++++++ .../src/cmd/query/tool/executor_error.rs | 27 ++++ .../src/cmd/query/tool/executor_mock.rs | 17 ++- .../jp_cli/src/cmd/query/tool/mcp_executor.rs | 2 +- .../src/cmd/query/tool/mcp_executor_tests.rs | 44 +++++- crates/jp_cli/src/cmd/query/turn_loop.rs | 16 ++ .../jp_cli/src/cmd/query/turn_loop_tests.rs | 107 +++++++++++++- crates/jp_cli/src/error.rs | 5 + crates/jp_mcp/src/server/http.rs | 32 +++- crates/jp_mcp/src/server/http_tests.rs | 49 +++++- 12 files changed, 547 insertions(+), 23 deletions(-) diff --git a/crates/jp_cli/src/cmd.rs b/crates/jp_cli/src/cmd.rs index 8c9120440..f29a58d6d 100644 --- a/crates/jp_cli/src/cmd.rs +++ b/crates/jp_cli/src/cmd.rs @@ -453,6 +453,19 @@ impl From for Error { Mcp(error) => return error.into(), McpEndpoint(error) => [("message", error.to_string())].into(), McpHost(error) => with_cause(&error, "MCP Host control failed"), + McpConnectionLost => [ + ( + "message", + "Lost the connection to the JP MCP Server, so the turn ended".to_owned(), + ), + ( + "suggestion", + "The tool calls it could not complete are recorded in the conversation. \ + Continue the conversation to retry them on a new connection." + .to_owned(), + ), + ] + .into(), Llm(error) => return error.into(), Io(error) => return error.into(), Url(error) => return error.into(), diff --git a/crates/jp_cli/src/cmd/query/tool/coordinator.rs b/crates/jp_cli/src/cmd/query/tool/coordinator.rs index f2cf51f07..32a873296 100644 --- a/crates/jp_cli/src/cmd/query/tool/coordinator.rs +++ b/crates/jp_cli/src/cmd/query/tool/coordinator.rs @@ -46,6 +46,7 @@ use std::{ collections::{BTreeMap, HashMap, VecDeque}, + fmt::Display, future::pending, sync::Arc, }; @@ -171,7 +172,14 @@ pub(crate) enum ToolEvent { /// A fanned-out call's operations were folded into the one response it is /// recorded with. - Folded { group: usize, review: Review }, + /// + /// `connection_lost` describes the failure when the call could not be + /// recorded because the Host lost its connection to the JP MCP Server. + Folded { + group: usize, + review: Review, + connection_lost: Option, + }, } #[derive(Debug, Default)] @@ -211,6 +219,13 @@ pub enum ExecutionOutcome { /// without a follow-up request. Stopped, + /// The Host lost its connection to the JP MCP Server, so no call could + /// complete. + /// Every call has a response recording that; the caller should record the + /// responses and end the turn with an error, rather than hand the failures + /// to the assistant to retry on a connection that is gone. + ConnectionLost, + /// The user escalated past the tool interrupt menu (cancelled it with /// Ctrl-C). /// The running tools have been cancelled; the caller should begin a @@ -306,6 +321,10 @@ struct Batch { /// Whether the progress row was claimed for the released calls. progress: bool, + + /// Whether the Host lost its connection to the JP MCP Server, after which + /// no call can complete. + connection_lost: bool, } impl Batch { @@ -325,6 +344,7 @@ impl Batch { stream_ended: false, released: false, progress: false, + connection_lost: false, } } @@ -1184,6 +1204,10 @@ impl ToolCoordinator { } } + if batch.connection_lost { + outcome.upgrade(ExecutionOutcome::ConnectionLost); + } + // Deregister the tool interrupt handler. drop(interrupt_guard); @@ -1431,9 +1455,26 @@ impl ToolCoordinator { self.render_result(&tool_name, &review.response, host.renderer); self.settle(batch, call, review, host); } - ToolEvent::Folded { group, review } => { - batch.groups[group].review = Some(review); - } + ToolEvent::Folded { + group, + review, + connection_lost, + } => self.on_folded(batch, group, review, connection_lost, host), + } + } + + /// Take the response a fanned-out call is recorded with. + fn on_folded( + &mut self, + batch: &mut Batch, + group: usize, + review: Review, + connection_lost: Option, + host: &mut Host<'_>, + ) { + batch.groups[group].review = Some(review); + if let Some(error) = connection_lost { + self.lose_connection(batch, &error, host); } } @@ -1445,8 +1486,22 @@ impl ToolCoordinator { result: ExecutorResult, host: &mut Host<'_>, ) { + let released = batch.calls[index].released; + if let ExecutorResult::Failed(error) | ExecutorResult::OutcomeUnknown(error) = &result + && error.is_connection_lost() + { + self.lose_connection(batch, error, host); + } + // Nothing can run once the connection is gone, so a call that gets this + // far ends here rather than waiting on the user or on a release. + // A result it completed with stands: it is what the call produced. + if batch.connection_lost && !matches!(result, ExecutorResult::Completed(_)) { + let may_have_run = released && matches!(result, ExecutorResult::OutcomeUnknown(_)); + self.settle_lost(batch, index, may_have_run, host); + return; + } + let call = &mut batch.calls[index]; - let released = call.released; let tool_id = call.tool_id.clone(); match result { ExecutorResult::AwaitingAdmission => call.wait = Wait::Admission, @@ -2052,25 +2107,27 @@ impl ToolCoordinator { let events = batch.events.clone(); let folds = batch.folds.clone(); tokio::spawn(async move { - let review = tokio::select! { + let (review, connection_lost) = tokio::select! { biased; () = folds.cancelled() => return, recorded = fan_out.recorded() => match recorded { - Ok(response) => Review::unchanged(response), + Ok(response) => (Review::unchanged(response), None), // Recorded as failed, so the conversation still pairs the // request with a response. Err(error) => { warn!(%error, tool_id, "A fanned-out call could not be completed."); - Review::replaced(ToolCallResponse { + let review = Review::replaced(ToolCallResponse { id: tool_id, result: Err(format!("Tool call could not be completed: {error}")), - }) + }); + (review, error.is_connection_lost().then(|| error.to_string())) } }, }; drop(events.send(ToolEvent::Folded { group: index, review, + connection_lost, })); }); } @@ -2261,6 +2318,20 @@ impl ToolCoordinator { may_have_run, "Tool call could not be completed." ); + self.settle_lost(batch, index, may_have_run, host); + } + + /// Record a call JP could not complete, with what the model needs to decide + /// what to do next. + fn settle_lost( + &mut self, + batch: &mut Batch, + index: usize, + may_have_run: bool, + host: &mut Host<'_>, + ) { + let call = &batch.calls[index]; + let tool_name = &call.tool_name; let message = if may_have_run { format!( "Tool '{tool_name}' may have run, but JP lost the call before its result arrived. \ @@ -2273,12 +2344,40 @@ impl ToolCoordinator { ) }; let response = ToolCallResponse { - id: batch.calls[index].tool_id.clone(), + id: call.tool_id.clone(), result: Err(message), }; self.settle(batch, index, Review::replaced(response), host); } + /// Stop the batch after the Host lost its connection to the JP MCP Server. + /// + /// Every call travels on that connection, so none still unsettled can + /// complete: each is recorded as lost now, rather than prompted for or + /// waited on. + /// `finish` then reports [`ExecutionOutcome::ConnectionLost`]. + fn lose_connection(&mut self, batch: &mut Batch, error: &dyn Display, host: &mut Host<'_>) { + if batch.connection_lost { + return; + } + batch.connection_lost = true; + let endpoint = self.endpoint(); + warn!( + %error, + endpoint = endpoint.as_ref().map_or("", Url::as_str), + "Lost the connection to the JP MCP Server; ending the turn without the remaining \ + tool calls." + ); + for index in 0..batch.calls.len() { + let call = &batch.calls[index]; + if matches!(call.wait, Wait::Done) { + continue; + } + let may_have_run = call.released; + self.settle_lost(batch, index, may_have_run, host); + } + } + /// Record a tool's question on the current turn, and return the inquiry id /// its answer is recorded under. /// diff --git a/crates/jp_cli/src/cmd/query/tool/coordinator_tests.rs b/crates/jp_cli/src/cmd/query/tool/coordinator_tests.rs index bf3f0297b..60a1cff51 100644 --- a/crates/jp_cli/src/cmd/query/tool/coordinator_tests.rs +++ b/crates/jp_cli/src/cmd/query/tool/coordinator_tests.rs @@ -746,6 +746,145 @@ async fn a_fanned_out_call_asks_about_each_operation_and_records_once() { ); } +/// A call that finds the Host's connection to the JP MCP Server gone as soon as +/// it is submitted. +struct Disconnected { + tool_id: String, + tool_name: String, +} + +#[async_trait] +impl Executor for Disconnected { + fn tool_id(&self) -> &str { + &self.tool_id + } + + fn tool_name(&self) -> &str { + &self.tool_name + } + + fn arguments(&self) -> Map { + Map::new() + } + + fn permission_info(&self) -> Option { + None + } + + fn set_arguments(&self, _: Value) {} + + async fn prepare(&self, _: bool, _: CancellationToken) -> ExecutorResult { + ExecutorResult::Failed(ExecutorError::ConnectionLost(Box::new(io::Error::other( + "Session expired (HTTP 404)", + )))) + } + + async fn execute( + &self, + _: &IndexMap, + _: CancellationToken, + _: Option, + ) -> ExecutorResult { + panic!("a call that was never admitted must not be released") + } +} + +/// Once the Host's connection to the JP MCP Server is lost, nothing in the +/// batch can run: every call is recorded as not executed, none is put to the +/// user, and the batch reports the lost connection so the turn ends. +/// +/// The second call asks before running, and the prompt backend holds no answer: +/// a prompt shown for it would settle it as a failed prompt instead. +#[tokio::test] +async fn a_lost_connection_settles_every_call_without_asking() { + let mut tools = ask_tools(); + tools.insert( + "lost_tool".to_owned(), + ToolConfig::from_partial( + serde_json::from_value(json!({"source": "builtin", "run": "allow"})).unwrap(), + vec![], + ) + .unwrap(), + ); + let source = TestExecutorSource::new() + .with_executor("lost_tool", |request| { + Box::new(Disconnected { + tool_id: request.id, + tool_name: request.name, + }) + }) + .with_executor("my_tool", |request| { + Box::new( + MockExecutor::completed(&request.id, &request.name, "ran").with_permission_info( + PermissionInfo { + tool_id: request.id.clone(), + state_key: request.id.clone(), + tool_name: request.name.clone(), + tool_source: ToolSource::Builtin { tool: None }, + run_mode: RunMode::Ask, + arguments: Value::Object(Map::new()), + }, + ), + ) + }); + let mut coordinator = ToolCoordinator::new(tools, Box::new(source)); + let (printer, _stdout, stderr) = Printer::memory(OutputFormat::TextPretty); + let printer = Arc::new(printer); + let (_workspace, lock) = test_lock(); + let mut turn_state = TurnState::default(); + + let result = drive( + &mut coordinator, + vec![ + ToolCallRequest { + id: "call_1".into(), + name: "lost_tool".into(), + arguments: Map::new(), + }, + ToolCallRequest { + id: "call_2".into(), + name: "my_tool".into(), + arguments: Map::new(), + }, + ], + Arc::new(MockPromptBackend::new()), + Arc::new(MockInquiryBackend::new(HashMap::new())), + &mut turn_state, + &lock.as_mut(), + &printer, + ) + .await; + + assert_eq!(result.outcome, ExecutionOutcome::ConnectionLost); + assert_eq!( + result + .reviews + .iter() + .map(|(id, review)| (id.as_str(), review.response.result.clone())) + .collect::>(), + vec![ + ( + "call_1", + Err( + "Tool 'lost_tool' was not executed: JP could not complete the call. You may \ + retry it." + .to_owned() + ) + ), + ( + "call_2", + Err( + "Tool 'my_tool' was not executed: JP could not complete the call. You may \ + retry it." + .to_owned() + ) + ), + ] + ); + printer.flush(); + assert_eq!(strip_ansi(&stderr.lock()), ""); +} + /// Tools named `name` whose calls an argument formatter describes, and which /// run without asking. fn described_tools(name: &str) -> ToolsConfig { diff --git a/crates/jp_cli/src/cmd/query/tool/executor_error.rs b/crates/jp_cli/src/cmd/query/tool/executor_error.rs index dc774bce8..dff541777 100644 --- a/crates/jp_cli/src/cmd/query/tool/executor_error.rs +++ b/crates/jp_cli/src/cmd/query/tool/executor_error.rs @@ -2,6 +2,7 @@ use std::error::Error as StdError; +use rmcp::ServiceError as McpCallError; use serde_json::Error as JsonError; use tokio::task::JoinError; @@ -35,6 +36,13 @@ pub(crate) enum ExecutorError { #[error("{0}")] Transport(#[source] Box), + /// The Host's connection to the JP MCP Server failed. + /// + /// Every call the Host submits travels on that one connection, so none + /// submitted after this can complete either. + #[error("Lost the connection to the JP MCP Server: {0}")] + ConnectionLost(#[source] Box), + /// The call stopped before the Host's reply reached the service. /// /// `operation` names the barrier that lapsed: `approval`, `release`, or @@ -78,6 +86,25 @@ pub(crate) enum ExecutorError { } impl ExecutorError { + /// Classify a failed MCP request the Host submitted. + /// + /// A request that could not be sent, or a connection that closed under it, + /// means the connection is gone. + /// Anything else is this request failing. + pub(crate) fn from_mcp_call(error: McpCallError) -> Self { + match error { + McpCallError::TransportSend(_) | McpCallError::TransportClosed => { + Self::ConnectionLost(Box::new(error)) + } + error => Self::Transport(Box::new(error)), + } + } + + /// Whether the Host lost its connection to the JP MCP Server. + pub(crate) fn is_connection_lost(&self) -> bool { + matches!(self, Self::ConnectionLost(_)) + } + /// Whether this belongs in the conversation as the call's outcome. /// /// A user stopping a tool is something that happened to the call, and the diff --git a/crates/jp_cli/src/cmd/query/tool/executor_mock.rs b/crates/jp_cli/src/cmd/query/tool/executor_mock.rs index 7afc7c5f1..e6e8703ec 100644 --- a/crates/jp_cli/src/cmd/query/tool/executor_mock.rs +++ b/crates/jp_cli/src/cmd/query/tool/executor_mock.rs @@ -102,15 +102,24 @@ impl MockExecutor { } fn new(tool_id: &str, tool_name: &str, result: Result) -> Self { + Self::ending_with( + tool_id, + tool_name, + ExecutorResult::Completed(ToolCallResponse { + id: tool_id.to_owned(), + result, + }), + ) + } + + /// Creates a mock executor whose execution ends in `result`. + pub(crate) fn ending_with(tool_id: &str, tool_name: &str, result: ExecutorResult) -> Self { Self { tool_id: tool_id.to_owned(), tool_name: tool_name.to_owned(), arguments: Map::new(), permission_info: None, - result: Mutex::new(Some(ExecutorResult::Completed(ToolCallResponse { - id: tool_id.to_owned(), - result, - }))), + result: Mutex::new(Some(result)), op: None, settlements: None, formatted: None, diff --git a/crates/jp_cli/src/cmd/query/tool/mcp_executor.rs b/crates/jp_cli/src/cmd/query/tool/mcp_executor.rs index 211765c96..821d7c204 100644 --- a/crates/jp_cli/src/cmd/query/tool/mcp_executor.rs +++ b/crates/jp_cli/src/cmd/query/tool/mcp_executor.rs @@ -1107,7 +1107,7 @@ impl PendingCall { result = task => { self.task = None; self.phase = Phase::Finished; - let result = result?.map_err(|error| ExecutorError::Transport(Box::new(error)))?; + let result = result?.map_err(ExecutorError::from_mcp_call)?; Ok(Received::Finished( from_mcp(result).map_err(ExecutorError::MalformedResult)?, )) diff --git a/crates/jp_cli/src/cmd/query/tool/mcp_executor_tests.rs b/crates/jp_cli/src/cmd/query/tool/mcp_executor_tests.rs index 49e3417ad..350cb5c6a 100644 --- a/crates/jp_cli/src/cmd/query/tool/mcp_executor_tests.rs +++ b/crates/jp_cli/src/cmd/query/tool/mcp_executor_tests.rs @@ -960,6 +960,42 @@ async fn a_protocol_failure_is_reported_as_a_failure_not_as_tool_output() { fixture.shutdown().await; } +/// A call on a Host connection the JP MCP Server no longer serves fails as a +/// lost connection, which no later call on it can recover from, rather than as +/// a failure of this one call. +#[tokio::test] +async fn a_call_on_a_lost_connection_reports_the_connection_lost() { + let mut fixture = Fixture::inquiring("allow").await; + // Stopping the endpoint under the Host's connection leaves the Host holding + // a session nothing serves, as an expired session does. + fixture + .owner + .endpoint + .take() + .expect("the fixture started an endpoint") + .shutdown() + .await + .unwrap(); + let executor = fixture.executor(&json!({})); + + let result = timeout( + Duration::from_secs(5), + executor.prepare(false, CancellationToken::new()), + ) + .await + .expect("a call on a lost connection must still end"); + + let ExecutorResult::Failed(error) = result else { + panic!("a call that never reached the server did not run, got {result:?}") + }; + assert!( + matches!(error, ExecutorError::ConnectionLost(_)), + "expected a lost connection, got {error:?}" + ); + assert_eq!(fixture.attempts(), 0); + fixture.shutdown().await; +} + /// A tool that does its work and then never reports back. struct VanishingTool(Arc); @@ -999,9 +1035,13 @@ async fn a_call_lost_after_release_is_not_reported_as_unexecuted() { .await .expect("a lost call must still end"); + let ExecutorResult::OutcomeUnknown(error) = result else { + panic!("a call lost after release may have run, got {result:?}") + }; + // One tool failing leaves the connection the other calls use intact. assert!( - matches!(result, ExecutorResult::OutcomeUnknown(_)), - "a call lost after release may have run, got {result:?}" + !error.is_connection_lost(), + "a failed tool is not a lost connection, got {error:?}" ); assert_eq!(fixture.attempts(), 1); fixture.shutdown().await; diff --git a/crates/jp_cli/src/cmd/query/turn_loop.rs b/crates/jp_cli/src/cmd/query/turn_loop.rs index a3bbdcdcd..b78ffbb6c 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop.rs @@ -1032,6 +1032,22 @@ async fn drive_turn( break; } + // The Host lost its connection to the JP MCP Server, so + // nothing more can run this turn. Persist the responses + // so every tool call keeps a matching response, then end + // the turn with an error instead of asking the assistant + // to retry on a connection that is gone. + ExecutionOutcome::ConnectionLost => { + commit_tool_responses( + execution_result, + tool_coordinator, + &mut turn_coordinator, + &mut conv, + ) + .await?; + return Err(Error::McpConnectionLost); + } + // The next loop iteration re-executes the cancelled // batch. ExecutionOutcome::Restart => { diff --git a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs index dc67b7555..b4ea0e93d 100644 --- a/crates/jp_cli/src/cmd/query/turn_loop_tests.rs +++ b/crates/jp_cli/src/cmd/query/turn_loop_tests.rs @@ -72,7 +72,7 @@ use crate::{ tool::{ ToolCoordinator, executor::{ - Executor, ExecutorResult, ExecutorSource, PermissionInfo, + Executor, ExecutorError, ExecutorResult, ExecutorSource, PermissionInfo, mock::{ MockExecutor, TestExecutorSource, asking_formatter, asking_formatter_runner, no_commands, @@ -2206,6 +2206,111 @@ async fn test_tool_stop_on_interrupt_commits_responses_without_follow_up() { assert!(test_result.is_ok(), "Test timed out after 10 seconds"); } +/// A tool call that loses the Host's connection to the JP MCP Server ends the +/// turn with an error naming that, instead of handing the failure to the +/// assistant as a tool result to retry against a connection that is gone. +/// +/// The call's response is still recorded, so the request stays paired, and no +/// follow-up request is sent. +#[tokio::test] +async fn a_lost_connection_ends_the_turn_with_an_error() { + let tmp = tempdir().unwrap(); + let root = tmp.path(); + let storage = root.join(".jp"); + + let mut config = AppConfig::new_test(); + config.conversation.tools.insert( + "lost_tool".to_string(), + ToolConfig::from_partial( + serde_json::from_value(json!({"source": "builtin", "run": "allow"})).unwrap(), + vec![], + ) + .unwrap(), + ); + + let fs = Arc::new(FsStorageBackend::new(&storage).expect("failed to create backend")); + let mut workspace = Workspace::in_memory(root).with_backend(fs.clone()); + let lock = workspace + .create_and_lock_conversation(Conversation::default(), Arc::new(config.clone()), None) + .unwrap(); + let conv_id = lock.id(); + + let provider = Arc::new(SequentialMockProvider::with_tool_then_message( + "call_lost", + "lost_tool", + "This follow-up should never be requested.", + )); + let model = provider + .model_details(&"test-model".parse().unwrap()) + .await + .unwrap(); + let (printer, _out, _err) = Printer::memory(OutputFormat::TextPretty); + let (router, _signals) = test_router(); + + let executor_source = TestExecutorSource::new().with_executor("lost_tool", |request| { + Box::new(MockExecutor::ending_with( + &request.id, + &request.name, + ExecutorResult::OutcomeUnknown(ExecutorError::ConnectionLost(Box::new( + std::io::Error::other("Session expired (HTTP 404)"), + ))), + )) + }); + + let result = timeout( + Duration::from_secs(10), + run_turn_loop( + Arc::clone(&provider) as Arc, + &model, + &config, + &router, + Utf8Path::new("/tmp"), + InvocationContext::default(), + false, + &[], + &lock, + ToolChoice::Auto, + &[], + Arc::new(printer), + Arc::new(MockPromptBackend::new()), + ToolCoordinator::new(config.conversation.tools.clone(), Box::new(executor_source)), + ChatRequest::from("Please use a tool"), + PendingStreamTrim::default(), + router.turn_interrupt(), + TurnInterrupts::none(), + ), + ) + .await + .expect("the turn ends"); + + assert!( + matches!(result, Err(Error::McpConnectionLost)), + "expected the lost connection to end the turn, got {result:?}" + ); + assert_eq!( + provider.call_index.load(Ordering::SeqCst), + 1, + "a lost connection must not trigger a follow-up request" + ); + assert!( + !router.shutdown_token().is_cancelled(), + "a lost connection ends the turn, not the process" + ); + + // Tool response content is base64-encoded in the raw events file. + let content = fs + .read_test_events_raw(&conv_id) + .expect("events should be persisted"); + let recorded = STANDARD.encode( + "Tool 'lost_tool' may have run, but JP lost the call before its result arrived. Check \ + whether its effects took place before calling it again.", + ); + assert!( + content.contains(&recorded), + "the lost call's response must be recorded.\nFile contents:\n{content}" + ); +} + /// A Ctrl-C pressed while a tool question prompt is active is declined by the /// tool handler and lands on the turn-level handler, which ends the turn /// gracefully once the tool completes: no follow-up request is sent. diff --git a/crates/jp_cli/src/error.rs b/crates/jp_cli/src/error.rs index e5c1b23c6..615d5d5a9 100644 --- a/crates/jp_cli/src/error.rs +++ b/crates/jp_cli/src/error.rs @@ -76,6 +76,11 @@ pub(crate) enum Error { #[error("MCP Host control failed: {0}")] McpHost(#[source] ExecutorError), + /// The Host lost its connection to the JP MCP Server mid-turn, so the turn + /// ended with the calls it could not complete. + #[error("Lost the connection to the JP MCP Server")] + McpConnectionLost, + #[error("LLM error")] Llm(#[from] jp_llm::Error), diff --git a/crates/jp_mcp/src/server/http.rs b/crates/jp_mcp/src/server/http.rs index 85c9a8356..bdf0df9c3 100644 --- a/crates/jp_mcp/src/server/http.rs +++ b/crates/jp_mcp/src/server/http.rs @@ -50,8 +50,7 @@ use super::{ /// A caller waiting on a call it has heard nothing from cannot tell a slow tool /// from a dead one, and clients commonly abandon such a call after a few /// minutes. -/// This sits well inside those limits, and inside the idle window the transport -/// applies to a session carrying no traffic. +/// This sits well inside those limits. const PROGRESS_HEARTBEAT: Duration = Duration::from_secs(30); /// Failure starting, connecting to, or stopping the in-process endpoint. @@ -108,6 +107,16 @@ impl Endpoint { async fn start_reporting_every( service: Service, heartbeat: Duration, + ) -> Result { + Self::start_with(service, heartbeat, session_manager()).await + } + + /// Start the endpoint, choosing how often a running call reports liveness + /// and how sessions are kept. + async fn start_with( + service: Service, + heartbeat: Duration, + sessions: LocalSessionManager, ) -> Result { let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))).await?; let address = listener.local_addr()?; @@ -133,7 +142,7 @@ impl Endpoint { heartbeat, }) }, - Arc::new(LocalSessionManager::default()), + Arc::new(sessions), config, ); let router = Router::new().nest_service("/mcp", transport); @@ -195,6 +204,23 @@ impl Drop for Endpoint { } } +/// How the endpoint keeps its MCP sessions. +/// +/// A session lasts until its client closes it or the endpoint shuts down, with +/// no idle window. +/// Nothing travels on the Host's session while a model is generating, which can +/// outlast any fixed window, and an expired session fails every later call on +/// it. +/// The endpoint lives only as long as its turn, so a session a client abandons +/// without closing is bounded by that. +fn session_manager() -> LocalSessionManager { + let mut sessions = LocalSessionManager::default(); + // Assigned on a default value because rmcp marks the config + // `#[non_exhaustive]`, which rules out struct-update syntax downstream. + sessions.session_config.keep_alive = None; + sessions +} + #[derive(Clone)] struct Handler { service: Arc, diff --git a/crates/jp_mcp/src/server/http_tests.rs b/crates/jp_mcp/src/server/http_tests.rs index 5b3cf3461..373f26ee0 100644 --- a/crates/jp_mcp/src/server/http_tests.rs +++ b/crates/jp_mcp/src/server/http_tests.rs @@ -11,9 +11,9 @@ use jp_config::{ }; use jp_tool::{Outcome, ToolDefinition, ToolDocs, ToolResult}; use reqwest::Client as HttpClient; -use rmcp::model::CallToolRequestParams; +use rmcp::{ServiceError as McpCallError, model::CallToolRequestParams}; use serde_json::{Map, Value, json}; -use tokio::time::{Duration, timeout}; +use tokio::time::{Duration, sleep, timeout}; use super::*; use crate::{ @@ -244,6 +244,51 @@ async fn endpoint_shutdown_cancels_waiting_call_and_closes_listener() { client.cancel().await.unwrap(); } +/// A session given an idle window is gone once the window passes with no +/// traffic, and every request on it fails from then on. +/// +/// This is how a turn lost its tools: a model took longer to answer than the +/// window, and nothing was sent on the session while it did. +#[tokio::test] +async fn a_session_with_an_idle_window_expires_once_it_passes() { + let (service, _host, count) = setup(); + let mut sessions = LocalSessionManager::default(); + sessions.session_config.keep_alive = Some(Duration::from_millis(100)); + let endpoint = Endpoint::start_with(service, PROGRESS_HEARTBEAT, sessions) + .await + .unwrap(); + let client = endpoint.connect().await.unwrap(); + client.peer().list_all_tools().await.unwrap(); + + sleep(Duration::from_millis(500)).await; + + let error = timeout(Duration::from_secs(2), client.peer().list_all_tools()) + .await + .unwrap() + .unwrap_err(); + let McpCallError::TransportSend(error) = error else { + panic!("expected a transport failure, got {error:?}") + }; + assert!( + error.to_string().contains("Session expired (HTTP 404)"), + "expected an expired session, got {error}" + ); + assert_eq!(count.load(Ordering::SeqCst), 0); + drop(client.cancel().await); + endpoint.shutdown().await.unwrap(); +} + +/// The endpoint's sessions have no idle window. +/// +/// A turn can wait on a model for longer than any fixed window, and nothing is +/// sent on the Host's session meanwhile. +/// The endpoint ends with its turn, which is what bounds a session's life +/// instead. +#[test] +fn the_endpoint_never_expires_an_idle_session() { + assert_eq!(session_manager().session_config.keep_alive, None); +} + #[tokio::test] async fn dropping_endpoint_stops_admission_even_if_host_is_still_connected() { let (service, _host, count) = setup(); From 1eb6e3ca8a8743769a57b8eefaa723b43933587a Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Wed, 21 Oct 2026 11:18:22 +0200 Subject: [PATCH 2/4] fixes Signed-off-by: Jean Mertz --- .config/rustqual/baseline-dry.json | 80 ++++++++++++++--------------- .config/rustqual/baseline.json | 82 +++++++++++++++--------------- 2 files changed, 81 insertions(+), 81 deletions(-) diff --git a/.config/rustqual/baseline-dry.json b/.config/rustqual/baseline-dry.json index cc499df38..a6ef15c7e 100644 --- a/.config/rustqual/baseline-dry.json +++ b/.config/rustqual/baseline-dry.json @@ -1,9 +1,9 @@ { "version": 2, - "quality_score": 0.6519302292023524, - "iosp_score": 0.8120950323974082, + "quality_score": 0.6525493395791738, + "iosp_score": 0.8125, "violations": 870, - "total": 6496, + "total": 6511, "complexity_warnings": 0, "magic_number_warnings": 0, "nesting_depth_warnings": 0, @@ -11,7 +11,7 @@ "unsafe_warnings": 0, "error_handling_warnings": 0, "duplicate_groups": 29, - "dead_code_warnings": 262, + "dead_code_warnings": 263, "dead_type_warnings": 28, "fragment_groups": 58, "boilerplate_warnings": 0, @@ -28,7 +28,7 @@ "tq_untested_logic_warnings": 0, "structural_srp_warnings": 0, "structural_coupling_warnings": 0, - "total_findings": 1292, + "total_findings": 1293, "violation_details": [ { "name": "ConversationStream::config_unsets", @@ -723,7 +723,7 @@ { "name": "render_available", "file": "crates/jp_config/src/types/api_key_env.rs", - "line": 63 + "line": 133 }, { "name": "PartialInterruptConfig::assign", @@ -783,22 +783,22 @@ { "name": "PartialGoogleConfig::assign", "file": "crates/jp_config/src/providers/llm/google.rs", - "line": 75 + "line": 83 }, { "name": "PartialOpenrouterConfig::assign", "file": "crates/jp_config/src/providers/llm/openrouter.rs", - "line": 84 + "line": 92 }, { "name": "PartialVllmConfig::assign", "file": "crates/jp_config/src/providers/llm/vllm.rs", - "line": 87 + "line": 95 }, { "name": "PartialCerebrasConfig::assign", "file": "crates/jp_config/src/providers/llm/cerebras.rs", - "line": 99 + "line": 107 }, { "name": "PartialOllamaConfig::assign", @@ -813,17 +813,17 @@ { "name": "PartialDeepseekConfig::assign", "file": "crates/jp_config/src/providers/llm/deepseek.rs", - "line": 75 + "line": 83 }, { "name": "PartialAnthropicConfig::assign", "file": "crates/jp_config/src/providers/llm/anthropic.rs", - "line": 157 + "line": 165 }, { "name": "PartialOpenaiConfig::assign", "file": "crates/jp_config/src/providers/llm/openai.rs", - "line": 122 + "line": 130 }, { "name": "PartialStdioConfig::assign", @@ -2423,12 +2423,12 @@ { "name": "read_setup_token", "file": "crates/jp_cli/src/cmd/provider.rs", - "line": 1041 + "line": 1054 }, { "name": "humanize", "file": "crates/jp_cli/src/cmd/provider.rs", - "line": 1136 + "line": 1149 }, { "name": "Composer::ask", @@ -2763,17 +2763,17 @@ { "name": "build_inquiry_backend", "file": "crates/jp_cli/src/cmd/query/turn_loop.rs", - "line": 1096 + "line": 1112 }, { "name": "build_inquiry_overrides", "file": "crates/jp_cli/src/cmd/query/turn_loop.rs", - "line": 1195 + "line": 1211 }, { "name": "commit_tool_responses", "file": "crates/jp_cli/src/cmd/query/turn_loop.rs", - "line": 1310 + "line": 1326 }, { "name": "build_execution_plan", @@ -2793,112 +2793,112 @@ { "name": "stderr_sink", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 245 + "line": 260 }, { "name": "ToolCoordinator::pre_render_for_prompt", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 672 + "line": 692 }, { "name": "ToolCoordinator::render_executor", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 692 + "line": 712 }, { "name": "ToolCoordinator::abandon", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 790 + "line": 810 }, { "name": "ToolCoordinator::decide_permission", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 885 + "line": 905 }, { "name": "ToolCoordinator::submit", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 974 + "line": 994 }, { "name": "ToolCoordinator::add_call", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1015 + "line": 1035 }, { "name": "ToolCoordinator::finish", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1092 + "line": 1112 }, { "name": "ToolCoordinator::dispatch", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1332 + "line": 1356 }, { "name": "ToolCoordinator::on_step", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1441 + "line": 1482 }, { "name": "ToolCoordinator::on_permission", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1491 + "line": 1546 }, { "name": "ToolCoordinator::pump", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1525 + "line": 1580 }, { "name": "ToolCoordinator::decide", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1583 + "line": 1638 }, { "name": "ToolCoordinator::announce", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1636 + "line": 1691 }, { "name": "ToolCoordinator::release", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1658 + "line": 1713 }, { "name": "ToolCoordinator::route", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1732 + "line": 1787 }, { "name": "ToolCoordinator::answer", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1830 + "line": 1885 }, { "name": "ToolCoordinator::serve", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1844 + "line": 1899 }, { "name": "ToolCoordinator::deliver", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1932 + "line": 1987 }, { "name": "ToolCoordinator::close", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1983 + "line": 2038 }, { "name": "ToolCoordinator::render_result", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2225 + "line": 2282 }, { "name": "ToolCoordinator::preset_answer", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2319 + "line": 2418 }, { "name": "route", diff --git a/.config/rustqual/baseline.json b/.config/rustqual/baseline.json index b5255b458..efdbc0e9b 100644 --- a/.config/rustqual/baseline.json +++ b/.config/rustqual/baseline.json @@ -1,13 +1,13 @@ { "version": 2, - "quality_score": 0.705261627659666, - "iosp_score": 0.9144580696202531, + "quality_score": 0.7060583570618308, + "iosp_score": 0.9147195109928029, "violations": 865, - "total": 13949, + "total": 13994, "complexity_warnings": 163, "magic_number_warnings": 238, "nesting_depth_warnings": 27, - "function_length_warnings": 430, + "function_length_warnings": 432, "unsafe_warnings": 74, "error_handling_warnings": 194, "duplicate_groups": 0, @@ -22,13 +22,13 @@ "coupling_warnings": 0, "coupling_cycles": 0, "tq_no_assertion_warnings": 0, - "tq_no_sut_warnings": 20, + "tq_no_sut_warnings": 19, "tq_untested_warnings": 95, "tq_uncovered_warnings": 0, "tq_untested_logic_warnings": 0, "structural_srp_warnings": 48, "structural_coupling_warnings": 8, - "total_findings": 2816, + "total_findings": 2817, "violation_details": [ { "name": "ConversationStream::config_unsets", @@ -723,7 +723,7 @@ { "name": "render_available", "file": "crates/jp_config/src/types/api_key_env.rs", - "line": 63 + "line": 133 }, { "name": "PartialInterruptConfig::assign", @@ -783,22 +783,22 @@ { "name": "PartialGoogleConfig::assign", "file": "crates/jp_config/src/providers/llm/google.rs", - "line": 75 + "line": 83 }, { "name": "PartialOpenrouterConfig::assign", "file": "crates/jp_config/src/providers/llm/openrouter.rs", - "line": 84 + "line": 92 }, { "name": "PartialVllmConfig::assign", "file": "crates/jp_config/src/providers/llm/vllm.rs", - "line": 87 + "line": 95 }, { "name": "PartialCerebrasConfig::assign", "file": "crates/jp_config/src/providers/llm/cerebras.rs", - "line": 99 + "line": 107 }, { "name": "PartialOllamaConfig::assign", @@ -813,17 +813,17 @@ { "name": "PartialDeepseekConfig::assign", "file": "crates/jp_config/src/providers/llm/deepseek.rs", - "line": 75 + "line": 83 }, { "name": "PartialAnthropicConfig::assign", "file": "crates/jp_config/src/providers/llm/anthropic.rs", - "line": 157 + "line": 165 }, { "name": "PartialOpenaiConfig::assign", "file": "crates/jp_config/src/providers/llm/openai.rs", - "line": 122 + "line": 130 }, { "name": "PartialStdioConfig::assign", @@ -2423,12 +2423,12 @@ { "name": "read_setup_token", "file": "crates/jp_cli/src/cmd/provider.rs", - "line": 1041 + "line": 1054 }, { "name": "humanize", "file": "crates/jp_cli/src/cmd/provider.rs", - "line": 1136 + "line": 1149 }, { "name": "Composer::ask", @@ -2758,17 +2758,17 @@ { "name": "build_inquiry_backend", "file": "crates/jp_cli/src/cmd/query/turn_loop.rs", - "line": 1096 + "line": 1112 }, { "name": "build_inquiry_overrides", "file": "crates/jp_cli/src/cmd/query/turn_loop.rs", - "line": 1195 + "line": 1211 }, { "name": "commit_tool_responses", "file": "crates/jp_cli/src/cmd/query/turn_loop.rs", - "line": 1310 + "line": 1326 }, { "name": "build_execution_plan", @@ -2788,112 +2788,112 @@ { "name": "stderr_sink", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 245 + "line": 260 }, { "name": "ToolCoordinator::pre_render_for_prompt", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 672 + "line": 692 }, { "name": "ToolCoordinator::render_executor", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 692 + "line": 712 }, { "name": "ToolCoordinator::abandon", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 790 + "line": 810 }, { "name": "ToolCoordinator::decide_permission", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 885 + "line": 905 }, { "name": "ToolCoordinator::submit", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 974 + "line": 994 }, { "name": "ToolCoordinator::add_call", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1015 + "line": 1035 }, { "name": "ToolCoordinator::finish", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1092 + "line": 1112 }, { "name": "ToolCoordinator::dispatch", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1332 + "line": 1356 }, { "name": "ToolCoordinator::on_step", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1441 + "line": 1482 }, { "name": "ToolCoordinator::on_permission", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1491 + "line": 1546 }, { "name": "ToolCoordinator::pump", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1525 + "line": 1580 }, { "name": "ToolCoordinator::decide", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1583 + "line": 1638 }, { "name": "ToolCoordinator::announce", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1636 + "line": 1691 }, { "name": "ToolCoordinator::release", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1658 + "line": 1713 }, { "name": "ToolCoordinator::route", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1732 + "line": 1787 }, { "name": "ToolCoordinator::answer", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1830 + "line": 1885 }, { "name": "ToolCoordinator::serve", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1844 + "line": 1899 }, { "name": "ToolCoordinator::deliver", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1932 + "line": 1987 }, { "name": "ToolCoordinator::close", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1983 + "line": 2038 }, { "name": "ToolCoordinator::render_result", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2225 + "line": 2282 }, { "name": "ToolCoordinator::preset_answer", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2319 + "line": 2418 }, { "name": "route", From 39998cf84b6955f7a189b4110bdcccb963929b10 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Sat, 24 Oct 2026 13:59:03 +0200 Subject: [PATCH 3/4] review feedback Signed-off-by: Jean Mertz --- .../jp_cli/src/cmd/query/tool/coordinator.rs | 37 ++++- .../src/cmd/query/tool/coordinator_tests.rs | 152 +++++++++++++++++- ...nator-close-a-prompt-it-no-longer-needs.md | 53 ++++++ 3 files changed, 229 insertions(+), 13 deletions(-) create mode 100644 docs/ticket/17qrhwx-let-the-tool-coordinator-close-a-prompt-it-no-longer-needs.md diff --git a/crates/jp_cli/src/cmd/query/tool/coordinator.rs b/crates/jp_cli/src/cmd/query/tool/coordinator.rs index 32a873296..6bad5f89e 100644 --- a/crates/jp_cli/src/cmd/query/tool/coordinator.rs +++ b/crates/jp_cli/src/cmd/query/tool/coordinator.rs @@ -1108,6 +1108,9 @@ impl ToolCoordinator { /// `interrupts` carries interrupts from a client driving the turn from /// outside the process; one is taken per call, and the rest stay queued. /// + /// A prompt still open once every call is settled is waited for: it owns + /// the terminal until answered, and its answer changes nothing. + /// /// [`submit`]: Self::submit pub(crate) async fn finish( &mut self, @@ -1140,7 +1143,11 @@ impl ToolCoordinator { let mut outcome = ExecutionOutcome::Completed; let mut stop = Stop::default(); - while !batch.settled() { + // A call can be settled while its prompt is open, as when the + // connection is lost or a client stops the calls. The prompt runs on a + // blocking thread nothing can stop, so it is waited for here rather + // than left on screen past the batch it reports to. + while !batch.settled() || batch.prompting { // One client interrupt per phase. Once the calls are being // cancelled, the answer they give back is settled, and a second // reply taken here would replace the first after both were reported @@ -1366,6 +1373,10 @@ impl ToolCoordinator { } ToolEvent::Permission { call, result } => { batch.prompting = false; + // Settled while its prompt was open; the answer comes too late. + if matches!(batch.calls[call].wait, Wait::Done) { + return; + } self.on_permission(batch, call, result, host); } ToolEvent::Answered { @@ -1450,10 +1461,7 @@ impl ToolCoordinator { } ToolEvent::Reviewed { call, review } => { batch.prompting = false; - let tool_name = batch.calls[call].tool_name.clone(); - host.renderer.focus(&batch.calls[call].tool_id); - self.render_result(&tool_name, &review.response, host.renderer); - self.settle(batch, call, review, host); + self.on_reviewed(batch, call, review, host); } ToolEvent::Folded { group, @@ -1463,6 +1471,25 @@ impl ToolCoordinator { } } + /// Take what the user settled on in a call's result review. + fn on_reviewed( + &mut self, + batch: &mut Batch, + index: usize, + review: Review, + host: &mut Host<'_>, + ) { + // Settled while its review was open; the response it was settled with + // stands. + if matches!(batch.calls[index].wait, Wait::Done) { + return; + } + let tool_name = batch.calls[index].tool_name.clone(); + host.renderer.focus(&batch.calls[index].tool_id); + self.render_result(&tool_name, &review.response, host.renderer); + self.settle(batch, index, review, host); + } + /// Take the response a fanned-out call is recorded with. fn on_folded( &mut self, diff --git a/crates/jp_cli/src/cmd/query/tool/coordinator_tests.rs b/crates/jp_cli/src/cmd/query/tool/coordinator_tests.rs index 60a1cff51..c4f8a3066 100644 --- a/crates/jp_cli/src/cmd/query/tool/coordinator_tests.rs +++ b/crates/jp_cli/src/cmd/query/tool/coordinator_tests.rs @@ -746,11 +746,15 @@ async fn a_fanned_out_call_asks_about_each_operation_and_records_once() { ); } -/// A call that finds the Host's connection to the JP MCP Server gone as soon as -/// it is submitted. +/// A call that finds the Host's connection to the JP MCP Server gone when it is +/// submitted. struct Disconnected { tool_id: String, tool_name: String, + + /// What the call waits for before reporting the lost connection, if + /// anything. + after: Option>, } #[async_trait] @@ -774,6 +778,9 @@ impl Executor for Disconnected { fn set_arguments(&self, _: Value) {} async fn prepare(&self, _: bool, _: CancellationToken) -> ExecutorResult { + if let Some(after) = &self.after { + after.notified().await; + } ExecutorResult::Failed(ExecutorError::ConnectionLost(Box::new(io::Error::other( "Session expired (HTTP 404)", )))) @@ -811,6 +818,7 @@ async fn a_lost_connection_settles_every_call_without_asking() { Box::new(Disconnected { tool_id: request.id, tool_name: request.name, + after: None, }) }) .with_executor("my_tool", |request| { @@ -885,6 +893,120 @@ async fn a_lost_connection_settles_every_call_without_asking() { assert_eq!(strip_ansi(&stderr.lock()), ""); } +/// A prompt already open when the connection is lost is waited for before the +/// calls are handed back, and its answer changes nothing. +/// +/// The prompt runs on a thread of its own and owns the terminal until it is +/// answered. +/// Returning while it is open would leave it on screen over the turn's error, +/// with its answer sent to a batch that no longer exists. +#[tokio::test] +async fn a_lost_connection_waits_for_an_open_prompt() { + let mut tools = ask_tools(); + tools.insert( + "lost_tool".to_owned(), + ToolConfig::from_partial( + serde_json::from_value(json!({"source": "builtin", "run": "allow"})).unwrap(), + vec![], + ) + .unwrap(), + ); + let (answer, answers) = std::sync::mpsc::channel(); + let prompts = Arc::new(HeldPromptBackend::new(answers)); + // The connection is lost only once the first call's approval prompt is + // open, so the loss always lands while it is waiting for an answer. + let opened = Arc::clone(&prompts.opened); + let source = TestExecutorSource::new() + .with_executor("my_tool", |request| { + Box::new( + MockExecutor::completed(&request.id, &request.name, "ran").with_permission_info( + PermissionInfo { + tool_id: request.id.clone(), + state_key: request.id.clone(), + tool_name: request.name.clone(), + tool_source: ToolSource::Builtin { tool: None }, + run_mode: RunMode::Ask, + arguments: Value::Object(Map::new()), + }, + ), + ) + }) + .with_executor("lost_tool", move |request| { + Box::new(Disconnected { + tool_id: request.id, + tool_name: request.name, + after: Some(Arc::clone(&opened)), + }) + }); + let mut coordinator = ToolCoordinator::new(tools, Box::new(source)); + let printer = Arc::new(Printer::sink()); + let (_workspace, lock) = test_lock(); + let conv = lock.as_mut(); + let mut turn_state = TurnState::default(); + + let driving = drive( + &mut coordinator, + vec![ + ToolCallRequest { + id: "call_1".into(), + name: "my_tool".into(), + arguments: Map::new(), + }, + ToolCallRequest { + id: "call_2".into(), + name: "lost_tool".into(), + arguments: Map::new(), + }, + ], + prompts.clone(), + Arc::new(MockInquiryBackend::new(HashMap::new())), + &mut turn_state, + &conv, + &printer, + ); + tokio::pin!(driving); + + assert!( + tokio::time::timeout(Duration::from_millis(500), &mut driving) + .await + .is_err(), + "the calls were handed back while the approval prompt was still open" + ); + answer.send('y').unwrap(); + let result = tokio::time::timeout(Duration::from_secs(5), driving) + .await + .expect("answering the prompt lets the calls be handed back"); + + assert_eq!(result.outcome, ExecutionOutcome::ConnectionLost); + assert_eq!( + result + .reviews + .iter() + .map(|(id, review)| (id.as_str(), review.response.result.clone())) + .collect::>(), + vec![ + ( + "call_1", + Err( + "Tool 'my_tool' was not executed: JP could not complete the call. You may \ + retry it." + .to_owned() + ) + ), + ( + "call_2", + Err( + "Tool 'lost_tool' was not executed: JP could not complete the call. You may \ + retry it." + .to_owned() + ) + ), + ], + "approving the call after the connection was lost does not run it" + ); + assert_eq!(prompts.shown.load(std::sync::atomic::Ordering::SeqCst), 1); +} + /// Tools named `name` whose calls an argument formatter describes, and which /// run without asking. fn described_tools(name: &str) -> ToolsConfig { @@ -1305,9 +1427,7 @@ async fn a_remembered_no_stops_a_later_calls_formatter() { drive( &mut coordinator, requests, - Arc::new(HeldPromptBackend { - answers: std::sync::Mutex::new(answers), - }), + Arc::new(HeldPromptBackend::new(answers)), Arc::new(MockInquiryBackend::new(HashMap::new())), &mut TurnState::default(), &lock.as_mut(), @@ -1462,6 +1582,22 @@ async fn an_unanswered_formatter_question_settles_the_call() { /// that fails the test if anything else is asked. struct HeldPromptBackend { answers: std::sync::Mutex>, + + /// Notified each time an inline select opens. + opened: Arc, + + /// How many inline selects have opened. + shown: std::sync::atomic::AtomicUsize, +} + +impl HeldPromptBackend { + fn new(answers: std::sync::mpsc::Receiver) -> Self { + Self { + answers: std::sync::Mutex::new(answers), + opened: Arc::new(Notify::new()), + shown: std::sync::atomic::AtomicUsize::new(0), + } + } } impl PromptBackend for HeldPromptBackend { @@ -1472,6 +1608,8 @@ impl PromptBackend for HeldPromptBackend { _default: Option, _writer: &mut dyn std::io::Write, ) -> Result { + self.shown.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + self.opened.notify_one(); self.answers .lock() .unwrap() @@ -1690,9 +1828,7 @@ async fn a_remembered_no_while_a_later_call_asks( }); let (answer, answers) = std::sync::mpsc::channel(); - let prompts: Arc = Arc::new(HeldPromptBackend { - answers: std::sync::Mutex::new(answers), - }); + let prompts: Arc = Arc::new(HeldPromptBackend::new(answers)); let assistant = Arc::new(SilentAssistant::default()); let inquiry_backend: Arc = assistant.clone(); let (printer, _stdout, stderr) = Printer::memory(OutputFormat::TextPretty); diff --git a/docs/ticket/17qrhwx-let-the-tool-coordinator-close-a-prompt-it-no-longer-needs.md b/docs/ticket/17qrhwx-let-the-tool-coordinator-close-a-prompt-it-no-longer-needs.md new file mode 100644 index 000000000..c9df31089 --- /dev/null +++ b/docs/ticket/17qrhwx-let-the-tool-coordinator-close-a-prompt-it-no-longer-needs.md @@ -0,0 +1,53 @@ +# Let the tool coordinator close a prompt it no longer needs + +- **Status**: Todo +- **Kind**: Feature +- **Authors**: jp +- **Date**: 2026-10-24 +- **Label**: client=cli +- **Label**: package=jp_cli +- **Label**: package=jp_inquire +- **Label**: type=follow-up + +A tool call can be settled while its approval, question, or result-review prompt +is still on screen. +Two paths reach that state: + +- The Host loses its connection to the JP MCP Server while another call's prompt + is open. + Every unsettled call is recorded as lost at once. +- A client driving the turn from outside the process stops or restarts the calls + while a local prompt is open. + `ToolCoordinator::finish` applies client interrupts even while a prompt is + active, and `abandon_parked` settles calls parked in `Wait::Approval` or + `Wait::Review`. + +`finish` waits for an open prompt before it hands the calls back, and an answer +given after its call was settled is ignored (PR #1247). +Nothing is lost, but the user still has to answer a prompt for a call that is +already decided before the turn can end. + +The prompts run in `spawn_blocking` through `ToolPrompter` and `PromptBackend` +(`inline_select`, `inline_reply`, `select`, `password`), and none of them can be +stopped from outside. +Closing a prompt the coordinator no longer needs would take: + +- A cancellation token on each `PromptBackend` method, threaded through + `ToolPrompter` and every test backend. +- Widgets that poll for input and check the token. + `InlineSelect` blocks inside inquire's `CustomType::prompt_with_writer`, which + appears to have no external abort; confirm that against inquire's source + before choosing between replacing the widget and wrapping it. + `inline_reply` is our own widget. +- Terminal cleanup when a prompt is stopped part-way: leave raw mode and erase + the partly drawn prompt. +- A decision for the `$EDITOR` path behind result editing, which can only be + stopped by ending the editor process. + +The wait in `finish` stays either way: a cancelled prompt still reports back +from its blocking task, and the coordinator has to receive that before the batch +goes away. + +Done when a call settled while its prompt is open closes that prompt without +user input, the terminal is left clean, and a test shows `finish` returning +without an answer being sent. From 7032a95ba91225d2417c3a912a1848210b11407c Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Sat, 24 Oct 2026 15:17:50 +0200 Subject: [PATCH 4/4] fixup! review feedback Signed-off-by: Jean Mertz --- .config/rustqual/baseline-dry.json | 45 +++++++++++++++------------- .config/rustqual/baseline.json | 47 +++++++++++++++++------------- 2 files changed, 51 insertions(+), 41 deletions(-) diff --git a/.config/rustqual/baseline-dry.json b/.config/rustqual/baseline-dry.json index a6ef15c7e..40f69566d 100644 --- a/.config/rustqual/baseline-dry.json +++ b/.config/rustqual/baseline-dry.json @@ -1,9 +1,9 @@ { "version": 2, - "quality_score": 0.6525493395791738, - "iosp_score": 0.8125, - "violations": 870, - "total": 6511, + "quality_score": 0.6522887459652152, + "iosp_score": 0.8123249299719888, + "violations": 871, + "total": 6512, "complexity_warnings": 0, "magic_number_warnings": 0, "nesting_depth_warnings": 0, @@ -28,7 +28,7 @@ "tq_untested_logic_warnings": 0, "structural_srp_warnings": 0, "structural_coupling_warnings": 0, - "total_findings": 1293, + "total_findings": 1294, "violation_details": [ { "name": "ConversationStream::config_unsets", @@ -2828,77 +2828,82 @@ { "name": "ToolCoordinator::finish", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1112 + "line": 1115 }, { "name": "ToolCoordinator::dispatch", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1356 + "line": 1363 + }, + { + "name": "ToolCoordinator::on_reviewed", + "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", + "line": 1475 }, { "name": "ToolCoordinator::on_step", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1482 + "line": 1509 }, { "name": "ToolCoordinator::on_permission", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1546 + "line": 1573 }, { "name": "ToolCoordinator::pump", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1580 + "line": 1607 }, { "name": "ToolCoordinator::decide", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1638 + "line": 1665 }, { "name": "ToolCoordinator::announce", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1691 + "line": 1718 }, { "name": "ToolCoordinator::release", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1713 + "line": 1740 }, { "name": "ToolCoordinator::route", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1787 + "line": 1814 }, { "name": "ToolCoordinator::answer", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1885 + "line": 1912 }, { "name": "ToolCoordinator::serve", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1899 + "line": 1926 }, { "name": "ToolCoordinator::deliver", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1987 + "line": 2014 }, { "name": "ToolCoordinator::close", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2038 + "line": 2065 }, { "name": "ToolCoordinator::render_result", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2282 + "line": 2309 }, { "name": "ToolCoordinator::preset_answer", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2418 + "line": 2445 }, { "name": "route", diff --git a/.config/rustqual/baseline.json b/.config/rustqual/baseline.json index efdbc0e9b..67757232b 100644 --- a/.config/rustqual/baseline.json +++ b/.config/rustqual/baseline.json @@ -1,13 +1,13 @@ { "version": 2, - "quality_score": 0.7060583570618308, - "iosp_score": 0.9147195109928029, - "violations": 865, - "total": 13994, + "quality_score": 0.7058772822297027, + "iosp_score": 0.9146377525874815, + "violations": 866, + "total": 13997, "complexity_warnings": 163, "magic_number_warnings": 238, "nesting_depth_warnings": 27, - "function_length_warnings": 432, + "function_length_warnings": 433, "unsafe_warnings": 74, "error_handling_warnings": 194, "duplicate_groups": 0, @@ -28,7 +28,7 @@ "tq_untested_logic_warnings": 0, "structural_srp_warnings": 48, "structural_coupling_warnings": 8, - "total_findings": 2817, + "total_findings": 2819, "violation_details": [ { "name": "ConversationStream::config_unsets", @@ -2823,77 +2823,82 @@ { "name": "ToolCoordinator::finish", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1112 + "line": 1115 }, { "name": "ToolCoordinator::dispatch", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1356 + "line": 1363 + }, + { + "name": "ToolCoordinator::on_reviewed", + "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", + "line": 1475 }, { "name": "ToolCoordinator::on_step", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1482 + "line": 1509 }, { "name": "ToolCoordinator::on_permission", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1546 + "line": 1573 }, { "name": "ToolCoordinator::pump", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1580 + "line": 1607 }, { "name": "ToolCoordinator::decide", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1638 + "line": 1665 }, { "name": "ToolCoordinator::announce", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1691 + "line": 1718 }, { "name": "ToolCoordinator::release", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1713 + "line": 1740 }, { "name": "ToolCoordinator::route", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1787 + "line": 1814 }, { "name": "ToolCoordinator::answer", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1885 + "line": 1912 }, { "name": "ToolCoordinator::serve", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1899 + "line": 1926 }, { "name": "ToolCoordinator::deliver", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 1987 + "line": 2014 }, { "name": "ToolCoordinator::close", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2038 + "line": 2065 }, { "name": "ToolCoordinator::render_result", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2282 + "line": 2309 }, { "name": "ToolCoordinator::preset_answer", "file": "crates/jp_cli/src/cmd/query/tool/coordinator.rs", - "line": 2418 + "line": 2445 }, { "name": "route",