feat: add semantic contract diff

Makefile

@@ -73,7 +73,7 @@ api-import: ## Import an API snapshot
 	$(BIN)/proxmox-api-contract import $(ARGS)
 
 api-diff: ## Compare API snapshots
-	@echo "API diff is scheduled for milestone B5" >&2; exit 2
+	$(BIN)/proxmox-api-contract diff $(ARGS)
 
 seed: ## Seed simulation data
 	@echo "Database seed is scheduled for milestone D2" >&2; exit 2

README.md

@@ -40,6 +40,15 @@ reject private address resolution, unsafe redirects, oversized responses, and
 unbounded retries. Imported revisions are addressed by their normalized
 snapshot checksum and are never overwritten.
 
+Normalized snapshots can be compared in text, JSON, Markdown, or HTML. The diff
+command exits with status 1 when it finds a breaking change, making it suitable
+for CI policy checks:
+
+```bash
+.venv/bin/proxmox-api-contract diff old-snapshot.json new-snapshot.json \
+  --format markdown
+```
+
 See [the architecture](docs/architecture.md) for component boundaries and
 durability decisions. Commands for not-yet-implemented milestones intentionally
 return a non-zero status instead of pretending to succeed.

app/contracts/cli.py

@@ -8,7 +8,16 @@ import json
 from datetime import UTC, datetime
 from pathlib import Path
 
+from app.contracts.diff import (
+    compare_snapshots,
+    has_breaking_changes,
+    render_html,
+    render_json,
+    render_markdown,
+    render_text,
+)
 from app.contracts.importer import RemoteSourceImporter
+from app.contracts.model import Snapshot
 from app.contracts.normalize import normalize_snapshot
 from app.contracts.source import ApiViewerParser, LocalFileImporter, SourceImporter
 from app.contracts.store import RevisionStore
@@ -28,6 +37,10 @@ def parser() -> argparse.ArgumentParser:
     commands.add_parser("list")
     show = commands.add_parser("show")
     show.add_argument("revision")
+    diff = commands.add_parser("diff")
+    diff.add_argument("before", type=Path)
+    diff.add_argument("after", type=Path)
+    diff.add_argument("--format", choices=("text", "json", "markdown", "html"), default="text")
     return root
 
 
@@ -40,6 +53,18 @@ async def run(arguments: argparse.Namespace) -> int:
     if arguments.command == "show":
         print(json.dumps(store.manifest(arguments.revision).model_dump(mode="json"), indent=2))
         return 0
+    if arguments.command == "diff":
+        before = Snapshot.model_validate_json(arguments.before.read_bytes())
+        after = Snapshot.model_validate_json(arguments.after.read_bytes())
+        changes = compare_snapshots(before, after)
+        renderers = {
+            "text": render_text,
+            "json": render_json,
+            "markdown": render_markdown,
+            "html": render_html,
+        }
+        print(renderers[arguments.format](changes))
+        return 1 if has_breaking_changes(changes) else 0
     if arguments.command == "validate":
         parsed = ApiViewerParser().parse(arguments.file.read_bytes())
         print(json.dumps({"nodes": len(parsed.nodes), "warnings": len(parsed.warnings)}))

app/contracts/diff.py

