Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 23 additions & 3 deletions codeframe/agents/worker_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -1397,9 +1397,21 @@ async def complete_task(self, task: Task, project_root: Optional[Any] = None) ->
# Commit both operations atomically
self.db.conn.commit()

# Build completion message with task classification info
from codeframe.core.models import QualityGateType
total_gates = len(QualityGateType)
skipped_count = quality_result.gates_skipped_count if hasattr(quality_result, 'gates_skipped_count') else 0
task_category = quality_result.task_category if hasattr(quality_result, 'task_category') else None

if skipped_count > 0 and task_category:
passed_count = total_gates - skipped_count
gates_msg = f"{passed_count} gates passed, {skipped_count} skipped (not applicable for {task_category} tasks)"
else:
gates_msg = "all quality gates passed"

logger.info(
f"Task {task.id} completed successfully - "
f"all quality gates passed and evidence {evidence_id} verified"
f"{gates_msg} and evidence {evidence_id} verified"
)

return {
Expand All @@ -1408,7 +1420,9 @@ async def complete_task(self, task: Task, project_root: Optional[Any] = None) ->
"quality_gate_result": quality_result,
"evidence_verified": True,
"evidence_id": evidence_id,
"message": "Task completed successfully - all quality gates passed and evidence verified",
"message": f"Task completed successfully - {gates_msg} and evidence verified",
"task_category": task_category,
"skipped_gates": quality_result.skipped_gates if hasattr(quality_result, 'skipped_gates') else [],
}
except Exception as e:
# Rollback both operations on any error to maintain consistency
Expand All @@ -1430,13 +1444,19 @@ async def complete_task(self, task: Task, project_root: Optional[Any] = None) ->
blocker_row = cursor.fetchone()
blocker_id = blocker_row[0] if blocker_row else None

# Include task category in failure message for context
task_category = quality_result.task_category if hasattr(quality_result, 'task_category') else None
category_note = f" (task category: {task_category})" if task_category else ""

return {
"success": False,
"status": "blocked",
"quality_gate_result": quality_result,
"blocker_id": blocker_id,
"message": f"Task blocked by quality gates - {len(quality_result.failures)} failures. "
"message": f"Task blocked by quality gates{category_note} - {len(quality_result.failures)} failures. "
f"Fix issues and try again.",
"task_category": task_category,
"skipped_gates": quality_result.skipped_gates if hasattr(quality_result, 'skipped_gates') else [],
}

def _create_quality_blocker(self, task: Task, failures: List[Any]) -> int:
Expand Down
141 changes: 141 additions & 0 deletions codeframe/config/quality_gates_config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
"""Configuration for Quality Gates task classification.

This module provides configuration options for controlling how quality gates
are applied based on task classification.

Environment Variables:
QUALITY_GATES_ENABLE_TASK_CLASSIFICATION: Enable/disable task classification (default: true)
QUALITY_GATES_STRICT_MODE: Run all gates regardless of task type (default: false)

Usage:
>>> from codeframe.config.quality_gates_config import get_quality_gates_config
>>> config = get_quality_gates_config()
>>> if config.enable_task_classification:
... # Use task classification logic
... pass
"""

import logging
import os
from dataclasses import dataclass
from typing import Dict, List, Optional

from codeframe.core.models import QualityGateType

logger = logging.getLogger(__name__)


@dataclass
class QualityGatesConfig:
"""Configuration for quality gates task classification.

Attributes:
enable_task_classification: Whether to use task classification for gate selection.
When enabled, tasks are classified and only applicable gates are run.
When disabled, all gates run for all tasks (legacy behavior).
strict_mode: When True, runs all gates regardless of task classification.
This is useful for ensuring comprehensive quality checks on all tasks.
custom_category_rules: Optional override for default category-to-gates mapping.
Keys are category names (e.g., "design"), values are lists of gate types.
"""

enable_task_classification: bool = True
strict_mode: bool = False
custom_category_rules: Optional[Dict[str, List[str]]] = None

def should_use_task_classification(self) -> bool:
"""Determine if task classification should be used.

Returns True if task classification is enabled AND strict mode is disabled.

Returns:
bool: True if task classification should be applied
"""
return self.enable_task_classification and not self.strict_mode

def get_custom_gates_for_category(self, category: str) -> Optional[List[QualityGateType]]:
"""Get custom gate list for a category if defined.

Args:
category: Task category name (e.g., "design", "code_implementation")

Returns:
List of QualityGateType if custom rules exist for category, None otherwise
"""
if not self.custom_category_rules:
return None

