Writing Custom Validators

Custom validators let you replace or extend the bundled quality gates with checks specific to your project or organisation.

The contract

A validator is a Python script that:

• Reads a JSON contract from stdin
• Writes a JSON result to stdout
• Exits 0 on pass, 1 on failure

The input contract:

{
  "skill_outputs": {
    "implement-next-phase": "...output from the skill...",
    "commit": "...output..."
  },
  "goal_state": "All phases implemented and committed, no unchecked tasks remaining",
  "loop_context": {
    "workspace_root": "/path/to/your/repo",
    "job_id": "feature/DM-123-add-rate-limiting",
    "task_id": "550e8400-e29b-41d4-a716-446655440000",
    "loop_name": "dev"
  }
}

The output result:

{
  "pass": true,
  "message": "All checks passed",
  "checks": [
    {
      "name": "tests_pass",
      "pass": true,
      "message": "Tests passed — ran `uv run pytest -q` (exit 0)"
    },
    {
      "name": "coverage_threshold",
      "pass": false,
      "message": "Coverage 71% is below threshold 80%"
    }
  ]
}

The top-level pass is the overall result. checks is an array of named results matching the check names declared in your loop config.

Where to put validators

Place validators in validators/{name}.py at your repository root. This overrides the bundled validator with the same name, or adds a new one.

your-project/
├── validators/
│   ├── run_tests.py        # overrides bundled run_tests
│   ├── coverage_check.py   # new validator
│   └── security_scan.py    # new validator
├── .dmx/
│   └── loops/
│       └── dev.yaml        # references coverage_check and security_scan
└── src/

A minimal example

#!/usr/bin/env python3
"""validators/lint_check.py — check that the staged diff has no lint errors."""

import json
import subprocess
import sys


def run(workspace_root: str) -> dict:
    result = subprocess.run(
        ["ruff", "check", "."],
        cwd=workspace_root,
        capture_output=True,
        text=True,
    )
    passed = result.returncode == 0
    return {
        "pass": passed,
        "message": "Lint passed" if passed else "Lint failed",
        "checks": [
            {
                "name": "no_lint_errors",
                "pass": passed,
                "message": result.stdout + result.stderr if not passed else "Clean",
            }
        ],
    }


if __name__ == "__main__":
    contract = json.loads(sys.stdin.read() or "{}")
    workspace_root = contract.get("loop_context", {}).get("workspace_root", ".")
    result = run(workspace_root)
    print(json.dumps(result))
    sys.exit(0 if result["pass"] else 1)

Reference it in your loop config:

validators:
  - tool: lint_check
    checks:
      - name: no_lint_errors
        required: true

An LLM-backed spec adherence validator

The bundled spec_adherence validator grades the structured validation-report.json that the validate skill writes from its own diff analysis (see Validators). For an independent judgment call on top of that report, replace it with an LLM-backed check:

#!/usr/bin/env python3
"""validators/spec_adherence.py — LLM-backed spec adherence check."""

import json
import sys
from pathlib import Path

# Use whichever LLM client your team has available
import anthropic


def run(workspace_root: str, skill_outputs: dict) -> dict:
    spec_path = Path(workspace_root) / ".dmx" / "spec.md"
    if not spec_path.exists():
        return {
            "pass": False,
            "message": "spec.md not found",
            "checks": [{"name": "scope_matches_spec", "pass": False}],
        }

    spec = spec_path.read_text()
    implementation_summary = "\n\n".join(
        f"### {skill}\n{output}" for skill, output in skill_outputs.items()
    )

    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=512,
        messages=[
            {
                "role": "user",
                "content": f"""Does this implementation match the spec?

SPEC:
{spec}

IMPLEMENTATION:
{implementation_summary}

Reply with JSON only:
{{"matches": true/false, "reason": "one sentence"}}""",
            }
        ],
    )

    result = json.loads(response.content[0].text)
    passed = result.get("matches", False)
    return {
        "pass": passed,
        "message": result.get("reason", ""),
        "checks": [
            {
                "name": "scope_matches_spec",
                "pass": passed,
                "message": result.get("reason", ""),
            }
        ],
    }


if __name__ == "__main__":
    contract = json.loads(sys.stdin.read() or "{}")
    workspace_root = contract.get("loop_context", {}).get("workspace_root", ".")
    result = run(workspace_root, contract.get("skill_outputs", {}))
    print(json.dumps(result))
    sys.exit(0 if result["pass"] else 1)

Timeout

Validators have a 120-second timeout by default. Long-running validators (full integration test suites, slow LLM calls) should be designed to return promptly or run as background tasks themselves.

Tips

Check names must match your loop config. The checks array in your validator output should include entries for every check name declared in the loop YAML. Undeclared checks are ignored; missing required checks are treated as failed.

Keep validators focused. One concern per validator. A run_tests validator should only check tests. A lint_check validator should only check lint. Separate concerns make it easy to understand which check is failing.

Return useful messages. The message field on each check is shown when the loop pauses. "coverage 71% is below threshold 80%" is much more actionable than "check failed".

Make validators idempotent. Validators may be re-run after you fix an issue and continue the loop. They should produce consistent results given the same state.