check_task.py

#!/usr/bin/env python3
"""
Minimal task.md structural check — runs in CI without depending on the
benchflow CLI (which is in flux: the multi-level dogfood that knows the
task.md format lives on an unreleased upstream branch).

Validates:
- task.md exists and has YAML frontmatter (--- ... ---) + a Markdown body
- Required frontmatter fields are present
- environment/Dockerfile exists and starts with `FROM `
- verifier/{test.sh, test_outputs.py, verifier.md} all exist
- verifier/rubrics/ contains at least one *.md file
- oracle/solve.sh exists

This is the same shape `bench tasks check --level publication-grade`
enforces locally; remove this script and swap CI to the benchflow CLI
once upstream lands the new format on main.

Exit code: 0 if every task validates, 1 if any task has issues.
"""
from __future__ import annotations

import re
import sys
from pathlib import Path
from typing import Iterable

FRONTMATTER_RE = re.compile(r"\A---\s*\n(.*?)\n---\s*(?:\n|$)", re.DOTALL)
REQUIRED_FRONTMATTER = (
    "version",
    "metadata",
    "agent",
    "verifier",
    "environment",
)
BENCHFLOW_FRONTMATTER = ("task", "metadata", "agent", "verifier", "sandbox")
REQUIRED_METADATA = (
    "author_name",
    "author_email",
    "category",
)


def parse_yaml_keys(block: str) -> set[str]:
    """Cheap top-level key extractor — avoids importing PyYAML in CI."""
    keys: set[str] = set()
    for line in block.splitlines():
        if not line or line.startswith(("#", " ", "\t", "-")):
            continue
        if ":" not in line:
            continue
        key = line.split(":", 1)[0].strip()
        if key:
            keys.add(key)
    return keys


def parse_metadata_keys(block: str) -> set[str]:
    """Extract keys nested under `metadata:` (2-space indent expected)."""
    keys: set[str] = set()
    in_metadata = False
    for line in block.splitlines():
        stripped = line.lstrip()
        if line.startswith("metadata:"):
            in_metadata = True
            continue
        if in_metadata:
            if line and not line.startswith((" ", "\t")):
                in_metadata = False
                continue
            if stripped.startswith("#") or not stripped:
                continue
            if ":" in stripped:
                keys.add(stripped.split(":", 1)[0].strip())
    return keys


def _check_environment(task_dir: Path) -> list[str]:
    issues: list[str] = []
    dockerfile = task_dir / "environment" / "Dockerfile"
    if not dockerfile.exists():
        issues.append("Missing required file: environment/Dockerfile")
    else:
        # Skip comment lines and blanks; the first executable instruction
        # must be FROM.
        first_instr = next(
            (
                line
                for line in dockerfile.read_text(encoding="utf-8").splitlines()
                if line.strip() and not line.lstrip().startswith("#")
            ),
            "",
        )
        if not first_instr.upper().startswith("FROM "):
            issues.append("environment/Dockerfile first instruction must be FROM")
    return issues


def check_task(task_dir: Path) -> list[str]:
    issues: list[str] = []

    # task.md ---------------------------------------------------------------
    task_md = task_dir / "task.md"
    if not task_md.exists():
        return [f"Missing required file: task.md"]

    text = task_md.read_text(encoding="utf-8")
    m = FRONTMATTER_RE.match(text)
    if not m:
        issues.append("task.md must start with YAML frontmatter (--- ... ---)")
    else:
        frontmatter = m.group(1)
        body = text[m.end():]
        top_keys = parse_yaml_keys(frontmatter)
        if "schema_version" in top_keys and "task" in top_keys:
            # BenchFlow-native task (schema 1.1): `bench tasks check` is the authority; the arena only
            # requires the executable pieces the pipeline snapshots and runs.
            for required in BENCHFLOW_FRONTMATTER:
                if required not in top_keys:
                    issues.append(f"task.md frontmatter missing: {required}")
            if "## prompt" not in body:
                issues.append("task.md body must contain a '## prompt' section")
            issues.extend(_check_environment(task_dir))
            if not (task_dir / "verifier" / "test.sh").exists():
                issues.append("Missing required file: verifier/test.sh")
            return issues
        for required in REQUIRED_FRONTMATTER:
            if required not in top_keys:
                issues.append(f"task.md frontmatter missing: {required}")
        metadata_keys = parse_metadata_keys(frontmatter)
        for required in REQUIRED_METADATA:
            if required not in metadata_keys:
                issues.append(f"task.md metadata.{required} required")
        if "## prompt" not in body:
            issues.append("task.md body must contain a '## prompt' section")

    issues.extend(_check_environment(task_dir))

    # verifier/ -------------------------------------------------------------
    for f in ("verifier/test.sh", "verifier/test_outputs.py", "verifier/verifier.md"):
        if not (task_dir / f).exists():
            issues.append(f"Missing required file: {f}")
    rubrics = task_dir / "verifier" / "rubrics"
    if not rubrics.is_dir():
        issues.append("Missing required directory: verifier/rubrics/")
    else:
        if not any(p.suffix == ".md" for p in rubrics.iterdir()):
            issues.append("verifier/rubrics/ must contain at least one *.md rubric")

    # oracle/ ---------------------------------------------------------------
    if not (task_dir / "oracle" / "solve.sh").exists():
        issues.append("Missing required file: oracle/solve.sh")

    return issues


def iter_task_dirs(roots: Iterable[str]) -> Iterable[Path]:
    for root in roots:
        p = Path(root)
        if p.is_dir() and (p / "task.md").exists():
            yield p
        elif p.is_dir():
            for child in sorted(p.iterdir()):
                if child.is_dir() and (child / "task.md").exists():
                    yield child


def main(argv: list[str]) -> int:
    targets = argv[1:] or ["starting-kit/examples", "starting-kit/template"]
    overall_ok = True
    any_seen = False
    for task_dir in iter_task_dirs(targets):
        any_seen = True
        issues = check_task(task_dir)
        if issues:
            overall_ok = False
            print(f"✗ {task_dir.name} — {len(issues)} issue(s):")
            for i in issues:
                print(f"  → {i}")
        else:
            print(f"✓ {task_dir.name} — valid")
    if not any_seen:
        print("no task directories found")
        return 1
    return 0 if overall_ok else 1


if __name__ == "__main__":
    sys.exit(main(sys.argv))