diff --git a/.claude/templates/coding_prompt.template.md b/.claude/templates/coding_prompt.template.md index 832eb5996..267440c23 100644 --- a/.claude/templates/coding_prompt.template.md +++ b/.claude/templates/coding_prompt.template.md @@ -146,6 +146,8 @@ 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 linters and type checkers. If it returns a "Quality gate failed" error, fix the reported errors and call it again. When quality gates are enabled, you can also run the same checks yourself at any time with `feature_verify_quality with feature_id=42` (the tool is only available, and only works, when quality gates are enabled for the project). + **NEVER:** - Delete features @@ -233,13 +235,17 @@ feature_mark_in_progress with feature_id={id} feature_mark_passing with feature_id={id} # 5. Mark a feature as failing (if you discover it's broken) -feature_mark_failing with feature_id={id} +feature_mark_failing with feature_id={id} and reason="what is broken" # 6. Skip a feature (moves to end of queue) - ONLY when blocked by external dependency 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 and store the results as evidence +# (only available when quality gates are enabled for the project) +feature_verify_quality with feature_id={id} ``` ### RULES: diff --git a/CLAUDE.md b/CLAUDE.md index 6f59910be..3a1a3d7c8 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 gates tests (30 tests) ``` ### React UI @@ -175,6 +176,7 @@ Publishing: `npm publish` (triggers `prepublishOnly` which builds UI, then publi - `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 +- `quality_gates.py` - Lint/type-check auto-detection and enforcement before `feature_mark_passing` ### Project Registry @@ -235,8 +237,9 @@ MCP tools available to the agent: - `feature_get_graph` - Get full dependency graph - `feature_claim_and_get` - Atomically claim next available feature (for parallel mode) - `feature_mark_in_progress` - Mark feature as in progress -- `feature_mark_passing` - Mark feature complete -- `feature_mark_failing` - Mark feature as failing +- `feature_mark_passing` - Mark feature complete (runs quality gates first when enabled) +- `feature_mark_failing` - Mark feature as failing (optional `reason`; increments `failure_count`) +- `feature_verify_quality` - Run detected linters/type checkers and store results as evidence (only when quality gates are enabled) - `feature_skip` - Move feature to end of queue - `feature_clear_in_progress` - Clear in-progress status - `feature_create_bulk` - Initialize all features (used by initializer) @@ -245,6 +248,27 @@ MCP tools available to the agent: - `feature_remove_dependency` - Remove a dependency - `feature_set_dependencies` - Set all dependencies for a feature at once +### Quality Gates + +Opt-in per project via `.autoforge/config.json` (disabled by default): + +```json +{ + "quality_gates": { + "enabled": true, + "strict_mode": true, + "checks": {"lint": true, "type_check": true}, + "timeout_seconds": 300 + } +} +``` + +- Auto-detects ESLint or Biome, TypeScript (`tsc`), ruff or flake8, and mypy. A tool runs only if the project both configures it and has it installed (`node_modules/.bin`, `.venv`/`venv`, or PATH). Commands are never read from config. +- When enabled, `feature_mark_passing` runs the checks first. Strict mode (default) refuses to mark the feature passing on failure; non-strict marks it passing and returns a warning. +- Results are stored on the feature (`quality_result`), along with `failure_reason` and `failure_count`. +- `feature_verify_quality` is only offered to coding/testing agents when gates are enabled, and refuses to run anything when they are disabled. +- The MCP server runs outside the bash sandbox, and linters load agent-written config files (e.g. `eslint.config.js`), so only enable this for projects you trust the agent with. + ### React UI (ui/) - Tech stack: React 19, TypeScript, Vite 7, TanStack Query, Tailwind CSS v4, Radix UI, dagre (graph layout), xterm.js (terminal) diff --git a/api/database.py b/api/database.py index 523ea2243..7db344cbe 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 of lint/type-check runs and failure tracking + quality_result = Column(JSON, nullable=True, default=None) # Last quality check report + failure_reason = Column(Text, nullable=True, default=None) # Why the feature last failed + failure_count = Column(Integer, nullable=False, default=0) # Number of recorded failures + 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_gate_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_gate_columns(engine) # Migrate to add schedules tables _migrate_add_schedules_tables(engine) diff --git a/client.py b/client.py index 26a710f50..ae07e2d36 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 load_config as load_quality_config from security import SENSITIVE_DIRECTORIES, bash_security_hook # Load environment variables from .env file if present @@ -190,6 +191,31 @@ def get_extra_read_paths() -> list[Path]: set(CODING_AGENT_TOOLS) | set(TESTING_AGENT_TOOLS) | set(INITIALIZER_AGENT_TOOLS) ) +# Runs project linters/type checkers from the feature MCP server, which sits +# outside the bash sandbox. Only exposed to coding/testing agents when quality +# gates are enabled for the project (.autoforge/config.json). +QUALITY_GATE_TOOLS = [ + "mcp__features__feature_verify_quality", +] + + +def get_feature_tools(agent_type: str, project_dir: Path) -> list[str]: + """Return the feature MCP tools an agent of this type may see. + + Quality gate tools are added for coding/testing agents only when the + project has quality gates enabled. + """ + feature_tools_map = { + "coding": CODING_AGENT_TOOLS, + "testing": TESTING_AGENT_TOOLS, + "initializer": INITIALIZER_AGENT_TOOLS, + } + tools = list(feature_tools_map.get(agent_type, CODING_AGENT_TOOLS)) + if agent_type != "initializer" and load_quality_config(project_dir).enabled: + tools.extend(QUALITY_GATE_TOOLS) + return tools + + # Built-in tools available to agents. # WebFetch and WebSearch are included so coding agents can look up current # documentation for frameworks and libraries they are implementing. @@ -234,12 +260,8 @@ def create_client( The Claude SDK auto-detects credentials from the Claude CLI configuration """ # Select the feature MCP tools appropriate for this agent type - feature_tools_map = { - "coding": CODING_AGENT_TOOLS, - "testing": TESTING_AGENT_TOOLS, - "initializer": INITIALIZER_AGENT_TOOLS, - } - feature_tools = feature_tools_map.get(agent_type, CODING_AGENT_TOOLS) + feature_tools = get_feature_tools(agent_type, project_dir) + quality_tools = [t for t in QUALITY_GATE_TOOLS if t in feature_tools] # Select max_turns based on agent type: # - coding/initializer: 300 turns (complex multi-step implementation) @@ -273,6 +295,7 @@ def create_client( "WebSearch(*)", # Allow Feature MCP tools for feature management *ALL_FEATURE_MCP_TOOLS, + *quality_tools, ] # Add extra read paths from environment variable (read-only access) diff --git a/mcp_server/feature_mcp.py b/mcp_server/feature_mcp.py index 71e217d41..fd738d997 100755 --- a/mcp_server/feature_mcp.py +++ b/mcp_server/feature_mcp.py @@ -22,6 +22,7 @@ - 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 the project's linters/type checkers (only when quality gates are enabled) Note: Feature selection (which feature to work on) is handled by the orchestrator, not by agents. Agents receive pre-assigned feature IDs. @@ -48,6 +49,7 @@ would_create_circular_dependency, ) from api.migration import migrate_json_to_sqlite +from quality_gates import format_failure_reason, load_config, run_quality_checks # Configuration from environment PROJECT_DIR = Path(os.environ.get("PROJECT_DIR", ".")).resolve() @@ -240,20 +242,64 @@ 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 (.autoforge/config.json), the + project's linters and type checkers run first. In strict mode the feature is + NOT marked passing when a check fails: fix the reported errors 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 + quality_gate details when quality gates are enabled. """ session = get_session() try: + config = load_config(PROJECT_DIR) + quality_report = None + if config.enabled: + # Check state first so we don't run slow checks for nothing + 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"}) + # Release the read transaction while checks run (they can take minutes) + session.rollback() + + quality_report = run_quality_checks(PROJECT_DIR, config) + if not quality_report.passed and config.strict_mode: + session.execute(text(""" + UPDATE features + SET quality_result = :quality_result, + failure_reason = :failure_reason, + failure_count = COALESCE(failure_count, 0) + 1 + WHERE id = :id + """), { + "id": feature_id, + "quality_result": json.dumps(quality_report.to_dict()), + "failure_reason": format_failure_reason(quality_report), + }) + session.commit() + return json.dumps({ + "error": ( + f"Quality gate failed for feature #{feature_id}: {quality_report.summary()}. " + "Fix the errors below, then call feature_mark_passing again." + ), + "quality_gate": quality_report.to_dict(), + }) + # 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, + quality_result = COALESCE(:quality_result, quality_result) WHERE id = :id AND passes = 0 - """), {"id": feature_id}) + """), { + "id": feature_id, + "quality_result": json.dumps(quality_report.to_dict()) if quality_report else None, + }) session.commit() if result.rowcount == 0: @@ -267,7 +313,18 @@ 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_gate"] = { + "passed": quality_report.passed, + "summary": quality_report.summary(), + } + if not quality_report.passed: + response["warning"] = ( + f"Quality checks failed ({quality_report.summary()}) but strict mode is off. " + "Fix the reported errors (see feature_verify_quality)." + ) + return json.dumps(response) except Exception as e: session.rollback() return json.dumps({"error": f"Failed to mark feature passing: {str(e)}"}) @@ -277,7 +334,11 @@ def feature_mark_passing( @mcp.tool() def feature_mark_failing( - feature_id: Annotated[int, Field(description="The ID of the feature to mark as failing", ge=1)] + feature_id: Annotated[int, Field(description="The ID of the feature to mark as failing", ge=1)], + reason: Annotated[ + str | None, + Field(description="Why the feature is failing (stored for retry tracking)", max_length=2000), + ] = None, ) -> str: """Mark a feature as failing after finding a regression. @@ -291,8 +352,12 @@ def feature_mark_failing( 3. Verify the fix 4. Call feature_mark_passing once fixed + Each call increments the feature's failure_count, and the optional reason + is stored in failure_reason so later sessions know what went wrong. + Args: feature_id: The ID of the feature to mark as failing + reason: Optional description of what is broken Returns: JSON with the updated feature details, or error if not found. @@ -307,9 +372,11 @@ def feature_mark_failing( # Atomic update for parallel safety session.execute(text(""" UPDATE features - SET passes = 0, in_progress = 0 + SET passes = 0, in_progress = 0, + failure_count = COALESCE(failure_count, 0) + 1, + failure_reason = COALESCE(:reason, failure_reason) WHERE id = :id - """), {"id": feature_id}) + """), {"id": feature_id, "reason": reason.strip() if reason and reason.strip() else None}) session.commit() # Refresh to get updated state @@ -326,6 +393,61 @@ def feature_mark_failing( session.close() +@mcp.tool() +def feature_verify_quality( + feature_id: Annotated[ + int | None, + Field(description="Feature to store the results on as evidence (optional)", ge=1), + ] = None +) -> str: + """Run the project's linters and type checkers and report the results. + + Only works when quality gates are enabled for the project in + .autoforge/config.json; otherwise it returns an error and runs nothing. + + Auto-detects ESLint or Biome, TypeScript (tsc), ruff or flake8, and mypy + from the project's config files. Use this before feature_mark_passing to + catch lint/type errors early. When feature_id is given, the report is + stored on the feature as evidence (quality_result). + + Args: + feature_id: Optional feature ID to attach the results to + + Returns: + JSON with: passed (bool), summary, checks (name, command, passed, output) + """ + # The checks run project tooling outside the bash sandbox, so they must + # stay behind the same opt-in as feature_mark_passing. + if not load_config(PROJECT_DIR).enabled: + return json.dumps({ + "error": "Quality gates are not enabled for this project (.autoforge/config.json)" + }) + + session = get_session() + try: + if feature_id is not None: + 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"}) + session.rollback() + + report = run_quality_checks(PROJECT_DIR) + report_dict = report.to_dict() + + if feature_id is not None: + session.execute(text(""" + UPDATE features SET quality_result = :quality_result WHERE id = :id + """), {"id": feature_id, "quality_result": json.dumps(report_dict)}) + session.commit() + + return json.dumps(report_dict) + except Exception as e: + session.rollback() + return json.dumps({"error": f"Failed to run quality checks: {str(e)}"}) + finally: + session.close() + + @mcp.tool() def feature_skip( feature_id: Annotated[int, Field(description="The ID of the feature to skip", ge=1)] diff --git a/package.json b/package.json index 9c547ddfc..adedd7390 100644 --- a/package.json +++ b/package.json @@ -31,6 +31,7 @@ "env_constants.py", "parallel_orchestrator.py", "progress.py", + "quality_gates.py", "prompts.py", "registry.py", "rate_limit_utils.py", diff --git a/quality_gates.py b/quality_gates.py new file mode 100644 index 000000000..25feb4e4a --- /dev/null +++ b/quality_gates.py @@ -0,0 +1,353 @@ +""" +Quality Gates +============= + +Auto-detects a project's linters and type checkers, runs them, and reports +the results. Used by the feature MCP server to verify code before a feature +is marked as passing. + +Quality gates are opt-in per project via ``.autoforge/config.json``:: + + { + "quality_gates": { + "enabled": true, + "strict_mode": true, + "checks": {"lint": true, "type_check": true}, + "timeout_seconds": 300 + } + } + +- ``enabled``: run checks automatically inside ``feature_mark_passing``. +- ``strict_mode``: refuse to mark a feature passing when a check fails. + When false, the feature is still marked passing and the failure is + returned as a warning (and stored as evidence). +- ``checks``: toggle linting / type checking individually. +- ``timeout_seconds``: per-check timeout. + +Only well-known tools are run, and only from the project's own install +(``node_modules/.bin``, ``.venv``/``venv``) or the PATH. Commands are never +read from the config file. Note that tools such as ESLint load configuration +files written by the agent, and the MCP server is not inside the bash +sandbox, so only enable quality gates for projects you trust the agent with. +""" + +import json +import os +import shutil +import subprocess +import sys +import time +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +# Keep stored evidence compact: only the tail of each tool's output is kept +MAX_OUTPUT_CHARS = 4000 +DEFAULT_TIMEOUT_SECONDS = 300 + +CONFIG_FILENAME = "config.json" + + +@dataclass +class QualityGateConfig: + """Per-project quality gate settings.""" + + enabled: bool = False + strict_mode: bool = True + lint: bool = True + type_check: bool = True + timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS + + +@dataclass +class QualityCheck: + """A detected check: a tool and the command that runs it.""" + + name: str + kind: str # "lint" or "type_check" + command: list[str] + + +@dataclass +class CheckResult: + name: str + kind: str + command: str + passed: bool + returncode: int | None + duration_seconds: float + output: str + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "kind": self.kind, + "command": self.command, + "passed": self.passed, + "returncode": self.returncode, + "duration_seconds": self.duration_seconds, + "output": self.output, + } + + +@dataclass +class QualityReport: + passed: bool + checks: list[CheckResult] = field(default_factory=list) + timestamp: str = "" + + @property + def failed_checks(self) -> list[CheckResult]: + return [c for c in self.checks if not c.passed] + + def summary(self) -> str: + if not self.checks: + return "No linters or type checkers detected" + parts = [f"{c.name}: {'passed' if c.passed else 'FAILED'}" for c in self.checks] + return ", ".join(parts) + + def to_dict(self) -> dict[str, Any]: + return { + "passed": self.passed, + "summary": self.summary(), + "timestamp": self.timestamp, + "checks": [c.to_dict() for c in self.checks], + } + + +def get_config_path(project_dir: Path) -> Path: + """Return the path to the project's ``.autoforge/config.json``.""" + from autoforge_paths import get_autoforge_dir + return get_autoforge_dir(project_dir) / CONFIG_FILENAME + + +def load_config(project_dir: Path) -> QualityGateConfig: + """Load quality gate settings. Missing or malformed config means disabled.""" + config_path = get_config_path(project_dir) + if not config_path.is_file(): + return QualityGateConfig() + + try: + data = json.loads(config_path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return QualityGateConfig() + + section = data.get("quality_gates") if isinstance(data, dict) else None + if not isinstance(section, dict): + return QualityGateConfig() + + checks = section.get("checks") + if not isinstance(checks, dict): + checks = {} + + timeout = section.get("timeout_seconds", DEFAULT_TIMEOUT_SECONDS) + if not isinstance(timeout, int) or isinstance(timeout, bool) or timeout < 1: + timeout = DEFAULT_TIMEOUT_SECONDS + + return QualityGateConfig( + enabled=section.get("enabled") is True, + strict_mode=section.get("strict_mode", True) is not False, + lint=checks.get("lint", True) is not False, + type_check=checks.get("type_check", True) is not False, + timeout_seconds=timeout, + ) + + +# --------------------------------------------------------------------------- +# Tool detection +# --------------------------------------------------------------------------- + +def _find_node_bin(project_dir: Path, name: str) -> str | None: + """Locate a locally installed Node tool (never npx, which may download).""" + bin_dir = project_dir / "node_modules" / ".bin" + candidates = [f"{name}.cmd", f"{name}.exe", 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 _find_python_bin(project_dir: Path, name: str) -> str | None: + """Locate a Python tool in the project's virtualenv, falling back to PATH.""" + if sys.platform == "win32": + rel_paths = [Path("Scripts") / f"{name}.exe"] + else: + rel_paths = [Path("bin") / name] + for venv_name in (".venv", "venv"): + for rel in rel_paths: + path = project_dir / venv_name / rel + if path.is_file(): + return str(path) + return shutil.which(name) + + +def _read_text(path: Path) -> str: + try: + return path.read_text(encoding="utf-8", errors="replace") + except OSError: + return "" + + +def _any_exists(project_dir: Path, names: list[str]) -> bool: + return any((project_dir / n).is_file() for n in names) + + +def _has_section(project_dir: Path, filename: str, marker: str) -> bool: + return marker in _read_text(project_dir / filename) + + +def detect_checks(project_dir: Path) -> list[QualityCheck]: + """Detect the linters and type checkers configured for a project. + + A tool is only selected when the project both configures it and has it + installed, so we never fail a gate because of a tool the project + doesn't use. + """ + checks: list[QualityCheck] = [] + + # --- JavaScript / TypeScript --- + if (project_dir / "package.json").is_file(): + biome = _find_node_bin(project_dir, "biome") + eslint = _find_node_bin(project_dir, "eslint") + has_biome_config = _any_exists(project_dir, ["biome.json", "biome.jsonc"]) + has_eslint_config = _any_exists(project_dir, [ + "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", + ]) or '"eslintConfig"' in _read_text(project_dir / "package.json") + + if has_biome_config and biome: + checks.append(QualityCheck("biome", "lint", [biome, "lint", "."])) + elif has_eslint_config and eslint: + checks.append(QualityCheck("eslint", "lint", [eslint, "."])) + + tsc = _find_node_bin(project_dir, "tsc") + if (project_dir / "tsconfig.json").is_file() and tsc: + if '"references"' in _read_text(project_dir / "tsconfig.json"): + # Solution-style config (e.g. Vite templates): a plain -p run + # checks no files, so build the referenced projects instead + command = [tsc, "-b", "--noEmit"] + else: + command = [tsc, "--noEmit", "-p", "tsconfig.json"] + checks.append(QualityCheck("tsc", "type_check", command)) + + # --- Python --- + has_ruff_config = _any_exists(project_dir, ["ruff.toml", ".ruff.toml"]) or _has_section( + project_dir, "pyproject.toml", "[tool.ruff" + ) + has_flake8_config = ( + (project_dir / ".flake8").is_file() + or _has_section(project_dir, "setup.cfg", "[flake8]") + or _has_section(project_dir, "tox.ini", "[flake8]") + ) + if has_ruff_config: + ruff = _find_python_bin(project_dir, "ruff") + if ruff: + checks.append(QualityCheck("ruff", "lint", [ruff, "check", "."])) + elif has_flake8_config: + flake8 = _find_python_bin(project_dir, "flake8") + if flake8: + checks.append(QualityCheck("flake8", "lint", [flake8, "."])) + + has_mypy_config = ( + _any_exists(project_dir, ["mypy.ini", ".mypy.ini"]) + or _has_section(project_dir, "pyproject.toml", "[tool.mypy") + or _has_section(project_dir, "setup.cfg", "[mypy") + ) + if has_mypy_config: + mypy = _find_python_bin(project_dir, "mypy") + if mypy: + checks.append(QualityCheck("mypy", "type_check", [mypy, "."])) + + return checks + + +# --------------------------------------------------------------------------- +# Running checks +# --------------------------------------------------------------------------- + +def _truncate(output: str) -> str: + if len(output) <= MAX_OUTPUT_CHARS: + return output + return "...(truncated)...\n" + output[-MAX_OUTPUT_CHARS:] + + +def _run_check(check: QualityCheck, project_dir: Path, timeout: int) -> CheckResult: + start = time.monotonic() + command_str = " ".join([Path(check.command[0]).name, *check.command[1:]]) + try: + proc = subprocess.run( + check.command, + cwd=project_dir, + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout, + env={**os.environ, "CI": "1", "NO_COLOR": "1"}, + ) + output = (proc.stdout or "") + (proc.stderr or "") + return CheckResult( + name=check.name, + kind=check.kind, + command=command_str, + passed=proc.returncode == 0, + returncode=proc.returncode, + duration_seconds=round(time.monotonic() - start, 2), + output=_truncate(output.strip()), + ) + except subprocess.TimeoutExpired: + return CheckResult( + name=check.name, + kind=check.kind, + command=command_str, + passed=False, + returncode=None, + duration_seconds=round(time.monotonic() - start, 2), + output=f"Timed out after {timeout} seconds", + ) + except OSError as e: + return CheckResult( + name=check.name, + kind=check.kind, + command=command_str, + passed=False, + returncode=None, + duration_seconds=round(time.monotonic() - start, 2), + output=f"Failed to run: {e}", + ) + + +def run_quality_checks(project_dir: Path, config: QualityGateConfig | None = None) -> QualityReport: + """Detect and run the project's quality checks. + + Checks disabled in the config are skipped. A project with no detected + tools passes trivially (there is nothing to verify). + """ + if config is None: + config = load_config(project_dir) + + checks = [ + c for c in detect_checks(project_dir) + if (c.kind == "lint" and config.lint) or (c.kind == "type_check" and config.type_check) + ] + results = [_run_check(c, project_dir, config.timeout_seconds) for c in checks] + + return QualityReport( + passed=all(r.passed for r in results), + checks=results, + timestamp=datetime.now(timezone.utc).isoformat(), + ) + + +def format_failure_reason(report: QualityReport) -> str: + """Short, human-readable reason for a failed report (stored in failure_reason).""" + failed = report.failed_checks + if not failed: + return "" + names = ", ".join(c.name for c in failed) + return f"Quality gate failed: {names}" diff --git a/server/services/assistant_chat_session.py b/server/services/assistant_chat_session.py index 1a16606d5..6f2d4ce9e 100755 --- a/server/services/assistant_chat_session.py +++ b/server/services/assistant_chat_session.py @@ -85,6 +85,8 @@ "mcp__features__feature_mark_in_progress", "mcp__features__feature_claim_and_get", "mcp__features__feature_clear_in_progress", + # Runs project tooling (linters/type checkers) and writes evidence + "mcp__features__feature_verify_quality", ] diff --git a/test_quality_gates.py b/test_quality_gates.py new file mode 100644 index 000000000..1797a4181 --- /dev/null +++ b/test_quality_gates.py @@ -0,0 +1,379 @@ +""" +Quality Gates Tests +=================== + +Tests for quality_gates.py (config loading, tool detection, running checks) +and the quality gate integration in the feature MCP server. + +Run with: python -m pytest test_quality_gates.py +""" + +import json +import stat +import sys +from pathlib import Path + +import pytest + +import quality_gates +from quality_gates import QualityGateConfig, detect_checks, load_config, run_quality_checks + +pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="uses POSIX shell scripts as fake tools") + + +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 _fake_tool(path: Path, exit_code: int, output: str = "") -> None: + """Create an executable script standing in for a linter/type checker.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f"#!/bin/sh\necho '{output}'\nexit {exit_code}\n") + path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + + +def _js_project(project_dir: Path, eslint_exit: int = 0, tsc_exit: int = 0) -> None: + (project_dir / "package.json").write_text("{}") + (project_dir / "eslint.config.js").write_text("export default [];") + (project_dir / "tsconfig.json").write_text("{}") + _fake_tool(project_dir / "node_modules" / ".bin" / "eslint", eslint_exit, "eslint output") + _fake_tool(project_dir / "node_modules" / ".bin" / "tsc", tsc_exit, "tsc output") + + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- + +class TestLoadConfig: + def test_missing_config_is_disabled(self, tmp_path): + assert load_config(tmp_path) == QualityGateConfig() + assert load_config(tmp_path).enabled is False + + def test_malformed_config_is_disabled(self, tmp_path): + (tmp_path / ".autoforge").mkdir() + (tmp_path / ".autoforge" / "config.json").write_text("{not json") + assert load_config(tmp_path).enabled is False + + def test_enabled_defaults_to_strict(self, tmp_path): + _write_config(tmp_path, {"enabled": True}) + config = load_config(tmp_path) + assert config.enabled is True + assert config.strict_mode is True + assert config.lint is True + assert config.type_check is True + + def test_full_config(self, tmp_path): + _write_config(tmp_path, { + "enabled": True, + "strict_mode": False, + "checks": {"lint": False, "type_check": True}, + "timeout_seconds": 30, + }) + config = load_config(tmp_path) + assert config.strict_mode is False + assert config.lint is False + assert config.type_check is True + assert config.timeout_seconds == 30 + + def test_truthy_non_bool_does_not_enable(self, tmp_path): + _write_config(tmp_path, {"enabled": "yes"}) + assert load_config(tmp_path).enabled is False + + def test_invalid_timeout_falls_back(self, tmp_path): + _write_config(tmp_path, {"enabled": True, "timeout_seconds": -5}) + assert load_config(tmp_path).timeout_seconds == quality_gates.DEFAULT_TIMEOUT_SECONDS + + +# --------------------------------------------------------------------------- +# Detection +# --------------------------------------------------------------------------- + +class TestDetectChecks: + def test_empty_project(self, tmp_path): + assert detect_checks(tmp_path) == [] + + def test_js_project(self, tmp_path): + _js_project(tmp_path) + names = [(c.name, c.kind) for c in detect_checks(tmp_path)] + assert names == [("eslint", "lint"), ("tsc", "type_check")] + + def test_tsc_build_mode_for_project_references(self, tmp_path): + _js_project(tmp_path) + (tmp_path / "tsconfig.json").write_text('{"files": [], "references": [{"path": "./tsconfig.app.json"}]}') + tsc = next(c for c in detect_checks(tmp_path) if c.name == "tsc") + assert tsc.command[1:] == ["-b", "--noEmit"] + + def test_configured_but_not_installed_is_skipped(self, tmp_path): + (tmp_path / "package.json").write_text("{}") + (tmp_path / "eslint.config.js").write_text("export default [];") + (tmp_path / "tsconfig.json").write_text("{}") + assert detect_checks(tmp_path) == [] + + def test_installed_but_not_configured_is_skipped(self, tmp_path): + (tmp_path / "package.json").write_text("{}") + _fake_tool(tmp_path / "node_modules" / ".bin" / "eslint", 0) + assert detect_checks(tmp_path) == [] + + def test_biome_preferred_over_eslint(self, tmp_path): + _js_project(tmp_path) + (tmp_path / "biome.json").write_text("{}") + _fake_tool(tmp_path / "node_modules" / ".bin" / "biome", 0) + names = [c.name for c in detect_checks(tmp_path)] + assert names == ["biome", "tsc"] + + def test_python_project_uses_project_venv(self, tmp_path): + (tmp_path / "pyproject.toml").write_text("[tool.ruff]\nline-length = 100\n\n[tool.mypy]\nstrict = true\n") + _fake_tool(tmp_path / ".venv" / "bin" / "ruff", 0) + _fake_tool(tmp_path / ".venv" / "bin" / "mypy", 0) + checks = detect_checks(tmp_path) + assert [c.name for c in checks] == ["ruff", "mypy"] + assert checks[0].command[0] == str(tmp_path / ".venv" / "bin" / "ruff") + + def test_flake8_from_setup_cfg(self, tmp_path): + (tmp_path / "setup.cfg").write_text("[flake8]\nmax-line-length = 120\n") + _fake_tool(tmp_path / "venv" / "bin" / "flake8", 0) + assert [c.name for c in detect_checks(tmp_path)] == ["flake8"] + + +# --------------------------------------------------------------------------- +# Running checks +# --------------------------------------------------------------------------- + +class TestRunQualityChecks: + def test_all_pass(self, tmp_path): + _js_project(tmp_path) + report = run_quality_checks(tmp_path, QualityGateConfig(enabled=True)) + assert report.passed is True + assert [c.passed for c in report.checks] == [True, True] + assert report.checks[0].output == "eslint output" + assert report.checks[0].command == "eslint ." + + def test_failure_is_reported(self, tmp_path): + _js_project(tmp_path, tsc_exit=2) + report = run_quality_checks(tmp_path, QualityGateConfig(enabled=True)) + assert report.passed is False + assert [c.name for c in report.failed_checks] == ["tsc"] + assert report.checks[1].returncode == 2 + assert "tsc: FAILED" in report.summary() + assert quality_gates.format_failure_reason(report) == "Quality gate failed: tsc" + + def test_disabled_check_kinds_are_skipped(self, tmp_path): + _js_project(tmp_path, tsc_exit=1) + report = run_quality_checks(tmp_path, QualityGateConfig(enabled=True, type_check=False)) + assert report.passed is True + assert [c.name for c in report.checks] == ["eslint"] + + def test_no_tools_passes(self, tmp_path): + report = run_quality_checks(tmp_path, QualityGateConfig(enabled=True)) + assert report.passed is True + assert report.checks == [] + + def test_timeout_fails_check(self, tmp_path): + _js_project(tmp_path) + (tmp_path / "node_modules" / ".bin" / "eslint").write_text("#!/bin/sh\nsleep 5\n") + report = run_quality_checks(tmp_path, QualityGateConfig(enabled=True, timeout_seconds=1, type_check=False)) + assert report.passed is False + assert "Timed out" in report.checks[0].output + + def test_output_is_truncated(self, tmp_path): + _js_project(tmp_path) + (tmp_path / "node_modules" / ".bin" / "eslint").write_text( + "#!/bin/sh\nhead -c 20000 /dev/zero | tr '\\0' 'x'\nexit 1\n" + ) + report = run_quality_checks(tmp_path, QualityGateConfig(enabled=True, type_check=False)) + assert len(report.checks[0].output) <= quality_gates.MAX_OUTPUT_CHARS + 50 + + +# --------------------------------------------------------------------------- +# Feature MCP integration +# --------------------------------------------------------------------------- + +@pytest.fixture +def mcp(tmp_path, monkeypatch): + """Feature MCP module wired to a fresh database in tmp_path with one feature.""" + from api.database import Feature, create_database + from mcp_server import feature_mcp + + engine, session_maker = create_database(tmp_path) + monkeypatch.setattr(feature_mcp, "PROJECT_DIR", tmp_path) + monkeypatch.setattr(feature_mcp, "_session_maker", session_maker) + + session = session_maker() + session.add(Feature(priority=1, category="core", name="Login", description="d", steps=["s"], in_progress=True)) + session.commit() + session.close() + + yield feature_mcp + + from api.database import dispose_engine + dispose_engine(tmp_path) + + +def _feature(mcp_module) -> dict: + feature: dict = json.loads(mcp_module.feature_get_by_id(1)) + return feature + + +class TestFeatureMcpIntegration: + def test_gates_disabled_marks_passing_without_checks(self, mcp, tmp_path): + _js_project(tmp_path, eslint_exit=1) + result = json.loads(mcp.feature_mark_passing(1)) + assert result["success"] is True + assert "quality_gate" not in result + assert _feature(mcp)["passes"] is True + assert _feature(mcp)["quality_result"] is None + + def test_strict_mode_blocks_failing_feature(self, mcp, tmp_path): + _js_project(tmp_path, eslint_exit=1) + _write_config(tmp_path, {"enabled": True}) + + result = json.loads(mcp.feature_mark_passing(1)) + assert "Quality gate failed" in result["error"] + assert result["quality_gate"]["passed"] is False + + feature = _feature(mcp) + assert feature["passes"] is False + assert feature["in_progress"] is True + assert feature["failure_count"] == 1 + assert feature["failure_reason"] == "Quality gate failed: eslint" + assert feature["quality_result"]["passed"] is False + + def test_strict_mode_passes_after_fix(self, mcp, tmp_path): + _js_project(tmp_path, eslint_exit=1) + _write_config(tmp_path, {"enabled": True}) + json.loads(mcp.feature_mark_passing(1)) + + _fake_tool(tmp_path / "node_modules" / ".bin" / "eslint", 0) + result = json.loads(mcp.feature_mark_passing(1)) + assert result["success"] is True + assert result["quality_gate"]["passed"] is True + + feature = _feature(mcp) + assert feature["passes"] is True + assert feature["in_progress"] is False + assert feature["failure_reason"] is None + assert feature["failure_count"] == 1 # history is kept + assert feature["quality_result"]["passed"] is True + + def test_non_strict_mode_warns(self, mcp, tmp_path): + _js_project(tmp_path, tsc_exit=1) + _write_config(tmp_path, {"enabled": True, "strict_mode": False}) + + result = json.loads(mcp.feature_mark_passing(1)) + assert result["success"] is True + assert result["quality_gate"]["passed"] is False + assert "warning" in result + feature = _feature(mcp) + assert feature["passes"] is True + assert feature["quality_result"]["passed"] is False + + def test_already_passing_skips_checks(self, mcp, tmp_path): + _write_config(tmp_path, {"enabled": True}) + mcp.feature_mark_passing(1) + result = json.loads(mcp.feature_mark_passing(1)) + assert "already passing" in result["error"] + + def test_verify_quality_stores_evidence(self, mcp, tmp_path): + _js_project(tmp_path, tsc_exit=1) + _write_config(tmp_path, {"enabled": True}) + result = json.loads(mcp.feature_verify_quality(1)) + assert result["passed"] is False + assert [c["name"] for c in result["checks"]] == ["eslint", "tsc"] + assert _feature(mcp)["quality_result"]["summary"] == result["summary"] + # Verifying never changes status + assert _feature(mcp)["passes"] is False + + def test_verify_quality_without_feature(self, mcp, tmp_path): + _write_config(tmp_path, {"enabled": True}) + result = json.loads(mcp.feature_verify_quality()) + assert result["passed"] is True + assert result["checks"] == [] + + def test_verify_quality_unknown_feature(self, mcp, tmp_path): + _write_config(tmp_path, {"enabled": True}) + assert "not found" in json.loads(mcp.feature_verify_quality(999))["error"] + + @pytest.mark.parametrize("section", [None, {"enabled": False}]) + def test_verify_quality_disabled_runs_nothing(self, mcp, tmp_path, monkeypatch, section): + _js_project(tmp_path, eslint_exit=1) + if section is not None: + _write_config(tmp_path, section) + + def fail(*args, **kwargs): + raise AssertionError("quality tooling must not run when gates are disabled") + + monkeypatch.setattr(quality_gates, "detect_checks", fail) + monkeypatch.setattr(quality_gates, "_run_check", fail) + monkeypatch.setattr(quality_gates.subprocess, "run", fail) + + for feature_id in (1, None): + result = json.loads(mcp.feature_verify_quality(feature_id)) + assert result["error"] == ( + "Quality gates are not enabled for this project (.autoforge/config.json)" + ) + assert _feature(mcp)["quality_result"] is None + + def test_mark_failing_tracks_reason_and_count(self, mcp): + mcp.feature_mark_failing(1, reason="Login button does nothing") + mcp.feature_mark_failing(1) + feature = _feature(mcp) + assert feature["failure_count"] == 2 + assert feature["failure_reason"] == "Login button does nothing" + + +def test_migration_adds_columns_to_existing_db(tmp_path): + """Databases created before quality gates get the new columns.""" + import sqlite3 + + from api.database import create_database, dispose_engine + + 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 VALUES (1, 1, 'c', 'n', 'd', '[]', 0)") + conn.commit() + conn.close() + + create_database(tmp_path) + dispose_engine(tmp_path) + + conn = sqlite3.connect(db_dir / "features.db") + columns = {row[1] for row in conn.execute("PRAGMA table_info(features)")} + row = conn.execute("SELECT failure_count, failure_reason, quality_result FROM features").fetchone() + conn.close() + assert {"quality_result", "failure_reason", "failure_count"} <= columns + assert row == (0, None, None) + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v", *sys.argv[1:]])) + + +class TestClientToolExposure: + """feature_verify_quality is only offered to agents when gates are enabled.""" + + TOOL = "mcp__features__feature_verify_quality" + + def test_hidden_without_config(self, tmp_path): + from client import get_feature_tools + for agent_type in ("coding", "testing", "initializer"): + assert self.TOOL not in get_feature_tools(agent_type, tmp_path) + + def test_hidden_when_disabled(self, tmp_path): + from client import get_feature_tools + _write_config(tmp_path, {"enabled": False}) + assert self.TOOL not in get_feature_tools("coding", tmp_path) + assert self.TOOL not in get_feature_tools("testing", tmp_path) + + def test_exposed_when_enabled(self, tmp_path): + from client import get_feature_tools + _write_config(tmp_path, {"enabled": True}) + assert self.TOOL in get_feature_tools("coding", tmp_path) + assert self.TOOL in get_feature_tools("testing", tmp_path) + assert self.TOOL not in get_feature_tools("initializer", tmp_path)