Writing Validators
Custom validators let you replace or extend the bundled quality gates with checks specific to your project or organization.
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": "DM-123",
"task_id": "550e8400-e29b-41d4-a716-446655440000",
"loop_name": "dev",
"branch": "feature-DM-123-add-rate-limiting",
"ticket_ref": "DM-123",
"timeout_seconds": 630
}
}timeout_seconds is the time limit the runner enforces for this validator. A validator that runs a slow command of its own should stop that command before the limit, so it can still report a useful message.
The output result, from a custom run_tests that also measures coverage:
{
"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/To use the same validator across many repositories, put it in a shared source instead: a validators/ directory in an org-wide git repo that each app repo vendors with /dmx/sync. A validator in your repo's own validators/ still takes precedence over a shared one with the same name.
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: trueAn 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
Each validator may run for 630 seconds by default. Set timeout_seconds on the validator in your loop YAML to change it (minimum 60). See timeout_seconds.
Validators run in the background, so a long one doesn't block your IDE's connection to dmx. When one runs past its limit, dmx stops it and every process it started, and the check fails with a message naming the limit. A process that starts its own session, such as some Docker-based test setups, isn't stopped, so clean those up in the validator itself.
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.