merge(upstream): integrate upstream/main with 'ours' strategy; adopt MCP unbounded channels, lock-scope fixes, and CI windows target; keep our TUI hooks intact

This commit is contained in:
James Peter
2025-08-30 06:35:54 +10:00
5 changed files with 26 additions and 22 deletions
+5
View File
@@ -102,12 +102,17 @@ jobs:
# Also run representative release builds on Mac and Linux because
# there could be release-only build errors we want to catch.
# Hopefully this also pre-populates the build cache to speed up
# releases.
- runner: macos-14
target: aarch64-apple-darwin
profile: release
- runner: ubuntu-24.04
target: x86_64-unknown-linux-musl
profile: release
- runner: windows-latest
target: x86_64-pc-windows-msvc
profile: release
steps:
- uses: actions/checkout@v5
+11 -6
View File
@@ -129,10 +129,7 @@ impl McpClient {
error!("failed to write newline to child stdin");
break;
}
if stdin.flush().await.is_err() {
error!("failed to flush child stdin");
break;
}
// No explicit flush needed on a pipe; write_all is sufficient.
}
Err(e) => error!("failed to serialize JSONRPCMessage: {e}"),
}
@@ -366,7 +363,11 @@ impl McpClient {
}
};
if let Some(tx) = pending.lock().await.remove(&id) {
let tx_opt = {
let mut guard = pending.lock().await;
guard.remove(&id)
};
if let Some(tx) = tx_opt {
// Ignore send errors – the receiver might have been dropped.
let _ = tx.send(JSONRPCMessage::Response(resp));
} else {
@@ -384,7 +385,11 @@ impl McpClient {
RequestId::String(_) => return, // see comment above
};
if let Some(tx) = pending.lock().await.remove(&id) {
let tx_opt = {
let mut guard = pending.lock().await;
guard.remove(&id)
};
if let Some(tx) = tx_opt {
let _ = tx.send(JSONRPCMessage::Error(err));
}
}
+1 -5
View File
@@ -59,7 +59,7 @@ pub async fn run_main(
// Set up channels.
let (incoming_tx, mut incoming_rx) = mpsc::channel::<JSONRPCMessage>(CHANNEL_CAPACITY);
let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<OutgoingMessage>(CHANNEL_CAPACITY);
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
// Task: read from stdin, push to `incoming_tx`.
let stdin_reader_handle = tokio::spawn({
@@ -134,10 +134,6 @@ pub async fn run_main(
error!("Failed to write newline to stdout: {e}");
break;
}
if let Err(e) = stdout.flush().await {
error!("Failed to flush stdout: {e}");
break;
}
}
Err(e) => error!("Failed to serialize JSONRPCMessage: {e}"),
}
+9 -9
View File
@@ -24,12 +24,12 @@ use crate::error_code::INTERNAL_ERROR_CODE;
/// Sends messages to the client and manages request callbacks.
pub(crate) struct OutgoingMessageSender {
next_request_id: AtomicI64,
sender: mpsc::Sender<OutgoingMessage>,
sender: mpsc::UnboundedSender<OutgoingMessage>,
request_id_to_callback: Mutex<HashMap<RequestId, oneshot::Sender<Result>>>,
}
impl OutgoingMessageSender {
pub(crate) fn new(sender: mpsc::Sender<OutgoingMessage>) -> Self {
pub(crate) fn new(sender: mpsc::UnboundedSender<OutgoingMessage>) -> Self {
Self {
next_request_id: AtomicI64::new(0),
sender,
@@ -55,7 +55,7 @@ impl OutgoingMessageSender {
method: method.to_string(),
params,
});
let _ = self.sender.send(outgoing_message).await;
let _ = self.sender.send(outgoing_message);
rx_approve
}
@@ -81,7 +81,7 @@ impl OutgoingMessageSender {
match serde_json::to_value(response) {
Ok(result) => {
let outgoing_message = OutgoingMessage::Response(OutgoingResponse { id, result });
let _ = self.sender.send(outgoing_message).await;
let _ = self.sender.send(outgoing_message);
}
Err(err) => {
self.send_error(
@@ -131,17 +131,17 @@ impl OutgoingMessageSender {
};
let outgoing_message =
OutgoingMessage::Notification(OutgoingNotification { method, params });
let _ = self.sender.send(outgoing_message).await;
let _ = self.sender.send(outgoing_message);
}
pub(crate) async fn send_notification(&self, notification: OutgoingNotification) {
let outgoing_message = OutgoingMessage::Notification(notification);
let _ = self.sender.send(outgoing_message).await;
let _ = self.sender.send(outgoing_message);
}
pub(crate) async fn send_error(&self, id: RequestId, error: JSONRPCErrorError) {
let outgoing_message = OutgoingMessage::Error(OutgoingError { id, error });
let _ = self.sender.send(outgoing_message).await;
let _ = self.sender.send(outgoing_message);
}
}
@@ -251,7 +251,7 @@ mod tests {
#[tokio::test]
async fn test_send_event_as_notification() {
let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<OutgoingMessage>(2);
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let event = Event {
@@ -282,7 +282,7 @@ mod tests {
#[tokio::test]
async fn test_send_event_as_notification_with_meta() {
let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<OutgoingMessage>(2);
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let session_configured_event = SessionConfiguredEvent {
-2
View File
@@ -73,8 +73,6 @@ mod chatwidget_stream_tests;
#[cfg(not(debug_assertions))]
mod updates;
#[cfg(not(debug_assertions))]
use color_eyre::owo_colors::OwoColorize;
pub use cli::Cli;