From 4a66e054ce705898def2aaaddc9f832d89e6df02 Mon Sep 17 00:00:00 2001 From: Benjamin Kapner Date: Thu, 2 Jul 2026 09:19:22 +0300 Subject: [PATCH 1/2] feat: add deterministic quality gate for SKILL.md submissions Add QualityGate with 6 deterministic checks: - Description quality: frontmatter validation (name, description) - Broken references: relative markdown links to missing files - File completeness: thin instruction.md, tests without assertions - Imprecise instructions: vague language (hedging, vague conditions) - Unfinished content: TODO, FIXME, placeholder markers, TBD - Generic advice: filler text (follow best practices, handle errors properly) Patterns adapted from harness-eval-lab (setup-eval). 44 tests. --- abevalflow/gates/quality/__init__.py | 2 + abevalflow/gates/quality/skillmd_quality.py | 174 ++++++++ abevalflow/quality/__init__.py | 1 + abevalflow/quality/skillmd_quality_scanner.py | 418 ++++++++++++++++++ pipeline/tasks/phases/test.yaml | 44 +- scripts/skillmd_quality_scan.py | 65 +++ tests/test_skillmd_quality_scanner.py | 349 +++++++++++++++ 7 files changed, 1051 insertions(+), 2 deletions(-) create mode 100644 abevalflow/gates/quality/skillmd_quality.py create mode 100644 abevalflow/quality/__init__.py create mode 100644 abevalflow/quality/skillmd_quality_scanner.py create mode 100644 scripts/skillmd_quality_scan.py create mode 100644 tests/test_skillmd_quality_scanner.py diff --git a/abevalflow/gates/quality/__init__.py b/abevalflow/gates/quality/__init__.py index ff227f0..d7655b1 100644 --- a/abevalflow/gates/quality/__init__.py +++ b/abevalflow/gates/quality/__init__.py @@ -49,6 +49,7 @@ def get_all_quality_gate_names() -> list[str]: from abevalflow.gates.quality.llm_review import LLMReviewGate +from abevalflow.gates.quality.skillmd_quality import SkillMdQualityGate __all__ = [ "register_quality_gate", @@ -56,4 +57,5 @@ def get_all_quality_gate_names() -> list[str]: "get_all_quality_gates", "get_all_quality_gate_names", "LLMReviewGate", + "SkillMdQualityGate", ] diff --git a/abevalflow/gates/quality/skillmd_quality.py b/abevalflow/gates/quality/skillmd_quality.py new file mode 100644 index 0000000..3316e0f --- /dev/null +++ b/abevalflow/gates/quality/skillmd_quality.py @@ -0,0 +1,174 @@ +"""Deterministic SKILL.md quality gate. + +Reads skillmd-quality-scan.json produced by the quality scan step. +Checks description quality, broken references, file completeness, +imprecise instructions, unfinished content, and generic advice. +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path + +from abevalflow.gates.base import Finding, GateMode, GateResult, GateType, Severity +from abevalflow.gates.quality import register_quality_gate +from abevalflow.gates.quality.base import QualityGate +from abevalflow.schemas import GatePolicy + +logger = logging.getLogger(__name__) + + +@register_quality_gate("skillmd-quality") +class SkillMdQualityGate(QualityGate): + """Deterministic quality checks for SKILL.md content. + + Reads skillmd-quality-scan.json. Pass/fail depends on mode: + - warn: always passes (findings are advisory) + - block: fails if HIGH or CRITICAL findings exist + """ + + name = "skillmd-quality" + + def evaluate( + self, + workspace_root: Path, + policy: GatePolicy, + ) -> GateResult: + """Evaluate deterministic quality scan results.""" + gate_policy = policy.get_gate_policy("quality") + + if gate_policy.mode == GateMode.DISABLED: + return GateResult( + gate_type=GateType.QUALITY, + gate_name="quality", + policy_key=self.name, + passed=True, + score=1.0, + mode=GateMode.DISABLED, + findings=[], + details={"scanner": self.name}, + message="SKILL.md quality scan disabled", + ) + + scan_path = workspace_root / "skillmd-quality-scan.json" + if not scan_path.exists(): + if gate_policy.mode == GateMode.BLOCK: + return GateResult( + gate_type=GateType.QUALITY, + gate_name="quality", + policy_key=self.name, + passed=False, + score=0.0, + mode=gate_policy.mode, + findings=[], + details={ + "scanner": self.name, + "scan_path": str(scan_path), + "status": "not_found", + }, + message=("FAIL: skillmd-quality-scan.json missing (required in block mode)"), + ) + return GateResult( + gate_type=GateType.QUALITY, + gate_name="quality", + policy_key=self.name, + passed=True, + score=1.0, + mode=gate_policy.mode, + findings=[], + details={ + "scanner": self.name, + "scan_path": str(scan_path), + "status": "not_found", + }, + message=("No skillmd-quality-scan.json found (scan may have been skipped)"), + ) + + try: + scan_data = json.loads(scan_path.read_text()) + except (json.JSONDecodeError, OSError) as e: + logger.error("Failed to read quality scan: %s", e) + return GateResult( + gate_type=GateType.QUALITY, + gate_name="quality", + policy_key=self.name, + passed=False, + score=0.0, + mode=gate_policy.mode, + findings=[], + details={"scanner": self.name, "error": str(e)}, + message=f"Failed to parse skillmd-quality-scan.json: {e}", + ) + + findings: list[Finding] = [] + severity_counts = { + "critical": 0, + "high": 0, + "medium": 0, + "low": 0, + "info": 0, + } + + for f in scan_data.get("findings", []): + sev_str = f.get("severity", "info").lower().strip() + try: + severity = Severity(sev_str) + except ValueError: + severity = Severity.INFO + + if severity.value in severity_counts: + severity_counts[severity.value] += 1 + + findings.append( + Finding( + severity=severity, + message=f.get("message", ""), + location=f.get("file_path"), + rule_id=f.get("rule_id", "unknown"), + details=f, + ) + ) + + high_or_critical = severity_counts["critical"] + severity_counts["high"] + + if gate_policy.mode == GateMode.BLOCK: + passed = high_or_critical == 0 + else: + passed = True + + total_findings = len(findings) + if total_findings == 0: + score = 1.0 + else: + weighted_score = ( + severity_counts["critical"] * 0.0 + + severity_counts["high"] * 0.25 + + severity_counts["medium"] * 0.5 + + severity_counts["low"] * 0.75 + + severity_counts["info"] * 0.9 + ) + score = weighted_score / total_findings + + message = ( + f"SKILL.md quality: {total_findings} findings " + f"(high={severity_counts['high']}," + f" medium={severity_counts['medium']}," + f" low={severity_counts['low']})" + ) + + return GateResult( + gate_type=GateType.QUALITY, + gate_name="quality", + policy_key=self.name, + passed=passed, + score=score, + mode=gate_policy.mode, + findings=findings, + details={ + "scanner": self.name, + "severity_counts": severity_counts, + "scan_path": str(scan_path), + }, + message=message, + ) diff --git a/abevalflow/quality/__init__.py b/abevalflow/quality/__init__.py new file mode 100644 index 0000000..ece1f2d --- /dev/null +++ b/abevalflow/quality/__init__.py @@ -0,0 +1 @@ +"""Quality scanning modules for skill submissions.""" diff --git a/abevalflow/quality/skillmd_quality_scanner.py b/abevalflow/quality/skillmd_quality_scanner.py new file mode 100644 index 0000000..968282f --- /dev/null +++ b/abevalflow/quality/skillmd_quality_scanner.py @@ -0,0 +1,418 @@ +"""SKILL.md quality scanner. + +Deterministic quality checks for skill submissions: description quality, +broken references, file completeness, imprecise instructions, unfinished +content, and generic advice. + +Patterns adapted from harness-eval-lab (setup-eval). +""" + +from __future__ import annotations + +import logging +import re +from pathlib import Path + +import yaml + +logger = logging.getLogger(__name__) + +_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---", re.DOTALL) +_RELATIVE_LINK_RE = re.compile(r"\[.*?\]\(([^)]+)\)") + +# --- Imprecise instruction patterns --- + +_HEDGING_PATTERNS: list[tuple[str, re.Pattern[str]]] = [ + ("try to", re.compile(r"\btry\s+to\s+\w+", re.I)), + ("if possible", re.compile(r"\bif\s+(?:at\s+all\s+)?possible\b", re.I)), + ("you might want to", re.compile(r"\byou\s+might\s+(?:want|wish)\s+to\b", re.I)), + ("perhaps consider", re.compile(r"\bperhaps\s+(?:consider|you)\b", re.I)), + ("consider using", re.compile(r"\bconsider\s+(?:using|adding|implementing)\b", re.I)), +] + +_VAGUE_CONDITION_PATTERNS: list[tuple[str, re.Pattern[str]]] = [ + ("if needed", re.compile(r"\bif\s+needed\b", re.I)), + ("if appropriate", re.compile(r"\bif\s+appropriate\b", re.I)), + ("as necessary", re.compile(r"\bas\s+necessary\b", re.I)), + ("if applicable", re.compile(r"\bif\s+applicable\b", re.I)), + ("when relevant", re.compile(r"\bwhen\s+relevant\b", re.I)), +] + +# --- Unfinished content patterns --- + +_UNFINISHED_PATTERNS: list[tuple[str, re.Pattern[str]]] = [ + ("TODO", re.compile(r"\bTODO\s*:", re.I)), + ("FIXME", re.compile(r"\bFIXME\s*:", re.I)), + ("XXX", re.compile(r"\bXXX\s*:", re.I)), + ("TBD", re.compile(r"\bTBD\b")), + ("coming soon", re.compile(r"\bcoming\s+soon\b", re.I)), + ("work in progress", re.compile(r"\bwork\s+in\s+progress\b", re.I)), + ("not yet implemented", re.compile(r"\bnot\s+yet\s+(?:implemented|done|complete)\b", re.I)), + ( + "placeholder bracket", + re.compile( + r"\[(?:INSERT|FILL\s+IN|CHANGE\s+THIS|UPDATE\s+THIS|REPLACE|YOUR)\s+[^\]]+\]", + re.I, + ), + ), + ("PLACEHOLDER", re.compile(r"\[PLACEHOLDER\]", re.I)), +] + +# --- Generic advice patterns --- + +_GENERIC_ADVICE_PATTERNS: list[tuple[str, re.Pattern[str]]] = [ + ("follow best practices", re.compile(r"follow\s+(the\s+)?best\s+practices", re.I)), + ("handle errors properly", re.compile(r"handle\s+errors\s+(?:properly|correctly|gracefully)", re.I)), + ("write clean code", re.compile(r"write\s+clean[\s,]+readable\s+code", re.I)), + ("ensure code quality", re.compile(r"ensure\s+(?:code\s+)?quality", re.I)), + ("consider edge cases", re.compile(r"consider\s+(all\s+)?edge\s+cases", re.I)), + ("use proper formatting", re.compile(r"use\s+proper\s+formatting", re.I)), + ("be thorough", re.compile(r"be\s+thorough\s+(?:in|and|with)", re.I)), +] + +_EXCLUDED_DIRS = {".git", "node_modules", "vendor", "__pycache__", ".venv"} + + +def _is_excluded(path: Path, base: Path) -> bool: + try: + parts = path.relative_to(base).parts + except ValueError: + return False + return bool(_EXCLUDED_DIRS.intersection(parts)) + + +def _is_in_code_fence_or_quote(line: str, in_code_fence: bool) -> bool: + return in_code_fence or line.lstrip().startswith(">") + + +# --- Check functions --- + + +def check_description_quality( + file_path: Path, + relative_to: Path | None = None, +) -> list[dict]: + """Check SKILL.md frontmatter for description quality issues.""" + try: + content = file_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return [] + + display_path = str(file_path.relative_to(relative_to)) if relative_to else str(file_path) + findings: list[dict] = [] + + fm_match = _FRONTMATTER_RE.match(content) + if not fm_match: + findings.append( + { + "severity": "medium", + "rule_id": "quality-no-frontmatter", + "message": "SKILL.md has no YAML frontmatter", + "file_path": display_path, + "category": "description_quality", + } + ) + return findings + + try: + fm = yaml.safe_load(fm_match.group(1)) + except yaml.YAMLError: + findings.append( + { + "severity": "medium", + "rule_id": "quality-invalid-frontmatter", + "message": "SKILL.md frontmatter is not valid YAML", + "file_path": display_path, + "category": "description_quality", + } + ) + return findings + + if not isinstance(fm, dict): + return findings + + if "name" not in fm: + findings.append( + { + "severity": "medium", + "rule_id": "quality-missing-name", + "message": "SKILL.md frontmatter is missing 'name' field", + "file_path": display_path, + "category": "description_quality", + } + ) + + desc = fm.get("description", "") + if not desc: + findings.append( + { + "severity": "medium", + "rule_id": "quality-missing-description", + "message": "SKILL.md frontmatter is missing 'description' field", + "file_path": display_path, + "category": "description_quality", + } + ) + elif len(str(desc)) < 10: + findings.append( + { + "severity": "medium", + "rule_id": "quality-short-description", + "message": (f"SKILL.md description is {len(str(desc))} characters, too short to be meaningful"), + "file_path": display_path, + "category": "description_quality", + } + ) + + name = fm.get("name", "") + if name and desc and str(name).strip().lower() == str(desc).strip().lower(): + findings.append( + { + "severity": "low", + "rule_id": "quality-description-equals-name", + "message": "SKILL.md description is identical to the name", + "file_path": display_path, + "category": "description_quality", + } + ) + + return findings + + +def check_broken_references( + file_path: Path, + relative_to: Path | None = None, +) -> list[dict]: + """Check markdown file for broken relative links.""" + try: + content = file_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return [] + + display_path = str(file_path.relative_to(relative_to)) if relative_to else str(file_path) + findings: list[dict] = [] + checked: set[str] = set() + + for line_num, line in enumerate(content.splitlines(), start=1): + for match in _RELATIVE_LINK_RE.finditer(line): + ref = match.group(1).strip() + + if ref.startswith(("http://", "https://", "#", "mailto:")): + continue + if "$" in ref or "{{" in ref: + continue + if ref in checked: + continue + checked.add(ref) + + ref_path = file_path.parent / ref + if not ref_path.exists(): + findings.append( + { + "severity": "medium", + "rule_id": "quality-broken-reference", + "message": (f"Line {line_num}: link to '{ref}' points to a file that does not exist"), + "file_path": display_path, + "category": "broken_reference", + "line": line_num, + } + ) + + return findings + + +def check_file_completeness(directory: Path) -> list[dict]: + """Check submission files for completeness issues.""" + findings: list[dict] = [] + + instruction = directory / "instruction.md" + if instruction.exists(): + try: + content = instruction.read_text(encoding="utf-8", errors="replace") + body = re.sub(r"^---.*?---\s*", "", content, flags=re.DOTALL).strip() + body = re.sub(r"^#.*\n?", "", body).strip() + if len(body) < 50: + findings.append( + { + "severity": "high" if len(body) < 10 else "low", + "rule_id": "quality-thin-instruction", + "message": (f"instruction.md has only {len(body)} characters of body text"), + "file_path": "instruction.md", + "category": "file_completeness", + } + ) + except OSError: + pass + + tests_dir = directory / "tests" + if tests_dir.is_dir(): + for py_file in tests_dir.glob("*.py"): + try: + content = py_file.read_text(encoding="utf-8", errors="replace") + if "assert" not in content and "pytest.raises" not in content: + rel = str(py_file.relative_to(directory)) + findings.append( + { + "severity": "medium", + "rule_id": "quality-no-assertions", + "message": f"{rel} has no assert statements", + "file_path": rel, + "category": "file_completeness", + } + ) + except OSError: + pass + + return findings + + +def check_imprecise_instructions( + file_path: Path, + relative_to: Path | None = None, +) -> list[dict]: + """Check for vague, hedging, or ambiguous language.""" + try: + content = file_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return [] + + display_path = str(file_path.relative_to(relative_to)) if relative_to else str(file_path) + findings: list[dict] = [] + in_code_fence = False + + all_patterns = [("hedging", label, pat) for label, pat in _HEDGING_PATTERNS] + [ + ("vague condition", label, pat) for label, pat in _VAGUE_CONDITION_PATTERNS + ] + + for line_num, line in enumerate(content.splitlines(), start=1): + stripped = line.strip() + if stripped.startswith("```"): + in_code_fence = not in_code_fence + continue + if _is_in_code_fence_or_quote(line, in_code_fence): + continue + + for category, label, pattern in all_patterns: + if pattern.search(line): + findings.append( + { + "severity": "low", + "rule_id": "quality-imprecise-instruction", + "message": f"Line {line_num}: '{label}' ({category})", + "file_path": display_path, + "category": "imprecise_instruction", + "line": line_num, + } + ) + break + + return findings + + +def check_unfinished_content( + file_path: Path, + relative_to: Path | None = None, +) -> list[dict]: + """Check for TODO, FIXME, placeholder markers, and deferred content.""" + try: + content = file_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return [] + + display_path = str(file_path.relative_to(relative_to)) if relative_to else str(file_path) + findings: list[dict] = [] + in_code_fence = False + + for line_num, line in enumerate(content.splitlines(), start=1): + stripped = line.strip() + if stripped.startswith("```"): + in_code_fence = not in_code_fence + continue + if _is_in_code_fence_or_quote(line, in_code_fence): + continue + + for label, pattern in _UNFINISHED_PATTERNS: + if pattern.search(line): + findings.append( + { + "severity": "medium", + "rule_id": "quality-unfinished-content", + "message": f"Line {line_num}: '{label}' marker found", + "file_path": display_path, + "category": "unfinished_content", + "line": line_num, + } + ) + break + + return findings + + +def check_generic_advice( + file_path: Path, + relative_to: Path | None = None, +) -> list[dict]: + """Check for generic filler text that adds no value.""" + try: + content = file_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return [] + + display_path = str(file_path.relative_to(relative_to)) if relative_to else str(file_path) + findings: list[dict] = [] + in_code_fence = False + + for line_num, line in enumerate(content.splitlines(), start=1): + stripped = line.strip() + if stripped.startswith("```"): + in_code_fence = not in_code_fence + continue + if _is_in_code_fence_or_quote(line, in_code_fence): + continue + + for label, pattern in _GENERIC_ADVICE_PATTERNS: + if pattern.search(line): + findings.append( + { + "severity": "low", + "rule_id": "quality-generic-advice", + "message": f"Line {line_num}: '{label}' is generic filler", + "file_path": display_path, + "category": "generic_advice", + "line": line_num, + } + ) + break + + return findings + + +# --- Main entry point --- + + +def scan_directory(directory: Path) -> dict: + """Run all deterministic quality checks on a submission directory.""" + if not directory.is_dir(): + logger.error("Not a directory: %s", directory) + return {"findings": []} + + all_findings: list[dict] = [] + + md_files = sorted(f for f in directory.rglob("*.md") if not _is_excluded(f, directory)) + + for md_file in md_files: + if md_file.name == "SKILL.md": + all_findings.extend(check_description_quality(md_file, relative_to=directory)) + all_findings.extend(check_broken_references(md_file, relative_to=directory)) + all_findings.extend(check_imprecise_instructions(md_file, relative_to=directory)) + all_findings.extend(check_unfinished_content(md_file, relative_to=directory)) + all_findings.extend(check_generic_advice(md_file, relative_to=directory)) + + all_findings.extend(check_file_completeness(directory)) + + logger.info( + "Quality scan: %d files, %d findings", + len(md_files), + len(all_findings), + ) + return {"findings": all_findings} diff --git a/pipeline/tasks/phases/test.yaml b/pipeline/tasks/phases/test.yaml index fe2c9d4..82a8f42 100644 --- a/pipeline/tasks/phases/test.yaml +++ b/pipeline/tasks/phases/test.yaml @@ -345,7 +345,47 @@ spec: echo "SKILL.md security scan complete" - # Step 4: Quality review + # Step 4: SKILL.md deterministic quality checks + - name: skillmd-quality-scan + image: registry.access.redhat.com/ubi9/python-311:9.6 + script: | + #!/usr/bin/env bash + set -euo pipefail + echo "=== TEST PHASE: SKILL.md Quality Scan ===" + + EVAL_ENGINE="$(params.eval-engine)" + if [ "$EVAL_ENGINE" = "mcpchecker" ]; then + echo "MCPChecker mode: SKILL.md quality scan skipped" + exit 0 + fi + + SUBMISSION_PATH="$(workspaces.source.path)/submissions/$(params.submission-dir)" + SUBMISSION_NAME="$(params.submission-name)" + WORKSPACE_ROOT="$(workspaces.source.path)" + JSON_PATH="$WORKSPACE_ROOT/skillmd-quality-scan.json" + + if [ -d "$SUBMISSION_PATH/skills" ] && [ -f "$SUBMISSION_PATH/skills/SKILL.md" ]; then + SUBMISSION_PATH="$SUBMISSION_PATH/skills" + fi + + PIPELINE_DIR="$(workspaces.source.path)/_pipeline" + pip install --quiet --no-cache-dir pydantic pyyaml 2>&1 | tail -3 + export PYTHONPATH="$PIPELINE_DIR" + + echo "Scanning: $SUBMISSION_PATH" + python3 "$PIPELINE_DIR/scripts/skillmd_quality_scan.py" \ + "$SUBMISSION_PATH" --output "$JSON_PATH" 2>&1 || true + + if [ -f "$JSON_PATH" ]; then + FINDINGS=$(python3 -c "import json; print(len(json.load(open('$JSON_PATH')).get('findings', [])))" 2>/dev/null || echo "0") + echo "SKILL.md quality scan: $FINDINGS findings" + else + echo "SKILL.md quality scan: no output (scanner may have failed)" + fi + + echo "SKILL.md quality scan complete" + + # Step 5: Quality review (LLM) - name: quality-review image: registry.access.redhat.com/ubi9/python-311:9.6 env: @@ -394,7 +434,7 @@ spec: echo "Quality review: passed=$PASSED" - # Step 5: Finalize results + # Step 6: Finalize results - name: finalize image: registry.access.redhat.com/ubi9/python-311:9.6 script: | diff --git a/scripts/skillmd_quality_scan.py b/scripts/skillmd_quality_scan.py new file mode 100644 index 0000000..3d1d6a2 --- /dev/null +++ b/scripts/skillmd_quality_scan.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +"""Run deterministic quality checks on a submission directory. + +Checks description quality, broken references, file completeness, +imprecise instructions, unfinished content, and generic advice. + +Produces a JSON report compatible with the QualityGate interface. + +Exit codes: 0 = scan completed, 1 = scan error. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys +from pathlib import Path + +from abevalflow.quality.skillmd_quality_scanner import scan_directory + +logger = logging.getLogger(__name__) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Run deterministic quality checks on skill submissions", + ) + parser.add_argument( + "submission_dir", + type=Path, + help="Path to the submission directory to check", + ) + parser.add_argument( + "--output", + type=Path, + required=True, + help="Path to write the JSON report", + ) + args = parser.parse_args(argv) + + if not args.submission_dir.is_dir(): + logger.error("Not a directory: %s", args.submission_dir) + return 1 + + try: + result = scan_directory(args.submission_dir) + except Exception: + logger.exception("Quality scan failed") + return 1 + + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(result, indent=2)) + + findings = result.get("findings", []) + logger.info("Quality scan complete: %d findings", len(findings)) + return 0 + + +if __name__ == "__main__": + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + ) + sys.exit(main()) diff --git a/tests/test_skillmd_quality_scanner.py b/tests/test_skillmd_quality_scanner.py new file mode 100644 index 0000000..8ac5912 --- /dev/null +++ b/tests/test_skillmd_quality_scanner.py @@ -0,0 +1,349 @@ +"""Tests for SKILL.md quality scanner and gate.""" + +import json + +import pytest + +from abevalflow.gates.base import GateMode +from abevalflow.gates.quality import ( + SkillMdQualityGate, + get_all_quality_gate_names, + get_quality_gate, +) +from abevalflow.quality.skillmd_quality_scanner import ( + check_broken_references, + check_description_quality, + check_file_completeness, + check_generic_advice, + check_imprecise_instructions, + check_unfinished_content, + scan_directory, +) +from abevalflow.schemas import GatePolicy, GatePolicyItem + +# --------------------------------------------------------------------------- +# Description quality tests +# --------------------------------------------------------------------------- + + +class TestDescriptionQuality: + def test_valid_frontmatter(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: my-skill\ndescription: A skill that does useful things\n---\n\nBody.") + assert len(check_description_quality(p)) == 0 + + def test_no_frontmatter(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("# My Skill\n\nNo frontmatter here.") + assert any(f["rule_id"] == "quality-no-frontmatter" for f in check_description_quality(p)) + + def test_missing_name(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\ndescription: A good description\n---\n\nBody.") + assert any(f["rule_id"] == "quality-missing-name" for f in check_description_quality(p)) + + def test_missing_description(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: my-skill\n---\n\nBody.") + assert any(f["rule_id"] == "quality-missing-description" for f in check_description_quality(p)) + + def test_short_description(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: my-skill\ndescription: Hi\n---\n\nBody.") + assert any(f["rule_id"] == "quality-short-description" for f in check_description_quality(p)) + + def test_description_equals_name(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: my-skill\ndescription: my-skill\n---\n\nBody.") + assert any(f["rule_id"] == "quality-description-equals-name" for f in check_description_quality(p)) + + def test_invalid_yaml(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\n: invalid: yaml: [[\n---\n\nBody.") + assert any(f["rule_id"] == "quality-invalid-frontmatter" for f in check_description_quality(p)) + + +# --------------------------------------------------------------------------- +# Broken references tests +# --------------------------------------------------------------------------- + + +class TestBrokenReferences: + def test_broken_relative_link(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("See [guide](./nonexistent.md) for details.") + assert any(f["rule_id"] == "quality-broken-reference" for f in check_broken_references(p)) + + def test_valid_relative_link(self, tmp_path): + (tmp_path / "guide.md").write_text("# Guide") + p = tmp_path / "SKILL.md" + p.write_text("See [guide](guide.md) for details.") + assert len(check_broken_references(p)) == 0 + + def test_skips_urls(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("See [docs](https://example.com) for details.") + assert len(check_broken_references(p)) == 0 + + def test_skips_anchors(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("See [section](#config) for details.") + assert len(check_broken_references(p)) == 0 + + def test_skips_templates(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("See [config](${PATH}/config.md) for details.") + assert len(check_broken_references(p)) == 0 + + def test_deduplication(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("See [a](missing.md) and [b](missing.md).") + broken = [f for f in check_broken_references(p) if f["rule_id"] == "quality-broken-reference"] + assert len(broken) == 1 + + +# --------------------------------------------------------------------------- +# File completeness tests +# --------------------------------------------------------------------------- + + +class TestFileCompleteness: + def test_thin_instruction(self, tmp_path): + (tmp_path / "instruction.md").write_text("# Task\n\nDo it.") + assert any(f["rule_id"] == "quality-thin-instruction" for f in check_file_completeness(tmp_path)) + + def test_good_instruction(self, tmp_path): + (tmp_path / "instruction.md").write_text( + "# Task\n\n" + "Write a Python function that takes a list of integers and returns " + "the sum of all even numbers in the list." + ) + assert not any("instruction" in f["rule_id"] for f in check_file_completeness(tmp_path)) + + def test_test_without_assertions(self, tmp_path): + tests_dir = tmp_path / "tests" + tests_dir.mkdir() + (tests_dir / "test_outputs.py").write_text("def test_something():\n pass\n") + assert any(f["rule_id"] == "quality-no-assertions" for f in check_file_completeness(tmp_path)) + + def test_test_with_assertions(self, tmp_path): + tests_dir = tmp_path / "tests" + tests_dir.mkdir() + (tests_dir / "test_outputs.py").write_text("def test_something():\n assert 1 == 1\n") + assert not any("assertion" in f["rule_id"] for f in check_file_completeness(tmp_path)) + + +# --------------------------------------------------------------------------- +# Imprecise instruction tests +# --------------------------------------------------------------------------- + + +class TestImpreciseInstructions: + def test_hedging_detected(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nTry to handle errors if possible.") + findings = check_imprecise_instructions(p) + assert any(f["rule_id"] == "quality-imprecise-instruction" for f in findings) + + def test_vague_condition_detected(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nAdd logging as necessary.") + findings = check_imprecise_instructions(p) + assert any(f["rule_id"] == "quality-imprecise-instruction" for f in findings) + + def test_clean_instruction(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nReturn the sum of all even numbers.") + assert len(check_imprecise_instructions(p)) == 0 + + def test_code_block_skipped(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\n```\ntry to do something\n```") + assert len(check_imprecise_instructions(p)) == 0 + + def test_blockquote_skipped(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\n> Try to handle this if possible.") + assert len(check_imprecise_instructions(p)) == 0 + + +# --------------------------------------------------------------------------- +# Unfinished content tests +# --------------------------------------------------------------------------- + + +class TestUnfinishedContent: + def test_todo_detected(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nTODO: finish this section") + findings = check_unfinished_content(p) + assert any(f["rule_id"] == "quality-unfinished-content" for f in findings) + + def test_fixme_detected(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nFIXME: broken logic here") + findings = check_unfinished_content(p) + assert any(f["rule_id"] == "quality-unfinished-content" for f in findings) + + def test_placeholder_detected(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\n[INSERT YOUR API KEY HERE]") + findings = check_unfinished_content(p) + assert any(f["rule_id"] == "quality-unfinished-content" for f in findings) + + def test_tbd_detected(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nThe output format is TBD") + findings = check_unfinished_content(p) + assert any(f["rule_id"] == "quality-unfinished-content" for f in findings) + + def test_clean_content(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nThis skill analyzes CSV data.") + assert len(check_unfinished_content(p)) == 0 + + def test_code_block_skipped(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\n```\n# TODO: implement\n```") + assert len(check_unfinished_content(p)) == 0 + + +# --------------------------------------------------------------------------- +# Generic advice tests +# --------------------------------------------------------------------------- + + +class TestGenericAdvice: + def test_best_practices_detected(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nFollow best practices when coding.") + findings = check_generic_advice(p) + assert any(f["rule_id"] == "quality-generic-advice" for f in findings) + + def test_handle_errors_detected(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nHandle errors gracefully.") + findings = check_generic_advice(p) + assert any(f["rule_id"] == "quality-generic-advice" for f in findings) + + def test_specific_content_clean(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\nReturn HTTP 400 for invalid input.") + assert len(check_generic_advice(p)) == 0 + + def test_code_block_skipped(self, tmp_path): + p = tmp_path / "SKILL.md" + p.write_text("---\nname: x\ndescription: y\n---\n\n```\n# follow best practices\n```") + assert len(check_generic_advice(p)) == 0 + + +# --------------------------------------------------------------------------- +# scan_directory tests +# --------------------------------------------------------------------------- + + +class TestScanDirectory: + def test_aggregates_all_checks(self, tmp_path): + (tmp_path / "SKILL.md").write_text("---\nname: x\n---\n\nTODO: finish this.") + (tmp_path / "instruction.md").write_text("# T\n\nDo.") + result = scan_directory(tmp_path) + assert len(result["findings"]) >= 2 + + def test_empty_directory(self, tmp_path): + result = scan_directory(tmp_path) + assert result["findings"] == [] + + def test_clean_submission(self, tmp_path): + (tmp_path / "SKILL.md").write_text( + "---\nname: my-skill\ndescription: A useful skill for data analysis\n---\n\n" + "This skill helps analyze CSV data and produce summary statistics." + ) + (tmp_path / "instruction.md").write_text( + "# Analyze CSV\n\n" + "Given a CSV file, read it with pandas, compute mean/median/std for " + "all numeric columns, and output a summary table." + ) + tests_dir = tmp_path / "tests" + tests_dir.mkdir() + (tests_dir / "test_outputs.py").write_text("def test_output():\n assert 'mean' in result\n") + result = scan_directory(tmp_path) + assert len(result["findings"]) == 0 + + +# --------------------------------------------------------------------------- +# Gate registration tests +# --------------------------------------------------------------------------- + + +class TestSkillMdQualityRegistration: + def test_registered(self): + assert "skillmd-quality" in get_all_quality_gate_names() + + def test_get_by_name(self): + gate = get_quality_gate("skillmd-quality") + assert isinstance(gate, SkillMdQualityGate) + + +# --------------------------------------------------------------------------- +# Gate evaluation tests +# --------------------------------------------------------------------------- + + +class TestSkillMdQualityGate: + def test_disabled(self, tmp_path): + policy = GatePolicy(gates={"quality": GatePolicyItem(mode=GateMode.DISABLED)}) + gate = SkillMdQualityGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is True + assert result.mode == GateMode.DISABLED + + def test_no_scan_file(self, tmp_path): + policy = GatePolicy() + gate = SkillMdQualityGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is True + assert "not_found" in result.details.get("status", "") + + def test_no_findings(self, tmp_path): + (tmp_path / "skillmd-quality-scan.json").write_text(json.dumps({"findings": []})) + policy = GatePolicy() + gate = SkillMdQualityGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is True + assert result.score == 1.0 + + def test_warn_mode_always_passes(self, tmp_path): + scan_data = {"findings": [{"severity": "high", "message": "Issue", "rule_id": "q-001"}]} + (tmp_path / "skillmd-quality-scan.json").write_text(json.dumps(scan_data)) + policy = GatePolicy(gates={"quality": GatePolicyItem(mode=GateMode.WARN)}) + gate = SkillMdQualityGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is True + + def test_block_mode_fails_on_high(self, tmp_path): + scan_data = {"findings": [{"severity": "high", "message": "Issue", "rule_id": "q-001"}]} + (tmp_path / "skillmd-quality-scan.json").write_text(json.dumps(scan_data)) + policy = GatePolicy(gates={"quality": GatePolicyItem(mode=GateMode.BLOCK)}) + gate = SkillMdQualityGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is False + + def test_block_mode_missing_fails(self, tmp_path): + policy = GatePolicy(gates={"quality": GatePolicyItem(mode=GateMode.BLOCK)}) + gate = SkillMdQualityGate() + result = gate.evaluate(tmp_path, policy) + assert result.passed is False + + def test_score_calculation(self, tmp_path): + scan_data = { + "findings": [ + {"severity": "medium", "message": "A", "rule_id": "q-001"}, + {"severity": "low", "message": "B", "rule_id": "q-002"}, + ] + } + (tmp_path / "skillmd-quality-scan.json").write_text(json.dumps(scan_data)) + policy = GatePolicy() + gate = SkillMdQualityGate() + result = gate.evaluate(tmp_path, policy) + expected = (0.5 + 0.75) / 2 + assert result.score == pytest.approx(expected) From 773754c232e6771b20245a0a8e860644e93a3ec3 Mon Sep 17 00:00:00 2001 From: Benjamin Kapner Date: Thu, 2 Jul 2026 11:41:05 +0300 Subject: [PATCH 2/2] feat: add circular reference detection between skills Detect circular reference chains between skills in submissions with multiple SKILL.md files (nested layout). Ported from harness-eval-lab circular_references.py. 4 new tests. --- abevalflow/quality/skillmd_quality_scanner.py | 103 ++++++++++++++++++ tests/test_skillmd_quality_scanner.py | 42 +++++++ 2 files changed, 145 insertions(+) diff --git a/abevalflow/quality/skillmd_quality_scanner.py b/abevalflow/quality/skillmd_quality_scanner.py index 968282f..e55df54 100644 --- a/abevalflow/quality/skillmd_quality_scanner.py +++ b/abevalflow/quality/skillmd_quality_scanner.py @@ -387,6 +387,108 @@ def check_generic_advice( return findings +# --- Circular reference patterns --- + +_SKILL_REF_PATTERNS = [ + re.compile(r"/(\w[\w-]+)(?:\s|$|[),\]])"), + re.compile(r"(?:skill|command)[:\s]+[\"']?(\w[\w-]+)[\"']?", re.I), + re.compile( + r"(?:invokes?|calls?|triggers?|runs?)\s+[\"'`]?/?(\w[\w-]+)[\"'`]?", + re.I, + ), +] + + +def _extract_skill_references(body: str, own_name: str) -> set[str]: + refs: set[str] = set() + for pattern in _SKILL_REF_PATTERNS: + for match in pattern.finditer(body): + name = match.group(1) + if name != own_name and len(name) > 1: + refs.add(name) + return refs + + +def _find_cycles(graph: dict[str, set[str]]) -> list[list[str]]: + visited: set[str] = set() + on_stack: set[str] = set() + cycles: list[list[str]] = [] + path: list[str] = [] + + def dfs(node: str) -> None: + visited.add(node) + on_stack.add(node) + path.append(node) + + for neighbor in graph.get(node, set()): + if neighbor not in graph: + continue + if neighbor not in visited: + dfs(neighbor) + elif neighbor in on_stack: + cycle_start = path.index(neighbor) + cycle = path[cycle_start:] + [neighbor] + cycles.append(cycle) + + path.pop() + on_stack.discard(node) + + for node in graph: + if node not in visited: + dfs(node) + + return cycles + + +def check_circular_references(directory: Path) -> list[dict]: + """Detect circular reference chains between skills in a submission.""" + skills_dir = directory / "skills" + if not skills_dir.is_dir(): + return [] + + graph: dict[str, set[str]] = {} + file_map: dict[str, str] = {} + + for skill_md in skills_dir.rglob("SKILL.md"): + skill_name = skill_md.parent.name + if skill_name == "skills": + skill_name = "root" + try: + content = skill_md.read_text(encoding="utf-8", errors="replace") + body = re.sub(r"^---.*?---\s*", "", content, flags=re.DOTALL) + refs = _extract_skill_references(body, skill_name) + graph[skill_name] = refs + file_map[skill_name] = str(skill_md.relative_to(directory)) + except OSError: + continue + + if len(graph) < 2: + return [] + + findings: list[dict] = [] + seen_cycles: set[str] = set() + + for cycle in _find_cycles(graph): + key = " -> ".join(sorted(set(cycle[:-1]))) + if key in seen_cycles: + continue + seen_cycles.add(key) + + chain = " -> ".join(cycle) + first = cycle[0] + findings.append( + { + "severity": "medium", + "rule_id": "quality-circular-reference", + "message": f"Circular reference detected: {chain}", + "file_path": file_map.get(first, ""), + "category": "circular_reference", + } + ) + + return findings + + # --- Main entry point --- @@ -409,6 +511,7 @@ def scan_directory(directory: Path) -> dict: all_findings.extend(check_generic_advice(md_file, relative_to=directory)) all_findings.extend(check_file_completeness(directory)) + all_findings.extend(check_circular_references(directory)) logger.info( "Quality scan: %d files, %d findings", diff --git a/tests/test_skillmd_quality_scanner.py b/tests/test_skillmd_quality_scanner.py index 8ac5912..5a9b9dd 100644 --- a/tests/test_skillmd_quality_scanner.py +++ b/tests/test_skillmd_quality_scanner.py @@ -12,6 +12,7 @@ ) from abevalflow.quality.skillmd_quality_scanner import ( check_broken_references, + check_circular_references, check_description_quality, check_file_completeness, check_generic_advice, @@ -237,6 +238,47 @@ def test_code_block_skipped(self, tmp_path): assert len(check_generic_advice(p)) == 0 +# --------------------------------------------------------------------------- +# scan_directory tests +# --------------------------------------------------------------------------- +# Circular references tests +# --------------------------------------------------------------------------- + + +class TestCircularReferences: + def test_no_cycle(self, tmp_path): + skills = tmp_path / "skills" + (skills / "skill-a").mkdir(parents=True) + (skills / "skill-b").mkdir(parents=True) + (skills / "skill-a" / "SKILL.md").write_text( + "---\nname: skill-a\ndescription: A\n---\n\nThis skill calls /skill-b." + ) + (skills / "skill-b" / "SKILL.md").write_text("---\nname: skill-b\ndescription: B\n---\n\nA standalone skill.") + assert len(check_circular_references(tmp_path)) == 0 + + def test_cycle_detected(self, tmp_path): + skills = tmp_path / "skills" + (skills / "skill-a").mkdir(parents=True) + (skills / "skill-b").mkdir(parents=True) + (skills / "skill-a" / "SKILL.md").write_text( + "---\nname: skill-a\ndescription: A\n---\n\nThis skill calls /skill-b." + ) + (skills / "skill-b" / "SKILL.md").write_text( + "---\nname: skill-b\ndescription: B\n---\n\nThis skill invokes /skill-a." + ) + findings = check_circular_references(tmp_path) + assert any(f["rule_id"] == "quality-circular-reference" for f in findings) + + def test_single_skill_no_cycle(self, tmp_path): + skills = tmp_path / "skills" + skills.mkdir() + (skills / "SKILL.md").write_text("---\nname: my-skill\ndescription: Solo\n---\n\nA single skill.") + assert len(check_circular_references(tmp_path)) == 0 + + def test_no_skills_dir(self, tmp_path): + assert len(check_circular_references(tmp_path)) == 0 + + # --------------------------------------------------------------------------- # scan_directory tests # ---------------------------------------------------------------------------