//! Embedded MCP server: agents read review comments, answer them in place,
//! and long-poll for the human's feedback. Tool handlers never touch the
//! review directly: every call is sent through the app event channel as an
//! [`McpRequest`] and answered by `App::handle_mcp` on the main loop, so the
//! app stays the single owner of all state (no locks).
use std::fmt;
use std::io::ErrorKind;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
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, ServerConfig};
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;
/// Author label stamped on replies the agent writes through MCP.
pub const AGENT_AUTHOR: &str = "agent";
/// Claude Code cuts a request at 120 s, so `MAX_WAIT_SECONDS` plus one
/// `REQUEST_TIMEOUT` must stay under it; the caller polls again to keep waiting.
const DEFAULT_WAIT_SECONDS: u64 = 25;
const MAX_WAIT_SECONDS: u64 = 55;
/// The app stalls on a call mostly while `$EDITOR` holds the terminal.
pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
/// One agent tool call in flight: the app answers on `reply`. `project`
/// names the tab a project-scoped call acts on, the one in front when unset.
pub struct McpRequest {
pub kind: McpRequestKind,
pub project: Option<String>,
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)
.field("project", &self.project)
.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,
},
/// A comment in the review the human has open; `as_human` makes the human
/// its author.
AddComment {
file: String,
line: u32,
line_end: Option<u32>,
body: String,
as_human: bool,
},
/// Refused for a human's comment and for a walkthrough stop or note,
/// which `publish_walkthrough` owns.
DeleteComment {
id: String,
},
/// Keeps status, replies and anchor. Same refusals as `DeleteComment`.
EditComment {
id: String,
body: String,
},
/// Open + replied comments for `wait_for_feedback` after an epoch bump.
Feedback,
/// Revises the walkthrough `id` names, or publishes a new one when `id`
/// is `None`.
PublishWalkthrough {
id: Option<String>,
title: String,
stops: Vec<StopParams>,
skipped: Option<String>,
summary: Option<String>,
},
/// The walkthrough `id` names, or the newest one when `id` is `None`.
GetWalkthrough {
id: Option<String>,
},
/// The agent's own words for the status bar's activity indicator.
ReportActivity {
focus: String,
file: Option<String>,
},
OpenProject {
path: String,
},
/// Takes the human to `target`, with `note` in the status line.
Focus {
target: FocusTarget,
note: Option<String>,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FocusTarget {
/// A comment id or a walkthrough id.
Id(String),
Code {
file: String,
line: Option<u32>,
line_end: Option<u32>,
review: Option<String>,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum McpResponse {
Status(ReviewStatusResponse),
Diff(String),
Comments(Vec<CommentInfo>),
Reviews(Vec<ReviewSummary>),
Replied {
status: String,
},
Ok,
Added {
id: String,
},
/// Domain refusal (unknown id/file): surfaces as a tool error.
Error(String),
WalkthroughPublished(WalkthroughPublished),
Walkthrough(Option<WalkthroughInfo>),
/// Answers [`McpRequestKind::Feedback`]: the open and replied comments.
Feedback {
comments: Vec<CommentInfo>,
},
/// Answers [`McpRequestKind::OpenProject`]: the tab now showing it.
ProjectOpened(ProjectInfo),
Focused(FocusResponse),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, schemars::JsonSchema)]
#[serde(rename_all = "lowercase")]
pub enum FocusView {
Diff,
File,
/// diffler is fetching what the review needs; the human's view moves
/// once it arrives.
Waiting,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct FocusResponse {
pub project: String,
/// The review on screen, e.g. "working tree" or "PR #12"; absent when
/// the file view shows the file as it is on disk.
#[serde(skip_serializing_if = "Option::is_none")]
pub review: Option<String>,
pub view: FocusView,
#[serde(skip_serializing_if = "Option::is_none")]
pub file: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
#[schemars(with = "Option<Count>")]
pub line: Option<u32>,
}
#[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,
/// Every persisted review and its comment counts; `files_changed` above is
/// the working-tree review, the default the human starts on.
pub reviews: Vec<ReviewSummary>,
/// Every walkthrough published in the repo, newest first; empty when none
/// has been published.
pub walkthroughs: Vec<WalkthroughSummary>,
/// Filenames under `.diffler/reviews/` that failed to parse and were
/// skipped: `reviews` and `walkthroughs` above may be missing an entry
/// because of one of these, rather than the repository having none.
/// Empty when every review file parsed.
pub corrupt_reviews: Vec<String>,
/// Every project open as a tab, with its comment counts. Comments and
/// feedback cover all of them; `project` targets a call at one.
pub projects: Vec<ProjectInfo>,
}
/// One project open as a tab in the human's diffler.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct ProjectInfo {
/// The name to pass as `project`: the repository's folder name.
pub name: String,
pub root: String,
/// The tab the human is looking at.
pub active: bool,
#[schemars(with = "Count")]
pub open_comments: usize,
#[schemars(with = "Count")]
pub replied_comments: usize,
}
/// What a fresh agent needs to know a walkthrough exists before reading
/// anything else: `get_walkthrough` carries the rest.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct WalkthroughSummary {
/// Pass this to `get_walkthrough` or `publish_walkthrough` to read or
/// revise this walkthrough.
pub id: String,
pub title: String,
#[schemars(with = "Count")]
pub stops: usize,
#[schemars(with = "Count")]
pub at: u64,
}
/// One review's provenance and comment tally: what was reviewed (working tree,
/// a commit, or a range) and how many comments sit at each status.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct ReviewSummary {
/// The project this review belongs to, when more than one is open.
#[serde(skip_serializing_if = "Option::is_none")]
pub project: Option<String>,
/// Stable source key (e.g. `working`, `commit-<oid>`, `range-<a>-<b>`).
pub source: String,
/// Human-facing description (e.g. "commit a1b2c3", "range a1b2c3..d4e5f6").
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,
/// The project this comment belongs to, when more than one is open.
#[serde(skip_serializing_if = "Option::is_none")]
pub project: Option<String>,
/// Stable key of the review this comment belongs to (see [`ReviewSummary`]).
pub source: String,
/// Human-facing description of that review (what the human was looking at).
pub source_label: String,
pub file: String,
#[schemars(with = "Option<Count>")]
pub line: Option<u32>,
#[schemars(with = "Option<Count>")]
pub line_end: Option<u32>,
/// Which side of the diff the line numbers count: "old" or "new".
pub side: String,
pub body: String,
pub status: String,
pub author: String,
pub replies: Vec<ReplyInfo>,
/// The anchored line changed or vanished since the comment was made.
pub outdated: bool,
/// Origin-prefixed diff snippet around the anchored line.
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,
}
/// One line of what `publish_walkthrough` checked or resolved. A refusal
/// (`too_many_stops`, `empty_stops`, `body_too_long`, `total_too_long`,
/// `anchor_unparsed`, `anchor_file_missing`, `nothing_to_anchor`,
/// `note_outside_stop`, `duplicate_id`) means nothing was stored; the rest
/// (`anchor_whole`, `figure_dropped`, `figure_simplified`) land alongside a
/// stored walkthrough so the agent learns what to fix without a broken
/// figure ever reaching the reader.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct ReceiptInfo {
#[schemars(with = "Option<Count>")]
pub stop: Option<usize>,
pub code: String,
pub detail: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct WalkthroughPublished {
pub id: String,
#[schemars(with = "Count")]
pub stops: usize,
pub receipts: Vec<ReceiptInfo>,
/// The full commit this walkthrough is pinned to, `None` when the repo
/// has no commits yet. Revise it once this no longer matches the branch.
pub rev: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct StopInfo {
/// The comment this stop is. Pass it back as a stop's `id` to revise the
/// stop and keep the thread hanging off it.
pub id: String,
pub title: String,
pub anchor: Option<String>,
pub body: String,
/// The extra comments this stop carries in its region. Pass a note's `id`
/// back to keep it; a note whose id you leave out is deleted.
pub notes: Vec<NoteInfo>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct NoteInfo {
pub id: String,
pub anchor: Option<String>,
pub body: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct WalkthroughInfo {
pub id: String,
pub title: String,
pub author: String,
#[schemars(with = "Count")]
pub at: u64,
pub skipped: Option<String>,
/// The walkthrough's own overview, when it has one: pass it back
/// unchanged on a revision to keep it.
pub summary: Option<String>,
/// The full commit this walkthrough is pinned to, `None` for one
/// published before `rev` existed.
pub rev: Option<String>,
pub stops: Vec<StopInfo>,
}
/// An MCP tool's structured output must itself describe an object, so a
/// possibly-absent walkthrough is a field here rather than the bare
/// `Option<WalkthroughInfo>` a top-level return would need.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct GetWalkthroughResponse {
pub walkthrough: Option<WalkthroughInfo>,
}
#[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 {
/// Restrict the diff to one file (repo-relative path).
pub file: Option<String>,
/// The project to act on: its folder name from `review_status`, or its
/// path. Omit it for the project the human is looking at.
pub project: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct GetCommentsParams {
/// Filter by comment status: "open", "replied", or "resolved".
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,
/// Short note on why the comment is addressed.
pub note: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct MarkViewedParams {
/// Repo-relative path of a file in the review diff.
pub file: String,
/// The project to act on: its folder name from `review_status`, or its
/// path. Omit it for the project the human is looking at.
pub project: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct AddCommentParams {
/// Repo-relative path of a file in the review diff.
pub file: String,
/// Line number in the diff (new-side, unless the line was deleted).
#[schemars(with = "Count")]
pub line: u32,
/// Last line of an inclusive range starting at `line`; omit for one line.
#[schemars(with = "Option<Count>")]
pub line_end: Option<u32>,
pub body: String,
/// Author the comment as the human, so it goes out untouched with their
/// next submitted review. Off by default, which leaves the comment the
/// agent's own for the human to answer.
pub as_human: Option<bool>,
/// The project to act on: its folder name from `review_status`, or its
/// path. Omit it for the project the human is looking at.
pub project: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct AddCommentResponse {
pub id: String,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct DeleteCommentParams {
/// The comment to delete, from `get_comments` or `add_comment`.
pub id: String,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct EditCommentParams {
/// The comment to rewrite, from `get_comments` or `add_comment`.
pub id: String,
pub body: String,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize, schemars::JsonSchema)]
pub struct StopParams {
/// The comment id of the stop this replaces, from `get_walkthrough`. Pass
/// it to keep that stop's thread; omit it for a new stop.
pub id: Option<String>,
/// A label of two to five words naming what the stop is about, no verb;
/// the sidebar shows about thirty characters.
pub title: String,
/// `path#symbol` where a symbol exists, `path:start-end` where none does,
/// `path:line` for one line, a bare `path` for a whole file, or omit for
/// a stop with no anchor.
pub anchor: Option<String>,
/// Markdown: two or three short bullets on what we chose and why, in full
/// sentences, every identifier in backticks, plain verbs for code (raise,
/// return, check) and no metaphor. A `mermaid` or `callstack` fence draws
/// a diagram.
pub body: String,
/// Extra remarks on this same stop: a second point about another part of
/// its region, or a diagram beside the prose. Each becomes its own comment
/// in the stop's slide.
pub notes: Option<Vec<NoteParams>>,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize, schemars::JsonSchema)]
pub struct NoteParams {
/// The comment id of the note this replaces, from `get_walkthrough`. Pass
/// it to keep that note's thread; omit it for a new note.
pub id: Option<String>,
/// `path:line` or `path:start-end` in the stop's own file, or omit to
/// anchor the note at the start of the stop's region.
pub anchor: Option<String>,
pub body: String,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct PublishWalkthroughParams {
/// The walkthrough to revise, from `get_walkthrough` or `review_status`.
/// Omit to publish a new one of its own.
pub id: Option<String>,
/// The walkthrough's name, as the status screen lists it.
pub title: String,
pub stops: Vec<StopParams>,
/// One line on what you left out and why.
pub skipped: Option<String>,
/// What the reader meets first: one short paragraph saying what the
/// change does, plus one diagram of the simplest shape that explains it,
/// five to eight nodes, participants or frames, naming real files or
/// functions. Pick the kind for what you are showing: a `mermaid`
/// `flowchart` for a branch, a state, or data moving through steps; a
/// `mermaid` `sequenceDiagram` for who calls whom over time, with
/// `link B: label @ path#symbol` tying participant `B` to its code; a
/// `callstack` fence for how a call path changed, one frame per line, two
/// spaces of indent per depth, `+ `/`- ` right before an added or removed
/// frame's label, and an optional ` @ path#symbol` after it. Never a list
/// of the stops or a repeat of their titles. Omit for a walkthrough with
/// no summary.
pub summary: Option<String>,
/// The project to act on: its folder name from `review_status`, or its
/// path. Omit it for the project the human is looking at.
pub project: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct OpenProjectParams {
/// Path of a git repository, absolute or starting with `~`.
pub path: String,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct FocusParams {
/// A comment id (from `get_comments`, `add_comment` or `get_walkthrough`) or a
/// walkthrough id. Takes the human to it in whichever review holds it.
pub id: Option<String>,
/// Repo-relative path to show, when you pass no `id`.
pub file: Option<String>,
/// New-side line to put the cursor on; omit for the top of the file.
#[schemars(with = "Option<Count>")]
pub line: Option<u32>,
/// Last line of an inclusive range starting at `line`, banded on screen.
#[schemars(with = "Option<Count>")]
pub line_end: Option<u32>,
/// The review to show `file` in: `working`, `commit:<rev>`,
/// `range:<oldest>..<newest>` (both commits included), `pr:<number>`,
/// `against:<rev>`, or a `source` from `list_reviews`. Omit it to use the
/// review on screen when it holds the file, else the working tree, else
/// the file as it is on disk.
pub review: Option<String>,
/// A few words the human reads as their view moves, e.g. "the retry loop
/// you asked about".
pub note: Option<String>,
/// The project to act on: its folder name from `review_status`, or its
/// path. Omit it for the project the human is looking at; an `id` finds
/// its own project.
pub project: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct GetWalkthroughParams {
/// The walkthrough to fetch, from `review_status`. Omit for the newest.
pub id: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct ReportActivityParams {
/// Short plain phrase for what you're doing right now, e.g. "writing the
/// walkthrough" or "fixing the failing test". Keep it to a few words;
/// the bar cuts a long one.
pub focus: String,
/// The file this is about, if any.
pub file: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct WaitForFeedbackParams {
/// Return once the feedback epoch exceeds this value. Defaults to the
/// current epoch, i.e. wait for the next human send.
#[schemars(with = "Option<Count>")]
pub since_epoch: Option<u64>,
/// Long-poll timeout in seconds (default 25), capped at 55; call again
/// to keep waiting.
#[schemars(with = "Option<Count>")]
pub timeout_seconds: Option<u64>,
}
/// Unsigned ints derive `format: "uint*"`, which JSON Schema does not register
/// and strict validators warn on. rmcp hardcodes its generator, so we opt each
/// field out with `#[schemars(with = "Count")]`.
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",
FileStatus::Unchanged => "unchanged",
}
}
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;
// outdated detection reads a range's end line, but we start the snippet at
// its first line so it reads top-down
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(),
project: None,
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,
}
}
/// MCP tool handler: forwards every call to the app over the event channel
/// and holds a feedback-epoch receiver for the long poll.
#[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, None, REQUEST_TIMEOUT).await
}
/// [`Self::request`] for a call that acts on one project's tab.
async fn request_in(
&self,
project: Option<String>,
kind: McpRequestKind,
) -> Result<McpResponse, ErrorData> {
self.request_with_timeout(kind, project, REQUEST_TIMEOUT)
.await
}
async fn request_with_timeout(
&self,
kind: McpRequestKind,
project: Option<String>,
timeout: Duration,
) -> Result<McpResponse, ErrorData> {
let (reply, response) = oneshot::channel();
self.tx
.send(AppEvent::Mcp(McpRequest {
kind,
project,
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_in(params.project, 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 = "Flag a comment as addressed: marks it replied. Writes nothing into the thread, so call it after reply_comment rather than repeating the answer. The optional note is used only when you did not reply. 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_in(params.project, kind).await? {
McpResponse::Ok => Ok(Json(OkResponse { ok: true })),
_ => Err(mismatch()),
}
}
#[tool(
description = "Write a review comment on a line or an inclusive line range in the review the human has open: the working tree, or the commit, range or PR on screen. To comment on another review, `focus` it first. The comment is anchored like the human's own, so it shows as outdated once its line changes. It is the agent's comment, for the human to answer in its thread; `as_human` makes it the human's own, sent with their next submitted review. Returns its `id`."
)]
async fn add_comment(
&self,
Parameters(params): Parameters<AddCommentParams>,
) -> Result<Json<AddCommentResponse>, ErrorData> {
let kind = McpRequestKind::AddComment {
file: params.file,
line: params.line,
line_end: params.line_end,
body: params.body,
as_human: params.as_human.unwrap_or(false),
};
match self.request_in(params.project, kind).await? {
McpResponse::Added { id } => Ok(Json(AddCommentResponse { id })),
_ => Err(mismatch()),
}
}
#[tool(
description = "Delete a comment, yours or the human's. Refused for one already posted to the forge, for one someone else has replied to (edit it instead), and for a walkthrough stop or note, which `publish_walkthrough` revises."
)]
async fn delete_comment(
&self,
Parameters(params): Parameters<DeleteCommentParams>,
) -> Result<Json<OkResponse>, ErrorData> {
let kind = McpRequestKind::DeleteComment { id: params.id };
match self.request(kind).await? {
McpResponse::Ok => Ok(Json(OkResponse { ok: true })),
_ => Err(mismatch()),
}
}
#[tool(
description = "Replace a comment's body, yours or the human's, keeping its author, status, replies and anchor. Refused for one already posted to the forge, and for a walkthrough stop or note, which `publish_walkthrough` revises."
)]
async fn edit_comment(
&self,
Parameters(params): Parameters<EditCommentParams>,
) -> Result<Json<OkResponse>, ErrorData> {
let kind = McpRequestKind::EditComment {
id: params.id,
body: params.body,
};
match self.request(kind).await? {
McpResponse::Ok => Ok(Json(OkResponse { ok: true })),
_ => Err(mismatch()),
}
}
#[tool(
description = "Publish a walkthrough: the order a reader should take through a change. Use one stop per real decision, as few as the change needs (five is common; more than ten means the change wants splitting). Each stop says why, never what the diff already shows, and becomes an agent comment the human answers in its thread. It describes the review open now, pinned to the commit checked out. Pass `id` to revise one in place, with each surviving stop's and note's `id` from `get_walkthrough`; any left out are deleted. Returns the walkthrough's `id` and receipts; a refusal names what to fix. Limits: 20 stops, 8 KB per body, 64 KB in all."
)]
async fn publish_walkthrough(
&self,
Parameters(params): Parameters<PublishWalkthroughParams>,
) -> Result<Json<WalkthroughPublished>, ErrorData> {
let kind = McpRequestKind::PublishWalkthrough {
id: params.id,
title: params.title,
stops: params.stops,
skipped: params.skipped,
summary: params.summary,
};
match self.request_in(params.project, kind).await? {
McpResponse::WalkthroughPublished(published) => Ok(Json(published)),
_ => Err(mismatch()),
}
}
#[tool(
description = "The walkthrough `id` names (from review_status), or the newest one when `id` is omitted, or null when the review has none."
)]
async fn get_walkthrough(
&self,
Parameters(params): Parameters<GetWalkthroughParams>,
) -> Result<Json<GetWalkthroughResponse>, ErrorData> {
match self
.request(McpRequestKind::GetWalkthrough { id: params.id })
.await?
{
McpResponse::Walkthrough(walkthrough) => {
Ok(Json(GetWalkthroughResponse { walkthrough }))
}
_ => Err(mismatch()),
}
}
#[tool(
description = "Show what you are doing in the human's status bar: a few words (\"writing the walkthrough\") and the file it is about, if any. Every other tool call already shows on its own, so call this before work no call covers, such as reading code or writing tests, or when a call's own label says too little."
)]
async fn report_activity(
&self,
Parameters(params): Parameters<ReportActivityParams>,
) -> Result<Json<OkResponse>, ErrorData> {
let kind = McpRequestKind::ReportActivity {
focus: params.focus,
file: params.file,
};
match self.request(kind).await? {
McpResponse::Ok => Ok(Json(OkResponse { ok: true })),
_ => Err(mismatch()),
}
}
#[tool(
description = "Open a git repository as a project tab in the human's diffler, so they can review its changes beside the others. Call it for every repository you changed outside the one diffler started in. Comments and feedback then cover it too, and `project` targets a call at it."
)]
async fn open_project(
&self,
Parameters(params): Parameters<OpenProjectParams>,
) -> Result<Json<ProjectInfo>, ErrorData> {
let kind = McpRequestKind::OpenProject { path: params.path };
match self.request(kind).await? {
McpResponse::ProjectOpened(project) => Ok(Json(project)),
_ => Err(mismatch()),
}
}
#[tool(
description = "Take the human's diffler to a place: a comment or walkthrough by `id`, or a `file` and line range in any review. It switches to the project's tab and opens that review with the cursor on the place. Use it when the human asks where something is, and after `publish_walkthrough` to open the walkthrough. To point at code with a remark, call `add_comment`, then focus the `id` it returns. Refused while the human is typing a comment or answering a dialog; a pull request that still has to load answers `waiting` and opens once it does."
)]
async fn focus(
&self,
Parameters(params): Parameters<FocusParams>,
) -> Result<Json<FocusResponse>, ErrorData> {
let target = match (params.id, params.file) {
(Some(id), None) => FocusTarget::Id(id),
(None, Some(file)) => FocusTarget::Code {
file,
line: params.line,
line_end: params.line_end,
review: params.review,
},
_ => {
return Err(ErrorData::invalid_params("pass either id or file", None));
}
};
let kind = McpRequestKind::Focus {
target,
note: params.note,
};
match self.request_in(params.project, kind).await? {
McpResponse::Focused(focused) => Ok(Json(focused)),
_ => Err(mismatch()),
}
}
#[tool(
description = "Wait until the human sends feedback: a comment, a reply, or the send key. Returns the new epoch and every open or replied comment, or `timed_out` while they are still reviewing; call again with the returned epoch to keep waiting. A reply on a walkthrough stop arrives as a reply on that stop's comment, so its id names the stop."
)]
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),
);
// we tell the app before blocking, since the poll sends no request
// until the human answers
let _ = self.tx.send(AppEvent::McpWaiting {
until: Instant::now() + timeout,
});
let waited = tokio::time::timeout(timeout, rx.wait_for(|epoch| *epoch > since))
.await
// copy the epoch out so the watch borrow ends before the match
.map(|result| result.map(|epoch| *epoch));
let (epoch, timed_out) = match waited {
Ok(Ok(epoch)) => (epoch, false),
Ok(Err(_)) => {
return Err(ErrorData::internal_error(
"the diffler TUI is not running",
None,
));
}
Err(_) => (*rx.borrow(), true),
};
if timed_out {
return Ok(Json(WaitForFeedbackResponse {
epoch,
timed_out: true,
comments: Vec::new(),
}));
}
match self.request(McpRequestKind::Feedback).await? {
McpResponse::Feedback { comments } => Ok(Json(WaitForFeedbackResponse {
epoch,
timed_out: false,
comments,
})),
_ => Err(mismatch()),
}
}
}
/// A skill file's prose with its YAML frontmatter dropped. We normalise CRLF
/// first, since on a Windows checkout `\n` matches no frontmatter and we would
/// ship the YAML header to the agent.
fn skill_body(doc: &str) -> String {
let doc = doc.replace("\r\n", "\n");
let body = doc
.strip_prefix("---\n")
.and_then(|rest| rest.split_once("\n---\n"))
.map_or(doc.as_str(), |(_, body)| body);
body.trim().to_owned()
}
// a published crate carries only its own directory, so the prompts live here
// and `agent_command_sync` pins them to the skill files the plugin ships
const DF_SKILL: &str = include_str!("../prompts/df.md");
const DFA_SKILL: &str = include_str!("../prompts/dfa.md");
const DFR_SKILL: &str = include_str!("../prompts/dfr.md");
#[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, skill_body(DF_SKILL))]
}
#[prompt(
name = "walkthrough",
description = "Walk the human through a change: publish a walkthrough, one stop per real decision, and answer their comments."
)]
async fn walkthrough(&self) -> Vec<PromptMessage> {
vec![PromptMessage::new_text(Role::User, skill_body(DFA_SKILL))]
}
#[prompt(
name = "critique",
description = "Review a change in diffler: read the diff and leave comments on real problems, one per issue, and never submit."
)]
async fn critique(&self) -> Vec<PromptMessage> {
vec![PromptMessage::new_text(Role::User, skill_body(DFR_SKILL))]
}
}
#[tool_handler(router = self.tool_router)]
#[prompt_handler(router = self.prompt_router)]
impl ServerHandler for DifflerMcp {
fn get_info(&self) -> ServerConfig {
ServerConfig::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 an \
answered one as addressed without adding text, and \
wait_for_feedback long-polls until the human sends \
new feedback. open_project opens another repository you \
changed as a tab in the same diffler; comments and feedback \
cover every open project, each tagged with its project. focus \
takes the human to a comment, a walkthrough, or a file and \
line in any review. The review prompt packages that loop as a \
command.",
)
}
}
pub struct McpHandle {
/// Actual bound port: the configured one, or an ephemeral fallback.
pub port: u16,
pub handle: JoinHandle<()>,
}
// SO_REUSEADDR lets a restart reclaim the port while the old socket lingers in
// TIME_WAIT. Unix only: on Windows it instead allows hijacking a live port and
// errors WSAEACCES, defeating the busy-port fallback.
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)
}
/// Serves `127.0.0.1:{port}/mcp`, or an ephemeral port when `port` is taken;
/// the handle carries the port that bound.
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 {
// serve only ends on listener errors; the TUI aborts this task on quit
let _ = axum::serve(listener, router).await;
});
Ok(McpHandle { port, handle })
}
const ENDPOINT_FILE: &str = "mcp.json";
fn endpoint_path(repo_root: &Path) -> PathBuf {
repo_root.join(".diffler").join(ENDPOINT_FILE)
}
#[derive(Serialize)]
struct EndpointFile {
port: u16,
url: String,
pid: u32,
}
/// The stdio proxy reads the bound port from `.diffler/mcp.json`, or from the
/// per-user registry when it starts outside any repo.
pub fn write_endpoint(repo_root: &Path, port: u16) -> std::io::Result<()> {
let pid = std::process::id();
let body = serde_json::to_string_pretty(&EndpointFile {
port,
url: format!("http://127.0.0.1:{port}/mcp"),
pid,
})
.map_err(std::io::Error::other)?;
diffler_core::store::write_file(repo_root, ENDPOINT_FILE, &body)?;
write_registry_entry(repo_root, port, pid);
Ok(())
}
/// We remove the file only while it names our port, since a second instance
/// in the same repo overwrites it and still needs it after we exit.
pub fn clear_endpoint(repo_root: &Path, port: u16) {
let path = endpoint_path(repo_root);
if let Ok(body) = std::fs::read_to_string(&path) {
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);
}
}
clear_registry_entry(repo_root, port);
}
#[derive(Serialize)]
struct RegistryEntry<'a> {
repo: &'a str,
port: u16,
pid: u32,
url: String,
}
/// `$XDG_STATE_HOME/diffler/instances`, defaulting to `~/.local/state`.
/// Tests set `DIFFLER_STATE_DIR` so they never touch the real home directory.
fn registry_dir() -> Option<PathBuf> {
let base = if let Some(dir) = non_empty_env("DIFFLER_STATE_DIR") {
PathBuf::from(dir)
} else if let Some(dir) = non_empty_env("XDG_STATE_HOME") {
PathBuf::from(dir)
} else {
PathBuf::from(non_empty_env("HOME")?)
.join(".local")
.join("state")
};
Some(base.join("diffler").join("instances"))
}
fn non_empty_env(key: &str) -> Option<std::ffi::OsString> {
std::env::var_os(key).filter(|v| !v.is_empty())
}
/// Best-effort: a same-repo proxy still finds `.diffler/mcp.json`, so we
/// ignore write failures here.
fn write_registry_entry(repo_root: &Path, port: u16, pid: u32) {
let Some(dir) = registry_dir() else {
return;
};
if std::fs::create_dir_all(&dir).is_err() {
return;
}
let repo = canonical_repo(repo_root);
let Some(repo) = repo.to_str() else {
return;
};
let Ok(body) = serde_json::to_string_pretty(&RegistryEntry {
repo,
port,
pid,
url: format!("http://127.0.0.1:{port}/mcp"),
}) else {
return;
};
let _ = std::fs::write(dir.join(registry_file(repo_root, port)), body);
}
pub(crate) fn canonical_repo(repo_root: &Path) -> PathBuf {
repo_root
.canonicalize()
.unwrap_or_else(|_| repo_root.to_path_buf())
}
/// One entry per project, since one diffler serves all its tabs on one port.
fn registry_file(repo_root: &Path, port: u16) -> String {
let mut hasher = std::hash::DefaultHasher::new();
std::hash::Hash::hash(&canonical_repo(repo_root), &mut hasher);
format!("{port}-{:016x}.json", std::hash::Hasher::finish(&hasher))
}
/// Same owner check as `clear_endpoint`.
fn clear_registry_entry(repo_root: &Path, port: u16) {
let Some(dir) = registry_dir() else {
return;
};
let path = dir.join(registry_file(repo_root, port));
let Ok(body) = std::fs::read_to_string(&path) else {
return;
};
let owner_port = serde_json::from_str::<serde_json::Value>(&body)
.ok()
.and_then(|v| v.get("port").and_then(serde_json::Value::as_u64));
if owner_port == 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(),
blobs: diffler_core::model::BlobIds::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(),
blobs: diffler_core::model::BlobIds::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();
// comment spans new lines 1-3; line_text snapshots line 3 ("three")
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,
None,
Duration::from_millis(50),
)
.await
.unwrap_err();
assert!(
err.message.contains("busy"),
"error should mention busy: {err:?}"
);
}
// a paused clock jumps to the next timer, so elapsed time is the effective timeout
#[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(120),
"the wait plus the reply it still owes must beat the client's 120 s cut"
);
}
#[test]
fn a_capped_wait_and_its_round_trip_fit_the_client_ceiling() {
assert!(MAX_WAIT_SECONDS + REQUEST_TIMEOUT.as_secs() < 120);
}
#[tokio::test]
async fn wait_for_feedback_answers_with_the_reviews_comments() {
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let (feedback_tx, feedback_rx) = tokio::sync::watch::channel(0u64);
feedback_tx.send_modify(|epoch| *epoch += 1);
let handler = DifflerMcp::new(tx, feedback_rx);
let requests = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let counted = requests.clone();
tokio::spawn(async move {
while let Some(event) = rx.recv().await {
let AppEvent::Mcp(request) = event else {
continue;
};
counted.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let response = match request.kind {
McpRequestKind::Feedback => McpResponse::Feedback {
comments: Vec::new(),
},
_ => McpResponse::Ok,
};
let _ = request.reply.send(response);
}
});
let Json(response) = handler
.wait_for_feedback(Parameters(WaitForFeedbackParams {
since_epoch: Some(0),
timeout_seconds: None,
}))
.await
.unwrap();
assert!(!response.timed_out);
assert_eq!(
requests.load(std::sync::atomic::Ordering::SeqCst),
1,
"the success path makes exactly one app request after the wait"
);
}
// `DIFFLER_STATE_DIR` is process-global, so registry tests serialize on this lock.
static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
#[allow(unsafe_code)]
fn with_state_dir<T>(dir: &Path, f: impl FnOnce() -> T) -> T {
let _guard = ENV_LOCK.lock().unwrap();
// SAFETY: ENV_LOCK serializes every test that reads or writes this var.
unsafe { std::env::set_var("DIFFLER_STATE_DIR", dir) };
let result = f();
unsafe { std::env::remove_var("DIFFLER_STATE_DIR") };
result
}
#[test]
fn endpoint_file_publishes_the_port_and_clears() {
let dir = tempfile::tempdir().expect("tempdir");
let state = tempfile::tempdir().expect("tempdir");
with_state_dir(state.path(), || {
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}");
assert!(body.contains("\"pid\""), "{body}");
// the .diffler dir self-gitignores like the session store
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");
let state = tempfile::tempdir().expect("tempdir");
with_state_dir(state.path(), || {
write_endpoint(dir.path(), 1111).expect("write first instance");
// a second diffler instance in the same repo overwrites the file
write_endpoint(dir.path(), 2222).expect("write second instance");
// the first instance shuts down and clears its own (stale) port
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 write_endpoint_registers_the_instance_for_cross_repo_discovery() {
let dir = tempfile::tempdir().expect("tempdir");
let state = tempfile::tempdir().expect("tempdir");
with_state_dir(state.path(), || {
write_endpoint(dir.path(), 8417).expect("write");
let entry_path = state
.path()
.join("diffler/instances")
.join(registry_file(dir.path(), 8417));
let body = std::fs::read_to_string(&entry_path).expect("registry entry written");
// we parse the entry because JSON escapes a Windows path's separators
let entry: serde_json::Value =
serde_json::from_str(&body).expect("registry entry is json");
let repo = dir.path().canonicalize().expect("canonicalize");
assert_eq!(entry["repo"].as_str(), repo.to_str(), "{body}");
assert_eq!(entry["port"].as_u64(), Some(8417), "{body}");
assert_eq!(
entry["pid"].as_u64(),
Some(u64::from(std::process::id())),
"{body}"
);
assert_eq!(
entry["url"].as_str(),
Some("http://127.0.0.1:8417/mcp"),
"{body}"
);
});
}
#[test]
fn clear_endpoint_removes_only_its_own_registry_entry() {
let dir = tempfile::tempdir().expect("tempdir");
let state = tempfile::tempdir().expect("tempdir");
with_state_dir(state.path(), || {
write_endpoint(dir.path(), 1111).expect("write first instance");
write_endpoint(dir.path(), 2222).expect("write second instance");
let instances = state.path().join("diffler/instances");
clear_endpoint(dir.path(), 1111);
assert!(
!instances.join(registry_file(dir.path(), 1111)).exists(),
"the cleared instance's own registry entry is gone"
);
assert!(
instances.join(registry_file(dir.path(), 2222)).exists(),
"a different instance's registry entry survives"
);
});
}
#[test]
fn one_diffler_registers_every_project_it_serves() {
let (first, second) = (
tempfile::tempdir().expect("tempdir"),
tempfile::tempdir().expect("tempdir"),
);
let state = tempfile::tempdir().expect("tempdir");
with_state_dir(state.path(), || {
write_endpoint(first.path(), 3333).expect("first project");
write_endpoint(second.path(), 3333).expect("second project");
let instances = state.path().join("diffler/instances");
assert_eq!(std::fs::read_dir(&instances).expect("registry").count(), 2);
clear_endpoint(second.path(), 3333);
assert!(instances.join(registry_file(first.path(), 3333)).exists());
assert!(!instances.join(registry_file(second.path(), 3333)).exists());
});
}
#[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 a_skill_loses_its_frontmatter_whatever_its_line_endings() {
assert_eq!(
super::skill_body("---\r\nname: df\r\n---\r\n\r\nthe body\r\n"),
"the body"
);
assert_eq!(
super::skill_body("---\nname: df\n---\n\nthe body\n"),
"the body"
);
}
/// Every agent command ships as a Claude Code skill, an `OpenCode` command
/// and a crate prompt, with only the frontmatter allowed to differ.
#[test]
fn both_agent_command_files_teach_the_same_loop() {
const DF_SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/df/SKILL.md"
));
const DF_OPENCODE: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../.opencode/commands/df.md"
));
const DFA_SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/dfa/SKILL.md"
));
const DFA_OPENCODE: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../.opencode/commands/dfa.md"
));
const DFR_SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/dfr/SKILL.md"
));
const DFR_OPENCODE: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../.opencode/commands/dfr.md"
));
for (name, skill, opencode, prompt) in [
("df", DF_SKILL, DF_OPENCODE, super::DF_SKILL),
("dfa", DFA_SKILL, DFA_OPENCODE, super::DFA_SKILL),
("dfr", DFR_SKILL, DFR_OPENCODE, super::DFR_SKILL),
] {
assert_eq!(
body(skill),
body(opencode),
"skills/{name}/SKILL.md and .opencode/commands/{name}.md have drifted; \
edit both or neither"
);
assert_eq!(
body(skill),
body(prompt),
"skills/{name}/SKILL.md and crates/diffler/prompts/{name}.md have drifted; \
the crate ships its own copy because a published crate carries only \
its own directory"
);
}
}
fn write_for_the_card(doc: &str) -> &str {
const HEADING: &str = "## Write for the card";
let Some(start) = doc.find(HEADING) else {
return "";
};
let after = &doc[start..];
let end = after[HEADING.len()..]
.find("\n## ")
.map_or(after.len(), |offset| offset + HEADING.len());
after[..end].trim()
}
/// All three skills write a card in the same pane, so they share its writing rules.
#[test]
fn every_skill_writes_the_card_the_same_way() {
const DF_SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/df/SKILL.md"
));
const DFA_SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/dfa/SKILL.md"
));
const DFR_SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/dfr/SKILL.md"
));
let df = write_for_the_card(DF_SKILL);
let dfa = write_for_the_card(DFA_SKILL);
let dfr = write_for_the_card(DFR_SKILL);
assert!(
!df.is_empty(),
"df must have a '## Write for the card' section"
);
assert_eq!(
df, dfa,
"the '## Write for the card' section must be identical in df and dfa"
);
assert_eq!(
df, dfr,
"the '## Write for the card' section must be identical in df and dfr"
);
}
/// Each MCP prompt must be its skill file's steps, verbatim.
#[tokio::test]
async fn the_prompts_are_their_skills_bodies() {
const DF_SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/df/SKILL.md"
));
const DFA_SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/dfa/SKILL.md"
));
const DFR_SKILL: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../skills/dfr/SKILL.md"
));
let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
let (_feedback_tx, feedback_rx) = tokio::sync::watch::channel(0u64);
let handler = super::DifflerMcp::new(tx, feedback_rx);
for (messages, skill) in [
(handler.review().await, DF_SKILL),
(handler.walkthrough().await, DFA_SKILL),
(handler.critique().await, DFR_SKILL),
] {
let text = &messages[0]
.content
.as_text()
.expect("prompt carries text")
.text;
assert_eq!(*text, body(skill));
}
}
}