From 6ca5f3e7863e15517cd17b3fcb36531dd1cb7c40 Mon Sep 17 00:00:00 2001 From: Devil Date: Wed, 30 Sep 2026 23:31:31 -0400 Subject: [PATCH] feat: spec-conformant MCP server and CLI fixes MCP server (rewritten, protocol 2025-11-25): - Negotiate the protocol version (echo a supported one, else our newest); advertise the Tasks capability only where the revision has it. - Implement the spec's experimental Tasks utility instead of the ad-hoc methods: task-augmented `clodex_build` (execution.taskSupport), plus tasks/get, tasks/result (blocks, returns the tool result), tasks/list (opaque cursors) and tasks/cancel, with spec shapes and -32602 errors. `tasks/update` was never part of the spec and is removed. - JSON-RPC hygiene: errors keep the request id, notifications and stray responses get no reply, ping, parse/invalid-request codes, batch refused, unexpected exceptions become -32603 with the id. - Argument problems are tool-level errors (isError) so the model can fix them; an unknown tool is a protocol error; the workflow is built lazily so a bad call creates no state. - Requests run in arrival order (clients may pipeline stateful calls); only long-running work (sync plan/build/audit, tasks/result waits) moves to a worker thread so ping/poll/cancel are never blocked. A cancelled-request notification or EOF releases a waiting tasks/result. - Repo root from CLODEX_REPO_ROOT or the git root; version from __version__; clodex_build exposes workspace and approval_profile. - Wire messages validated against the official SDK's 2025-11-25 schema models (optional `mcp` test extra; CI installs it). CLI: - `audit --base REF` / `--commit SHA` (and --diff as the explicit default); an empty diff reports nothing-to-audit instead of running reviewers. - Plain-text output without --json (doctor shows [ok]/[FAIL]/[warn] lines with fixes); exit codes 0 ok / 1 blocked / 2 usage / 3 failed / 4 cancelled; unknown run ids and bad refs are one-line errors (CLODEX_DEBUG=1 for the traceback). - Every command finds the repo root from any subdirectory instead of using the cwd. Co-Authored-By: Claude Sonnet 5.5 --- .github/workflows/ci.yml | 2 + README.md | 35 ++- clodex/artifacts.py | 32 +++ clodex/cli.py | 123 +++++++-- clodex/config.py | 7 + clodex/mcp_server.py | 455 ++++++++++++++++++++++++++++------ clodex/state.py | 7 +- clodex/workflow.py | 29 ++- pyproject.toml | 4 + tests/test_cli_behavior.py | 231 +++++++++++++++++ tests/test_mcp_conformance.py | 391 +++++++++++++++++++++++++++++ tests/test_mcp_sdk.py | 65 +++++ tests/test_mcp_server.py | 4 +- 13 files changed, 1270 insertions(+), 115 deletions(-) create mode 100644 tests/test_cli_behavior.py create mode 100644 tests/test_mcp_conformance.py create mode 100644 tests/test_mcp_sdk.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 54c2142..fe5b0d8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,6 +30,8 @@ jobs: run: | git config --global user.email "ci@example.com" git config --global user.name "CI" + - name: Install the MCP SDK (enables wire-schema tests; they skip if unavailable) + run: python -m pip install mcp || echo "mcp SDK unavailable on this Python; wire-schema tests will skip" - name: Unit tests run: python -m unittest discover -s tests -v - name: package.json is valid JSON diff --git a/README.md b/README.md index 9b31ae4..b93258f 100644 --- a/README.md +++ b/README.md @@ -115,7 +115,7 @@ PowerShell: | `clodex init --migrate [--split-claude] [--dry-run]` | Update `CLODEX.md` settings that no longer work (retired models, unsupported efforts) | | `clodex plan ""` | Run Claude planning only | | `clodex build ""` | Run plan, implementation, and dual audit loop in an isolated worktree | -| `clodex audit --diff` | Audit current uncommitted changes | +| `clodex audit [--diff \| --base REF \| --commit SHA]` | Audit the uncommitted diff (default), everything since `REF` diverged from `HEAD`, or one commit | | `clodex run ""` | Alias for `build` | | `clodex apply ` | Apply an approved worktree patch back to the source checkout (`--check` to dry-run, `--force` for unapproved runs) | | `clodex clean ` | Remove the kept git worktree of a finished run (artifacts and patch stay) | @@ -176,6 +176,14 @@ approved diff hash; a local-workspace run is reported as already in the working that fails unexpectedly is marked `failed` (traceback in `error.txt`) and its worktree is removed; approved and blocked runs keep theirs until `clodex clean `. +### Exit codes and output + +`0` success; `1` the work was refused or rejected (`blocked`, `apply-refused`, ...); `2` usage or +configuration error (including unknown run ids and bad git refs, reported as one line; set +`CLODEX_DEBUG=1` for the traceback); `3` the run failed unexpectedly; `4` it was cancelled. +Without `--json`, output is plain text; `--json` is the stable machine-readable form. Commands +find the repository from any subdirectory (or `CLODEX_REPO_ROOT`). + ### Models and `clodex doctor` `clodex doctor` validates the configuration against what is actually installed: @@ -226,15 +234,34 @@ The MCP server exposes: - `clodex_handoff_get` - `clodex_handoff_decide` -The server also handles MCP-style `tasks/get`, `tasks/update`, and -`tasks/cancel` JSON-RPC methods using the Clodex `run_id` as the task id. - Start it with: ```bash python -m clodex mcp-server ``` +The server speaks MCP `2025-11-25` (and negotiates down to `2025-06-18`, `2025-03-26` or +`2024-11-05`). It works on the repository containing its working directory, or the one named by +`CLODEX_REPO_ROOT`, which is the reliable way to point a client at a project. + +- **Tool errors vs protocol errors:** wrong or missing tool arguments come back as a tool result + with `isError: true` (so the model can correct itself); an unknown tool or malformed JSON-RPC is + a protocol error. +- **Responsiveness:** requests are handled in arrival order, so a client may pipeline + `handoff_create` -> `handoff_update` -> `handoff_get`. Only long-running work (a synchronous + plan/build/audit, or waiting in `tasks/result`) runs on a worker thread, so pings, polling and + cancellation are never blocked. +- **Tasks (experimental in the spec):** `clodex_build` supports task augmentation. Call + `tools/call` with a `task` object and it returns a task immediately, backed by a durable worker; + then use `tasks/get`, `tasks/result` (blocks until done, returns the tool result), `tasks/list` + and `tasks/cancel`. The Clodex `run_id` is the task id. Run statuses map to task statuses: + `queued`/`running`/`planning`/`auditing`/`needs-fix` = `working`, `approved`/`applied` = + `completed`, `blocked`/`failed` = `failed`, `cancelled` = `cancelled`. Task ids are not + access-controlled: this is a local, single-user server, so only expose it to clients you trust. + The pre-spec `tasks/update` method is gone. +- `clodex_audit` accepts `base` (everything since that ref diverged) and `commit` (one commit) as + well as the default uncommitted diff. + ## Safety Clodex defaults to git worktree isolation, Codex `workspace-write` sandboxing, diff --git a/clodex/artifacts.py b/clodex/artifacts.py index af88303..cc8c01a 100644 --- a/clodex/artifacts.py +++ b/clodex/artifacts.py @@ -66,5 +66,37 @@ def current_diff(repo_root: Path) -> str: return result.stdout.decode("utf-8", errors="surrogateescape") +def _git(repo_root: Path, *args: str) -> subprocess.CompletedProcess[bytes]: + return subprocess.run(["git", *args], cwd=repo_root, capture_output=True, check=False) + + +def _verified_commit(repo_root: Path, ref: str) -> str: + result = _git(repo_root, "rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}") + if result.returncode != 0: + raise ValueError(f"Unknown git ref: {ref}") + return result.stdout.decode("utf-8", errors="replace").strip() + + +def audit_diff(repo_root: Path, base: str | None = None, commit: str | None = None) -> str: + """The change to audit: the uncommitted diff (default), everything since `base` diverged, or one commit.""" + if base and commit: + raise ValueError("Use either a base ref or a commit, not both") + if commit: + sha = _verified_commit(repo_root, commit) + result = _git(repo_root, "show", "--binary", "--format=", sha) + elif base: + _verified_commit(repo_root, base) + merge_base = _git(repo_root, "merge-base", base, "HEAD") + if merge_base.returncode != 0: + raise ValueError(f"No common ancestor between {base} and HEAD") + # Working tree vs the merge-base: the branch's commits plus any uncommitted edits. + result = _git(repo_root, "diff", "--binary", merge_base.stdout.decode("utf-8", errors="replace").strip()) + else: + return current_diff(repo_root) + if result.returncode != 0: + raise ValueError(f"git could not produce the diff: {result.stderr.decode('utf-8', errors='replace').strip()}") + return result.stdout.decode("utf-8", errors="surrogateescape") + + def hash_text(text: str) -> str: return hashlib.sha256(text.encode("utf-8", errors="surrogateescape")).hexdigest() diff --git a/clodex/cli.py b/clodex/cli.py index d99b352..ca4dd04 100644 --- a/clodex/cli.py +++ b/clodex/cli.py @@ -2,11 +2,12 @@ import argparse import json +import os import sys from pathlib import Path from typing import Any -from .config import ConfigError +from .config import ConfigError, resolve_repo_root from .doctor import run_doctor from .evals import run_local_evals from .hooks import hook_config, ingest_hook_event @@ -27,13 +28,16 @@ def main(argv: list[str] | None = None) -> int: try: return _main(argv) - except (ConfigError, ModelRetiredError) as exc: + except (ConfigError, ModelRetiredError, ValueError) as exc: + if os.environ.get("CLODEX_DEBUG"): + raise + # Deliberate user-facing errors (unknown run id, bad ref, bad config): one line, no traceback. print(f"clodex: error: {exc}", file=sys.stderr) - return 2 + return EXIT_USAGE def handle_migrate(args: argparse.Namespace) -> int: - contract = Path.cwd() / "CLODEX.md" + contract = resolve_repo_root() / "CLODEX.md" if not contract.is_file(): print_output({"ok": False, "error": f"{contract} not found"}, args.json) return 1 @@ -82,9 +86,12 @@ def _main(argv: list[str] | None = None) -> int: build = sub.add_parser("build", help="Run plan, implementation, and dual audit") add_build_args(build) - audit = sub.add_parser("audit", help="Audit current uncommitted changes") + audit = sub.add_parser("audit", help="Audit current uncommitted changes, everything since a base ref, or one commit") audit.add_argument("--dry-run", action="store_true") - audit.add_argument("--diff", action="store_true", help="Accepted for compatibility; audit always uses git diff") + audit_target = audit.add_mutually_exclusive_group() + audit_target.add_argument("--diff", action="store_true", help="Audit the current uncommitted diff (the default)") + audit_target.add_argument("--base", metavar="REF", help="Audit everything since this ref diverged from HEAD, including uncommitted edits") + audit_target.add_argument("--commit", metavar="SHA", help="Audit the changes introduced by one commit") run = sub.add_parser("run", help="Alias for build") add_build_args(run) @@ -150,7 +157,7 @@ def _main(argv: list[str] | None = None) -> int: if args.command == "mcp-server": return mcp_main() if args.command == "doctor": - exit_code, data = run_doctor(strict=args.strict) + exit_code, data = run_doctor(resolve_repo_root(), strict=args.strict) print_output(data, args.json) return exit_code if args.command == "init" and args.migrate: @@ -200,12 +207,13 @@ def _main(argv: list[str] | None = None) -> int: print_output(data, args.json) return exit_code - workflow = ClodexWorkflow(Path.cwd()) + repo = resolve_repo_root() + workflow = ClodexWorkflow(repo) if args.command == "plan": result = workflow.plan(" ".join(args.task), dry_run=args.dry_run) print_result(result.__dict__, args.json) - return 0 + return exit_code_for(result.status) if args.command in {"build", "run"}: result = workflow.build( " ".join(args.task), @@ -215,25 +223,25 @@ def _main(argv: list[str] | None = None) -> int: apply_changes=args.apply_changes, ) print_result(result.__dict__, args.json) - return 0 if result.status not in {"blocked"} else 1 + return exit_code_for(result.status) if args.command == "audit": - result = workflow.audit(dry_run=args.dry_run) + result = workflow.audit(dry_run=args.dry_run, base=args.base, commit=args.commit) print_result(result.__dict__, args.json) - return 0 if result.status not in {"blocked"} else 1 + return exit_code_for(result.status) if args.command == "task": return handle_task(args, workflow, args.json) if args.command == "apply": result = workflow.apply_run(args.run_id, check=args.check, force=args.force) print_result(result.__dict__, args.json) - return 0 if result.status not in {"apply-failed", "apply-check-failed", "apply-refused"} else 1 + return exit_code_for(result.status) if args.command == "clean": result = workflow.clean_run(args.run_id) print_result(result.__dict__, args.json) - return 0 if result.status != "clean-refused" else 1 + return exit_code_for(result.status) if args.command == "trace": return handle_trace(args, workflow, args.json) if args.command == "eval": - data = run_local_evals(Path.cwd()) + data = run_local_evals(repo) print_output(data, args.json) return 0 if data["passed"] else 1 if args.command == "hooks": @@ -266,7 +274,7 @@ def add_build_args(parser: argparse.ArgumentParser) -> None: def handle_task(args: argparse.Namespace, workflow: ClodexWorkflow, as_json: bool) -> int: - manager = TaskManager(Path.cwd()) + manager = TaskManager(resolve_repo_root()) if args.task_command == "start": result = manager.start(" ".join(args.task), workspace_backend=args.workspace, approval_profile=args.approval_profile, dry_run=args.dry_run) print_result(result.__dict__, as_json) @@ -289,7 +297,7 @@ def handle_task(args: argparse.Namespace, workflow: ClodexWorkflow, as_json: boo with Heartbeat(workflow.state, args.run_id): result = workflow.run_existing(args.run_id, workspace_backend=args.workspace, approval_profile=args.approval_profile) print_result(result.__dict__, as_json) - return 0 if result.status not in {"blocked"} else 1 + return exit_code_for(result.status) return 2 @@ -308,18 +316,37 @@ def handle_trace(args: argparse.Namespace, workflow: ClodexWorkflow, as_json: bo def handle_hooks(args: argparse.Namespace, as_json: bool) -> int: if args.hooks_command == "print": - print_output(hook_config(Path.cwd()), as_json) + print_output(hook_config(resolve_repo_root()), as_json) return 0 if args.hooks_command == "install": - print_output({"dry_run": True, "config": hook_config(Path.cwd())}, as_json) + print_output({"dry_run": True, "config": hook_config(resolve_repo_root())}, as_json) return 0 if args.hooks_command == "ingest": payload = json.loads(sys.stdin.read() or "{}") - print_output(ingest_hook_event(Path.cwd(), args.run_id, payload), as_json) + print_output(ingest_hook_event(resolve_repo_root(), args.run_id, payload), as_json) return 0 return 2 +# Exit codes: 0 success, 1 the work was refused or rejected (blocked, nothing applied), +# 2 usage or configuration error, 3 failed unexpectedly, 4 cancelled. +EXIT_BLOCKED = 1 +EXIT_USAGE = 2 +EXIT_FAILED = 3 +EXIT_CANCELLED = 4 +_BLOCKED_STATUSES = {"blocked", "apply-refused", "apply-failed", "apply-check-failed", "clean-refused"} + + +def exit_code_for(status: str) -> int: + if status in _BLOCKED_STATUSES: + return EXIT_BLOCKED + if status == "failed": + return EXIT_FAILED + if status == "cancelled": + return EXIT_CANCELLED + return 0 + + def print_result(data: dict[str, Any], as_json: bool) -> None: if as_json: print(json.dumps(data, indent=2, sort_keys=True)) @@ -338,5 +365,59 @@ def print_result(data: dict[str, Any], as_json: bool) -> None: def print_output(data: dict[str, Any], as_json: bool) -> None: if as_json: print(json.dumps(data, indent=2, sort_keys=True)) + elif "diagnostics" in data and "python" in data: + print("\n".join(format_doctor(data))) else: - print(json.dumps(data, indent=2, sort_keys=True)) + print("\n".join(format_human(data)) or "(nothing to show)") + + +def _scalar(value: Any) -> str: + if value is None: + return "-" + if isinstance(value, bool): + return "yes" if value else "no" + return str(value) + + +def format_human(data: Any, indent: int = 0) -> list[str]: + """Plain-text rendering of nested dicts/lists for people (machine consumers use --json).""" + pad = " " * indent + lines: list[str] = [] + if isinstance(data, dict): + for key, value in data.items(): + if isinstance(value, (dict, list)) and value: + lines.append(f"{pad}{key}:") + lines.extend(format_human(value, indent + 1)) + else: + lines.append(f"{pad}{key}: {_scalar(value) if not isinstance(value, (dict, list)) else '(none)'}") + elif isinstance(data, list): + for item in data: + if isinstance(item, dict): + head = next((str(item[key]) for key in ("id", "name", "run_id", "taskId") if key in item), "") + lines.append(f"{pad}- {head}".rstrip()) + lines.extend(format_human({k: v for k, v in item.items() if str(v) != head}, indent + 1)) + else: + lines.append(f"{pad}- {_scalar(item)}") + else: + lines.append(f"{pad}{_scalar(data)}") + return lines + + +def format_doctor(data: dict[str, Any]) -> list[str]: + def mark(ok: bool) -> str: + return "[ok] " if ok else "[FAIL]" + + lines = [f"{mark(data['python']['ok'])} python {data['python']['version']}", f"{mark(data['contract']['ok'])} CLODEX.md ({data['contract']['path']})"] + for name in ("git", "claude", "codex"): + info = data[name] + lines.append(f"{mark(info['ok'])} {name}: {info.get('version') or 'not found'}") + for name, auth in (data.get("auth") or {}).items(): + lines.append(f"{mark(auth['status'] != 'logged-out')} {name} login: {auth['status']}") + for item in data["diagnostics"]: + tag = "[FAIL]" if item["level"] == "error" else "[warn]" + lines.append(f"{tag} {item['where']}: {item['message']}") + if item.get("fix"): + lines.append(f" fix: {item['fix']}") + lines.append("") + lines.append("Everything looks good." if data["ok"] else "Problems found; see above.") + return lines diff --git a/clodex/config.py b/clodex/config.py index 0275b33..92e4ceb 100644 --- a/clodex/config.py +++ b/clodex/config.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os import sys from dataclasses import dataclass, field from pathlib import Path @@ -153,6 +154,12 @@ def tracing(self) -> dict[str, Any]: return dict(DEFAULT_CONFIG["tracing"] | self.raw.get("tracing", {})) +def resolve_repo_root() -> Path: + """The repo Clodex should work on: $CLODEX_REPO_ROOT if set (MCP clients start servers from anywhere), else the git root above the cwd.""" + override = os.environ.get("CLODEX_REPO_ROOT") + return find_repo_root(Path(override)) if override else find_repo_root() + + def find_repo_root(start: Path | None = None) -> Path: current = (start or Path.cwd()).resolve() for candidate in [current, *current.parents]: diff --git a/clodex/mcp_server.py b/clodex/mcp_server.py index cfed53d..f832e66 100644 --- a/clodex/mcp_server.py +++ b/clodex/mcp_server.py @@ -1,12 +1,18 @@ from __future__ import annotations +import base64 import json import sqlite3 import sys +import threading import uuid +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path from typing import Any from . import __version__ +from .config import resolve_repo_root +from .schemas import validate as validate_schema from .tasks import TaskManager from .workflow import ClodexWorkflow @@ -21,14 +27,31 @@ { "name": "clodex_build", "title": "Build with Clodex", - "description": "Run Claude planning, Codex implementation, and dual audit.", - "inputSchema": {"type": "object", "properties": {"task": {"type": "string"}, "dry_run": {"type": "boolean"}}, "required": ["task"]}, + "description": "Run Claude planning, Codex implementation, and dual audit. Supports task augmentation: pass `task` in tools/call params to run it as a durable MCP task.", + "inputSchema": { + "type": "object", + "properties": { + "task": {"type": "string"}, + "dry_run": {"type": "boolean"}, + "workspace": {"type": "string", "enum": ["git-worktree", "local"]}, + "approval_profile": {"type": "string", "enum": ["ci", "local", "auto_review"]}, + }, + "required": ["task"], + }, + "execution": {"taskSupport": "optional"}, }, { "name": "clodex_audit", "title": "Audit with Clodex", - "description": "Run Claude and Codex adversarial audit over current changes.", - "inputSchema": {"type": "object", "properties": {"dry_run": {"type": "boolean"}}}, + "description": "Run Claude and Codex adversarial audit over current uncommitted changes, everything since a base ref, or one commit.", + "inputSchema": { + "type": "object", + "properties": { + "dry_run": {"type": "boolean"}, + "base": {"type": "string", "description": "Audit everything since this ref diverged from HEAD"}, + "commit": {"type": "string", "description": "Audit a single commit"}, + }, + }, }, { "name": "clodex_status", @@ -128,78 +151,358 @@ } -def main() -> int: - for line in sys.stdin: - if not line.strip(): - continue +SUPPORTED_PROTOCOL_VERSIONS = ("2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05") +LATEST_PROTOCOL_VERSION = SUPPORTED_PROTOCOL_VERSIONS[0] +TASKS_MIN_VERSION = "2025-11-25" # the Tasks utility (experimental) first appears in this revision +POLL_INTERVAL_MS = 2000 +RELATED_TASK_KEY = "io.modelcontextprotocol/related-task" +IMMEDIATE_RESPONSE_KEY = "io.modelcontextprotocol/model-immediate-response" +TASK_PAGE_SIZE = 20 + +PARSE_ERROR = -32700 +INVALID_REQUEST = -32600 +METHOD_NOT_FOUND = -32601 +INVALID_PARAMS = -32602 +INTERNAL_ERROR = -32603 + +TOOL_INDEX = {tool["name"]: tool for tool in TOOLS} +# Tools that run through the durable worker and can therefore be task-augmented. +TASK_TOOLS = {name for name, tool in TOOL_INDEX.items() if (tool.get("execution") or {}).get("taskSupport") in {"optional", "required"}} +# Handled by TaskManager alone; everything else needs the workflow/state. +TASK_MANAGER_TOOLS = {"clodex_task_start", "clodex_task_get", "clodex_task_cancel"} + + +class RpcError(Exception): + def __init__(self, code: int, message: str, data: Any = None): + super().__init__(message) + self.code = code + self.message = message + self.data = data + + +class _Aborted(Exception): + """The client cancelled the request (or disconnected); send no response.""" + + +def error_response(request_id: Any, code: int, message: str, data: Any = None) -> dict[str, Any]: + error: dict[str, Any] = {"code": code, "message": message} + if data is not None: + error["data"] = data + return {"jsonrpc": "2.0", "id": request_id, "error": error} + + +def task_status(run_status: str) -> str: + """Map a Clodex run status onto the MCP task status enum.""" + if run_status in {"approved", "applied", "completed"}: + return "completed" + if run_status in {"blocked", "failed"}: + return "failed" + if run_status == "cancelled": + return "cancelled" + return "working" + + +def task_object(run: dict[str, Any]) -> dict[str, Any]: + status = task_status(str(run.get("status"))) + message = f"run {run.get('status')}" + if status == "failed": + message = str(run.get("error") or run.get("blocked_reason") or message) + return { + "taskId": run["id"], + "status": status, + "statusMessage": message, + "createdAt": run.get("created_at"), + "lastUpdatedAt": run.get("updated_at"), + "ttl": None, # runs are kept until deleted, i.e. unlimited + "pollInterval": POLL_INTERVAL_MS, + } + + +def encode_cursor(offset: int) -> str: + return base64.urlsafe_b64encode(f"o:{offset}".encode("ascii")).decode("ascii") + + +def decode_cursor(cursor: Any) -> int: + try: + text = base64.urlsafe_b64decode(str(cursor).encode("ascii")).decode("ascii") + prefix, number = text.split(":", 1) + offset = int(number) + if prefix != "o" or offset < 0: + raise ValueError + return offset + except (ValueError, UnicodeError): + raise RpcError(INVALID_PARAMS, "Invalid cursor") from None + + +def validate_arguments(tool: dict[str, Any], arguments: dict[str, Any]) -> str | None: + """A message for the model if the arguments are wrong, else None.""" + schema = tool["inputSchema"] + for key in schema.get("required", []): + if key not in arguments or arguments[key] is None: + return f"Missing required argument: {key}" + if tool["name"] in HANDOFF_TOOL_NAMES: + return None # the handoff tools report their own, more specific, type errors + errors = validate_schema(arguments, schema) + return "Invalid arguments: " + "; ".join(errors[:5]) if errors else None + + +class McpServer: + """Line-delimited JSON-RPC over stdio. + + Requests are handled in order on the reader thread; only long-running work (a synchronous + build/plan/audit, or waiting in tasks/result) moves to a worker thread, so it can never + stall pings, polling or cancellation. + """ + + def __init__(self, out: Any = None, workers: int = 8): + self.out = out or sys.stdout + self.workers = workers + self.version = LATEST_PROTOCOL_VERSION + self._write_lock = threading.Lock() + self._closing = threading.Event() + self._inflight: dict[Any, threading.Event] = {} + self._inflight_lock = threading.Lock() + + # ------------------------------------------------------------ transport + + def send(self, message: dict[str, Any]) -> None: + with self._write_lock: + self.out.write(json.dumps(message) + "\n") + self.out.flush() + + def serve(self, lines: Any) -> int: + with ThreadPoolExecutor(max_workers=self.workers, thread_name_prefix="clodex-mcp") as pool: + for line in lines: + if line.strip(): + self.dispatch_line(line, pool) + self._closing.set() # client went away: unblock anything waiting on a task + return 0 + + def dispatch_line(self, line: str, pool: ThreadPoolExecutor) -> None: + try: + message = json.loads(line) + except ValueError: + self.send(error_response(None, PARSE_ERROR, "Parse error")) + return + if isinstance(message, list): + self.send(error_response(None, INVALID_REQUEST, "Batch requests are not supported")) + return + if not isinstance(message, dict): + self.send(error_response(None, INVALID_REQUEST, "Invalid request")) + return + method = message.get("method") + if not isinstance(method, str): + if "result" in message or "error" in message: + return # a response to a request of ours; we never send any + self.send(error_response(message.get("id"), INVALID_REQUEST, "Invalid request: missing method")) + return + params = message.get("params") + if "id" not in message: + self.handle_notification(method, params) # notifications never get a response + return + request_id = message["id"] + if self.is_blocking(method, params): + pool.submit(self.run_request, request_id, method, params) + else: + # In arrival order, so a client may pipeline create -> update -> get and rely on it. + self.run_request(request_id, method, params) + + # Work that can take minutes. Everything else is quick and stays inline. + LONG_TOOLS = frozenset({"clodex_plan", "clodex_build", "clodex_audit"}) + + def is_blocking(self, method: str, params: Any) -> bool: + if method == "tasks/result": + return True # waits for the task to finish + if method != "tools/call" or not isinstance(params, dict) or params.get("name") not in self.LONG_TOOLS: + return False + arguments = params.get("arguments") + if isinstance(arguments, dict) and arguments.get("dry_run"): + return False + # A task-augmented build only starts the worker and returns immediately. + return not (isinstance(params.get("task"), dict) and self.tasks_enabled and params.get("name") in TASK_TOOLS) + + def handle_notification(self, method: str, params: Any) -> None: + if method == "notifications/cancelled" and isinstance(params, dict): + with self._inflight_lock: + event = self._inflight.get(params.get("requestId")) + if event is not None: + event.set() + + def run_request(self, request_id: Any, method: str, params: Any) -> None: + cancelled = threading.Event() + with self._inflight_lock: + self._inflight[request_id] = cancelled try: - request = json.loads(line) - response = handle_request(request) - except Exception as exc: # MCP servers must not crash on bad client input. - response = {"jsonrpc": "2.0", "id": None, "error": {"code": -32603, "message": str(exc)}} - if response is not None: - print(json.dumps(response), flush=True) - return 0 - - -def handle_request(request: dict[str, Any]) -> dict[str, Any] | None: - method = request.get("method") - request_id = request.get("id") - if method == "initialize": + try: + result = self.handle(method, params, cancelled) + response: dict[str, Any] = {"jsonrpc": "2.0", "id": request_id, "result": result} + except RpcError as exc: + response = error_response(request_id, exc.code, exc.message, exc.data) + except _Aborted: + return + except Exception as exc: # noqa: BLE001 - a bad request must never take the server down + response = error_response(request_id, INTERNAL_ERROR, f"{type(exc).__name__}: {exc}") + if not cancelled.is_set(): + self.send(response) + finally: + with self._inflight_lock: + self._inflight.pop(request_id, None) + + # ------------------------------------------------------------ methods + + @property + def tasks_enabled(self) -> bool: + return self.version >= TASKS_MIN_VERSION + + def handle(self, method: str, params: Any, cancelled: threading.Event) -> dict[str, Any]: + if params is not None and not isinstance(params, dict): + raise RpcError(INVALID_PARAMS, "params must be an object") + params = params or {} + if method == "initialize": + return self.initialize(params) + if method == "ping": + return {} + if method == "tools/list": + return {"tools": TOOLS} + if method == "tools/call": + return self.tools_call(params) + if self.tasks_enabled: + if method == "tasks/get": + return task_object(self.task_run(params)) + if method == "tasks/result": + return self.tasks_result(params, cancelled) + if method == "tasks/list": + return self.tasks_list(params) + if method == "tasks/cancel": + return self.tasks_cancel(params) + raise RpcError(METHOD_NOT_FOUND, f"Method not found: {method}") + + def initialize(self, params: dict[str, Any]) -> dict[str, Any]: + requested = params.get("protocolVersion") + if not isinstance(requested, str): + raise RpcError(INVALID_PARAMS, "initialize requires a protocolVersion") + # Echo the client's version if we speak it, else answer with our newest. + self.version = requested if requested in SUPPORTED_PROTOCOL_VERSIONS else LATEST_PROTOCOL_VERSION + capabilities: dict[str, Any] = {"tools": {"listChanged": False}} + if self.tasks_enabled: + capabilities["tasks"] = {"list": {}, "cancel": {}, "requests": {"tools": {"call": {}}}} return { - "jsonrpc": "2.0", - "id": request_id, - "result": { - "protocolVersion": "2025-11-25", - "capabilities": {"tools": {"listChanged": False}, "tasks": {}}, - "serverInfo": {"name": "clodex-mcp-server", "version": __version__}, - }, + "protocolVersion": self.version, + "capabilities": capabilities, + "serverInfo": {"name": "clodex-mcp-server", "title": "Clodex", "version": __version__}, } - if method == "notifications/initialized": - return None - if method == "tools/list": - return {"jsonrpc": "2.0", "id": request_id, "result": {"tools": TOOLS}} - if method == "tools/call": - params = request.get("params") or {} + + # ------------------------------------------------------------ tools + + def tools_call(self, params: dict[str, Any]) -> dict[str, Any]: name = params.get("name") - arguments = params.get("arguments") or {} - return {"jsonrpc": "2.0", "id": request_id, "result": tool_call(name, arguments)} - if method == "tasks/get": - params = request.get("params") or {} - run_id = str(params.get("id") or params.get("taskId") or "") - data = TaskManager().get(run_id) - if data is None: - return {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32004, "message": f"Unknown task: {run_id}"}} - return {"jsonrpc": "2.0", "id": request_id, "result": task_result(data)} - if method == "tasks/cancel": - params = request.get("params") or {} - run_id = str(params.get("id") or params.get("taskId") or "") - try: - result = TaskManager().cancel(run_id) - except ValueError as exc: - return {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32004, "message": str(exc)}} - return {"jsonrpc": "2.0", "id": request_id, "result": task_result({"run": result.__dict__})} - if method == "tasks/update": - params = request.get("params") or {} - run_id = str(params.get("id") or params.get("taskId") or "") - data = TaskManager().get(run_id) - if data is None: - return {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32004, "message": f"Unknown task: {run_id}"}} - return {"jsonrpc": "2.0", "id": request_id, "result": task_result(data)} - return {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32601, "message": f"Method not found: {method}"}} + if not isinstance(name, str): + raise RpcError(INVALID_PARAMS, "tools/call requires a tool name") + tool = TOOL_INDEX.get(name) + if tool is None: + raise RpcError(INVALID_PARAMS, f"Unknown tool: {name}") + arguments = params.get("arguments") + if arguments is None: + arguments = {} + if not isinstance(arguments, dict): + return call_text("Arguments must be an object", is_error=True) + problem = validate_arguments(tool, arguments) + if problem: + return call_text(problem, is_error=True) # tool-level error so the model can correct itself + if isinstance(params.get("task"), dict) and self.tasks_enabled: + if name not in TASK_TOOLS: + raise RpcError(METHOD_NOT_FOUND, f"Tool does not support task augmentation: {name}") + return self.create_task(name, arguments) + return tool_call(name, arguments) + + def create_task(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]: + if arguments.get("dry_run"): + raise RpcError(INVALID_PARAMS, "dry_run is not supported for task-augmented calls") + manager = TaskManager(resolve_repo_root()) + started = manager.start( + str(arguments["task"]), + workspace_backend=arguments.get("workspace"), + approval_profile=arguments.get("approval_profile"), + ) + run = manager.state.get_run(started.run_id) + return { + "task": task_object(run), + "_meta": {IMMEDIATE_RESPONSE_KEY: f"Clodex build started as task {started.run_id}. Poll tasks/get; fetch the outcome with tasks/result."}, + } + # ------------------------------------------------------------ tasks + + def task_run(self, params: dict[str, Any]) -> dict[str, Any]: + task_id = params.get("taskId", params.get("id")) # `id` is the pre-spec spelling + if not isinstance(task_id, str) or not task_id: + raise RpcError(INVALID_PARAMS, "taskId is required") + data = TaskManager(resolve_repo_root()).get(task_id) + # Only runs started through the worker are tasks (they are the ones with a process). + if data is None or not data["run"].get("pid"): + raise RpcError(INVALID_PARAMS, "Failed to retrieve task: Task not found") + return data["run"] + + def tasks_result(self, params: dict[str, Any], cancelled: threading.Event) -> dict[str, Any]: + run = self.task_run(params) + manager = TaskManager(resolve_repo_root()) + while task_status(str(run["status"])) == "working": + if cancelled.wait(0.5) or self._closing.is_set(): + raise _Aborted + data = manager.get(str(run["id"])) + if data is None: + raise RpcError(INVALID_PARAMS, "Failed to retrieve task: Task not found") + run = data["run"] + summary: dict[str, Any] = {key: run.get(key) for key in ("id", "status", "diff_hash", "error", "blocked_reason", "artifacts_dir", "workspace_path")} + agreement = Path(str(run.get("artifacts_dir") or "")) / "05-agreement.json" + if agreement.is_file(): + try: + summary["agreement"] = json.loads(agreement.read_text(encoding="utf-8")) + except (OSError, ValueError): + pass + result = call_json(summary, is_error=task_status(str(run["status"])) != "completed") + result["_meta"] = {RELATED_TASK_KEY: {"taskId": run["id"]}} + return result + + def tasks_list(self, params: dict[str, Any]) -> dict[str, Any]: + offset = decode_cursor(params["cursor"]) if params.get("cursor") is not None else 0 + runs = TaskManager(resolve_repo_root()).state.list_runs(limit=TASK_PAGE_SIZE + 1, offset=offset, with_worker=True) + page = runs[:TASK_PAGE_SIZE] + result: dict[str, Any] = {"tasks": [task_object(run) for run in page]} + if len(runs) > TASK_PAGE_SIZE: + result["nextCursor"] = encode_cursor(offset + TASK_PAGE_SIZE) + return result + + def tasks_cancel(self, params: dict[str, Any]) -> dict[str, Any]: + run = self.task_run(params) + status = task_status(str(run["status"])) + if status != "working": + raise RpcError(INVALID_PARAMS, f"Cannot cancel task: already in terminal status '{status}'") + manager = TaskManager(resolve_repo_root()) + manager.cancel(str(run["id"])) + return task_object(manager.state.get_run(str(run["id"])) or run) -def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]: - if name in HANDOFF_TOOL_NAMES and not isinstance(arguments, dict): - return call_text("Arguments must be an object", is_error=True) - workflow = ClodexWorkflow() +def main() -> int: + return McpServer().serve(sys.stdin) + + +def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]: + root = resolve_repo_root() + workflow = None if name in TASK_MANAGER_TOOLS else ClodexWorkflow(root) if name == "clodex_plan": result = workflow.plan(str(arguments["task"]), dry_run=bool(arguments.get("dry_run", False))) elif name == "clodex_build": - result = workflow.build(str(arguments["task"]), dry_run=bool(arguments.get("dry_run", False))) + result = workflow.build( + str(arguments["task"]), + dry_run=bool(arguments.get("dry_run", False)), + workspace_backend=arguments.get("workspace"), + approval_profile=arguments.get("approval_profile"), + ) elif name == "clodex_audit": - result = workflow.audit(dry_run=bool(arguments.get("dry_run", False))) + try: + result = workflow.audit(dry_run=bool(arguments.get("dry_run", False)), base=arguments.get("base"), commit=arguments.get("commit")) + except ValueError as exc: # bad ref etc.: let the model correct itself + return call_text(str(exc), is_error=True) elif name == "clodex_status": data = {"tasks": workflow.state.list_tasks(), "runs": workflow.state.list_runs()} return {"content": [{"type": "text", "text": json.dumps(data, indent=2)}], "isError": False} @@ -210,20 +513,23 @@ def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]: workflow.state.update_task(str(arguments["id"]), str(arguments["status"])) return {"content": [{"type": "text", "text": "task updated"}], "isError": False} elif name == "clodex_task_start": - result = TaskManager().start( + result = TaskManager(root).start( str(arguments["task"]), workspace_backend=arguments.get("workspace"), approval_profile=arguments.get("approval_profile"), dry_run=bool(arguments.get("dry_run", False)), ) - return {"content": [{"type": "text", "text": json.dumps(task_result({"run": result.__dict__}), indent=2)}], "isError": False} + return {"content": [{"type": "text", "text": json.dumps({"id": result.run_id, "status": result.status, "result": {"run": result.__dict__}}, indent=2)}], "isError": False} elif name == "clodex_task_get": - data = TaskManager().get(str(arguments["run_id"])) + data = TaskManager(root).get(str(arguments["run_id"])) if data is None: return {"content": [{"type": "text", "text": f"Unknown run: {arguments['run_id']}"}], "isError": True} return {"content": [{"type": "text", "text": json.dumps(data, indent=2)}], "isError": False} elif name == "clodex_task_cancel": - result = TaskManager().cancel(str(arguments["run_id"])) + try: + result = TaskManager(root).cancel(str(arguments["run_id"])) + except ValueError as exc: + return call_text(str(exc), is_error=True) return {"content": [{"type": "text", "text": json.dumps(result.__dict__, indent=2)}], "isError": False} elif name == "clodex_handoff_create": try: @@ -390,16 +696,5 @@ def expected_handoff_error(exc: Exception) -> str: return str(exc) -def task_result(data: dict[str, Any]) -> dict[str, Any]: - run = data.get("run") or {} - run_id = run.get("id") or run.get("run_id") - status = run.get("status", "unknown") - return { - "id": run_id, - "status": status, - "result": data, - } - - if __name__ == "__main__": raise SystemExit(main()) diff --git a/clodex/state.py b/clodex/state.py index daca4e0..b5f417c 100644 --- a/clodex/state.py +++ b/clodex/state.py @@ -568,6 +568,9 @@ def list_tasks(self) -> list[dict[str, Any]]: with self.session() as con: return [dict(row) for row in con.execute("select * from tasks order by updated_at desc")] - def list_runs(self) -> list[dict[str, Any]]: + def list_runs(self, limit: int = 20, offset: int = 0, with_worker: bool = False) -> list[dict[str, Any]]: + """Newest first. `with_worker` keeps only runs started through a worker process (the MCP "tasks").""" + where = "where pid is not null" if with_worker else "" with self.session() as con: - return [dict(row) for row in con.execute("select * from runs order by updated_at desc limit 20")] + rows = con.execute(f"select * from runs {where} order by updated_at desc, id desc limit ? offset ?", (limit, offset)) + return [dict(row) for row in rows] diff --git a/clodex/workflow.py b/clodex/workflow.py index 049812b..0b3f20e 100644 --- a/clodex/workflow.py +++ b/clodex/workflow.py @@ -9,7 +9,7 @@ from typing import Any from .agents import AgentRunner -from .artifacts import ArtifactStore, current_diff, hash_text, make_run_id, make_task_id +from .artifacts import ArtifactStore, audit_diff, current_diff, hash_text, make_run_id, make_task_id from .commands import claude_audit_command, claude_plan_command, codex_exec_command, codex_review_command from .config import ClodexConfig, load_config from .jsonutil import AgentEnvelopeError, extract_json_object @@ -126,11 +126,23 @@ def run_existing( return WorkflowResult("failed", run_id, str(run["task_id"]), str(artifacts.path), {"error": str(exc)}) return self._execute_build(str(run["prompt"]), str(run["task_id"]), run_id, artifacts, workspace_backend, approval_profile, apply_changes) - def audit(self, dry_run: bool = False) -> WorkflowResult: + def audit(self, dry_run: bool = False, base: str | None = None, commit: str | None = None) -> WorkflowResult: + """Audit the uncommitted diff, everything since `base`, or a single `commit`.""" if dry_run: return WorkflowResult("dry-run", None, None, None, {"commands": self.dry_run_commands()}) ensure_usable(self.config) - task = "Audit current uncommitted changes" + + def diff_fn() -> str: + return audit_diff(self.repo_root, base=base, commit=commit) + + if not diff_fn().strip(): + return WorkflowResult("nothing-to-audit", None, None, None, {"message": "There are no changes to audit."}) + if commit: + task = f"Audit commit {commit}" + elif base: + task = f"Audit changes since {base}" + else: + task = "Audit current uncommitted changes" task_id = make_task_id(task) run_id = make_run_id(task_id) artifacts = ArtifactStore(self.config, run_id, self.state) @@ -144,7 +156,10 @@ def audit(self, dry_run: bool = False) -> WorkflowResult: task_id, artifacts, trace, - lambda: self._audit_loop(task_id, run_id, artifacts, plan_json, AgentRunner(self.repo_root), self.repo_root, trace, None), + lambda: self._audit_loop( + task_id, run_id, artifacts, plan_json, AgentRunner(self.repo_root), self.repo_root, trace, None, + diff_fn=diff_fn, allow_fixes=commit is None, # a historical commit cannot be fixed in place + ), ) def apply_run(self, run_id: str | None, check: bool = False, force: bool = False) -> WorkflowResult: @@ -269,6 +284,8 @@ def _audit_loop( trace: TraceWriter, approval_profile: str | None, include_untracked: bool = False, + diff_fn: Callable[[], str] | None = None, + allow_fixes: bool = True, ) -> WorkflowResult: agreement: dict[str, Any] = {} block_error: str | None = None @@ -276,7 +293,7 @@ def _audit_loop( for attempt in range(self.config.max_fix_loops + 1): if include_untracked: self._include_untracked(diff_root) - diff = current_diff(diff_root) + diff = diff_fn() if diff_fn else current_diff(diff_root) diff_hash = hash_text(diff) artifacts.write_text("changes.diff", diff, exact=True) verdicts = self._run_reviewers(artifacts, plan_json, diff, diff_hash, runner, trace, attempt) @@ -293,7 +310,7 @@ def _audit_loop( # A reviewer that could not run is an infrastructure problem, not something Codex can fix. block_error = "Required reviewer failed: " + "; ".join(f"{v['reviewer_id']}: {v['_error']}" for v in failed) break - if attempt >= self.config.max_fix_loops: + if attempt >= self.config.max_fix_loops or not allow_fixes: break findings = self._required_fixes(*verdicts, diff_hash=diff_hash) fix = runner.run(codex_exec_command(self.config, diff_root, approval_profile=approval_profile), fix_prompt(plan_json, findings)) diff --git a/pyproject.toml b/pyproject.toml index edd7a8c..55fb048 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,10 @@ requires-python = ">=3.12" dependencies = ["pyyaml>=6"] classifiers = ["License :: Other/Proprietary License"] +[project.optional-dependencies] +# The official MCP SDK: enables tests that validate wire messages against its schema models. +test = ["mcp>=2"] + [project.scripts] clodex = "clodex.cli:main" clodex-mcp-server = "clodex.mcp_server:main" diff --git a/tests/test_cli_behavior.py b/tests/test_cli_behavior.py new file mode 100644 index 0000000..b5b5d5b --- /dev/null +++ b/tests/test_cli_behavior.py @@ -0,0 +1,231 @@ +"""Tests: CLI behavior (audit targets, exit codes, errors, repo root, human output).""" + +from __future__ import annotations + +import contextlib +import io +import json +import os +import subprocess +import sys +import unittest +from pathlib import Path +from unittest import mock + +from clodex import cli +from clodex.cli import exit_code_for, format_human +from clodex.workflow import ClodexWorkflow +from tests.support import ROOT, FakeCliPath, TempRepo + + +def git(repo: Path, *args: str) -> str: + return subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True, check=True).stdout.strip() + + +def run_main(repo: Path, *args: str, env: dict | None = None) -> tuple[int, str, str]: + """Run clodex.cli.main in-process against `repo`; returns (exit code, stdout, stderr).""" + out, err = io.StringIO(), io.StringIO() + with mock.patch.dict(os.environ, {"CLODEX_REPO_ROOT": str(repo), **(env or {})}), contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + try: + code = cli.main(list(args)) + except SystemExit as exc: # argparse usage errors + code = int(exc.code) + return code, out.getvalue(), err.getvalue() + + +def run_cli(repo: Path, *args: str, cwd: Path | None = None, env: dict | None = None) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, "-m", "clodex", *args], + cwd=cwd or repo, + env={**os.environ, "PYTHONPATH": str(ROOT), **(env or {})}, + capture_output=True, + text=True, + stdin=subprocess.DEVNULL, + ) + + +class ExitCodeTests(unittest.TestCase): + def test_status_to_exit_code(self): + for status in ("approved", "planned", "applied", "apply-check", "dry-run", "queued", "cleaned", "nothing-to-clean", "nothing-to-audit"): + self.assertEqual(exit_code_for(status), 0, status) + for status in ("blocked", "apply-refused", "apply-failed", "apply-check-failed", "clean-refused"): + self.assertEqual(exit_code_for(status), 1, status) + self.assertEqual(exit_code_for("failed"), 3) + self.assertEqual(exit_code_for("cancelled"), 4) + + def test_build_exit_codes_through_main(self): + with TempRepo() as repo, FakeCliPath(): + code, out, _ = run_main(repo, "--json", "build", "--workspace", "local", "implement fixture") + self.assertEqual((code, json.loads(out)["status"]), (0, "approved")) + with TempRepo() as repo, FakeCliPath(reject_reviewers=("codex-architecture",)): + (repo / "CLODEX.md").write_text("---\nmax_fix_loops: 0\n---\nbody\n", encoding="utf-8") + code, out, _ = run_main(repo, "--json", "build", "--workspace", "local", "x") + self.assertEqual((code, json.loads(out)["status"]), (1, "blocked")) + with TempRepo() as repo, FakeCliPath(): + with mock.patch.object(ClodexWorkflow, "_run_json_with_retry", side_effect=RuntimeError("boom")): + code, out, _ = run_main(repo, "--json", "plan", "x") + self.assertEqual((code, json.loads(out)["status"]), (3, "failed")) + + +class ErrorHandlingTests(unittest.TestCase): + def test_unknown_run_ids_are_one_line_errors_not_tracebacks(self): + with TempRepo() as repo: + for args in (("apply", "nope"), ("clean", "nope"), ("task", "cancel", "nope")): + result = run_cli(repo, *args) + self.assertEqual(result.returncode, 2, args) + self.assertIn("clodex: error: Unknown run: nope", result.stderr) + self.assertNotIn("Traceback", result.stderr) + + def test_debug_env_restores_the_traceback(self): + with TempRepo() as repo: + result = run_cli(repo, "apply", "nope", env={"CLODEX_DEBUG": "1"}) + self.assertNotEqual(result.returncode, 0) + self.assertIn("Traceback", result.stderr) + + def test_usage_errors_exit_2(self): + with TempRepo() as repo: + self.assertEqual(run_cli(repo, "audit", "--base", "x", "--commit", "y").returncode, 2) + self.assertEqual(run_cli(repo, "nonsense").returncode, 2) + + +class AuditTargetTests(unittest.TestCase): + def commit_change(self, repo: Path, name: str = "feature.txt", text: str = "feature\n") -> str: + (repo / name).write_text(text, encoding="utf-8") + git(repo, "add", name) + git(repo, "commit", "-m", f"add {name}") + return git(repo, "rev-parse", "HEAD") + + def test_nothing_to_audit_creates_no_run(self): + with TempRepo() as repo, FakeCliPath(): + code, out, _ = run_main(repo, "--json", "audit") + self.assertEqual((code, json.loads(out)["status"]), (0, "nothing-to-audit")) + self.assertEqual(ClodexWorkflow(repo).state.list_runs(), []) + + def test_uncommitted_diff_is_the_default_and_diff_flag_is_accepted(self): + with TempRepo() as repo, FakeCliPath(): + (repo / "seed.txt").write_text("changed\n", encoding="utf-8") + for flag in ([], ["--diff"]): + code, out, _ = run_main(repo, "--json", "audit", *flag) + self.assertEqual((code, json.loads(out)["status"]), (0, "approved"), flag) + + def test_audit_a_single_commit(self): + with TempRepo() as repo, FakeCliPath(): + sha = self.commit_change(repo) + result = ClodexWorkflow(repo).audit(commit=sha) + self.assertEqual(result.status, "approved") + diff = (Path(result.artifacts_dir) / "changes.diff").read_text(encoding="utf-8") + self.assertIn("feature.txt", diff) + self.assertNotIn("seed.txt", diff) + + def test_audit_of_a_commit_never_runs_a_fix_attempt(self): + with TempRepo() as repo, FakeCliPath(reject_reviewers=("codex-architecture",)): + (repo / "CLODEX.md").write_text("---\nmax_fix_loops: 2\n---\nbody\n", encoding="utf-8") + git(repo, "add", "CLODEX.md") + git(repo, "commit", "-m", "allow fixes") + sha = self.commit_change(repo) + result = ClodexWorkflow(repo).audit(commit=sha) + self.assertEqual(result.status, "blocked") + self.assertFalse(list(Path(result.artifacts_dir).glob("fix-attempt-*"))) + + def test_audit_everything_since_a_base_ref_including_uncommitted_edits(self): + with TempRepo() as repo, FakeCliPath(): + base = git(repo, "rev-parse", "HEAD") + self.commit_change(repo, "one.txt") + (repo / "two.txt").write_text("uncommitted\n", encoding="utf-8") + git(repo, "add", "two.txt") + result = ClodexWorkflow(repo).audit(base=base) + self.assertEqual(result.status, "approved") + diff = (Path(result.artifacts_dir) / "changes.diff").read_text(encoding="utf-8") + self.assertIn("one.txt", diff) + self.assertIn("two.txt", diff) + + def test_bad_refs_and_conflicting_targets_are_value_errors(self): + with TempRepo() as repo, FakeCliPath(): + workflow = ClodexWorkflow(repo) + with self.assertRaisesRegex(ValueError, "Unknown git ref: nope"): + workflow.audit(base="nope") + with self.assertRaisesRegex(ValueError, "Unknown git ref: nope"): + workflow.audit(commit="nope") + with self.assertRaisesRegex(ValueError, "either a base ref or a commit"): + workflow.audit(base="HEAD", commit="HEAD") + result = run_cli(repo, "audit", "--commit", "nope") + self.assertEqual(result.returncode, 2) + self.assertIn("Unknown git ref", result.stderr) + + +class RepoRootTests(unittest.TestCase): + def test_commands_from_a_subdirectory_use_the_repo_root(self): + with TempRepo() as repo: + sub = repo / "pkg" / "deep" + sub.mkdir(parents=True) + result = run_cli(repo, "--json", "status", cwd=sub) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue((repo / ".clodex" / "state.sqlite3").is_file()) + self.assertFalse((sub / ".clodex").exists()) + + def test_repo_root_env_overrides_the_working_directory(self): + with TempRepo() as repo, TempRepo() as elsewhere: + result = run_cli(repo, "--json", "status", cwd=elsewhere, env={"CLODEX_REPO_ROOT": str(repo)}) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue((repo / ".clodex" / "state.sqlite3").is_file()) + self.assertFalse((elsewhere / ".clodex").exists()) + + +class HumanOutputTests(unittest.TestCase): + def test_format_human_renders_nested_data_readably(self): + lines = format_human({"tasks": [{"id": "t1", "status": "done"}], "ok": True, "note": None, "empty": [], "deep": {"a": 1}}) + text = "\n".join(lines) + self.assertIn("tasks:\n - t1\n status: done", text) + self.assertIn("ok: yes", text) + self.assertIn("note: -", text) + self.assertIn("empty: (none)", text) + self.assertIn("deep:\n a: 1", text) + self.assertNotIn("{", text) + + def test_non_json_output_is_text_and_json_flag_still_json(self): + with TempRepo() as repo, FakeCliPath(): + plain = run_cli(repo, "status") + self.assertEqual(plain.returncode, 0, plain.stderr) + with self.assertRaises(ValueError): + json.loads(plain.stdout) + self.assertIn("tasks: (none)", plain.stdout) + self.assertIn("runs: (none)", plain.stdout) + self.assertIsInstance(json.loads(run_cli(repo, "--json", "status").stdout), dict) + + def test_doctor_human_output_lists_checks_and_fixes(self): + with TempRepo() as repo, FakeCliPath(): + (repo / "CLODEX.md").write_text("---\ncodex:\n model: gpt-5.4\n---\nbody\n", encoding="utf-8") + result = run_cli(repo, "doctor") + self.assertEqual(result.returncode, 1) + self.assertIn("[ok] python", result.stdout) + self.assertIn("[ok] claude:", result.stdout) + self.assertIn("[FAIL] codex.model", result.stdout) + self.assertIn("fix:", result.stdout) + self.assertIn("Problems found", result.stdout) + with TempRepo() as repo, FakeCliPath(): + result = run_cli(repo, "doctor") + self.assertEqual(result.returncode, 0, result.stdout) + self.assertIn("Everything looks good.", result.stdout) + + +class McpAuditToolTests(unittest.TestCase): + def test_audit_tool_takes_base_and_commit_and_reports_bad_refs_as_tool_errors(self): + from clodex import mcp_server + + with TempRepo() as repo, FakeCliPath(), mock.patch.dict(os.environ, {"CLODEX_REPO_ROOT": str(repo)}): + properties = mcp_server.TOOL_INDEX["clodex_audit"]["inputSchema"]["properties"] + self.assertIn("base", properties) + self.assertIn("commit", properties) + bad = mcp_server.tool_call("clodex_audit", {"commit": "nope"}) + self.assertTrue(bad["isError"]) + self.assertIn("Unknown git ref", bad["content"][0]["text"]) + (repo / "new.txt").write_text("x\n", encoding="utf-8") + git(repo, "add", "new.txt") + git(repo, "commit", "-m", "new") + good = mcp_server.tool_call("clodex_audit", {"commit": "HEAD"}) + self.assertFalse(good["isError"]) + self.assertIn("approved", good["content"][0]["text"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_conformance.py b/tests/test_mcp_conformance.py new file mode 100644 index 0000000..8305ff8 --- /dev/null +++ b/tests/test_mcp_conformance.py @@ -0,0 +1,391 @@ +"""Tests: MCP protocol conformance (lifecycle, errors, tools, spec Tasks utility).""" + +from __future__ import annotations + +import json +import os +import queue +import threading +import time +import unittest +import warnings +from pathlib import Path +from unittest import mock + +from clodex import __version__ +from clodex import mcp_server +from clodex.mcp_server import RELATED_TASK_KEY, McpServer, task_status +from clodex.tasks import TaskManager +from tests.support import FakeCliPath, TempRepo + + +class Sink: + """Collects the server's output, one JSON message per line.""" + + def __init__(self): + self.messages: list[dict] = [] + self.cond = threading.Condition() + + def write(self, text: str) -> None: + for line in text.splitlines(): + if line.strip(): + with self.cond: + self.messages.append(json.loads(line)) + self.cond.notify_all() + + def flush(self) -> None: + pass + + def wait_for(self, request_id, timeout: float = 30.0) -> dict | None: + deadline = time.monotonic() + timeout + with self.cond: + while True: + for message in self.messages: + if message.get("id") == request_id: + return message + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + self.cond.wait(remaining) + + +class LiveServer: + """An McpServer running on a thread, fed through a queue so tests can interleave input and output.""" + + def __init__(self, protocol: str = "2025-11-25", initialize: bool = True): + self.sink = Sink() + self.feed: queue.Queue = queue.Queue() + self.server = McpServer(out=self.sink) + self.thread = threading.Thread(target=self.server.serve, args=(iter(self.feed.get, None),), daemon=True) + self.thread.start() + self._next_id = 100 + if initialize: + response = self.call("initialize", {"protocolVersion": protocol, "capabilities": {}, "clientInfo": {"name": "t", "version": "1"}}) + assert "result" in response, response + + def send_raw(self, text: str) -> None: + self.feed.put(text) + + def send(self, method: str, params=None, request_id=None, notification: bool = False) -> int | None: + message: dict = {"jsonrpc": "2.0", "method": method} + if params is not None: + message["params"] = params + if not notification: + request_id = self._next_id if request_id is None else request_id + self._next_id = max(self._next_id, request_id) + 1 + message["id"] = request_id + self.feed.put(json.dumps(message)) + return request_id + + def call(self, method: str, params=None, timeout: float = 30.0) -> dict: + request_id = self.send(method, params) + response = self.sink.wait_for(request_id, timeout) + assert response is not None, f"no response to {method}" + return response + + def tool(self, name: str, arguments=None, **extra) -> dict: + params = {"name": name, "arguments": arguments or {}, **extra} + return self.call("tools/call", params) + + def silence(self, seconds: float = 0.4) -> list[dict]: + before = len(self.sink.messages) + time.sleep(seconds) + return self.sink.messages[before:] + + def close(self, timeout: float = 30.0) -> bool: + self.feed.put(None) + self.thread.join(timeout) + return not self.thread.is_alive() + + +class ServerCase(unittest.TestCase): + def setUp(self): + # Task workers are detached processes whose Popen handles are never reaped here. + warning_context = warnings.catch_warnings() + warning_context.__enter__() + self.addCleanup(warning_context.__exit__, None, None, None) + warnings.simplefilter("ignore", ResourceWarning) + self.repo_cm = TempRepo() + self.repo = self.repo_cm.__enter__() + self.addCleanup(self.repo_cm.__exit__, None, None, None) + env = mock.patch.dict(os.environ, {"CLODEX_REPO_ROOT": str(self.repo)}) + env.start() + self.addCleanup(env.stop) + self.servers: list[LiveServer] = [] + self.addCleanup(self.cleanup_servers) + + def start(self, **kwargs) -> LiveServer: + live = LiveServer(**kwargs) + self.servers.append(live) + return live + + def cleanup_servers(self): + for live in self.servers: + live.close(timeout=60) + manager = TaskManager(self.repo) + for run in manager.state.list_runs(limit=100): + if str(run["status"]) in {"queued", "running", "planning", "auditing", "needs-fix", "cancel_requested", "handoff"}: + try: + manager.cancel(str(run["id"])) + except Exception: # noqa: BLE001 - best-effort cleanup + pass + + +class LifecycleTests(ServerCase): + def test_version_negotiation_and_capabilities(self): + live = self.start(initialize=False) + result = live.call("initialize", {"protocolVersion": "2025-11-25"})["result"] + self.assertEqual(result["protocolVersion"], "2025-11-25") + self.assertEqual(result["capabilities"]["tasks"], {"list": {}, "cancel": {}, "requests": {"tools": {"call": {}}}}) + self.assertEqual(result["serverInfo"]["version"], __version__) + old = self.start(initialize=False).call("initialize", {"protocolVersion": "2025-06-18"})["result"] + self.assertEqual(old["protocolVersion"], "2025-06-18") + self.assertNotIn("tasks", old["capabilities"], "tasks only exist from 2025-11-25") + unknown = self.start(initialize=False).call("initialize", {"protocolVersion": "1999-01-01"})["result"] + self.assertEqual(unknown["protocolVersion"], "2025-11-25", "answer with our newest") + missing = self.start(initialize=False).call("initialize", {}) + self.assertEqual(missing["error"]["code"], -32602) + + def test_ping_and_unknown_method_keep_the_request_id(self): + live = self.start() + self.assertEqual(live.call("ping")["result"], {}) + response = live.call("nope/nothing") + self.assertEqual((response["id"], response["error"]["code"]), (live._next_id - 1, -32601)) + + def test_malformed_input_gets_spec_errors(self): + live = self.start() + live.send_raw("{not json") + live.send_raw("[1, 2]") + live.send_raw('"just a string"') + live.send_raw(json.dumps({"jsonrpc": "2.0", "id": 77})) + deadline = time.monotonic() + 10 + while len(live.sink.messages) < 5 and time.monotonic() < deadline: # +1 for initialize + time.sleep(0.05) + codes = [(m.get("id"), m["error"]["code"]) for m in live.sink.messages if "error" in m] + self.assertEqual(codes, [(None, -32700), (None, -32600), (None, -32600), (77, -32600)]) + + def test_notifications_and_stray_responses_get_no_reply(self): + live = self.start() + live.send("notifications/initialized", notification=True) + live.send("made/up", {"x": 1}, notification=True) + live.send("notifications/cancelled", {"requestId": 12345}, notification=True) + live.send_raw(json.dumps({"jsonrpc": "2.0", "id": 5, "result": {}})) # a reply to a request we never sent + self.assertEqual(live.silence(), []) + + def test_params_must_be_an_object_and_errors_keep_ids(self): + live = self.start() + response = live.call("tools/call", "oops") + self.assertEqual(response["error"]["code"], -32602) + self.assertEqual(live.call("tools/call", {"arguments": {}})["error"]["code"], -32602) + + def test_unexpected_exceptions_become_internal_errors_with_the_id(self): + live = self.start() + with mock.patch.object(mcp_server, "tool_call", side_effect=RuntimeError("boom")): + response = live.tool("clodex_status") + self.assertEqual(response["error"]["code"], -32603) + self.assertIn("boom", response["error"]["message"]) + self.assertEqual(live.call("ping")["result"], {}, "the server survives") + + +class ToolTests(ServerCase): + def test_tool_list_marks_only_build_as_task_capable(self): + live = self.start() + tools = {tool["name"]: tool for tool in live.call("tools/list")["result"]["tools"]} + self.assertEqual(tools["clodex_build"]["execution"], {"taskSupport": "optional"}) + self.assertEqual([name for name, tool in tools.items() if "execution" in tool], ["clodex_build"]) + self.assertIn("workspace", tools["clodex_build"]["inputSchema"]["properties"]) + self.assertIn("approval_profile", tools["clodex_build"]["inputSchema"]["properties"]) + + def test_unknown_tool_is_a_protocol_error_and_touches_no_state(self): + live = self.start() + response = live.tool("clodex_nope") + self.assertEqual(response["error"]["code"], -32602) + self.assertFalse((self.repo / ".clodex" / "state.sqlite3").exists()) + + def test_bad_arguments_are_tool_errors_the_model_can_fix(self): + live = self.start() + for name, arguments, expected in ( + ("clodex_build", {}, "Missing required argument: task"), + ("clodex_plan", {"task": None}, "Missing required argument: task"), + ("clodex_build", {"task": "x", "dry_run": "yes"}, "Invalid arguments"), + ("clodex_build", {"task": "x", "workspace": "docker"}, "Invalid arguments"), + ("clodex_task_get", {}, "Missing required argument: run_id"), + ): + result = live.tool(name, arguments)["result"] + self.assertTrue(result["isError"], name) + self.assertIn(expected, result["content"][0]["text"]) + not_object = live.call("tools/call", {"name": "clodex_status", "arguments": [1]})["result"] + self.assertTrue(not_object["isError"]) + + def test_repo_root_comes_from_the_environment_not_the_cwd(self): + live = self.start() + result = live.tool("clodex_status")["result"] + self.assertFalse(result["isError"]) + self.assertTrue((self.repo / ".clodex" / "state.sqlite3").is_file()) + + def test_dry_run_build_accepts_workspace_and_profile(self): + live = self.start() + result = live.tool("clodex_build", {"task": "x", "dry_run": True, "workspace": "local", "approval_profile": "auto_review"})["result"] + self.assertFalse(result["isError"]) + self.assertIn('"status": "dry-run"', result["content"][0]["text"]) + + +class OrderingAndConcurrencyTests(ServerCase): + def test_pipelined_state_changes_are_applied_in_order(self): + live = self.start() + live.send("tools/call", {"name": "clodex_handoff_create", "arguments": {"run_id": "h1", "task": "t"}}, request_id=1) + live.send("tools/call", {"name": "clodex_handoff_update", "arguments": {"run_id": "h1", "phase": "audit", "actor": "claude"}}, request_id=2) + live.send("tools/call", {"name": "clodex_handoff_get", "arguments": {"run_id": "h1"}}, request_id=3) + got = live.sink.wait_for(3) + self.assertFalse(got["result"]["isError"], got) + self.assertEqual(json.loads(got["result"]["content"][0]["text"])["run"]["phase"], "audit") + order = [m["id"] for m in live.sink.messages if m.get("id") in {1, 2, 3}] + self.assertEqual(order, [1, 2, 3]) + + def test_a_long_synchronous_build_does_not_block_other_requests(self): + with FakeCliPath(sleep_seconds=3): + live = self.start() + build_id = live.send("tools/call", {"name": "clodex_build", "arguments": {"task": "slow", "workspace": "local"}}) + time.sleep(0.3) + started = time.monotonic() + self.assertEqual(live.call("ping", timeout=5)["result"], {}) + self.assertLess(time.monotonic() - started, 2.5, "ping must not wait for the build") + self.assertIsNone(live.sink.wait_for(build_id, 0.01), "the build is still running") + self.assertIsNotNone(live.sink.wait_for(build_id, 120), "the build eventually answers") + + +class TaskTests(ServerCase): + def start_task(self, live: LiveServer, **arguments) -> dict: + args = {"task": "implement fixture", "workspace": "local", **arguments} + response = live.call("tools/call", {"name": "clodex_build", "arguments": args, "task": {"ttl": 60000}}) + self.assertIn("result", response, response) + return response["result"] + + def test_task_augmented_build_returns_a_task_immediately(self): + with FakeCliPath(): + live = self.start() + result = self.start_task(live) + task = result["task"] + self.assertEqual(task["status"], "working") + self.assertIsInstance(task["taskId"], str) + self.assertIsNone(task["ttl"]) + self.assertEqual(task["pollInterval"], mcp_server.POLL_INTERVAL_MS) + for key in ("createdAt", "lastUpdatedAt", "statusMessage"): + self.assertTrue(task[key]) + self.assertIn(mcp_server.IMMEDIATE_RESPONSE_KEY, result["_meta"]) + fetched = live.call("tasks/get", {"taskId": task["taskId"]})["result"] + self.assertEqual(fetched["taskId"], task["taskId"]) + self.assertIn(fetched["status"], {"working", "completed"}) + + def test_tasks_result_blocks_until_done_then_returns_the_tool_result(self): + with FakeCliPath(): + live = self.start() + task_id = self.start_task(live)["task"]["taskId"] + response = live.call("tasks/result", {"taskId": task_id}, timeout=90) + result = response["result"] + self.assertFalse(result["isError"], result) + self.assertEqual(result["_meta"][RELATED_TASK_KEY], {"taskId": task_id}) + summary = json.loads(result["content"][0]["text"]) + self.assertEqual(summary["status"], "approved") + self.assertTrue(summary["agreement"]["approved"]) + self.assertEqual(live.call("tasks/get", {"taskId": task_id})["result"]["status"], "completed") + + def test_failed_run_is_a_failed_task_with_an_error_result(self): + with FakeCliPath(reject_reviewers=("codex-architecture",)): + (self.repo / "CLODEX.md").write_text("---\nmax_fix_loops: 0\n---\nbody\n", encoding="utf-8") + live = self.start() + task_id = self.start_task(live)["task"]["taskId"] + result = live.call("tasks/result", {"taskId": task_id}, timeout=90)["result"] + self.assertTrue(result["isError"]) + task = live.call("tasks/get", {"taskId": task_id})["result"] + self.assertEqual(task["status"], "failed") + + def test_unknown_and_non_worker_tasks_are_invalid_params(self): + live = self.start() + for method in ("tasks/get", "tasks/result", "tasks/cancel"): + for params in ({"taskId": "missing"}, {}, {"taskId": 5}): + self.assertEqual(live.call(method, params)["error"]["code"], -32602, (method, params)) + TaskManager(self.repo).state.upsert_task("t", "t", "done") + TaskManager(self.repo).state.create_run("sync-run", "t", "p", "approved") # no worker pid: not a task + self.assertEqual(live.call("tasks/get", {"taskId": "sync-run"})["error"]["code"], -32602) + + def test_tasks_list_paginates_with_opaque_cursors(self): + manager = TaskManager(self.repo) + manager.state.upsert_task("t", "t", "done") + for index in range(23): + manager.state.create_run(f"run-{index:02d}", "t", "p", "approved", pid=1) + live = self.start() + first = live.call("tasks/list")["result"] + self.assertEqual(len(first["tasks"]), 20) + self.assertIn("nextCursor", first) + second = live.call("tasks/list", {"cursor": first["nextCursor"]})["result"] + self.assertEqual(len(second["tasks"]), 3) + self.assertNotIn("nextCursor", second) + ids = [t["taskId"] for t in first["tasks"] + second["tasks"]] + self.assertEqual(len(set(ids)), 23) + self.assertEqual(live.call("tasks/list", {"cursor": "garbage"})["error"]["code"], -32602) + self.assertEqual(live.call("tasks/get", {"taskId": ids[0]})["result"]["taskId"], ids[0], "listed tasks are gettable") + + def test_cancel_moves_a_working_task_to_cancelled_and_rejects_repeats(self): + with FakeCliPath(sleep_seconds=60): + live = self.start() + task_id = self.start_task(live)["task"]["taskId"] + cancelled = live.call("tasks/cancel", {"taskId": task_id}, timeout=60)["result"] + self.assertEqual((cancelled["taskId"], cancelled["status"]), (task_id, "cancelled")) + again = live.call("tasks/cancel", {"taskId": task_id}) + self.assertEqual(again["error"]["code"], -32602) + self.assertIn("terminal status 'cancelled'", again["error"]["message"]) + result = live.call("tasks/result", {"taskId": task_id}, timeout=10)["result"] + self.assertTrue(result["isError"]) + self.assertEqual(json.loads(result["content"][0]["text"])["status"], "cancelled") + + def test_cancelled_notification_aborts_a_waiting_tasks_result_without_a_response(self): + with FakeCliPath(sleep_seconds=60): + live = self.start() + task_id = self.start_task(live)["task"]["taskId"] + waiting = live.send("tasks/result", {"taskId": task_id}) + self.assertEqual(live.call("ping")["result"], {}, "waiting must not block the loop") + live.send("notifications/cancelled", {"requestId": waiting}, notification=True) + self.assertIsNone(live.sink.wait_for(waiting, 2.0), "a cancelled request gets no response") + + def test_closing_the_connection_releases_a_waiting_tasks_result(self): + with FakeCliPath(sleep_seconds=60): + live = self.start() + task_id = self.start_task(live)["task"]["taskId"] + live.send("tasks/result", {"taskId": task_id}) + time.sleep(0.5) + started = time.monotonic() + self.assertTrue(live.close(timeout=15), "serve() must return after EOF") + self.assertLess(time.monotonic() - started, 10) + + def test_task_augmentation_rules(self): + with FakeCliPath(): + live = self.start() + not_supported = live.call("tools/call", {"name": "clodex_plan", "arguments": {"task": "x"}, "task": {}}) + self.assertEqual(not_supported["error"]["code"], -32601) + dry = live.call("tools/call", {"name": "clodex_build", "arguments": {"task": "x", "dry_run": True}, "task": {}}) + self.assertEqual(dry["error"]["code"], -32602) + self.assertEqual(live.call("tools/call", {"name": "clodex_build", "arguments": {}, "task": {}})["result"]["isError"], True) + + def test_older_protocol_versions_have_no_tasks(self): + live = self.start(protocol="2025-06-18") + self.assertEqual(live.call("tasks/get", {"taskId": "x"})["error"]["code"], -32601) + self.assertEqual(live.call("tasks/list")["error"]["code"], -32601) + ignored = live.call("tools/call", {"name": "clodex_build", "arguments": {"task": "x", "dry_run": True}, "task": {"ttl": 1}}) + self.assertFalse(ignored["result"]["isError"], "task metadata is ignored when the capability was not negotiated") + + +class StatusMappingTests(unittest.TestCase): + def test_run_status_to_task_status(self): + expected = { + "queued": "working", "running": "working", "planning": "working", "auditing": "working", + "needs-fix": "working", "cancel_requested": "working", "handoff": "working", + "approved": "completed", "applied": "completed", "completed": "completed", + "blocked": "failed", "failed": "failed", "cancelled": "cancelled", + } + for run_status, wanted in expected.items(): + self.assertEqual(task_status(run_status), wanted, run_status) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_sdk.py b/tests/test_mcp_sdk.py new file mode 100644 index 0000000..3f544ba --- /dev/null +++ b/tests/test_mcp_sdk.py @@ -0,0 +1,65 @@ +"""Tests: validate the server's wire messages against the official MCP SDK's 2025-11-25 schema models. + +Skipped unless the `mcp` package (which ships `mcp_types`) is installed: `pip install mcp`. +These catch a different class of bug than the hand-written conformance tests: a field the +spec requires, a wrong type, or an unexpected extra property. +""" + +from __future__ import annotations + +import unittest + +from tests.support import FakeCliPath +from tests.test_mcp_conformance import ServerCase + +try: + from mcp_types import _v2025_11_25 as wire +except ImportError: # pragma: no cover - depends on the environment + wire = None + + +def check(testcase: unittest.TestCase, model_name: str, payload: dict) -> None: + model = getattr(wire, model_name) + try: + model.model_validate(payload) + except Exception as exc: # noqa: BLE001 - surface any pydantic error with the payload + testcase.fail(f"{model_name} rejected {payload!r}: {exc}") + + +@unittest.skipIf(wire is None, "the `mcp` SDK is not installed") +class WireSchemaTests(ServerCase): + def test_handshake_and_tool_messages_match_the_sdk_schema(self): + live = self.start(initialize=False) + initialize = live.call("initialize", {"protocolVersion": "2025-11-25", "capabilities": {}, "clientInfo": {"name": "t", "version": "1"}}) + check(self, "InitializeResult", initialize["result"]) + check(self, "ListToolsResult", live.call("tools/list")["result"]) + check(self, "CallToolResult", live.tool("clodex_status")["result"]) + check(self, "CallToolResult", live.tool("clodex_build", {})["result"]) # a tool-level error is still a CallToolResult + check(self, "CallToolResult", live.tool("clodex_build", {"task": "x", "dry_run": True})["result"]) + + def test_task_messages_match_the_sdk_schema(self): + with FakeCliPath(): + live = self.start() + created = live.call("tools/call", {"name": "clodex_build", "arguments": {"task": "implement fixture", "workspace": "local"}, "task": {"ttl": 60000}})["result"] + check(self, "CreateTaskResult", created) + task_id = created["task"]["taskId"] + check(self, "GetTaskResult", live.call("tasks/get", {"taskId": task_id})["result"]) + check(self, "ListTasksResult", live.call("tasks/list")["result"]) + check(self, "CallToolResult", live.call("tasks/result", {"taskId": task_id}, timeout=90)["result"]) + check(self, "GetTaskResult", live.call("tasks/get", {"taskId": task_id})["result"]) + + def test_cancelled_task_matches_the_sdk_schema(self): + with FakeCliPath(sleep_seconds=60): + live = self.start() + created = live.call("tools/call", {"name": "clodex_build", "arguments": {"task": "slow", "workspace": "local"}, "task": {}})["result"] + check(self, "CancelTaskResult", live.call("tasks/cancel", {"taskId": created["task"]["taskId"]}, timeout=60)["result"]) + + def test_error_responses_match_the_jsonrpc_schema(self): + live = self.start() + for method, params in (("nope/x", None), ("tasks/get", {"taskId": "missing"}), ("tools/call", {"name": "clodex_nope"})): + response = live.call(method, params) + check(self, "JSONRPCErrorResponse", response) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 512cf2c..b2a79a3 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -54,7 +54,7 @@ def test_mcp_tools_list_includes_handoff_tools(self): self.assertIn("clodex_handoff_decide", names) def test_mcp_tasks_get_unknown_run(self): - request = {"jsonrpc": "2.0", "id": 1, "method": "tasks/get", "params": {"id": "missing"}} + request = {"jsonrpc": "2.0", "id": 1, "method": "tasks/get", "params": {"taskId": "missing"}} result = subprocess.run( [sys.executable, "-m", "clodex", "mcp-server"], input=json.dumps(request) + "\n", @@ -67,7 +67,7 @@ def test_mcp_tasks_get_unknown_run(self): ) self.assertEqual(result.returncode, 0, result.stderr) response = json.loads(result.stdout.splitlines()[0]) - self.assertEqual(response["error"]["code"], -32004) + self.assertEqual(response["error"]["code"], -32602) # spec: invalid or unknown taskId def test_mcp_handoff_create_update_get_and_decide(self): with tempfile.TemporaryDirectory() as tmp: