use std::fmt;
use std::io::ErrorKind;
use std::path::Path;
use std::time::Duration;
use diffler_core::feedback;
use diffler_core::model::{DiffModel, FileStatus};
use diffler_core::session::{Comment, CommentStatus};
use diffler_core::source::ReviewSource;
use rmcp::handler::server::router::tool::ToolRouter;
use rmcp::handler::server::wrapper::Parameters;
use rmcp::model::{ServerCapabilities, ServerInfo};
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};
use rmcp::{ErrorData, Json, ServerHandler, schemars, tool, tool_handler, tool_router};
use serde::Serialize;
use tokio::sync::mpsc::UnboundedSender;
use tokio::sync::{oneshot, watch};
use tokio::task::JoinHandle;
use crate::event::AppEvent;
pub const AGENT_AUTHOR: &str = "agent";
const DEFAULT_WAIT_SECONDS: u64 = 300;
const MAX_WAIT_SECONDS: u64 = 540;
pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
pub struct McpRequest {
pub kind: McpRequestKind,
pub reply: oneshot::Sender<McpResponse>,
}
impl fmt::Debug for McpRequest {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("McpRequest")
.field("kind", &self.kind)
.finish_non_exhaustive()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum McpRequestKind {
ReviewStatus,
GetDiff {
file: Option<String>,
},
GetComments {
status: Option<CommentStatus>,
},
ListReviews,
ReplyComment {
id: String,
body: String,
},
ProposeResolve {
id: String,
note: Option<String>,
},
MarkViewed {
file: String,
},
Feedback,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum McpResponse {
Status(ReviewStatusResponse),
Diff(String),
Comments(Vec<CommentInfo>),
Reviews(Vec<ReviewSummary>),
Replied {
status: String,
},
Ok,
Error(String),
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct ReviewStatusResponse {
pub repo: String,
pub branch: Option<String>,
pub oid7: String,
pub files_changed: Vec<FileEntry>,
pub open_comments: usize,
pub replied_comments: usize,
pub resolved_comments: usize,
pub feedback_epoch: u64,
pub reviews: Vec<ReviewSummary>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct ReviewSummary {
pub source: String,
pub label: String,
pub open_comments: usize,
pub replied_comments: usize,
pub resolved_comments: usize,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct FileEntry {
pub path: String,
pub status: String,
pub viewed: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct DiffResponse {
pub diff: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct CommentInfo {
pub id: String,
pub source: String,
pub source_label: String,
pub file: String,
pub line: Option<u32>,
pub line_end: Option<u32>,
pub side: String,
pub body: String,
pub status: String,
pub author: String,
pub replies: Vec<ReplyInfo>,
pub outdated: bool,
pub context: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct ReplyInfo {
pub author: String,
pub body: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct CommentsResponse {
pub comments: Vec<CommentInfo>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct ReviewsResponse {
pub reviews: Vec<ReviewSummary>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct ReplyResponse {
pub ok: bool,
pub status: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct OkResponse {
pub ok: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct WaitForFeedbackResponse {
pub epoch: u64,
pub timed_out: bool,
pub comments: Vec<CommentInfo>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct GetDiffParams {
pub file: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct GetCommentsParams {
pub status: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct ReplyCommentParams {
pub id: String,
pub body: String,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct ProposeResolveParams {
pub id: String,
pub note: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct MarkViewedParams {
pub file: String,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct WaitForFeedbackParams {
pub since_epoch: Option<u64>,
pub timeout_seconds: Option<u64>,
}
pub const fn comment_status_name(status: CommentStatus) -> &'static str {
match status {
CommentStatus::Open => "open",
CommentStatus::Replied => "replied",
CommentStatus::Resolved => "resolved",
}
}
pub const fn file_status_name(status: FileStatus) -> &'static str {
match status {
FileStatus::Added => "added",
FileStatus::Modified => "modified",
FileStatus::Deleted => "deleted",
FileStatus::Renamed => "renamed",
FileStatus::Untracked => "untracked",
}
}
pub fn render_unified(model: &DiffModel, file: Option<&str>) -> Result<String, String> {
use std::fmt::Write as _;
if let Some(path) = file
&& !model.files.iter().any(|f| f.path == path)
{
return Err(format!("unknown file: {path}"));
}
let mut out = String::new();
for f in &model.files {
if file.is_some_and(|path| path != f.path) {
continue;
}
let old = match f.status {
FileStatus::Added | FileStatus::Untracked => "/dev/null".to_owned(),
_ => format!("a/{}", f.old_path.as_deref().unwrap_or(&f.path)),
};
let new = match f.status {
FileStatus::Deleted => "/dev/null".to_owned(),
_ => format!("b/{}", f.path),
};
let _ = writeln!(out, "--- {old}");
let _ = writeln!(out, "+++ {new}");
if f.binary {
out.push_str("Binary files differ\n");
continue;
}
for hunk in &f.hunks {
let _ = writeln!(out, "{}", hunk.header());
for line in &hunk.lines {
let _ = writeln!(out, "{}{}", line.kind.origin(), line.text);
}
}
}
Ok(out)
}
pub fn comment_info(comment: &Comment, model: &DiffModel, source: &ReviewSource) -> CommentInfo {
let anchor = &comment.anchor;
let context = anchor
.line
.and_then(|line| feedback::context_snippet(model, &anchor.file, line, anchor.on_old_side))
.map(|snippet| {
snippet
.iter()
.map(|(origin, text)| format!("{origin}{text}"))
.collect::<Vec<_>>()
.join("\n")
});
let outdated = anchor.is_outdated(model);
CommentInfo {
id: comment.id.clone(),
source: source.key(),
source_label: source.label(),
file: anchor.file.clone(),
line: anchor.line,
line_end: anchor.line_end,
side: if anchor.on_old_side { "old" } else { "new" }.to_owned(),
body: comment.body.clone(),
status: comment_status_name(comment.status).to_owned(),
author: comment.author.clone(),
replies: comment
.replies
.iter()
.map(|r| ReplyInfo {
author: r.author.clone(),
body: r.body.clone(),
})
.collect(),
outdated,
context,
}
}
#[derive(Clone)]
pub struct DifflerMcp {
tx: UnboundedSender<AppEvent>,
feedback_rx: watch::Receiver<u64>,
tool_router: ToolRouter<Self>,
}
fn mismatch() -> ErrorData {
ErrorData::internal_error("unexpected response from the diffler app", None)
}
fn parse_status(status: &str) -> Result<CommentStatus, ErrorData> {
match status {
"open" => Ok(CommentStatus::Open),
"replied" => Ok(CommentStatus::Replied),
"resolved" => Ok(CommentStatus::Resolved),
other => Err(ErrorData::invalid_params(
format!("unknown status {other:?} (expected open, replied, or resolved)"),
None,
)),
}
}
impl DifflerMcp {
pub fn new(tx: UnboundedSender<AppEvent>, feedback_rx: watch::Receiver<u64>) -> Self {
Self {
tx,
feedback_rx,
tool_router: Self::tool_router(),
}
}
async fn request(&self, kind: McpRequestKind) -> Result<McpResponse, ErrorData> {
self.request_with_timeout(kind, REQUEST_TIMEOUT).await
}
async fn request_with_timeout(
&self,
kind: McpRequestKind,
timeout: Duration,
) -> Result<McpResponse, ErrorData> {
let (reply, response) = oneshot::channel();
self.tx
.send(AppEvent::Mcp(McpRequest { kind, reply }))
.map_err(|_| ErrorData::internal_error("the diffler TUI is not running", None))?;
match tokio::time::timeout(timeout, response).await {
Ok(Ok(McpResponse::Error(message))) => Err(ErrorData::invalid_params(message, None)),
Ok(Ok(response)) => Ok(response),
Ok(Err(_)) => Err(ErrorData::internal_error(
"the diffler TUI dropped the request",
None,
)),
Err(_) => Err(ErrorData::internal_error(
"diffler is busy (editor open or loop stalled) — retry",
None,
)),
}
}
}
#[tool_router]
impl DifflerMcp {
#[tool(
description = "Current review state: repo, branch, changed files with viewed marks, comment counts, and the feedback epoch for wait_for_feedback."
)]
async fn review_status(&self) -> Result<Json<ReviewStatusResponse>, ErrorData> {
match self.request(McpRequestKind::ReviewStatus).await? {
McpResponse::Status(status) => Ok(Json(status)),
_ => Err(mismatch()),
}
}
#[tool(
description = "Unified diff text of the working tree under review, optionally restricted to one file."
)]
async fn get_diff(
&self,
Parameters(params): Parameters<GetDiffParams>,
) -> Result<Json<DiffResponse>, ErrorData> {
let kind = McpRequestKind::GetDiff { file: params.file };
match self.request(kind).await? {
McpResponse::Diff(diff) => Ok(Json(DiffResponse { diff })),
_ => Err(mismatch()),
}
}
#[tool(
description = "Review comments across every review (working tree, commits, ranges), each tagged with its source and source_label; anchors, diff context, and threads included. Optionally filtered by status (open, replied, resolved)."
)]
async fn get_comments(
&self,
Parameters(params): Parameters<GetCommentsParams>,
) -> Result<Json<CommentsResponse>, ErrorData> {
let status = params.status.as_deref().map(parse_status).transpose()?;
match self.request(McpRequestKind::GetComments { status }).await? {
McpResponse::Comments(comments) => Ok(Json(CommentsResponse { comments })),
_ => Err(mismatch()),
}
}
#[tool(
description = "List every review the human has — the working tree, individual commits, and commit ranges — each with its comment counts, so you can tell where feedback came from."
)]
async fn list_reviews(&self) -> Result<Json<ReviewsResponse>, ErrorData> {
match self.request(McpRequestKind::ListReviews).await? {
McpResponse::Reviews(reviews) => Ok(Json(ReviewsResponse { reviews })),
_ => Err(mismatch()),
}
}
#[tool(description = "Answer a review comment in place; the human sees the reply immediately.")]
async fn reply_comment(
&self,
Parameters(params): Parameters<ReplyCommentParams>,
) -> Result<Json<ReplyResponse>, ErrorData> {
let kind = McpRequestKind::ReplyComment {
id: params.id,
body: params.body,
};
match self.request(kind).await? {
McpResponse::Replied { status } => Ok(Json(ReplyResponse { ok: true, status })),
_ => Err(mismatch()),
}
}
#[tool(
description = "Propose a comment as resolved: marks it replied with an agent note. Only the human can resolve it, in the TUI."
)]
async fn propose_resolve(
&self,
Parameters(params): Parameters<ProposeResolveParams>,
) -> Result<Json<ReplyResponse>, ErrorData> {
let kind = McpRequestKind::ProposeResolve {
id: params.id,
note: params.note,
};
match self.request(kind).await? {
McpResponse::Replied { status } => Ok(Json(ReplyResponse { ok: true, status })),
_ => Err(mismatch()),
}
}
#[tool(
description = "Mark a file as viewed in the review the human is currently looking at (working tree, or the open commit/range diff)."
)]
async fn mark_viewed(
&self,
Parameters(params): Parameters<MarkViewedParams>,
) -> Result<Json<OkResponse>, ErrorData> {
let kind = McpRequestKind::MarkViewed { file: params.file };
match self.request(kind).await? {
McpResponse::Ok => Ok(Json(OkResponse { ok: true })),
_ => Err(mismatch()),
}
}
#[tool(
description = "Long-poll until the human sends feedback (comments, replies, or the send key). Returns the new epoch and all open/replied comments, or timed_out."
)]
async fn wait_for_feedback(
&self,
Parameters(params): Parameters<WaitForFeedbackParams>,
) -> Result<Json<WaitForFeedbackResponse>, ErrorData> {
let mut rx = self.feedback_rx.clone();
let since = params.since_epoch.unwrap_or_else(|| *rx.borrow());
let timeout = Duration::from_secs(
params
.timeout_seconds
.unwrap_or(DEFAULT_WAIT_SECONDS)
.min(MAX_WAIT_SECONDS),
);
let waited = tokio::time::timeout(timeout, rx.wait_for(|epoch| *epoch > since))
.await
.map(|result| result.map(|epoch| *epoch));
let epoch = match waited {
Ok(Ok(epoch)) => epoch,
Ok(Err(_)) => {
return Err(ErrorData::internal_error(
"the diffler TUI is not running",
None,
));
}
Err(_) => {
let epoch = *rx.borrow();
return Ok(Json(WaitForFeedbackResponse {
epoch,
timed_out: true,
comments: Vec::new(),
}));
}
};
match self.request(McpRequestKind::Feedback).await? {
McpResponse::Comments(comments) => Ok(Json(WaitForFeedbackResponse {
epoch,
timed_out: false,
comments,
})),
_ => Err(mismatch()),
}
}
}
#[tool_handler(router = self.tool_router)]
impl ServerHandler for DifflerMcp {
fn get_info(&self) -> ServerInfo {
ServerInfo::new(ServerCapabilities::builder().enable_tools().build()).with_instructions(
"diffler code review: get_comments reads the human's diff comments, \
reply_comment answers them in place, propose_resolve flags them as \
addressed, and wait_for_feedback long-polls until the human sends \
new feedback.",
)
}
}
pub struct McpHandle {
pub port: u16,
pub handle: JoinHandle<()>,
}
pub async fn spawn_mcp(
tx: UnboundedSender<AppEvent>,
feedback_rx: watch::Receiver<u64>,
port: u16,
) -> std::io::Result<McpHandle> {
let listener = match tokio::net::TcpListener::bind(("127.0.0.1", port)).await {
Ok(listener) => listener,
Err(e) if e.kind() == ErrorKind::AddrInUse => {
tokio::net::TcpListener::bind(("127.0.0.1", 0)).await?
}
Err(e) => return Err(e),
};
let port = listener.local_addr()?.port();
let service: StreamableHttpService<DifflerMcp, LocalSessionManager> =
StreamableHttpService::new(
move || Ok(DifflerMcp::new(tx.clone(), feedback_rx.clone())),
std::sync::Arc::default(),
StreamableHttpServerConfig::default(),
);
let router = axum::Router::new().nest_service("/mcp", service);
let handle = tokio::spawn(async move {
let _ = axum::serve(listener, router).await;
});
Ok(McpHandle { port, handle })
}
pub fn connect_hint(port: u16) -> String {
format!("mcp :{port} — claude mcp add --transport http diffler http://127.0.0.1:{port}/mcp")
}
const ENDPOINT_FILE: &str = "mcp.json";
fn endpoint_path(repo_root: &Path) -> std::path::PathBuf {
repo_root.join(".diffler").join(ENDPOINT_FILE)
}
pub fn write_endpoint(repo_root: &Path, port: u16) -> std::io::Result<()> {
let dir = repo_root.join(".diffler");
std::fs::create_dir_all(&dir)?;
let gitignore = dir.join(".gitignore");
if !gitignore.exists() {
std::fs::write(&gitignore, "*\n")?;
}
let body = format!("{{\n \"port\": {port},\n \"url\": \"http://127.0.0.1:{port}/mcp\"\n}}\n");
std::fs::write(endpoint_path(repo_root), body)
}
pub fn clear_endpoint(repo_root: &Path) {
let _ = std::fs::remove_file(endpoint_path(repo_root));
}
#[cfg(test)]
mod tests {
use diffler_core::model::{DiffLine, FileDiff, Hunk, HunkId, LineKind};
use diffler_core::session::{Anchor, Session};
use super::*;
fn diff_line(kind: LineKind, old_no: Option<u32>, new_no: Option<u32>, text: &str) -> DiffLine {
DiffLine::new(kind, old_no, new_no, text.to_owned())
}
fn sample_model() -> DiffModel {
DiffModel {
files: vec![
FileDiff {
path: "src/auth.py".into(),
old_path: None,
status: FileStatus::Modified,
binary: false,
old_text: None,
new_text: Some("one\nTWO\nthree\n".into()),
hunks: vec![Hunk {
id: HunkId("h1".into()),
old_start: 1,
old_lines: 3,
new_start: 1,
new_lines: 3,
context: String::new(),
lines: vec![
diff_line(LineKind::Context, Some(1), Some(1), "one"),
diff_line(LineKind::Deleted, Some(2), None, "two"),
diff_line(LineKind::Added, None, Some(2), "TWO"),
diff_line(LineKind::Context, Some(3), Some(3), "three"),
],
}],
},
FileDiff {
path: "logo.png".into(),
old_path: None,
status: FileStatus::Added,
binary: true,
old_text: None,
new_text: None,
hunks: vec![],
},
],
}
}
fn anchor(file: &str, line: Option<u32>) -> Anchor {
Anchor {
file: file.to_owned(),
line,
line_end: None,
on_old_side: false,
hunk: None,
line_text: None,
}
}
#[test]
fn render_unified_emits_headers_hunks_and_origins() {
let text = render_unified(&sample_model(), None).unwrap();
assert!(text.contains("--- a/src/auth.py\n+++ b/src/auth.py\n"));
assert!(text.contains("@@ -1,3 +1,3 @@\n one\n-two\n+TWO\n three\n"));
assert!(text.contains("--- /dev/null\n+++ b/logo.png\nBinary files differ\n"));
}
#[test]
fn render_unified_filters_to_one_file() {
let text = render_unified(&sample_model(), Some("src/auth.py")).unwrap();
assert!(text.contains("src/auth.py"));
assert!(!text.contains("logo.png"));
}
#[test]
fn render_unified_unknown_file_is_an_error() {
let err = render_unified(&sample_model(), Some("nope.rs")).unwrap_err();
assert!(err.contains("nope.rs"));
}
#[test]
fn render_unified_marks_deleted_files_with_dev_null() {
let mut model = sample_model();
model.files[0].status = FileStatus::Deleted;
let text = render_unified(&model, Some("src/auth.py")).unwrap();
assert!(text.contains("--- a/src/auth.py\n+++ /dev/null\n"));
}
#[test]
fn render_unified_renamed_file_pairs_old_and_new_paths() {
let mut model = sample_model();
model.files[0].status = FileStatus::Renamed;
model.files[0].old_path = Some("src/auth_v1.py".to_owned());
let text = render_unified(&model, Some("src/auth.py")).unwrap();
assert!(
text.contains("--- a/src/auth_v1.py\n+++ b/src/auth.py\n"),
"rename header must pair both paths: {text}"
);
}
#[test]
fn render_unified_empty_model_is_an_empty_diff() {
let model = DiffModel { files: vec![] };
assert_eq!(render_unified(&model, None).unwrap(), "");
}
#[test]
fn comment_info_old_side_anchor_reports_side_and_context() {
let mut session = Session::default();
let mut a = anchor("src/auth.py", Some(2));
a.on_old_side = true;
a.line_text = Some("two".to_owned());
session.add_comment("human", a, "what was wrong with two?");
let info = comment_info(
&session.comments[0],
&sample_model(),
&ReviewSource::WorkingTree,
);
assert_eq!(info.side, "old");
assert_eq!(info.line, Some(2));
assert_eq!(info.line_end, None);
assert!(!info.outdated, "the deleted line is still in the diff");
assert_eq!(info.context.as_deref(), Some(" one\n-two\n+TWO"));
}
#[test]
fn comment_info_carries_context_and_thread() {
let mut session = Session::default();
let mut a = anchor("src/auth.py", Some(2));
a.line_text = Some("TWO".to_owned());
let id = session.add_comment("human", a, "why uppercase?").id.clone();
session.reply(&id, AGENT_AUTHOR, "legacy API");
let info = comment_info(
&session.comments[0],
&sample_model(),
&ReviewSource::WorkingTree,
);
assert_eq!(info.file, "src/auth.py");
assert_eq!(info.line, Some(2));
assert_eq!(info.side, "new");
assert_eq!(info.status, "replied");
assert_eq!(info.context.as_deref(), Some("-two\n+TWO\n three"));
assert!(!info.outdated);
assert_eq!(info.replies.len(), 1);
assert_eq!(info.replies[0].author, AGENT_AUTHOR);
}
#[test]
fn comment_info_flags_departed_lines_outdated() {
let mut session = Session::default();
session.add_comment("human", anchor("src/auth.py", Some(99)), "moved on");
let info = comment_info(
&session.comments[0],
&sample_model(),
&ReviewSource::WorkingTree,
);
assert!(info.outdated);
assert_eq!(info.context, None);
}
#[test]
fn comment_info_flags_drifted_line_text_outdated() {
let mut session = Session::default();
let mut a = anchor("src/auth.py", Some(2));
a.line_text = Some("old text".to_owned());
session.add_comment("human", a, "stale");
let info = comment_info(
&session.comments[0],
&sample_model(),
&ReviewSource::WorkingTree,
);
assert!(info.outdated);
assert!(info.context.is_some(), "context still renders");
}
#[test]
fn file_level_comment_outdated_only_when_file_left_the_diff() {
let mut session = Session::default();
session.add_comment("human", anchor("src/auth.py", None), "overall");
session.add_comment("human", anchor("gone.py", None), "gone");
let model = sample_model();
assert!(!comment_info(&session.comments[0], &model, &ReviewSource::WorkingTree).outdated);
assert!(comment_info(&session.comments[1], &model, &ReviewSource::WorkingTree).outdated);
}
#[test]
fn range_comment_not_outdated_when_end_line_matches() {
let mut session = Session::default();
let mut a = Anchor {
file: "src/auth.py".to_owned(),
line: Some(1),
line_end: Some(3),
on_old_side: false,
hunk: None,
line_text: Some("three".to_owned()),
};
assert_ne!(a.line_text.as_deref(), Some("one"));
session.add_comment("human", a.clone(), "range comment");
let info = comment_info(
&session.comments[0],
&sample_model(),
&ReviewSource::WorkingTree,
);
assert!(
!info.outdated,
"end line text matches snapshot — must NOT be outdated"
);
a.line_text = Some("changed".to_owned());
session.comments[0].anchor = a;
let info2 = comment_info(
&session.comments[0],
&sample_model(),
&ReviewSource::WorkingTree,
);
assert!(info2.outdated, "end line text drifted — must be outdated");
}
#[tokio::test]
async fn request_times_out_when_app_does_not_answer() {
let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
let (_feedback_tx, feedback_rx) = tokio::sync::watch::channel(0u64);
let handler = DifflerMcp::new(tx, feedback_rx);
let err = handler
.request_with_timeout(McpRequestKind::ReviewStatus, Duration::from_millis(50))
.await
.unwrap_err();
assert!(
err.message.contains("busy"),
"error should mention busy: {err:?}"
);
}
#[tokio::test(start_paused = true)]
async fn wait_for_feedback_clamps_the_timeout_to_the_cap() {
let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
let (_feedback_tx, feedback_rx) = tokio::sync::watch::channel(0u64);
let handler = DifflerMcp::new(tx, feedback_rx);
let started = tokio::time::Instant::now();
let Json(response) = handler
.wait_for_feedback(Parameters(WaitForFeedbackParams {
since_epoch: None,
timeout_seconds: Some(MAX_WAIT_SECONDS * 10),
}))
.await
.unwrap();
assert!(response.timed_out, "no feedback ever arrives");
assert_eq!(
started.elapsed(),
Duration::from_secs(MAX_WAIT_SECONDS),
"a timeout above the cap must clamp to {MAX_WAIT_SECONDS}s"
);
}
#[test]
fn endpoint_file_publishes_the_port_and_clears() {
let dir = tempfile::tempdir().expect("tempdir");
write_endpoint(dir.path(), 8417).expect("write");
let path = dir.path().join(".diffler/mcp.json");
let body = std::fs::read_to_string(&path).expect("read");
assert!(body.contains("\"port\": 8417"), "{body}");
assert!(body.contains("http://127.0.0.1:8417/mcp"), "{body}");
let gitignore =
std::fs::read_to_string(dir.path().join(".diffler/.gitignore")).expect("gitignore");
assert_eq!(gitignore, "*\n");
clear_endpoint(dir.path());
assert!(!path.exists(), "endpoint file removed on shutdown");
}
#[tokio::test]
async fn spawn_mcp_fallback_on_addr_in_use() {
let holder = tokio::net::TcpListener::bind(("127.0.0.1", 0))
.await
.unwrap();
let busy_port = holder.local_addr().unwrap().port();
let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
let (_feedback_tx, feedback_rx) = tokio::sync::watch::channel(0u64);
let handle = spawn_mcp(tx, feedback_rx, busy_port).await.unwrap();
assert_ne!(
handle.port, busy_port,
"should have bound to a different ephemeral port"
);
handle.handle.abort();
}
}