From 3ddfd042e9836a9221ff62f8c9771eb0f76de0f1 Mon Sep 17 00:00:00 2001 From: mohamed zwanski Date: Thu, 22 Oct 2026 21:52:01 +0100 Subject: [PATCH 1/2] Add quality gates: enforce lint/type-check before feature_mark_passing (#96) New quality_gates.py auto-detects ESLint, Biome, ruff, flake8, tsc and mypy from project config files and runs them with a timeout. Node tools only run from the project's node_modules/.bin (never npx), Python tools from the project's venv or PATH. When enabled in .autoforge/config.json (off by default), feature_mark_passing runs the checks first; in strict mode (default) failing checks block marking the feature passing. Results are stored as evidence in a new quality_result column, and failures are tracked via failure_reason/failure_count. New MCP tools: feature_verify_quality and feature_report_failure. Co-Authored-By: Claude Opus 5.5 --- .claude/templates/coding_prompt.template.md | 11 + .claude/templates/testing_prompt.template.md | 3 +- .github/workflows/ci.yml | 2 + CLAUDE.md | 14 + api/database.py | 25 ++ client.py | 3 + examples/project_config.json | 11 + mcp_server/feature_mcp.py | 169 +++++++- package.json | 1 + quality_gates.py | 363 ++++++++++++++++ test_quality_gates.py | 423 +++++++++++++++++++ 11 files changed, 1022 insertions(+), 3 deletions(-) create mode 100644 examples/project_config.json create mode 100644 quality_gates.py create mode 100644 test_quality_gates.py diff --git a/.claude/templates/coding_prompt.template.md b/.claude/templates/coding_prompt.template.md index 832eb5996..371d4c88a 100644 --- a/.claude/templates/coding_prompt.template.md +++ b/.claude/templates/coding_prompt.template.md @@ -146,6 +146,11 @@ After thorough verification, mark the feature as passing: Use the feature_mark_passing tool with feature_id=42 ``` +If the project has quality gates enabled, `feature_mark_passing` first runs the project's +lint and type checks. If it responds that quality gates failed, the feature was NOT marked +passing: fix the reported errors and call it again. You can run the same checks earlier with +`feature_verify_quality`. + **NEVER:** - Delete features @@ -240,6 +245,12 @@ feature_skip with feature_id={id} # 7. Clear in-progress status (when abandoning a feature) feature_clear_in_progress with feature_id={id} + +# 8. Run lint/type-check quality gates (before marking passing) +feature_verify_quality with feature_id={id} + +# 9. Record why an attempt failed (before skipping/abandoning a feature) +feature_report_failure with feature_id={id} and reason="..." ``` ### RULES: diff --git a/.claude/templates/testing_prompt.template.md b/.claude/templates/testing_prompt.template.md index ee6a08fa2..29aa1620d 100644 --- a/.claude/templates/testing_prompt.template.md +++ b/.claude/templates/testing_prompt.template.md @@ -110,7 +110,8 @@ A regression has been introduced. You MUST fix it: - `feature_get_stats` - Get progress overview (passing/in_progress/total counts) - `feature_get_by_id` - Get your assigned feature details - `feature_mark_failing` - Mark a feature as failing (when you find a regression) -- `feature_mark_passing` - Mark a feature as passing (after fixing a regression) +- `feature_mark_passing` - Mark a feature as passing (after fixing a regression; may be rejected if quality gates fail) +- `feature_verify_quality` - Run the project's lint/type-check quality gates ### Browser Automation (Playwright CLI) Use `playwright-cli` commands for browser interaction. Key commands: diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2c0a6eb4c..999bfae96 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,6 +20,8 @@ jobs: run: ruff check . - name: Run security tests run: python test_security.py + - name: Run quality gate tests + run: python -m pytest test_quality_gates.py ui: runs-on: ubuntu-latest diff --git a/CLAUDE.md b/CLAUDE.md index 6f59910be..f240a9624 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -120,6 +120,7 @@ python test_security_integration.py # Integration tests (9 tests) python -m pytest test_client.py # Client tests (20 tests) python -m pytest test_dependency_resolver.py # Dependency resolver tests (12 tests) python -m pytest test_rate_limit_utils.py # Rate limit tests (22 tests) +python -m pytest test_quality_gates.py # Quality gate tests ``` ### React UI @@ -172,6 +173,7 @@ Publishing: `npm publish` (triggers `prepublishOnly` which builds UI, then publi - `auth.py` - Authentication error detection for Claude CLI - `env_constants.py` - Shared environment variable constants (API_ENV_VARS) used by client.py and chat sessions - `rate_limit_utils.py` - Rate limit detection, retry parsing, exponential backoff with jitter +- `quality_gates.py` - Lint/type-check auto-detection and enforcement before `feature_mark_passing` - `api/database.py` - SQLAlchemy models (Feature, Schedule, ScheduleOverride) - `api/dependency_resolver.py` - Cycle detection (Kahn's algorithm + DFS) and dependency validation - `api/migration.py` - JSON-to-SQLite migration utility @@ -244,6 +246,18 @@ MCP tools available to the agent: - `feature_add_dependency` - Add dependency between features (with cycle detection) - `feature_remove_dependency` - Remove a dependency - `feature_set_dependencies` - Set all dependencies for a feature at once +- `feature_verify_quality` - Run lint/type-check quality gates and store the result on the feature +- `feature_report_failure` - Record why a feature attempt failed (increments `failure_count`) + +#### Quality Gates + +`quality_gates.py` auto-detects a project's linters (ESLint, Biome, ruff, flake8) and type checkers (tsc, mypy) from its config files and runs them. Node tools only run from the project's `node_modules/.bin`; Python tools from the project's `.venv`/`venv` or PATH. Disabled by default; enable per project in `.autoforge/config.json` (see `examples/project_config.json`): + +```json +{"quality_gates": {"enabled": true, "strict_mode": true, "timeout_seconds": 120, "checks": {"lint": true, "type_check": true}}} +``` + +When enabled, `feature_mark_passing` runs the checks first. In `strict_mode` (default) a failing check blocks marking the feature passing, records `failure_reason`, and increments `failure_count`; with `strict_mode: false` it only warns. The latest report is stored in the feature's `quality_result` column as test evidence. ### React UI (ui/) diff --git a/api/database.py b/api/database.py index 523ea2243..b508a5d09 100644 --- a/api/database.py +++ b/api/database.py @@ -66,6 +66,11 @@ class Feature(Base): human_input_request = Column(JSON, nullable=True, default=None) # Agent's structured request human_input_response = Column(JSON, nullable=True, default=None) # Human's response + # Quality gates: evidence from the last lint/type-check run and failure tracking + quality_result = Column(JSON, nullable=True, default=None) # Last quality report (see quality_gates.py) + failure_reason = Column(Text, nullable=True, default=None) # Why the last attempt failed + failure_count = Column(Integer, nullable=False, default=0) # Number of failed attempts + def to_dict(self) -> dict: """Convert feature to dictionary for JSON serialization.""" return { @@ -84,6 +89,10 @@ def to_dict(self) -> dict: "needs_human_input": self.needs_human_input if self.needs_human_input is not None else False, "human_input_request": self.human_input_request, "human_input_response": self.human_input_response, + # Quality gate fields + "quality_result": self.quality_result, + "failure_reason": self.failure_reason, + "failure_count": self.failure_count if self.failure_count is not None else 0, } def get_dependencies_safe(self) -> list[int]: @@ -326,6 +335,21 @@ def _migrate_add_human_input_columns(engine) -> None: conn.commit() +def _migrate_add_quality_columns(engine) -> None: + """Add quality gate columns to existing databases that don't have them.""" + with engine.connect() as conn: + result = conn.execute(text("PRAGMA table_info(features)")) + columns = [row[1] for row in result.fetchall()] + + if "quality_result" not in columns: + conn.execute(text("ALTER TABLE features ADD COLUMN quality_result TEXT DEFAULT NULL")) + if "failure_reason" not in columns: + conn.execute(text("ALTER TABLE features ADD COLUMN failure_reason TEXT DEFAULT NULL")) + if "failure_count" not in columns: + conn.execute(text("ALTER TABLE features ADD COLUMN failure_count INTEGER NOT NULL DEFAULT 0")) + conn.commit() + + def _migrate_add_schedules_tables(engine) -> None: """Create schedules and schedule_overrides tables if they don't exist.""" from sqlalchemy import inspect @@ -450,6 +474,7 @@ def create_database(project_dir: Path) -> tuple: _migrate_add_dependencies_column(engine) _migrate_add_testing_columns(engine) _migrate_add_human_input_columns(engine) + _migrate_add_quality_columns(engine) # Migrate to add schedules tables _migrate_add_schedules_tables(engine) diff --git a/client.py b/client.py index 26a710f50..e66dc4fce 100644 --- a/client.py +++ b/client.py @@ -159,6 +159,8 @@ def get_extra_read_paths() -> list[Path]: "mcp__features__feature_mark_failing", "mcp__features__feature_skip", "mcp__features__feature_clear_in_progress", + "mcp__features__feature_verify_quality", + "mcp__features__feature_report_failure", ] TESTING_AGENT_TOOLS = [ @@ -170,6 +172,7 @@ def get_extra_read_paths() -> list[Path]: "mcp__features__feature_get_graph", "mcp__features__feature_mark_passing", "mcp__features__feature_mark_failing", + "mcp__features__feature_verify_quality", ] INITIALIZER_AGENT_TOOLS = [ diff --git a/examples/project_config.json b/examples/project_config.json new file mode 100644 index 000000000..2436ab047 --- /dev/null +++ b/examples/project_config.json @@ -0,0 +1,11 @@ +{ + "quality_gates": { + "enabled": true, + "strict_mode": true, + "timeout_seconds": 120, + "checks": { + "lint": true, + "type_check": true + } + } +} diff --git a/mcp_server/feature_mcp.py b/mcp_server/feature_mcp.py index 71e217d41..c94346086 100755 --- a/mcp_server/feature_mcp.py +++ b/mcp_server/feature_mcp.py @@ -22,6 +22,8 @@ - feature_get_ready: Get features ready to implement - feature_get_blocked: Get features blocked by dependencies (with limit) - feature_get_graph: Get the dependency graph +- feature_verify_quality: Run lint/type-check quality gates and record the result +- feature_report_failure: Record why a feature attempt failed Note: Feature selection (which feature to work on) is handled by the orchestrator, not by agents. Agents receive pre-assigned feature IDs. @@ -48,6 +50,8 @@ would_create_circular_dependency, ) from api.migration import migrate_json_to_sqlite +from quality_gates import load_config as load_quality_config +from quality_gates import run_quality_checks # Configuration from environment PROJECT_DIR = Path(os.environ.get("PROJECT_DIR", ".")).resolve() @@ -240,12 +244,41 @@ def feature_mark_passing( Updates the feature's passes field to true and clears the in_progress flag. Use this after you have implemented the feature and verified it works correctly. + If quality gates are enabled for the project, lint and type checks run first. + In strict mode the feature is NOT marked passing while any check fails; fix + the reported problems and call this tool again. + Args: feature_id: The ID of the feature to mark as passing Returns: - JSON with success confirmation: {success, feature_id, name} + JSON with success confirmation: {success, feature_id, name}, plus a + quality_gates summary when quality gates are enabled. """ + quality_config = load_quality_config(PROJECT_DIR) + quality_report = None + if quality_config.enabled: + session = get_session() + try: + feature = session.query(Feature).filter(Feature.id == feature_id).first() + if feature is None: + return json.dumps({"error": f"Feature with ID {feature_id} not found"}) + if feature.passes: + return json.dumps({"error": f"Feature with ID {feature_id} is already passing"}) + finally: + session.close() + + # Run checks outside any DB transaction - they can take a while + quality_report = run_quality_checks(PROJECT_DIR, quality_config) + if not quality_report.passed and quality_config.strict_mode: + reason = f"Quality gates failed: {quality_report.message}" + _store_quality_result(feature_id, quality_report.to_dict(), failure_reason=reason) + return json.dumps({ + "error": f"Feature #{feature_id} was NOT marked passing. {reason}. " + "Fix the problems below and call feature_mark_passing again.", + "quality_gates": _summarize_report(quality_report.to_dict()), + }) + session = get_session() try: # Atomic update with state guard - prevents double-pass in parallel mode @@ -254,6 +287,11 @@ def feature_mark_passing( SET passes = 1, in_progress = 0 WHERE id = :id AND passes = 0 """), {"id": feature_id}) + if quality_report is not None: + session.execute( + text("UPDATE features SET quality_result = :qr WHERE id = :id"), + {"id": feature_id, "qr": json.dumps(quality_report.to_dict())}, + ) session.commit() if result.rowcount == 0: @@ -267,7 +305,12 @@ def feature_mark_passing( # Get the feature name for the response feature = session.query(Feature).filter(Feature.id == feature_id).first() - return json.dumps({"success": True, "feature_id": feature_id, "name": feature.name}) + response: dict = {"success": True, "feature_id": feature_id, "name": feature.name} + if quality_report is not None: + response["quality_gates"] = _summarize_report(quality_report.to_dict()) + if not quality_report.passed: + response["warning"] = f"Quality gates failed (strict_mode off): {quality_report.message}" + return json.dumps(response) except Exception as e: session.rollback() return json.dumps({"error": f"Failed to mark feature passing: {str(e)}"}) @@ -275,6 +318,128 @@ def feature_mark_passing( session.close() +def _summarize_report(report: dict) -> dict: + """Return a compact view of a quality report for MCP responses.""" + return { + "passed": report["passed"], + "message": report["message"], + "checks": [ + { + "name": c["name"], + "category": c["category"], + "passed": c["passed"], + # Only include output for failures to keep responses small + **({"output": c["output"]} if not c["passed"] else {}), + } + for c in report["checks"] + ], + } + + +def _store_quality_result(feature_id: int, report: dict, failure_reason: str | None = None) -> None: + """Persist a quality report on a feature; a failure_reason also bumps failure_count.""" + session = get_session() + try: + if failure_reason is None: + session.execute( + text("UPDATE features SET quality_result = :qr WHERE id = :id"), + {"id": feature_id, "qr": json.dumps(report)}, + ) + else: + session.execute(text(""" + UPDATE features + SET quality_result = :qr, + failure_reason = :reason, + failure_count = COALESCE(failure_count, 0) + 1 + WHERE id = :id + """), {"id": feature_id, "qr": json.dumps(report), "reason": failure_reason}) + session.commit() + except Exception: + session.rollback() + raise + finally: + session.close() + + +@mcp.tool() +def feature_verify_quality( + feature_id: Annotated[int, Field(description="The ID of the feature the checks are for", ge=1)] +) -> str: + """Run the project's lint and type checks and record the result on a feature. + + Auto-detects ESLint, Biome, ruff, flake8, TypeScript (tsc) and mypy from the + project's config files. Use this before feature_mark_passing to find and fix + lint/type errors early. The result is stored on the feature as evidence. + + Runs even if quality gates are not enabled in .autoforge/config.json. + + Args: + feature_id: The ID of the feature the checks are for + + Returns: + JSON with: passed (bool), message, checks (name, category, passed, output on failure) + """ + session = get_session() + try: + feature = session.query(Feature).filter(Feature.id == feature_id).first() + if feature is None: + return json.dumps({"error": f"Feature with ID {feature_id} not found"}) + finally: + session.close() + + try: + report = run_quality_checks(PROJECT_DIR).to_dict() + _store_quality_result(feature_id, report) + except Exception as e: + return json.dumps({"error": f"Failed to run quality checks: {str(e)}"}) + + return json.dumps(_summarize_report(report)) + + +@mcp.tool() +def feature_report_failure( + feature_id: Annotated[int, Field(description="The ID of the feature that failed", ge=1)], + reason: Annotated[str, Field(min_length=1, max_length=2000, description="Why the attempt failed")] +) -> str: + """Record that an attempt at a feature failed, and why. + + Stores the reason and increments the feature's failure_count so repeated + failures are visible to humans and later sessions. Does not change the + feature's passes or in_progress status. + + Args: + feature_id: The ID of the feature that failed + reason: Short explanation of what went wrong + + Returns: + JSON with: feature_id, failure_count, failure_reason + """ + session = get_session() + try: + result = session.execute(text(""" + UPDATE features + SET failure_reason = :reason, + failure_count = COALESCE(failure_count, 0) + 1 + WHERE id = :id + """), {"id": feature_id, "reason": reason}) + session.commit() + + if result.rowcount == 0: + return json.dumps({"error": f"Feature with ID {feature_id} not found"}) + + feature = session.query(Feature).filter(Feature.id == feature_id).first() + return json.dumps({ + "feature_id": feature_id, + "failure_count": feature.failure_count, + "failure_reason": feature.failure_reason, + }) + except Exception as e: + session.rollback() + return json.dumps({"error": f"Failed to report failure: {str(e)}"}) + finally: + session.close() + + @mcp.tool() def feature_mark_failing( feature_id: Annotated[int, Field(description="The ID of the feature to mark as failing", ge=1)] diff --git a/package.json b/package.json index 9c547ddfc..fcc64260b 100644 --- a/package.json +++ b/package.json @@ -32,6 +32,7 @@ "parallel_orchestrator.py", "progress.py", "prompts.py", + "quality_gates.py", "registry.py", "rate_limit_utils.py", "security.py", diff --git a/quality_gates.py b/quality_gates.py new file mode 100644 index 000000000..306d0380a --- /dev/null +++ b/quality_gates.py @@ -0,0 +1,363 @@ +""" +Quality Gates +============= + +Auto-detects a project's linters and type checkers, runs them, and reports +structured results. Used by the feature MCP server to verify code before a +feature is marked as passing. + +Configuration lives in ``{project_dir}/.autoforge/config.json``:: + + { + "quality_gates": { + "enabled": true, + "strict_mode": true, + "timeout_seconds": 120, + "checks": {"lint": true, "type_check": true} + } + } + +Quality gates are disabled by default. When enabled with ``strict_mode`` (the +default once enabled), ``feature_mark_passing`` refuses to mark a feature as +passing while any check fails. With ``strict_mode: false`` the checks still +run and are recorded, but failures are reported as warnings only. + +Detected tools (only run when the project is configured for them): + +- Lint: ESLint, Biome, ruff, flake8 +- Type check: TypeScript (``tsc --noEmit``, or ``tsc --build --noEmit`` for + project references), mypy + +Node tools are only run from the project's ``node_modules/.bin`` and Python +tools only when installed, so a check never triggers a package download. +""" + +from __future__ import annotations + +import json +import os +import re +import shutil +import subprocess +import sys +import time +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +DEFAULT_TIMEOUT_SECONDS = 120 +MAX_TIMEOUT_SECONDS = 600 +# Keep stored output small - it lands in the features database and in MCP responses +MAX_OUTPUT_CHARS = 4000 + +ESLINT_CONFIGS = ( + "eslint.config.js", "eslint.config.mjs", "eslint.config.cjs", + "eslint.config.ts", "eslint.config.mts", "eslint.config.cts", + ".eslintrc", ".eslintrc.js", ".eslintrc.cjs", ".eslintrc.json", + ".eslintrc.yml", ".eslintrc.yaml", +) +BIOME_CONFIGS = ("biome.json", "biome.jsonc") +FLAKE8_CONFIGS = (".flake8",) + +ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]") + + +@dataclass +class QualityGateConfig: + """Effective quality-gate settings for a project.""" + + enabled: bool = False + strict_mode: bool = True + lint: bool = True + type_check: bool = True + timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS + + +@dataclass +class QualityCheck: + """A single check to run: a name, a category and the command line.""" + + name: str + category: str # "lint" or "type_check" + command: list[str] + + +@dataclass +class CheckResult: + """Outcome of running one quality check.""" + + name: str + category: str + command: str + passed: bool + exit_code: int | None + duration_seconds: float + output: str + timed_out: bool = False + + +@dataclass +class QualityReport: + """Aggregate result of all quality checks for a project.""" + + passed: bool + checks: list[CheckResult] = field(default_factory=list) + strict_mode: bool = True + timestamp: str = "" + message: str = "" + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +def _config_path(project_dir: Path) -> Path: + return project_dir / ".autoforge" / "config.json" + + +def load_config(project_dir: Path) -> QualityGateConfig: + """Load quality gate settings from ``.autoforge/config.json``. + + Missing or malformed config falls back to defaults (disabled). + """ + config = QualityGateConfig() + path = _config_path(project_dir) + if not path.is_file(): + return config + + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as e: + print(f"[quality_gates] Ignoring unreadable {path}: {e}", file=sys.stderr) + return config + + section = data.get("quality_gates") if isinstance(data, dict) else None + if not isinstance(section, dict): + return config + + if isinstance(section.get("enabled"), bool): + config.enabled = section["enabled"] + if isinstance(section.get("strict_mode"), bool): + config.strict_mode = section["strict_mode"] + + timeout = section.get("timeout_seconds") + if isinstance(timeout, int) and not isinstance(timeout, bool) and timeout > 0: + config.timeout_seconds = min(timeout, MAX_TIMEOUT_SECONDS) + + checks = section.get("checks") + if isinstance(checks, dict): + if isinstance(checks.get("lint"), bool): + config.lint = checks["lint"] + if isinstance(checks.get("type_check"), bool): + config.type_check = checks["type_check"] + + return config + + +def _node_bin(project_dir: Path, name: str) -> str | None: + """Return the project-local Node binary, or None if not installed.""" + bin_dir = project_dir / "node_modules" / ".bin" + candidates = [f"{name}.cmd", name] if sys.platform == "win32" else [name] + for candidate in candidates: + path = bin_dir / candidate + if path.is_file(): + return str(path) + return None + + +def _python_tool(project_dir: Path, name: str) -> list[str] | None: + """Return a command prefix for a Python tool, or None if unavailable. + + Prefers the project's own virtualenv so the tool sees the project's + dependencies, then falls back to PATH. + """ + scripts = "Scripts" if sys.platform == "win32" else "bin" + exe = f"{name}.exe" if sys.platform == "win32" else name + for venv in (".venv", "venv"): + path = project_dir / venv / scripts / exe + if path.is_file(): + return [str(path)] + found = shutil.which(name) + if found: + return [found] + return None + + +def _read_text(path: Path) -> str: + try: + return path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return "" + + +def _package_json(project_dir: Path) -> dict[str, Any]: + text = _read_text(project_dir / "package.json") + if not text: + return {} + try: + data = json.loads(text) + except ValueError: + return {} + return data if isinstance(data, dict) else {} + + +def _has_python_sources(project_dir: Path) -> bool: + markers = ("pyproject.toml", "setup.py", "setup.cfg", "requirements.txt") + return any((project_dir / m).is_file() for m in markers) + + +def detect_checks(project_dir: Path, config: QualityGateConfig | None = None) -> list[QualityCheck]: + """Detect which linters and type checkers apply to a project.""" + # Absolute paths: the commands run with cwd=project_dir + project_dir = project_dir.resolve() + if config is None: + config = load_config(project_dir) + + checks: list[QualityCheck] = [] + pkg = _package_json(project_dir) + pyproject = _read_text(project_dir / "pyproject.toml") + setup_cfg = _read_text(project_dir / "setup.cfg") + tox_ini = _read_text(project_dir / "tox.ini") + is_python = _has_python_sources(project_dir) + + if config.lint: + has_eslint_config = ( + any((project_dir / c).is_file() for c in ESLINT_CONFIGS) or "eslintConfig" in pkg + ) + eslint = _node_bin(project_dir, "eslint") + if has_eslint_config and eslint: + checks.append(QualityCheck("eslint", "lint", [eslint, "."])) + + biome = _node_bin(project_dir, "biome") + if any((project_dir / c).is_file() for c in BIOME_CONFIGS) and biome: + checks.append(QualityCheck("biome", "lint", [biome, "lint", "."])) + + has_ruff_config = ( + (project_dir / "ruff.toml").is_file() + or (project_dir / ".ruff.toml").is_file() + or "[tool.ruff" in pyproject + ) + has_flake8_config = ( + any((project_dir / c).is_file() for c in FLAKE8_CONFIGS) + or "[flake8]" in setup_cfg + or "[flake8]" in tox_ini + ) + ruff = _python_tool(project_dir, "ruff") + flake8 = _python_tool(project_dir, "flake8") + if has_ruff_config and ruff: + checks.append(QualityCheck("ruff", "lint", [*ruff, "check", "."])) + elif has_flake8_config and flake8: + checks.append(QualityCheck("flake8", "lint", [*flake8, "."])) + + if config.type_check: + tsc = _node_bin(project_dir, "tsc") + tsconfig = project_dir / "tsconfig.json" + if tsconfig.is_file() and tsc: + # Solution-style configs (Vite's default) list no files of their own and + # only reference sub-projects, which plain `tsc` would skip + if '"references"' in _read_text(tsconfig): + checks.append(QualityCheck("tsc", "type_check", [tsc, "--build", "--noEmit"])) + else: + checks.append(QualityCheck("tsc", "type_check", [tsc, "--noEmit"])) + + has_mypy_config = ( + (project_dir / "mypy.ini").is_file() + or (project_dir / ".mypy.ini").is_file() + or "[tool.mypy" in pyproject + or "[mypy" in setup_cfg + ) + mypy = _python_tool(project_dir, "mypy") + if is_python and has_mypy_config and mypy: + checks.append(QualityCheck("mypy", "type_check", [*mypy, "."])) + + return checks + + +def _clean_output(output: str) -> str: + """Strip color codes and keep stored output to a bounded size.""" + output = ANSI_ESCAPE_RE.sub("", output).strip() + if len(output) <= MAX_OUTPUT_CHARS: + return output + # Keep the tail - linters usually print the summary last + return "...(truncated)...\n" + output[-MAX_OUTPUT_CHARS:] + + +def run_check(check: QualityCheck, project_dir: Path, timeout: int) -> CheckResult: + """Run a single check in the project directory.""" + start = time.monotonic() + env = {**os.environ, "CI": "1", "NO_COLOR": "1"} + # Some tools treat any FORCE_COLOR value (even "0") as "force color on" + env.pop("FORCE_COLOR", None) + try: + proc = subprocess.run( + check.command, + cwd=project_dir, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout, + env=env, + stdin=subprocess.DEVNULL, + ) + output = (proc.stdout or "") + (proc.stderr or "") + return CheckResult( + name=check.name, + category=check.category, + command=" ".join([Path(check.command[0]).name, *check.command[1:]]), + passed=proc.returncode == 0, + exit_code=proc.returncode, + duration_seconds=round(time.monotonic() - start, 2), + output=_clean_output(output), + ) + except subprocess.TimeoutExpired as e: + partial = e.stdout or "" + if isinstance(partial, bytes): + partial = partial.decode("utf-8", errors="replace") + return CheckResult( + name=check.name, + category=check.category, + command=" ".join([Path(check.command[0]).name, *check.command[1:]]), + passed=False, + exit_code=None, + duration_seconds=round(time.monotonic() - start, 2), + output=_clean_output(f"{partial}\nTimed out after {timeout}s"), + timed_out=True, + ) + except OSError as e: + return CheckResult( + name=check.name, + category=check.category, + command=" ".join([Path(check.command[0]).name, *check.command[1:]]), + passed=False, + exit_code=None, + duration_seconds=round(time.monotonic() - start, 2), + output=f"Failed to run {check.name}: {e}", + ) + + +def run_quality_checks(project_dir: Path, config: QualityGateConfig | None = None) -> QualityReport: + """Detect and run all applicable quality checks for a project.""" + project_dir = project_dir.resolve() + if config is None: + config = load_config(project_dir) + + checks = detect_checks(project_dir, config) + results = [run_check(c, project_dir, config.timeout_seconds) for c in checks] + passed = all(r.passed for r in results) + + if not results: + message = "No linters or type checkers detected; nothing to verify" + elif passed: + message = f"All {len(results)} quality check(s) passed: " + ", ".join(r.name for r in results) + else: + failed = [r.name for r in results if not r.passed] + message = f"{len(failed)} of {len(results)} quality check(s) failed: " + ", ".join(failed) + + return QualityReport( + passed=passed, + checks=results, + strict_mode=config.strict_mode, + timestamp=datetime.now(timezone.utc).isoformat(), + message=message, + ) diff --git a/test_quality_gates.py b/test_quality_gates.py new file mode 100644 index 000000000..a2f9a845b --- /dev/null +++ b/test_quality_gates.py @@ -0,0 +1,423 @@ +""" +Quality Gates Tests +=================== + +Tests for quality_gates.py and the quality gate integration in the feature +MCP server (feature_mark_passing, feature_verify_quality, feature_report_failure). + +Run with: python -m pytest test_quality_gates.py +""" + +import json +import sqlite3 +import sys +from pathlib import Path + +import pytest + +import quality_gates +from api.database import Feature, create_database, dispose_engine +from mcp_server import feature_mcp +from quality_gates import QualityCheck, QualityGateConfig + + +def _write_config(project_dir: Path, section: dict) -> None: + autoforge_dir = project_dir / ".autoforge" + autoforge_dir.mkdir(parents=True, exist_ok=True) + (autoforge_dir / "config.json").write_text(json.dumps({"quality_gates": section})) + + +def _py_check(name: str, category: str, code: str) -> QualityCheck: + return QualityCheck(name, category, [sys.executable, "-c", code]) + + +# ============================================================================= +# Config loading +# ============================================================================= + + +def test_config_defaults_to_disabled(tmp_path): + config = quality_gates.load_config(tmp_path) + assert config.enabled is False + assert config.strict_mode is True + assert config.lint is True + assert config.type_check is True + + +def test_config_reads_settings(tmp_path): + _write_config(tmp_path, { + "enabled": True, + "strict_mode": False, + "timeout_seconds": 30, + "checks": {"lint": False, "type_check": True}, + }) + config = quality_gates.load_config(tmp_path) + assert config.enabled is True + assert config.strict_mode is False + assert config.timeout_seconds == 30 + assert config.lint is False + assert config.type_check is True + + +def test_config_ignores_bad_values(tmp_path): + _write_config(tmp_path, {"enabled": "yes", "timeout_seconds": -5, "checks": "all"}) + config = quality_gates.load_config(tmp_path) + assert config.enabled is False + assert config.timeout_seconds == quality_gates.DEFAULT_TIMEOUT_SECONDS + + +def test_config_caps_timeout(tmp_path): + _write_config(tmp_path, {"timeout_seconds": 99999}) + assert quality_gates.load_config(tmp_path).timeout_seconds == quality_gates.MAX_TIMEOUT_SECONDS + + +def test_config_tolerates_malformed_json(tmp_path): + (tmp_path / ".autoforge").mkdir() + (tmp_path / ".autoforge" / "config.json").write_text("{not json") + assert quality_gates.load_config(tmp_path).enabled is False + + +# ============================================================================= +# Detection +# ============================================================================= + + +def _fake_node_bin(project_dir: Path, name: str) -> None: + bin_dir = project_dir / "node_modules" / ".bin" + bin_dir.mkdir(parents=True, exist_ok=True) + filename = f"{name}.cmd" if sys.platform == "win32" else name + (bin_dir / filename).write_text("") + + +def test_detects_nothing_in_empty_project(tmp_path): + assert quality_gates.detect_checks(tmp_path, QualityGateConfig(enabled=True)) == [] + + +def test_detects_eslint_and_tsc(tmp_path): + (tmp_path / "eslint.config.js").write_text("export default []") + (tmp_path / "tsconfig.json").write_text("{}") + _fake_node_bin(tmp_path, "eslint") + _fake_node_bin(tmp_path, "tsc") + + checks = quality_gates.detect_checks(tmp_path, QualityGateConfig(enabled=True)) + assert [c.name for c in checks] == ["eslint", "tsc"] + assert checks[1].command[1:] == ["--noEmit"] + + +def test_tsc_uses_build_mode_for_project_references(tmp_path): + (tmp_path / "tsconfig.json").write_text('{"files": [], "references": [{"path": "./tsconfig.app.json"}]}') + _fake_node_bin(tmp_path, "tsc") + checks = quality_gates.detect_checks(tmp_path, QualityGateConfig(enabled=True)) + assert checks[0].command[1:] == ["--build", "--noEmit"] + + +def test_node_tools_require_local_install(tmp_path): + """Config alone is not enough: we never fall back to npx (which may download packages).""" + (tmp_path / "eslint.config.js").write_text("export default []") + (tmp_path / "tsconfig.json").write_text("{}") + assert quality_gates.detect_checks(tmp_path, QualityGateConfig(enabled=True)) == [] + + +def test_detects_biome(tmp_path): + (tmp_path / "biome.json").write_text("{}") + _fake_node_bin(tmp_path, "biome") + checks = quality_gates.detect_checks(tmp_path, QualityGateConfig(enabled=True)) + assert [c.name for c in checks] == ["biome"] + assert checks[0].command[1:] == ["lint", "."] + + +def test_detects_python_tools(tmp_path, monkeypatch): + (tmp_path / "pyproject.toml").write_text("[tool.ruff]\nline-length = 100\n\n[tool.mypy]\n") + monkeypatch.setattr(quality_gates.shutil, "which", lambda name: f"/usr/bin/{name}") + + checks = quality_gates.detect_checks(tmp_path, QualityGateConfig(enabled=True)) + assert [(c.name, c.category) for c in checks] == [("ruff", "lint"), ("mypy", "type_check")] + + +def test_prefers_ruff_over_flake8(tmp_path, monkeypatch): + (tmp_path / "pyproject.toml").write_text("[tool.ruff]\n") + (tmp_path / ".flake8").write_text("[flake8]\n") + monkeypatch.setattr(quality_gates.shutil, "which", lambda name: f"/usr/bin/{name}") + + names = [c.name for c in quality_gates.detect_checks(tmp_path, QualityGateConfig(enabled=True))] + assert names == ["ruff"] + + +def test_detects_flake8(tmp_path, monkeypatch): + (tmp_path / "setup.cfg").write_text("[flake8]\nmax-line-length = 100\n") + monkeypatch.setattr(quality_gates.shutil, "which", lambda name: f"/usr/bin/{name}") + + names = [c.name for c in quality_gates.detect_checks(tmp_path, QualityGateConfig(enabled=True))] + assert names == ["flake8"] + + +def test_prefers_project_venv(tmp_path, monkeypatch): + (tmp_path / "ruff.toml").write_text("") + scripts = tmp_path / ".venv" / ("Scripts" if sys.platform == "win32" else "bin") + scripts.mkdir(parents=True) + exe = scripts / ("ruff.exe" if sys.platform == "win32" else "ruff") + exe.write_text("") + monkeypatch.setattr(quality_gates.shutil, "which", lambda name: f"/usr/bin/{name}") + + checks = quality_gates.detect_checks(tmp_path, QualityGateConfig(enabled=True)) + assert checks[0].command[0] == str(exe) + + +def test_respects_disabled_check_categories(tmp_path): + (tmp_path / "eslint.config.js").write_text("export default []") + (tmp_path / "tsconfig.json").write_text("{}") + _fake_node_bin(tmp_path, "eslint") + _fake_node_bin(tmp_path, "tsc") + + lint_only = QualityGateConfig(enabled=True, type_check=False) + assert [c.name for c in quality_gates.detect_checks(tmp_path, lint_only)] == ["eslint"] + + types_only = QualityGateConfig(enabled=True, lint=False) + assert [c.name for c in quality_gates.detect_checks(tmp_path, types_only)] == ["tsc"] + + +# ============================================================================= +# Running checks +# ============================================================================= + + +def test_run_check_pass_and_fail(tmp_path): + ok = quality_gates.run_check(_py_check("ok", "lint", "print('clean')"), tmp_path, 30) + assert ok.passed is True + assert ok.exit_code == 0 + assert "clean" in ok.output + + bad = quality_gates.run_check( + _py_check("bad", "lint", "import sys; print('error: x'); sys.exit(2)"), tmp_path, 30 + ) + assert bad.passed is False + assert bad.exit_code == 2 + assert "error: x" in bad.output + + +def test_run_check_timeout(tmp_path): + result = quality_gates.run_check(_py_check("slow", "lint", "import time; time.sleep(5)"), tmp_path, 1) + assert result.passed is False + assert result.timed_out is True + assert "Timed out" in result.output + + +def test_run_check_missing_binary(tmp_path): + check = QualityCheck("ghost", "lint", [str(tmp_path / "does-not-exist")]) + result = quality_gates.run_check(check, tmp_path, 5) + assert result.passed is False + assert "Failed to run ghost" in result.output + + +def test_run_check_truncates_output(tmp_path): + code = f"print('x' * {quality_gates.MAX_OUTPUT_CHARS * 2})" + result = quality_gates.run_check(_py_check("noisy", "lint", code), tmp_path, 30) + assert len(result.output) < quality_gates.MAX_OUTPUT_CHARS + 100 + + +def test_report_with_no_checks_passes(tmp_path): + report = quality_gates.run_quality_checks(tmp_path, QualityGateConfig(enabled=True)) + assert report.passed is True + assert report.checks == [] + assert "No linters" in report.message + + +# ============================================================================= +# Database migration +# ============================================================================= + + +def test_migration_adds_quality_columns(tmp_path): + db_dir = tmp_path / ".autoforge" + db_dir.mkdir() + conn = sqlite3.connect(db_dir / "features.db") + conn.execute(""" + CREATE TABLE features ( + id INTEGER PRIMARY KEY, priority INTEGER NOT NULL, category VARCHAR(100) NOT NULL, + name VARCHAR(255) NOT NULL, description TEXT NOT NULL, steps JSON NOT NULL, + passes BOOLEAN NOT NULL + ) + """) + conn.execute( + "INSERT INTO features (priority, category, name, description, steps, passes) " + "VALUES (1, 'c', 'old', 'd', '[]', 0)" + ) + conn.commit() + conn.close() + + try: + _, session_maker = create_database(tmp_path) + session = session_maker() + try: + feature = session.query(Feature).first() + data = feature.to_dict() + assert data["failure_count"] == 0 + assert data["failure_reason"] is None + assert data["quality_result"] is None + finally: + session.close() + finally: + dispose_engine(tmp_path) + + +# ============================================================================= +# MCP integration +# ============================================================================= + + +@pytest.fixture +def mcp_project(tmp_path, monkeypatch): + """A project with one in-progress feature, wired into the feature MCP module.""" + engine, session_maker = create_database(tmp_path) + session = session_maker() + session.add(Feature( + id=1, priority=1, category="core", name="Login", description="d", + steps=["s"], passes=False, in_progress=True, + )) + session.commit() + session.close() + + monkeypatch.setattr(feature_mcp, "PROJECT_DIR", tmp_path) + monkeypatch.setattr(feature_mcp, "_session_maker", session_maker) + yield tmp_path, session_maker + dispose_engine(tmp_path) + + +def _feature(session_maker) -> Feature: + session = session_maker() + try: + feature: Feature = session.query(Feature).filter(Feature.id == 1).one() + return feature + finally: + session.close() + + +def _set_checks(monkeypatch, checks: list[QualityCheck]) -> None: + monkeypatch.setattr(quality_gates, "detect_checks", lambda project_dir, config=None: checks) + + +FAILING = _py_check("ruff", "lint", "import sys; print('E501 line too long'); sys.exit(1)") +PASSING = _py_check("tsc", "type_check", "pass") + + +def test_mark_passing_unchanged_when_disabled(mcp_project, monkeypatch): + project_dir, session_maker = mcp_project + _set_checks(monkeypatch, [FAILING]) + + result = json.loads(feature_mcp.feature_mark_passing(1)) + assert result["success"] is True + assert "quality_gates" not in result + feature = _feature(session_maker) + assert feature.passes is True + assert feature.quality_result is None + + +def test_strict_mode_blocks_failing_checks(mcp_project, monkeypatch): + project_dir, session_maker = mcp_project + _write_config(project_dir, {"enabled": True}) + _set_checks(monkeypatch, [PASSING, FAILING]) + + result = json.loads(feature_mcp.feature_mark_passing(1)) + assert "error" in result + assert "NOT marked passing" in result["error"] + failed = [c for c in result["quality_gates"]["checks"] if not c["passed"]] + assert [c["name"] for c in failed] == ["ruff"] + assert "E501" in failed[0]["output"] + + feature = _feature(session_maker) + assert feature.passes is False + assert feature.in_progress is True # Agent keeps the feature to fix it + assert feature.failure_count == 1 + assert "ruff" in feature.failure_reason + assert feature.quality_result["passed"] is False + + # A second blocked attempt increments the counter + feature_mcp.feature_mark_passing(1) + assert _feature(session_maker).failure_count == 2 + + +def test_strict_mode_passes_when_checks_pass(mcp_project, monkeypatch): + project_dir, session_maker = mcp_project + _write_config(project_dir, {"enabled": True}) + _set_checks(monkeypatch, [PASSING]) + + result = json.loads(feature_mcp.feature_mark_passing(1)) + assert result["success"] is True + assert result["quality_gates"]["passed"] is True + + feature = _feature(session_maker) + assert feature.passes is True + assert feature.in_progress is False + assert feature.quality_result["checks"][0]["name"] == "tsc" + + +def test_non_strict_mode_warns_but_marks_passing(mcp_project, monkeypatch): + project_dir, session_maker = mcp_project + _write_config(project_dir, {"enabled": True, "strict_mode": False}) + _set_checks(monkeypatch, [FAILING]) + + result = json.loads(feature_mcp.feature_mark_passing(1)) + assert result["success"] is True + assert "warning" in result + feature = _feature(session_maker) + assert feature.passes is True + assert feature.quality_result["passed"] is False + + +def test_mark_passing_already_passing_skips_checks(mcp_project, monkeypatch): + project_dir, _ = mcp_project + _write_config(project_dir, {"enabled": True}) + _set_checks(monkeypatch, [PASSING]) + feature_mcp.feature_mark_passing(1) + + def boom(*args, **kwargs): + raise AssertionError("checks should not run for an already-passing feature") + + monkeypatch.setattr(quality_gates, "detect_checks", boom) + result = json.loads(feature_mcp.feature_mark_passing(1)) + assert "already passing" in result["error"] + + +def test_mark_passing_unknown_feature_with_gates(mcp_project): + project_dir, _ = mcp_project + _write_config(project_dir, {"enabled": True}) + result = json.loads(feature_mcp.feature_mark_passing(999)) + assert "not found" in result["error"] + + +def test_verify_quality_records_evidence(mcp_project, monkeypatch): + _, session_maker = mcp_project + _set_checks(monkeypatch, [FAILING]) + + result = json.loads(feature_mcp.feature_verify_quality(1)) + assert result["passed"] is False + feature = _feature(session_maker) + assert feature.quality_result["passed"] is False + # Verifying alone is not a failed attempt + assert feature.failure_count == 0 + assert feature.passes is False + + +def test_verify_quality_unknown_feature(mcp_project): + result = json.loads(feature_mcp.feature_verify_quality(999)) + assert "not found" in result["error"] + + +def test_report_failure(mcp_project): + _, session_maker = mcp_project + + result = json.loads(feature_mcp.feature_report_failure(1, "Login API returns 500")) + assert result == {"feature_id": 1, "failure_count": 1, "failure_reason": "Login API returns 500"} + + result = json.loads(feature_mcp.feature_report_failure(1, "Still 500")) + assert result["failure_count"] == 2 + + feature = _feature(session_maker) + assert feature.failure_reason == "Still 500" + assert feature.in_progress is True + assert feature.passes is False + + +def test_report_failure_unknown_feature(mcp_project): + result = json.loads(feature_mcp.feature_report_failure(999, "x")) + assert "not found" in result["error"] From bdddead8279e442d378fd773b107e51a203948ba Mon Sep 17 00:00:00 2001 From: mohamed zwanski Date: Thu, 22 Oct 2026 22:22:17 +0100 Subject: [PATCH 2/2] Quality gates: take the on/off switch from the host env, not project files QA found that checks run inside the feature MCP server, outside the Bash sandbox and allowlist, and that the agent could both write the tooling (node_modules/.bin/eslint) and enable the gates via .autoforge/config.json, or call feature_verify_quality while gates were disabled. - Settings now come only from AUTOFORGE_QUALITY_GATES(_TIMEOUT|_CHECKS) in the AutoForge process environment; client.py forwards them to the MCP server. .autoforge/config.json is no longer read. - feature_verify_quality returns an error and runs nothing when disabled; run_quality_checks also refuses to run while disabled. - Document that enabling gates is a trust decision (unsandboxed project tooling) in quality_gates.py, CLAUDE.md and .env.example. - Clear failure_reason when a feature passes; only store quality_result when the UPDATE actually marked the row passing. - Tests for the env config, the disabled no-op, and a regression test reproducing the QA scenario. Co-Authored-By: Claude Opus 5.5 --- .claude/templates/coding_prompt.template.md | 2 +- .env.example | 10 ++ CLAUDE.md | 14 +- client.py | 4 + examples/project_config.json | 11 -- mcp_server/feature_mcp.py | 17 ++- quality_gates.py | 122 ++++++++------- test_quality_gates.py | 155 ++++++++++++++++---- 8 files changed, 235 insertions(+), 100 deletions(-) delete mode 100644 examples/project_config.json diff --git a/.claude/templates/coding_prompt.template.md b/.claude/templates/coding_prompt.template.md index 371d4c88a..03985e697 100644 --- a/.claude/templates/coding_prompt.template.md +++ b/.claude/templates/coding_prompt.template.md @@ -149,7 +149,7 @@ Use the feature_mark_passing tool with feature_id=42 If the project has quality gates enabled, `feature_mark_passing` first runs the project's lint and type checks. If it responds that quality gates failed, the feature was NOT marked passing: fix the reported errors and call it again. You can run the same checks earlier with -`feature_verify_quality`. +`feature_verify_quality` (it returns an error when quality gates are not enabled; that is fine, just continue). **NEVER:** diff --git a/.env.example b/.env.example index ed163bacf..8dfd3d00a 100644 --- a/.env.example +++ b/.env.example @@ -61,3 +61,13 @@ # ANTHROPIC_DEFAULT_OPUS_MODEL=kimi-k2.5 # ANTHROPIC_DEFAULT_SONNET_MODEL=kimi-k2.5 # ANTHROPIC_DEFAULT_HAIKU_MODEL=kimi-k2.5 + +# Quality Gates (optional) +# Run the project's linters/type checkers (ESLint, Biome, ruff, flake8, tsc, mypy) +# before a feature can be marked passing. off (default) | strict | warn +# WARNING: checks run project-local tooling and config (node_modules/.bin, .venv, +# eslint.config.js, ...) OUTSIDE the agent's Bash sandbox. Only enable this for +# projects you trust to run unsandboxed on your machine. +# AUTOFORGE_QUALITY_GATES=strict +# AUTOFORGE_QUALITY_GATES_TIMEOUT=120 +# AUTOFORGE_QUALITY_GATES_CHECKS=lint,type_check diff --git a/CLAUDE.md b/CLAUDE.md index f240a9624..138d67434 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -251,13 +251,19 @@ MCP tools available to the agent: #### Quality Gates -`quality_gates.py` auto-detects a project's linters (ESLint, Biome, ruff, flake8) and type checkers (tsc, mypy) from its config files and runs them. Node tools only run from the project's `node_modules/.bin`; Python tools from the project's `.venv`/`venv` or PATH. Disabled by default; enable per project in `.autoforge/config.json` (see `examples/project_config.json`): +`quality_gates.py` auto-detects a project's linters (ESLint, Biome, ruff, flake8) and type checkers (tsc, mypy) from its config files and runs them. Node tools only run from the project's `node_modules/.bin`; Python tools from the project's `.venv`/`venv` or PATH. -```json -{"quality_gates": {"enabled": true, "strict_mode": true, "timeout_seconds": 120, "checks": {"lint": true, "type_check": true}}} +Disabled by default. Configured **only** via environment variables of the AutoForge process (e.g. `~/.autoforge/.env`, see `.env.example`), which `client.py` forwards to the feature MCP server. Nothing inside the project directory (which the agent can write) can enable or weaken the gates: + +```bash +AUTOFORGE_QUALITY_GATES=strict # off (default) | strict | warn +AUTOFORGE_QUALITY_GATES_TIMEOUT=120 # per-check timeout in seconds (max 600) +AUTOFORGE_QUALITY_GATES_CHECKS=lint,type_check ``` -When enabled, `feature_mark_passing` runs the checks first. In `strict_mode` (default) a failing check blocks marking the feature passing, records `failure_reason`, and increments `failure_count`; with `strict_mode: false` it only warns. The latest report is stored in the feature's `quality_result` column as test evidence. +When enabled, `feature_mark_passing` runs the checks first. In `strict` mode a failing check blocks marking the feature passing, records `failure_reason`, and increments `failure_count`; in `warn` mode it only warns. The latest report is stored in the feature's `quality_result` column as test evidence. While disabled, `feature_verify_quality` returns an error and runs nothing. + +**Security / trust:** the checks run inside the feature MCP server, outside the agent's Bash sandbox and the `security.py` allowlist, and execute project-local tooling and config the agent can modify (`node_modules/.bin/*`, `.venv`, `eslint.config.js`, ...). Enabling quality gates is a trust decision; only enable them for projects you trust to run unsandboxed. ### React UI (ui/) diff --git a/client.py b/client.py index e66dc4fce..402a3e7df 100644 --- a/client.py +++ b/client.py @@ -16,6 +16,7 @@ from claude_agent_sdk.types import HookContext, HookInput, HookMatcher, SyncHookJSONOutput from dotenv import load_dotenv +from quality_gates import QUALITY_GATE_ENV_VARS from security import SENSITIVE_DIRECTORIES, bash_security_hook # Load environment variables from .env file if present @@ -338,6 +339,9 @@ def create_client( # (subprocess inherits parent environment automatically) "PROJECT_DIR": str(project_dir.resolve()), "PYTHONPATH": str(Path(__file__).parent.resolve()), + # Trusted quality-gate switch: comes from the host environment only, + # never from files the agent can write inside the project + **{k: os.environ[k] for k in QUALITY_GATE_ENV_VARS if k in os.environ}, }, }, } diff --git a/examples/project_config.json b/examples/project_config.json deleted file mode 100644 index 2436ab047..000000000 --- a/examples/project_config.json +++ /dev/null @@ -1,11 +0,0 @@ -{ - "quality_gates": { - "enabled": true, - "strict_mode": true, - "timeout_seconds": 120, - "checks": { - "lint": true, - "type_check": true - } - } -} diff --git a/mcp_server/feature_mcp.py b/mcp_server/feature_mcp.py index c94346086..f2c147678 100755 --- a/mcp_server/feature_mcp.py +++ b/mcp_server/feature_mcp.py @@ -255,7 +255,8 @@ def feature_mark_passing( JSON with success confirmation: {success, feature_id, name}, plus a quality_gates summary when quality gates are enabled. """ - quality_config = load_quality_config(PROJECT_DIR) + # Settings come from the MCP server's environment only (never from project files) + quality_config = load_quality_config() quality_report = None if quality_config.enabled: session = get_session() @@ -284,10 +285,10 @@ def feature_mark_passing( # Atomic update with state guard - prevents double-pass in parallel mode result = session.execute(text(""" UPDATE features - SET passes = 1, in_progress = 0 + SET passes = 1, in_progress = 0, failure_reason = NULL WHERE id = :id AND passes = 0 """), {"id": feature_id}) - if quality_report is not None: + if quality_report is not None and result.rowcount > 0: session.execute( text("UPDATE features SET quality_result = :qr WHERE id = :id"), {"id": feature_id, "qr": json.dumps(quality_report.to_dict())}, @@ -371,7 +372,8 @@ def feature_verify_quality( project's config files. Use this before feature_mark_passing to find and fix lint/type errors early. The result is stored on the feature as evidence. - Runs even if quality gates are not enabled in .autoforge/config.json. + Only available when quality gates are enabled for this AutoForge + installation; otherwise returns an error and runs nothing. Args: feature_id: The ID of the feature the checks are for @@ -379,6 +381,11 @@ def feature_verify_quality( Returns: JSON with: passed (bool), message, checks (name, category, passed, output on failure) """ + quality_config = load_quality_config() + if not quality_config.enabled: + # Checks run outside the Bash sandbox, so they need the user's opt-in + return json.dumps({"error": "Quality gates are not enabled for this project"}) + session = get_session() try: feature = session.query(Feature).filter(Feature.id == feature_id).first() @@ -388,7 +395,7 @@ def feature_verify_quality( session.close() try: - report = run_quality_checks(PROJECT_DIR).to_dict() + report = run_quality_checks(PROJECT_DIR, quality_config).to_dict() _store_quality_result(feature_id, report) except Exception as e: return json.dumps({"error": f"Failed to run quality checks: {str(e)}"}) diff --git a/quality_gates.py b/quality_gates.py index 306d0380a..e1cc32a35 100644 --- a/quality_gates.py +++ b/quality_gates.py @@ -6,21 +6,27 @@ structured results. Used by the feature MCP server to verify code before a feature is marked as passing. -Configuration lives in ``{project_dir}/.autoforge/config.json``:: - - { - "quality_gates": { - "enabled": true, - "strict_mode": true, - "timeout_seconds": 120, - "checks": {"lint": true, "type_check": true} - } - } - -Quality gates are disabled by default. When enabled with ``strict_mode`` (the -default once enabled), ``feature_mark_passing`` refuses to mark a feature as -passing while any check fails. With ``strict_mode: false`` the checks still -run and are recorded, but failures are reported as warnings only. +Quality gates are disabled by default. They are configured ONLY through +environment variables of the AutoForge process (e.g. ``~/.autoforge/.env``), +never through files inside the project, because the coding agent can write +anything in the project directory and must not be able to turn the gates on +(or weaken them):: + + AUTOFORGE_QUALITY_GATES=strict # off (default) | strict (or 1/true/on) | warn + AUTOFORGE_QUALITY_GATES_TIMEOUT=120 # per-check timeout in seconds (max 600) + AUTOFORGE_QUALITY_GATES_CHECKS=lint,type_check + +In ``strict`` mode ``feature_mark_passing`` refuses to mark a feature as +passing while any check fails. In ``warn`` mode the checks still run and are +recorded, but failures are reported as warnings only. + +SECURITY / TRUST: the checks run inside the feature MCP server process, which +is NOT covered by the agent's Bash sandbox or the command allowlist in +``security.py``. They execute project-local tooling and configuration that the +agent can modify (``node_modules/.bin/*``, ``.venv``/``venv``, +``eslint.config.js`` and similar). Enabling quality gates is therefore a trust +decision: only enable them for projects whose contents you trust to run +unsandboxed on your machine. Detected tools (only run when the project is configured for them): @@ -111,45 +117,52 @@ def to_dict(self) -> dict[str, Any]: return asdict(self) -def _config_path(project_dir: Path) -> Path: - return project_dir / ".autoforge" / "config.json" +ENV_ENABLED = "AUTOFORGE_QUALITY_GATES" +ENV_TIMEOUT = "AUTOFORGE_QUALITY_GATES_TIMEOUT" +ENV_CHECKS = "AUTOFORGE_QUALITY_GATES_CHECKS" +# Environment variables forwarded to the feature MCP server (see client.py) +QUALITY_GATE_ENV_VARS = (ENV_ENABLED, ENV_TIMEOUT, ENV_CHECKS) -def load_config(project_dir: Path) -> QualityGateConfig: - """Load quality gate settings from ``.autoforge/config.json``. +_STRICT_VALUES = {"1", "true", "yes", "on", "strict"} +_WARN_VALUES = {"warn", "warning"} - Missing or malformed config falls back to defaults (disabled). + +def load_config(env: dict[str, str] | None = None) -> QualityGateConfig: + """Load quality gate settings from environment variables. + + Deliberately ignores any file in the project directory: the agent can + write there, and enabling the gates runs project tooling unsandboxed. + Unknown or malformed values fall back to safe defaults (disabled). """ + if env is None: + env = dict(os.environ) config = QualityGateConfig() - path = _config_path(project_dir) - if not path.is_file(): - return config - try: - data = json.loads(path.read_text(encoding="utf-8")) - except (OSError, ValueError) as e: - print(f"[quality_gates] Ignoring unreadable {path}: {e}", file=sys.stderr) - return config - - section = data.get("quality_gates") if isinstance(data, dict) else None - if not isinstance(section, dict): - return config - - if isinstance(section.get("enabled"), bool): - config.enabled = section["enabled"] - if isinstance(section.get("strict_mode"), bool): - config.strict_mode = section["strict_mode"] - - timeout = section.get("timeout_seconds") - if isinstance(timeout, int) and not isinstance(timeout, bool) and timeout > 0: - config.timeout_seconds = min(timeout, MAX_TIMEOUT_SECONDS) - - checks = section.get("checks") - if isinstance(checks, dict): - if isinstance(checks.get("lint"), bool): - config.lint = checks["lint"] - if isinstance(checks.get("type_check"), bool): - config.type_check = checks["type_check"] + mode = env.get(ENV_ENABLED, "").strip().lower() + if mode in _STRICT_VALUES: + config.enabled = True + config.strict_mode = True + elif mode in _WARN_VALUES: + config.enabled = True + config.strict_mode = False + elif mode not in ("", "0", "false", "no", "off"): + print(f"[quality_gates] Ignoring unknown {ENV_ENABLED}={mode!r}; gates disabled", file=sys.stderr) + + timeout = env.get(ENV_TIMEOUT, "").strip() + if timeout: + try: + value = int(timeout) + except ValueError: + value = 0 + if value > 0: + config.timeout_seconds = min(value, MAX_TIMEOUT_SECONDS) + + checks = env.get(ENV_CHECKS) + if checks is not None and checks.strip(): + names = {c.strip().lower().replace("-", "_") for c in checks.split(",")} + config.lint = "lint" in names + config.type_check = "type_check" in names return config @@ -211,7 +224,7 @@ def detect_checks(project_dir: Path, config: QualityGateConfig | None = None) -> # Absolute paths: the commands run with cwd=project_dir project_dir = project_dir.resolve() if config is None: - config = load_config(project_dir) + config = load_config() checks: list[QualityCheck] = [] pkg = _package_json(project_dir) @@ -340,7 +353,16 @@ def run_quality_checks(project_dir: Path, config: QualityGateConfig | None = Non """Detect and run all applicable quality checks for a project.""" project_dir = project_dir.resolve() if config is None: - config = load_config(project_dir) + config = load_config() + + # Defense in depth: never execute project tooling unless the trusted switch is on + if not config.enabled: + return QualityReport( + passed=True, + strict_mode=config.strict_mode, + timestamp=datetime.now(timezone.utc).isoformat(), + message="Quality gates are disabled; nothing was run", + ) checks = detect_checks(project_dir, config) results = [run_check(c, project_dir, config.timeout_seconds) for c in checks] diff --git a/test_quality_gates.py b/test_quality_gates.py index a2f9a845b..cfa1be539 100644 --- a/test_quality_gates.py +++ b/test_quality_gates.py @@ -21,60 +21,104 @@ from quality_gates import QualityCheck, QualityGateConfig -def _write_config(project_dir: Path, section: dict) -> None: +def _write_project_config(project_dir: Path, section: dict) -> None: + """Write the (ignored) legacy per-project config an agent could create.""" autoforge_dir = project_dir / ".autoforge" autoforge_dir.mkdir(parents=True, exist_ok=True) (autoforge_dir / "config.json").write_text(json.dumps({"quality_gates": section})) +def _enable(monkeypatch, mode: str = "strict") -> None: + monkeypatch.setenv(quality_gates.ENV_ENABLED, mode) + + def _py_check(name: str, category: str, code: str) -> QualityCheck: return QualityCheck(name, category, [sys.executable, "-c", code]) +@pytest.fixture(autouse=True) +def _clean_quality_env(monkeypatch): + """Never let the developer's own environment leak into the tests.""" + for var in quality_gates.QUALITY_GATE_ENV_VARS: + monkeypatch.delenv(var, raising=False) + + # ============================================================================= # Config loading # ============================================================================= -def test_config_defaults_to_disabled(tmp_path): - config = quality_gates.load_config(tmp_path) +def test_config_defaults_to_disabled(): + config = quality_gates.load_config({}) assert config.enabled is False assert config.strict_mode is True assert config.lint is True assert config.type_check is True -def test_config_reads_settings(tmp_path): - _write_config(tmp_path, { - "enabled": True, - "strict_mode": False, - "timeout_seconds": 30, - "checks": {"lint": False, "type_check": True}, - }) - config = quality_gates.load_config(tmp_path) +@pytest.mark.parametrize("value", ["strict", "1", "true", "ON", " Strict "]) +def test_config_strict_values(value): + config = quality_gates.load_config({quality_gates.ENV_ENABLED: value}) + assert config.enabled is True + assert config.strict_mode is True + + +def test_config_warn_mode(): + config = quality_gates.load_config({quality_gates.ENV_ENABLED: "warn"}) assert config.enabled is True assert config.strict_mode is False + + +@pytest.mark.parametrize("value", ["", "0", "off", "false", "maybe"]) +def test_config_disabled_values(value): + assert quality_gates.load_config({quality_gates.ENV_ENABLED: value}).enabled is False + + +def test_config_reads_timeout_and_checks(): + config = quality_gates.load_config({ + quality_gates.ENV_ENABLED: "strict", + quality_gates.ENV_TIMEOUT: "30", + quality_gates.ENV_CHECKS: "type-check", + }) assert config.timeout_seconds == 30 assert config.lint is False assert config.type_check is True -def test_config_ignores_bad_values(tmp_path): - _write_config(tmp_path, {"enabled": "yes", "timeout_seconds": -5, "checks": "all"}) - config = quality_gates.load_config(tmp_path) - assert config.enabled is False +def test_config_ignores_bad_timeout(): + config = quality_gates.load_config({quality_gates.ENV_TIMEOUT: "-5"}) assert config.timeout_seconds == quality_gates.DEFAULT_TIMEOUT_SECONDS + config = quality_gates.load_config({quality_gates.ENV_TIMEOUT: "soon"}) + assert config.timeout_seconds == quality_gates.DEFAULT_TIMEOUT_SECONDS + + +def test_config_caps_timeout(): + config = quality_gates.load_config({quality_gates.ENV_TIMEOUT: "99999"}) + assert config.timeout_seconds == quality_gates.MAX_TIMEOUT_SECONDS -def test_config_caps_timeout(tmp_path): - _write_config(tmp_path, {"timeout_seconds": 99999}) - assert quality_gates.load_config(tmp_path).timeout_seconds == quality_gates.MAX_TIMEOUT_SECONDS +def test_config_reads_process_environment(monkeypatch): + _enable(monkeypatch, "warn") + config = quality_gates.load_config() + assert config.enabled is True + assert config.strict_mode is False + +def test_project_config_file_cannot_enable_gates(tmp_path, monkeypatch): + """The agent can write .autoforge/config.json, so it must have no effect.""" + _write_project_config(tmp_path, {"enabled": True, "strict_mode": True}) + monkeypatch.chdir(tmp_path) + assert quality_gates.load_config().enabled is False + + +def test_disabled_config_runs_nothing(tmp_path, monkeypatch): + def boom(*args, **kwargs): + raise AssertionError("no checks may be detected or run while disabled") -def test_config_tolerates_malformed_json(tmp_path): - (tmp_path / ".autoforge").mkdir() - (tmp_path / ".autoforge" / "config.json").write_text("{not json") - assert quality_gates.load_config(tmp_path).enabled is False + monkeypatch.setattr(quality_gates, "detect_checks", boom) + report = quality_gates.run_quality_checks(tmp_path, QualityGateConfig(enabled=False)) + assert report.checks == [] + assert "disabled" in report.message # ============================================================================= @@ -314,7 +358,7 @@ def test_mark_passing_unchanged_when_disabled(mcp_project, monkeypatch): def test_strict_mode_blocks_failing_checks(mcp_project, monkeypatch): project_dir, session_maker = mcp_project - _write_config(project_dir, {"enabled": True}) + _enable(monkeypatch) _set_checks(monkeypatch, [PASSING, FAILING]) result = json.loads(feature_mcp.feature_mark_passing(1)) @@ -338,7 +382,7 @@ def test_strict_mode_blocks_failing_checks(mcp_project, monkeypatch): def test_strict_mode_passes_when_checks_pass(mcp_project, monkeypatch): project_dir, session_maker = mcp_project - _write_config(project_dir, {"enabled": True}) + _enable(monkeypatch) _set_checks(monkeypatch, [PASSING]) result = json.loads(feature_mcp.feature_mark_passing(1)) @@ -353,7 +397,7 @@ def test_strict_mode_passes_when_checks_pass(mcp_project, monkeypatch): def test_non_strict_mode_warns_but_marks_passing(mcp_project, monkeypatch): project_dir, session_maker = mcp_project - _write_config(project_dir, {"enabled": True, "strict_mode": False}) + _enable(monkeypatch, "warn") _set_checks(monkeypatch, [FAILING]) result = json.loads(feature_mcp.feature_mark_passing(1)) @@ -366,7 +410,7 @@ def test_non_strict_mode_warns_but_marks_passing(mcp_project, monkeypatch): def test_mark_passing_already_passing_skips_checks(mcp_project, monkeypatch): project_dir, _ = mcp_project - _write_config(project_dir, {"enabled": True}) + _enable(monkeypatch) _set_checks(monkeypatch, [PASSING]) feature_mcp.feature_mark_passing(1) @@ -378,15 +422,16 @@ def boom(*args, **kwargs): assert "already passing" in result["error"] -def test_mark_passing_unknown_feature_with_gates(mcp_project): +def test_mark_passing_unknown_feature_with_gates(mcp_project, monkeypatch): project_dir, _ = mcp_project - _write_config(project_dir, {"enabled": True}) + _enable(monkeypatch) result = json.loads(feature_mcp.feature_mark_passing(999)) assert "not found" in result["error"] def test_verify_quality_records_evidence(mcp_project, monkeypatch): _, session_maker = mcp_project + _enable(monkeypatch) _set_checks(monkeypatch, [FAILING]) result = json.loads(feature_mcp.feature_verify_quality(1)) @@ -398,7 +443,8 @@ def test_verify_quality_records_evidence(mcp_project, monkeypatch): assert feature.passes is False -def test_verify_quality_unknown_feature(mcp_project): +def test_verify_quality_unknown_feature(mcp_project, monkeypatch): + _enable(monkeypatch) result = json.loads(feature_mcp.feature_verify_quality(999)) assert "not found" in result["error"] @@ -421,3 +467,54 @@ def test_report_failure(mcp_project): def test_report_failure_unknown_feature(mcp_project): result = json.loads(feature_mcp.feature_report_failure(999, "x")) assert "not found" in result["error"] + + +def test_failure_reason_cleared_when_feature_passes(mcp_project, monkeypatch): + _, session_maker = mcp_project + _enable(monkeypatch) + _set_checks(monkeypatch, [FAILING]) + feature_mcp.feature_mark_passing(1) + assert _feature(session_maker).failure_reason is not None + + _set_checks(monkeypatch, [PASSING]) + assert json.loads(feature_mcp.feature_mark_passing(1))["success"] is True + feature = _feature(session_maker) + assert feature.failure_reason is None + assert feature.failure_count == 1 # History is kept + + +def test_verify_quality_disabled_runs_nothing(mcp_project, monkeypatch): + _, session_maker = mcp_project + + def boom(*args, **kwargs): + raise AssertionError("checks must not run while gates are disabled") + + monkeypatch.setattr(quality_gates, "detect_checks", boom) + result = json.loads(feature_mcp.feature_verify_quality(1)) + assert result == {"error": "Quality gates are not enabled for this project"} + assert _feature(session_maker).quality_result is None + + +@pytest.mark.skipif(sys.platform == "win32", reason="uses an executable shell-style script") +def test_agent_written_tooling_never_runs_unless_trusted(mcp_project, tmp_path_factory): + """Regression test for the QA finding on PR #334. + + An agent can create an executable node_modules/.bin/eslint, an eslint config + and .autoforge/config.json enabling the gates. Without the trusted env + switch, neither MCP tool may execute it. + """ + project_dir, session_maker = mcp_project + marker = tmp_path_factory.mktemp("outside") / "pwned" + + (project_dir / ".eslintrc.json").write_text("{}") + bin_dir = project_dir / "node_modules" / ".bin" + bin_dir.mkdir(parents=True) + script = bin_dir / "eslint" + script.write_text(f"#!{sys.executable}\nopen({str(marker)!r}, 'w').write('x')\n") + script.chmod(0o755) + _write_project_config(project_dir, {"enabled": True, "strict_mode": True}) + + assert "not enabled" in json.loads(feature_mcp.feature_verify_quality(1))["error"] + assert json.loads(feature_mcp.feature_mark_passing(1))["success"] is True + assert not marker.exists() + assert _feature(session_maker).quality_result is None