gate_names = self.custom_category_rules.get(category)
if gate_names is None:
return None

# Convert string names to QualityGateType enum values
gates = []
for name in gate_names:
try:
gates.append(QualityGateType(name))
except ValueError:
# Invalid gate name - log warning and skip
valid_names = [g.value for g in QualityGateType]
logger.warning(
f"Invalid gate name '{name}' in custom_category_rules for category '{category}'. "
f"Valid gate names are: {valid_names}"
)

return gates if gates else None


def _parse_bool_env(key: str, default: bool) -> bool:
"""Parse boolean environment variable.

Args:
key: Environment variable name
default: Default value if not set

Returns:
Parsed boolean value
"""
value = os.environ.get(key, "").lower()
if value in ("true", "1", "yes", "on"):
return True
if value in ("false", "0", "no", "off"):
return False
return default


# Singleton config instance
_config: Optional[QualityGatesConfig] = None


def get_quality_gates_config() -> QualityGatesConfig:
"""Get the quality gates configuration.

Configuration is loaded from environment variables on first call and cached.

Environment Variables:
QUALITY_GATES_ENABLE_TASK_CLASSIFICATION: Enable task classification (default: true)
QUALITY_GATES_STRICT_MODE: Run all gates regardless of task type (default: false)

Returns:
QualityGatesConfig instance
"""
global _config

if _config is None:
_config = QualityGatesConfig(
enable_task_classification=_parse_bool_env(
"QUALITY_GATES_ENABLE_TASK_CLASSIFICATION", True
),
strict_mode=_parse_bool_env("QUALITY_GATES_STRICT_MODE", False),
)

return _config


def reset_config() -> None:
"""Reset the cached configuration.

Useful for testing when environment variables change.
"""
global _config
_config = None
19 changes: 18 additions & 1 deletion codeframe/core/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -904,13 +904,25 @@ class QualityGateFailure(BaseModel):


class QualityGateResult(BaseModel):
"""Result of running quality gates for a task (Sprint 10)."""
"""Result of running quality gates for a task (Sprint 10).

Includes information about which gates were run based on task classification,
and which gates were skipped as not applicable.
"""

task_id: int
status: str = Field(..., description="passed or failed")
failures: List[QualityGateFailure] = Field(default_factory=list)
execution_time_seconds: float = Field(..., ge=0.0)
timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
skipped_gates: List[str] = Field(
default_factory=list,
description="Gates that were skipped based on task classification"
)
task_category: Optional[str] = Field(
default=None,
description="Task category used for gate selection (e.g., 'design', 'code_implementation')"
)

@property
def passed(self) -> bool:
Expand All @@ -922,6 +934,11 @@ def has_critical_failures(self) -> bool:
"""Whether any failures are critical."""
return any(f.severity == Severity.CRITICAL for f in self.failures)

@property
def gates_skipped_count(self) -> int:
"""Number of gates that were skipped."""
return len(self.skipped_gates)


class CheckpointMetadata(BaseModel):
"""Metadata stored in checkpoint for quick inspection (Sprint 10)."""
Expand Down
159 changes: 159 additions & 0 deletions codeframe/lib/quality_gate_rules.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
"""Quality Gate Rules module for task-category-based gate applicability.

This module defines which quality gates should be applied based on the task
category. Different task types have different quality requirements:

- CODE_IMPLEMENTATION: Full quality gates (all 6)
- DESIGN: Only code review (for design document quality)
- DOCUMENTATION: Only linting (for markdown/doc linting)
- CONFIGURATION: Linting and type check
- TESTING: Tests, coverage, skip detection
- REFACTORING: Full quality gates (all 6)
- MIXED: Full quality gates (conservative approach)

Usage:
>>> from codeframe.lib.quality_gate_rules import QualityGateRules
>>> from codeframe.lib.task_classifier import TaskCategory
>>> rules = QualityGateRules()
>>> gates = rules.get_applicable_gates(TaskCategory.DESIGN)
>>> # gates = [QualityGateType.CODE_REVIEW]
"""

from typing import List, Optional

from codeframe.core.models import QualityGateType
from codeframe.lib.task_classifier import TaskCategory


