use std::fmt;
use std::io::ErrorKind;
use std::net::SocketAddr;
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::prompt::PromptRouter;
use rmcp::handler::server::router::tool::ToolRouter;
use rmcp::handler::server::wrapper::Parameters;
use rmcp::model::{PromptMessage, Role, ServerCapabilities, ServerInfo};
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};
use rmcp::{
ErrorData, Json, ServerHandler, prompt, prompt_handler, prompt_router, 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 = 25;
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>,
#[schemars(with = "Count")]
pub open_comments: usize,
#[schemars(with = "Count")]
pub replied_comments: usize,
#[schemars(with = "Count")]
pub resolved_comments: usize,
#[schemars(with = "Count")]
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,
#[schemars(with = "Count")]
pub open_comments: usize,
#[schemars(with = "Count")]
pub replied_comments: usize,
#[schemars(with = "Count")]
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,
#[schemars(with = "Option<Count>")]
pub line: Option<u32>,
#[schemars(with = "Option<Count>")]
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 {
#[schemars(with = "Count")]
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 {
#[schemars(with = "Option<Count>")]
pub since_epoch: Option<u64>,
#[schemars(with = "Option<Count>")]
pub timeout_seconds: Option<u64>,
}
struct Count;
impl schemars::JsonSchema for Count {
fn inline_schema() -> bool {
true
}
fn schema_name() -> std::borrow::Cow<'static, str> {
"Count".into()
}
fn schema_id() -> std::borrow::Cow<'static, str> {
"diffler::mcp::Count".into()
}
fn json_schema(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
schemars::json_schema!({ "type": "integer", "minimum": 0 })
}
}
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>,
prompt_router: PromptRouter<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(),
prompt_router: Self::prompt_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. A timed_out result means the human is still reviewing: call again with the epoch it returned to keep waiting."
)]
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()),
}
}
}
#[prompt_router]
impl DifflerMcp {
#[prompt(
name = "review",
description = "Check the diffler review: read the human's open comments, address them in code, reply, and wait for the next round."
)]
async fn review(&self) -> Vec<PromptMessage> {
vec![PromptMessage::new_text(
Role::User,
"Check the diffler review and respond to the human's feedback:\n\
1. Call review_status for the active review and its changed files.\n\
2. Call get_comments with status \"open\" and read each comment in place.\n\
3. Address every comment in the code it anchors to.\n\
4. Answer each with reply_comment (what you changed and why), then \
propose_resolve; only the human can resolve for real, in the TUI.\n\
5. Call wait_for_feedback with the latest epoch and start over when it \
returns: the human just sent new feedback. It answers within a minute; \
a timed_out result means they are still reviewing, so call it again. \
If the call itself fails, check review_status: diffler is closed only \
when that fails too, otherwise keep waiting.",
)]
}
}
#[tool_handler(router = self.tool_router)]
#[prompt_handler(router = self.prompt_router)]
impl ServerHandler for DifflerMcp {
fn get_info(&self) -> ServerInfo {
ServerInfo::new(
ServerCapabilities::builder()
.enable_tools()
.enable_prompts()
.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. The review prompt packages that loop as a command.",
)
}
}
pub struct McpHandle {
pub port: u16,
pub handle: JoinHandle<()>,
}
fn bind_reusable(addr: SocketAddr) -> std::io::Result<tokio::net::TcpListener> {
let socket = tokio::net::TcpSocket::new_v4()?;
#[cfg(unix)]
socket.set_reuseaddr(true)?;
socket.bind(addr)?;
socket.listen(1024)
}
pub fn spawn_mcp(
tx: UnboundedSender<AppEvent>,
feedback_rx: watch::Receiver<u64>,
port: u16,
) -> std::io::Result<McpHandle> {
let listener = match bind_reusable(SocketAddr::from(([127, 0, 0, 1], port))) {
Ok(listener) => listener,
Err(e) if e.kind() == ErrorKind::AddrInUse => {
bind_reusable(SocketAddr::from(([127, 0, 0, 1], 0)))?
}
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 })
}
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, port: u16) {
let path = endpoint_path(repo_root);
let Ok(body) = std::fs::read_to_string(&path) else {
return;
};
let current_owner = serde_json::from_str::<serde_json::Value>(&body)
.ok()
.and_then(|v| v.get("port").and_then(serde_json::Value::as_u64));
if current_owner == Some(u64::from(port)) {
let _ = std::fs::remove_file(&path);
}
}
#[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"),
],
}],
hashes: diffler_core::model::HashCache::default(),
},
FileDiff {
path: "logo.png".into(),
old_path: None,
status: FileStatus::Added,
binary: true,
old_text: None,
new_text: None,
hunks: vec![],
hashes: diffler_core::model::HashCache::default(),
},
],
}
}
fn anchor(file: &str, line: Option<u32>) -> Anchor {
Anchor {
file: file.to_owned(),
line,
line_end: None,
on_old_side: false,
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(a, "human", "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(a, "human", "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(anchor("src/auth.py", Some(99)), "human", "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(a, "human", "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(anchor("src/auth.py", None), "human", "overall");
session.add_comment(anchor("gone.py", None), "human", "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,
line_text: Some("three".to_owned()),
};
assert_ne!(a.line_text.as_deref(), Some("one"));
session.add_comment(a.clone(), "human", "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"
);
}
#[tokio::test(start_paused = true)]
async fn wait_for_feedback_defaults_inside_the_client_request_timeout() {
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: None,
}))
.await
.unwrap();
assert!(response.timed_out, "no feedback ever arrives");
assert!(
started.elapsed() + REQUEST_TIMEOUT < Duration::from_secs(60),
"the wait plus the reply it still owes must beat a 60 s client cut"
);
}
#[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(), 8417);
assert!(!path.exists(), "endpoint file removed on shutdown");
}
#[test]
fn clear_endpoint_leaves_a_newer_owner_alone() {
let dir = tempfile::tempdir().expect("tempdir");
write_endpoint(dir.path(), 1111).expect("write first instance");
write_endpoint(dir.path(), 2222).expect("write second instance");
clear_endpoint(dir.path(), 1111);
let path = dir.path().join(".diffler/mcp.json");
let body = std::fs::read_to_string(&path)
.expect("file survives: it names the still-running instance");
assert!(body.contains("\"port\": 2222"), "{body}");
}
#[test]
fn tool_schemas_use_no_unregistered_format() {
for tool in DifflerMcp::tool_router().list_all() {
for (part, schema) in [
("input", Some(&*tool.input_schema)),
("output", tool.output_schema.as_deref()),
] {
let Some(schema) = schema else { continue };
let text = serde_json::to_string(schema).expect("schema serializes");
assert!(
!text.contains("\"format\""),
"{} {part} schema carries a format keyword: {text}",
tool.name
);
}
}
}
#[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).unwrap();
assert_ne!(
handle.port, busy_port,
"should have bound to a different ephemeral port"
);
handle.handle.abort();
}
}
#[cfg(test)]
mod agent_command_sync {
fn body(doc: &str) -> String {
let doc = doc.replace("\r\n", "\n");
let after_frontmatter = doc
.strip_prefix("---\n")
.and_then(|rest| rest.split_once("\n---\n"))
.map_or(doc.as_str(), |(_, body)| body);
after_frontmatter.trim().to_owned()
}
#[test]
fn both_agent_command_files_teach_the_same_review_loop() {
const SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/diffler/SKILL.md"
));
const OPENCODE: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../.opencode/commands/diffler.md"
));
assert_eq!(
body(SKILL),
body(OPENCODE),
"skills/diffler/SKILL.md and .opencode/commands/diffler.md have drifted; \
edit both or neither"
);
}
}