from __future__ import annotations
import argparse
import dataclasses
import datetime as dt
import difflib
import hashlib
import json
import os
import shutil
import subprocess
import sys
import textwrap
from pathlib import Path
from typing import Callable, Iterable
REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_TEST_ENV = Path("/home/nsabaj/Code/test-env").resolve()
DEFAULT_MODEL = "minimax-m2.7:cloud"
QA_MARKER = ".mermaid-qa-root"
IGNORED_PARTS = {
".git",
".mermaid",
".pytest_cache",
"__pycache__",
".qa-runs",
"node_modules",
"dist",
}
class QaFailure(RuntimeError):
pass
@dataclasses.dataclass
class CommandResult:
args: list[str]
cwd: str
returncode: int
stdout: str
stderr: str
def to_json(self) -> dict:
return dataclasses.asdict(self)
@dataclasses.dataclass
class ScenarioResult:
name: str
ok: bool
duration_seconds: float
workspace: str | None = None
mermaid_returncode: int | None = None
mermaid_errors: list[str] = dataclasses.field(default_factory=list)
checks: list[str] = dataclasses.field(default_factory=list)
failure: str | None = None
def to_json(self) -> dict:
return dataclasses.asdict(self)
@dataclasses.dataclass(frozen=True)
class Scenario:
name: str
tier: str
fixture: str | None
prompt: str | None
checker: Callable[["QaContext", Path, dict, CommandResult | None], list[str]]
allow_mermaid_errors: bool = False
@dataclasses.dataclass
class QaContext:
args: argparse.Namespace
run_dir: Path
workspaces_dir: Path
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run Mermaid self-QA scenarios")
parser.add_argument(
"--test-env",
default=os.environ.get("MERMAID_QA_TEST_ENV", str(DEFAULT_TEST_ENV)),
help="Fixture root to wipe and recreate",
)
parser.add_argument(
"--model",
default=os.environ.get("MERMAID_QA_MODEL", DEFAULT_MODEL),
help="Model passed to `mermaid --model`",
)
parser.add_argument(
"--mermaid-bin",
default=str(REPO_ROOT / "target" / "debug" / "mermaid"),
help="Mermaid binary to test",
)
parser.add_argument(
"--scenario",
action="append",
choices=sorted(SCENARIOS.keys()) if "SCENARIOS" in globals() else None,
help="Scenario to run; may be repeated",
)
parser.add_argument(
"--tier",
choices=["fast", "real", "all"],
default="fast",
help="Scenario tier to run when --scenario is omitted",
)
parser.add_argument(
"--timeout",
type=int,
default=25 * 60,
help="Per Mermaid run timeout in seconds",
)
parser.add_argument(
"--skip-build",
action="store_true",
help="Do not run `cargo build --bin mermaid` before QA",
)
parser.add_argument(
"--keep-workspace",
action="store_true",
help="Do not restore scenario workspaces after checks",
)
parser.add_argument(
"--wipe-runs",
action="store_true",
help="Also delete prior .qa-runs logs when resetting the QA root",
)
parser.add_argument(
"--json",
action="store_true",
help="Print only the final JSON report path and status",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
test_env = Path(args.test_env).expanduser().resolve()
mermaid_bin = Path(args.mermaid_bin).expanduser().resolve()
args.test_env = str(test_env)
args.mermaid_bin = str(mermaid_bin)
started = dt.datetime.now(dt.UTC)
run_id = started.strftime("%Y%m%d_%H%M%S")
try:
guard_test_env(test_env)
reset_test_env(test_env, wipe_runs=args.wipe_runs)
except Exception as exc: failure = {
"ok": False,
"started_at": started.isoformat(),
"finished_at": dt.datetime.now(dt.UTC).isoformat(),
"model": args.model,
"mermaid_bin": str(mermaid_bin),
"test_env": str(test_env),
"fatal_error": str(exc),
"results": [],
}
if args.json:
print(json.dumps(failure, sort_keys=True))
else:
print("Mermaid self-QA")
print(f"Fatal: {exc}")
print()
print("FAIL")
return 1
run_dir = test_env / ".qa-runs" / run_id
run_dir.mkdir(parents=True, exist_ok=True)
workspaces_dir = test_env / "workspaces"
workspaces_dir.mkdir(parents=True, exist_ok=True)
ctx = QaContext(args=args, run_dir=run_dir, workspaces_dir=workspaces_dir)
report = {
"ok": False,
"started_at": started.isoformat(),
"model": args.model,
"mermaid_bin": str(mermaid_bin),
"test_env": str(test_env),
"run_dir": str(run_dir),
"results": [],
}
try:
if not args.skip_build:
build_result = run_command(
["cargo", "build", "--bin", "mermaid"],
cwd=REPO_ROOT,
timeout=10 * 60,
)
write_command(run_dir / "cargo_build.json", build_result)
if build_result.returncode != 0:
raise QaFailure("cargo build --bin mermaid failed")
selected = args.scenario or scenarios_for_tier(args.tier)
results = [run_scenario(ctx, SCENARIOS[name]) for name in selected]
report["results"] = [result.to_json() for result in results]
report["ok"] = all(result.ok for result in results)
except Exception as exc: report["fatal_error"] = str(exc)
report["finished_at"] = dt.datetime.now(dt.UTC).isoformat()
report_path = run_dir / "report.json"
report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
if args.json:
print(json.dumps({"ok": report["ok"], "report": str(report_path)}, sort_keys=True))
else:
print_summary(report, report_path)
return 0 if report["ok"] else 1
def guard_test_env(test_env: Path) -> None:
home = Path.home().resolve()
forbidden = {
Path("/").resolve(),
home,
REPO_ROOT.resolve(),
REPO_ROOT.parent.resolve(),
}
if not str(test_env):
raise QaFailure("test-env path is empty")
if test_env in forbidden:
raise QaFailure(f"refusing to use dangerous test-env path: {test_env}")
if test_env != DEFAULT_TEST_ENV:
marker = test_env / QA_MARKER
if not marker.exists():
raise QaFailure(
f"custom test-env {test_env} is not initialized; "
f"create {marker} first if this path is intentional"
)
if test_env.parent != DEFAULT_TEST_ENV.parent and not (test_env / QA_MARKER).exists():
raise QaFailure(f"refusing unmarked QA root outside {DEFAULT_TEST_ENV.parent}: {test_env}")
def reset_test_env(test_env: Path, wipe_runs: bool) -> None:
test_env.mkdir(parents=True, exist_ok=True)
for child in list(test_env.iterdir()):
if child.name == ".qa-runs" and not wipe_runs:
continue
remove_path(child)
(test_env / QA_MARKER).write_text("Mermaid QA root. Safe to reset by scripts/qa_mermaid.py.\n")
(test_env / ".qa-runs").mkdir(exist_ok=True)
def remove_path(path: Path) -> None:
if path.is_dir() and not path.is_symlink():
shutil.rmtree(path)
else:
path.unlink(missing_ok=True)
def run_scenario(ctx: QaContext, scenario: Scenario) -> ScenarioResult:
start = dt.datetime.now()
scenario_dir = ctx.run_dir / scenario.name
scenario_dir.mkdir(parents=True, exist_ok=True)
workspace = ctx.workspaces_dir / scenario.name
snapshot = ctx.run_dir / "snapshots" / scenario.name
try:
if scenario.fixture is not None:
create_fixture(workspace, scenario.fixture)
copy_tree(workspace, snapshot)
before = manifest(workspace) if workspace.exists() else {}
mermaid_result = None
mermaid_json: dict = {}
if scenario.prompt is not None:
prompt = scenario.prompt.strip()
(scenario_dir / "prompt.txt").write_text(prompt + "\n")
mermaid_result = run_mermaid(ctx, workspace, prompt)
write_command(scenario_dir / "mermaid_command.json", mermaid_result)
(scenario_dir / "stdout.txt").write_text(mermaid_result.stdout)
(scenario_dir / "stderr.txt").write_text(mermaid_result.stderr)
mermaid_json = parse_mermaid_json(mermaid_result.stdout)
(scenario_dir / "mermaid.json").write_text(
json.dumps(mermaid_json, indent=2, sort_keys=True) + "\n"
)
errors = list(mermaid_json.get("errors") or [])
if mermaid_result.returncode != 0 and not scenario.allow_mermaid_errors:
raise QaFailure(f"mermaid exited {mermaid_result.returncode}: {errors}")
if errors and not scenario.allow_mermaid_errors:
raise QaFailure(f"mermaid reported tool errors: {errors}")
if not str(mermaid_json.get("response") or "").strip():
raise QaFailure("mermaid returned an empty response")
checks = scenario.checker(ctx, workspace, before, mermaid_result)
after = manifest(workspace) if workspace.exists() else {}
write_manifest(scenario_dir / "before_manifest.json", before)
write_manifest(scenario_dir / "after_manifest.json", after)
write_diff(scenario_dir / "diff.patch", snapshot, workspace)
result = ScenarioResult(
name=scenario.name,
ok=True,
duration_seconds=elapsed_seconds(start),
workspace=str(workspace) if workspace.exists() else None,
mermaid_returncode=mermaid_result.returncode if mermaid_result else None,
mermaid_errors=list(mermaid_json.get("errors") or []),
checks=checks,
)
except Exception as exc: result = ScenarioResult(
name=scenario.name,
ok=False,
duration_seconds=elapsed_seconds(start),
workspace=str(workspace) if workspace.exists() else None,
failure=str(exc),
)
finally:
if workspace.exists() and snapshot.exists() and not ctx.args.keep_workspace:
restore_tree(snapshot, workspace)
return result
def run_mermaid(ctx: QaContext, workspace: Path, prompt: str) -> CommandResult:
return run_command(
[
ctx.args.mermaid_bin,
"--model",
ctx.args.model,
"--path",
str(workspace),
"run",
"-f",
"json",
prompt,
],
cwd=REPO_ROOT,
timeout=ctx.args.timeout,
)
def create_fixture(workspace: Path, fixture: str) -> None:
remove_path(workspace) if workspace.exists() else None
workspace.mkdir(parents=True, exist_ok=True)
files = fixture_files(fixture)
for rel, content in files.items():
path = workspace / rel
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(textwrap.dedent(content).lstrip())
def fixture_files(fixture: str) -> dict[str, str]:
common = {
"README.md": """
# Mermaid QA Arithmetic Fixture
This is a tiny Python project used to test Mermaid.
Run checks with:
```bash
python3 -m unittest discover -s tests
```
""",
"MERMAID.md": """
# Mermaid QA Fixture
- This is a disposable QA project.
- Use `python3 -m unittest discover -s tests` for validation.
- Do not edit tests unless the user explicitly asks.
- Keep implementation changes small and explain the command result.
""",
"src/__init__.py": "",
"tests/__init__.py": "",
}
if fixture == "clean":
common.update(
{
"src/calculator.py": """
def add(a, b):
return a + b
def subtract(a, b):
return a - b
""",
"tests/test_calculator.py": """
import unittest
from src.calculator import add, subtract
class CalculatorTests(unittest.TestCase):
def test_add(self):
self.assertEqual(add(2, 3), 5)
def test_subtract(self):
self.assertEqual(subtract(7, 4), 3)
if __name__ == "__main__":
unittest.main()
""",
}
)
elif fixture == "buggy_add":
common.update(
{
"src/calculator.py": """
def add(a, b):
return a - b
def subtract(a, b):
return a - b
""",
"tests/test_calculator.py": """
import unittest
from src.calculator import add, subtract
class CalculatorTests(unittest.TestCase):
def test_add(self):
self.assertEqual(add(2, 3), 5)
self.assertEqual(add(-1, 1), 0)
def test_subtract(self):
self.assertEqual(subtract(7, 4), 3)
if __name__ == "__main__":
unittest.main()
""",
}
)
elif fixture == "missing_multiply":
common.update(
{
"src/calculator.py": """
def add(a, b):
return a + b
def subtract(a, b):
return a - b
""",
"tests/test_calculator.py": """
import unittest
from src.calculator import add, multiply, subtract
class CalculatorTests(unittest.TestCase):
def test_add(self):
self.assertEqual(add(2, 3), 5)
def test_subtract(self):
self.assertEqual(subtract(7, 4), 3)
def test_multiply(self):
self.assertEqual(multiply(6, 7), 42)
self.assertEqual(multiply(-3, 4), -12)
if __name__ == "__main__":
unittest.main()
""",
}
)
else:
raise QaFailure(f"unknown fixture: {fixture}")
return common
def check_read_only(
_ctx: QaContext,
workspace: Path,
before: dict,
_mermaid: CommandResult | None,
) -> list[str]:
after = manifest(workspace)
if after != before:
raise QaFailure("read-only project scan modified files")
return ["workspace unchanged"]
def check_fix_failing_test(
_ctx: QaContext,
workspace: Path,
before: dict,
_mermaid: CommandResult | None,
) -> list[str]:
result = run_command(["python3", "-m", "unittest", "discover", "-s", "tests"], cwd=workspace)
if result.returncode != 0:
raise QaFailure(f"fixture tests still fail:\n{result.stdout}\n{result.stderr}")
after = manifest(workspace)
changed = changed_files(before, after)
if changed != ["src/calculator.py"]:
raise QaFailure(f"expected only src/calculator.py to change, got {changed}")
source = (workspace / "src" / "calculator.py").read_text()
if "return a + b" not in source:
raise QaFailure("calculator.add was not repaired to use addition")
return ["tests pass", "only implementation changed", "add uses addition"]
def check_add_small_feature(
_ctx: QaContext,
workspace: Path,
before: dict,
_mermaid: CommandResult | None,
) -> list[str]:
result = run_command(["python3", "-m", "unittest", "discover", "-s", "tests"], cwd=workspace)
if result.returncode != 0:
raise QaFailure(f"fixture tests still fail:\n{result.stdout}\n{result.stderr}")
after = manifest(workspace)
changed = changed_files(before, after)
if changed != ["src/calculator.py"]:
raise QaFailure(f"expected only src/calculator.py to change, got {changed}")
source = (workspace / "src" / "calculator.py").read_text()
if "def multiply" not in source:
raise QaFailure("multiply function was not implemented")
return ["tests pass", "only implementation changed", "multiply implemented"]
def check_runtime_smoke(
ctx: QaContext,
_workspace: Path,
_before: dict,
_mermaid: CommandResult | None,
) -> list[str]:
result = run_command([ctx.args.mermaid_bin, "tasks", "--limit", "5"], cwd=REPO_ROOT)
write_command(ctx.run_dir / "runtime_tasks.json", result)
if result.returncode != 0:
raise QaFailure(f"`mermaid tasks` failed: {result.stderr}")
if "Mermaid runtime tasks" not in result.stdout:
raise QaFailure("runtime task output did not include the expected heading")
return ["runtime tasks command exits cleanly"]
def check_compact_smoke(
ctx: QaContext,
workspace: Path,
_before: dict,
_mermaid: CommandResult | None,
) -> list[str]:
workspace.mkdir(parents=True, exist_ok=True)
result = run_command(
[
ctx.args.mermaid_bin,
"--path",
workspace,
"qa",
"compact-smoke",
"--turns",
"6",
"--format",
"json",
],
cwd=REPO_ROOT,
)
write_command(ctx.run_dir / "compact_smoke_command.json", result)
if result.returncode != 0:
raise QaFailure(f"`mermaid qa compact-smoke` failed: {result.stderr}")
payload = parse_mermaid_json(result.stdout)
if payload.get("ok") is not True:
raise QaFailure(f"compact-smoke reported failure: {payload}")
if int(payload.get("archived_messages") or 0) <= 0:
raise QaFailure("compact-smoke archived no messages")
if int(payload.get("preserved_messages") or 0) <= 0:
raise QaFailure("compact-smoke preserved no messages")
if int(payload.get("replacement_messages") or 0) < 3:
raise QaFailure("compact-smoke produced too few replacement messages")
for key in ("conversation_path", "archive_path"):
path = payload.get(key)
if not path or not Path(path).exists():
raise QaFailure(f"compact-smoke missing {key}: {path}")
return [
"compact smoke command exits cleanly",
"archives and preserves messages",
"conversation and archive files exist",
]
SCENARIOS: dict[str, Scenario] = {
"compact_smoke": Scenario(
name="compact_smoke",
tier="fast",
fixture=None,
prompt=None,
checker=check_compact_smoke,
),
"runtime_smoke": Scenario(
name="runtime_smoke",
tier="fast",
fixture=None,
prompt=None,
checker=check_runtime_smoke,
),
"read_only_project_scan": Scenario(
name="read_only_project_scan",
tier="real",
fixture="clean",
prompt="""
Inspect this small Python project. Do not modify any files.
Report the project purpose, important files, the validation command,
and one possible next improvement.
""",
checker=check_read_only,
),
"fix_failing_test": Scenario(
name="fix_failing_test",
tier="real",
fixture="buggy_add",
prompt="""
Run `python3 -m unittest discover -s tests`, fix the implementation
bug that causes the failure, rerun the same command, and report the
result. Do not edit tests.
""",
checker=check_fix_failing_test,
allow_mermaid_errors=True,
),
"add_small_feature": Scenario(
name="add_small_feature",
tier="real",
fixture="missing_multiply",
prompt="""
Run `python3 -m unittest discover -s tests`, implement the missing
calculator feature needed by the tests, rerun the same command, and
report the result. Do not edit tests.
""",
checker=check_add_small_feature,
allow_mermaid_errors=True,
),
}
def scenarios_for_tier(tier: str) -> list[str]:
if tier == "all":
return list(SCENARIOS.keys())
return [name for name, scenario in SCENARIOS.items() if scenario.tier == tier]
def run_command(
args: Iterable[str | Path],
cwd: Path,
timeout: int = 120,
) -> CommandResult:
argv = [str(arg) for arg in args]
completed = subprocess.run(
argv,
cwd=str(cwd),
text=True,
capture_output=True,
timeout=timeout,
check=False,
)
return CommandResult(
args=argv,
cwd=str(cwd),
returncode=completed.returncode,
stdout=completed.stdout,
stderr=completed.stderr,
)
def parse_mermaid_json(stdout: str) -> dict:
try:
value = json.loads(stdout)
except json.JSONDecodeError as exc:
raise QaFailure(f"Mermaid stdout was not valid JSON: {exc}\n{stdout[:1000]}") from exc
if not isinstance(value, dict):
raise QaFailure("Mermaid JSON output was not an object")
return value
def manifest(root: Path) -> dict[str, str]:
out: dict[str, str] = {}
if not root.exists():
return out
for path in sorted(p for p in root.rglob("*") if p.is_file()):
rel = path.relative_to(root).as_posix()
if should_ignore(rel):
continue
out[rel] = sha256_file(path)
return out
def should_ignore(rel: str) -> bool:
parts = set(Path(rel).parts)
return bool(parts & IGNORED_PARTS)
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def changed_files(before: dict[str, str], after: dict[str, str]) -> list[str]:
keys = sorted(set(before) | set(after))
return [key for key in keys if before.get(key) != after.get(key)]
def write_manifest(path: Path, data: dict[str, str]) -> None:
path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n")
def write_command(path: Path, result: CommandResult) -> None:
path.write_text(json.dumps(result.to_json(), indent=2, sort_keys=True) + "\n")
def write_diff(path: Path, before: Path, after: Path) -> None:
lines: list[str] = []
before_files = text_files(before)
after_files = text_files(after)
for rel in sorted(set(before_files) | set(after_files)):
old = before_files.get(rel, [])
new = after_files.get(rel, [])
if old == new:
continue
lines.extend(
difflib.unified_diff(
old,
new,
fromfile=f"before/{rel}",
tofile=f"after/{rel}",
lineterm="",
)
)
lines.append("")
path.write_text("\n".join(lines).rstrip() + ("\n" if lines else ""))
def text_files(root: Path) -> dict[str, list[str]]:
out: dict[str, list[str]] = {}
if not root.exists():
return out
for path in sorted(p for p in root.rglob("*") if p.is_file()):
rel = path.relative_to(root).as_posix()
if should_ignore(rel):
continue
try:
out[rel] = path.read_text().splitlines()
except UnicodeDecodeError:
out[rel] = ["<binary>"]
return out
def copy_tree(src: Path, dst: Path) -> None:
remove_path(dst) if dst.exists() else None
shutil.copytree(src, dst)
def restore_tree(snapshot: Path, workspace: Path) -> None:
remove_path(workspace) if workspace.exists() else None
shutil.copytree(snapshot, workspace)
def elapsed_seconds(start: dt.datetime) -> float:
return (dt.datetime.now() - start).total_seconds()
def print_summary(report: dict, report_path: Path) -> None:
print("Mermaid self-QA")
print(f"Report: {report_path}")
print(f"Model: {report['model']}")
print(f"Test env: {report['test_env']}")
print()
for item in report.get("results", []):
status = "PASS" if item.get("ok") else "FAIL"
print(f"{status} {item['name']} ({item['duration_seconds']:.1f}s)")
if item.get("checks"):
for check in item["checks"]:
print(f" - {check}")
if item.get("mermaid_errors"):
print(f" - mermaid tool errors observed: {len(item['mermaid_errors'])}")
if item.get("failure"):
print(f" - {item['failure']}")
if report.get("fatal_error"):
print()
print(f"Fatal: {report['fatal_error']}")
print()
print("PASS" if report.get("ok") else "FAIL")
if __name__ == "__main__":
sys.exit(main())