# Gate applicability mapping
# Based on the Quality Gate Applicability Matrix
_GATE_RULES: dict[TaskCategory, List[QualityGateType]] = {
TaskCategory.CODE_IMPLEMENTATION: [
QualityGateType.TESTS,
QualityGateType.COVERAGE,
QualityGateType.TYPE_CHECK,
QualityGateType.LINTING,
QualityGateType.CODE_REVIEW,
QualityGateType.SKIP_DETECTION,
],
TaskCategory.DESIGN: [
QualityGateType.CODE_REVIEW,
],
TaskCategory.DOCUMENTATION: [
QualityGateType.LINTING,
],
TaskCategory.CONFIGURATION: [
QualityGateType.TYPE_CHECK,
QualityGateType.LINTING,
],
TaskCategory.TESTING: [
QualityGateType.TESTS,
QualityGateType.COVERAGE,
QualityGateType.SKIP_DETECTION,
],
TaskCategory.REFACTORING: [
QualityGateType.TESTS,
QualityGateType.COVERAGE,
QualityGateType.TYPE_CHECK,
QualityGateType.LINTING,
QualityGateType.CODE_REVIEW,
QualityGateType.SKIP_DETECTION,
],
TaskCategory.MIXED: [
QualityGateType.TESTS,
QualityGateType.COVERAGE,
QualityGateType.TYPE_CHECK,
QualityGateType.LINTING,
QualityGateType.CODE_REVIEW,
QualityGateType.SKIP_DETECTION,
],
}

# Skip reasons for each category/gate combination
_SKIP_REASONS: dict[TaskCategory, dict[QualityGateType, str]] = {
TaskCategory.DESIGN: {
QualityGateType.TESTS: "Design tasks do not produce executable code to test",
QualityGateType.COVERAGE: "Design tasks do not produce code requiring coverage",
QualityGateType.TYPE_CHECK: "Design tasks do not produce typed code",
QualityGateType.LINTING: "Design tasks may not produce lintable code",
QualityGateType.SKIP_DETECTION: "Design tasks do not include test files",
},
TaskCategory.DOCUMENTATION: {
QualityGateType.TESTS: "Documentation tasks do not produce executable code to test",
QualityGateType.COVERAGE: "Documentation tasks do not produce code requiring coverage",
QualityGateType.TYPE_CHECK: "Documentation tasks do not produce typed code",
QualityGateType.CODE_REVIEW: "Documentation tasks are reviewed through linting",
QualityGateType.SKIP_DETECTION: "Documentation tasks do not include test files",
},
TaskCategory.CONFIGURATION: {
QualityGateType.TESTS: "Configuration tasks typically don't require unit tests",
QualityGateType.COVERAGE: "Configuration tasks don't require coverage metrics",
QualityGateType.CODE_REVIEW: "Configuration is reviewed through type checking",
QualityGateType.SKIP_DETECTION: "Configuration tasks do not include test files",
},
TaskCategory.TESTING: {
QualityGateType.TYPE_CHECK: "Test code may use dynamic patterns that fail type checks",
QualityGateType.LINTING: "Test code may use patterns that trigger linting warnings",
QualityGateType.CODE_REVIEW: "Test code is validated through execution rather than review",
},
}


class QualityGateRules:
"""Rules engine for determining applicable quality gates per task category.

This class encapsulates the logic for determining which quality gates should
be executed for a given task category. It provides both positive (get gates)
and negative (should skip) queries.
"""

@property
def all_gates(self) -> List[QualityGateType]:
"""Return all quality gate types."""
return list(QualityGateType)

def get_applicable_gates(self, category: TaskCategory) -> List[QualityGateType]:
"""Get the list of quality gates that should be applied for a task category.

Args:
category: The task category

Returns:
List of QualityGateType values that should be applied (copy to prevent mutation)
"""
# Return a copy to prevent callers from mutating the internal rules
gates = _GATE_RULES.get(category, _GATE_RULES[TaskCategory.MIXED])
return list(gates)

def should_skip_gate(self, category: TaskCategory, gate: QualityGateType) -> bool:
"""Check if a specific gate should be skipped for a task category.

Args:
category: The task category
gate: The quality gate type to check

Returns:
True if the gate should be skipped, False if it should run
"""
applicable_gates = self.get_applicable_gates(category)
return gate not in applicable_gates

def get_skip_reason(
self, category: TaskCategory, gate: QualityGateType
) -> Optional[str]:
"""Get the reason why a gate is skipped for a category.

Args:
category: The task category
gate: The quality gate type

Returns:
String explaining why the gate is skipped, or None if gate is applicable
"""
if not self.should_skip_gate(category, gate):
return None

category_reasons = _SKIP_REASONS.get(category, {})
return category_reasons.get(
gate, f"{gate.value} gate is not applicable for {category.value} tasks"
)
Loading
Loading