@@ -0,0 +1,224 @@
+"""Deterministic semantic differences between normalized snapshots."""
+
+from __future__ import annotations
+
+import html
+import json
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Any
+
+from app.contracts.model import Method, Parameter, PathContract, Snapshot
+
+
+class Severity(StrEnum):
+    BREAKING = "breaking"
+    NON_BREAKING = "non-breaking"
+    DOCUMENTATION = "documentation"
+
+
+@dataclass(frozen=True, slots=True, order=True)
+class Change:
+    path: str
+    method: str
+    category: str
+    severity: Severity
+    detail: str
+    before: str | None = None
+    after: str | None = None
+
+
+def _stable(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def _methods(snapshot: Snapshot) -> dict[tuple[str, str], Method]:
+    return {(path.path, method.verb): method for path in snapshot.paths for method in path.methods}
+
+
+def _paths(snapshot: Snapshot) -> dict[str, PathContract]:
+    return {path.path: path for path in snapshot.paths}
+
+
+def compare_snapshots(before: Snapshot, after: Snapshot) -> tuple[Change, ...]:
+    changes: list[Change] = []
+    old_paths, new_paths = _paths(before), _paths(after)
+    for path in sorted(old_paths.keys() - new_paths.keys()):
+        changes.append(Change(path, "", "path", Severity.BREAKING, "path removed"))
+    for path in sorted(new_paths.keys() - old_paths.keys()):
+        changes.append(Change(path, "", "path", Severity.NON_BREAKING, "path added"))
+
+    old_methods, new_methods = _methods(before), _methods(after)
+    for path, verb in sorted(old_methods.keys() - new_methods.keys()):
+        changes.append(Change(path, verb, "method", Severity.BREAKING, "method removed"))
+    for path, verb in sorted(new_methods.keys() - old_methods.keys()):
+        changes.append(Change(path, verb, "method", Severity.NON_BREAKING, "method added"))
+    for key in sorted(old_methods.keys() & new_methods.keys()):
+        _compare_method(key, old_methods[key], new_methods[key], changes)
+    return tuple(sorted(changes))
+
+
+def _compare_method(
+    key: tuple[str, str], before: Method, after: Method, changes: list[Change]
+) -> None:
+    path, verb = key
+    if before.description != after.description:
+        changes.append(
+            Change(
+                path,
+                verb,
+                "documentation",
+                Severity.DOCUMENTATION,
+                "description changed",
+                before.description,
+                after.description,
+            )
+        )
+    if before.permissions != after.permissions:
+        changes.append(
+            Change(
+                path,
+                verb,
+                "permissions",
+                Severity.BREAKING,
+                "permissions changed",
+                _stable(before.permissions.model_dump(mode="json") if before.permissions else None),
+                _stable(after.permissions.model_dump(mode="json") if after.permissions else None),
+            )
+        )
+    _compare_parameters(path, verb, before.parameters, after.parameters, changes)
+    _compare_schema(
+        path,
+        verb,
+        "returns",
+        before.returns.model_dump(mode="json"),
+        after.returns.model_dump(mode="json"),
+        changes,
+    )
+
+
+def _compare_parameters(
+    path: str,
+    verb: str,
+    before: tuple[Parameter, ...],
+    after: tuple[Parameter, ...],
+    changes: list[Change],
+) -> None:
+    old = {parameter.name: parameter for parameter in before}
+    new = {parameter.name: parameter for parameter in after}
+    for name in sorted(old.keys() - new.keys()):
+        changes.append(Change(path, verb, "parameter", Severity.BREAKING, f"removed: {name}"))
+    for name in sorted(new.keys() - old.keys()):
+        severity = Severity.NON_BREAKING if new[name].definition.optional else Severity.BREAKING
+        changes.append(Change(path, verb, "parameter", severity, f"added: {name}"))
+    for name in sorted(old.keys() & new.keys()):
+        _compare_schema(
+            path,
+            verb,
+            f"parameter:{name}",
+            old[name].definition.model_dump(mode="json"),
+            new[name].definition.model_dump(mode="json"),
+            changes,
+        )
+
+
+def _compare_schema(
+    path: str,
+    verb: str,
+    label: str,
+    before: dict[str, Any],
+    after: dict[str, Any],
+    changes: list[Change],
+) -> None:
+    groups = {
+        "schema": {"type", "properties", "items", "enum", "format", "pattern"},
+        "default": {"default", "optional"},
+        "constraint": {"minimum", "maximum", "min_length", "max_length"},
+        "documentation": {"description"},
+    }
+    for category, fields in groups.items():
+        old = {field: before.get(field) for field in fields}
+        new = {field: after.get(field) for field in fields}
+        if old != new:
+            severity = Severity.DOCUMENTATION if category == "documentation" else Severity.BREAKING
+            changes.append(
+                Change(
+                    path,
+                    verb,
+                    category,
+                    severity,
+                    f"{label} {category} changed",
+                    _stable(old),
+                    _stable(new),
+                )
+            )
+
+
+def render_json(changes: tuple[Change, ...]) -> str:
+    return json.dumps(
+        [
+            {
+                "after": change.after,
+                "before": change.before,
+                "category": change.category,
+                "detail": change.detail,
+                "method": change.method,
+                "path": change.path,
+                "severity": change.severity,
+            }
+            for change in changes
+        ],
+        ensure_ascii=False,
+        indent=2,
+    )
+
+
+def render_text(changes: tuple[Change, ...]) -> str:
+    return "\n".join(
+        (
+            f"{change.severity}: {change.method} {change.path} [{change.category}] {change.detail}"
+        ).strip()
+        for change in changes
+    )
+
+
+def render_markdown(changes: tuple[Change, ...]) -> str:
+    lines = [
+        "# API contract diff",
+        "",
+        "| Severity | Method | Path | Category | Detail |",
+        "|---|---|---|---|---|",
+    ]
+    lines.extend(
+        (
+            f"| {change.severity} | {change.method} | `{change.path}` | "
+            f"{change.category} | {change.detail} |"
+        )
+        for change in changes
+    )
+    return "\n".join(lines)
+
+
+def render_html(changes: tuple[Change, ...]) -> str:
+    rows = "".join(
+        "<tr>"
+        + "".join(
+            f"<td>{html.escape(str(value))}</td>"
+            for value in (
+                change.severity,
+                change.method,
+                change.path,
+                change.category,
+                change.detail,
+            )
+        )
+        + "</tr>"
+        for change in changes
+    )
+    return (
+        f"<!doctype html><meta charset=utf-8><title>API contract diff</title><table>{rows}</table>"
+    )
+
+
+def has_breaking_changes(changes: tuple[Change, ...]) -> bool:
+    return any(change.severity is Severity.BREAKING for change in changes)

tests/unit/test_contract_diff.py

@@ -0,0 +1,87 @@
+"""Semantic contract diff classification and rendering tests."""
+
+import json
+from datetime import UTC, datetime
+
+from app.contracts.diff import (
+    Severity,
+    compare_snapshots,
+    has_breaking_changes,
+    render_html,
+    render_json,
+    render_markdown,
+    render_text,
+)
+from app.contracts.model import Method, PathContract, Schema, Snapshot
+
+
+def snapshot(paths: tuple[PathContract, ...]) -> Snapshot:
+    return Snapshot(
+        source_version="test",
+        retrieved_at=datetime(2026, 1, 1, tzinfo=UTC),
+        raw_sha256="0" * 64,
+        paths=paths,
+        path_count=len(paths),
+        method_count=sum(len(path.methods) for path in paths),
+    )
+
+
+def method(description: str = "old", returns: Schema | None = None) -> Method:
+    return Method(
+        verb="GET",
+        name="read",
+        description=description,
+        returns=returns or Schema(type="string"),
+        checksum="1" * 64,
+    )
+
+
+def test_classifies_added_removed_and_changed_contracts() -> None:
+    before = snapshot(
+        (
+            PathContract(path="/removed", methods=(method(),)),
+            PathContract(path="/version", methods=(method(),)),
+        )
+    )
+    after = snapshot(
+        (
+            PathContract(path="/added", methods=(method(),)),
+            PathContract(
+                path="/version",
+                methods=(method("new", Schema(type="integer", minimum=1)),),
+            ),
+        )
+    )
+
+    changes = compare_snapshots(before, after)
+
+    assert changes == tuple(sorted(changes))
+    assert {change.category for change in changes} >= {
+        "path",
+        "method",
+        "documentation",
+        "schema",
+        "constraint",
+    }
+    assert has_breaking_changes(changes)
+    assert any(change.severity is Severity.NON_BREAKING for change in changes)
+
+
+def test_renderers_are_stable_and_escape_html() -> None:
+    before = snapshot((PathContract(path="/<old>", methods=(method(),)),))
+    after = snapshot(())
+    changes = compare_snapshots(before, after)
+
+    assert render_text(changes).startswith("breaking:")
+    assert "| breaking |" in render_markdown(changes)
+    assert "&lt;old&gt;" in render_html(changes)
+    decoded = json.loads(render_json(changes))
+    assert decoded[0]["severity"] == "breaking"
+    assert render_json(changes) == render_json(changes)
+
+
+def test_no_changes_has_clean_ci_policy() -> None:
+    value = snapshot((PathContract(path="/version", methods=(method(),)),))
+
+    assert compare_snapshots(value, value) == ()
+    assert not has_breaking_changes(())