diff --git a/.egg/phase-permissions.json b/.egg/phase-permissions.json new file mode 100644 index 0000000000..28f22f0873 --- /dev/null +++ b/.egg/phase-permissions.json @@ -0,0 +1,147 @@ +{ + "schemaVersion": "1.0", + "phases": { + "refine": { + "allowed_operations": [ + { + "type": "gh", + "pattern": "issue comment *", + "description": "Comment on issues" + }, + { + "type": "gh", + "pattern": "issue edit *", + "description": "Edit issues" + }, + { + "type": "egg-contract", + "pattern": "add-decision *", + "description": "Create HITL decision points" + }, + { + "type": "egg-contract", + "pattern": "show *", + "description": "View contract state" + } + ], + "blocked_operations": [ + { + "type": "git", + "pattern": "push *", + "description": "Cannot push code during refine phase" + }, + { + "type": "gh", + "pattern": "pr create*", + "description": "Cannot create PRs during refine phase" + } + ], + "exit_requires": "human" + }, + "plan": { + "allowed_operations": [ + { + "type": "gh", + "pattern": "issue comment *", + "description": "Comment on issues" + }, + { + "type": "gh", + "pattern": "issue edit *", + "description": "Edit issues" + }, + { + "type": "egg-contract", + "pattern": "add-decision *", + "description": "Create HITL decision points" + }, + { + "type": "egg-contract", + "pattern": "show *", + "description": "View contract state" + } + ], + "blocked_operations": [ + { + "type": "git", + "pattern": "push *", + "description": "Cannot push code during plan phase" + }, + { + "type": "gh", + "pattern": "pr create*", + "description": "Cannot create PRs during plan phase" + } + ], + "exit_requires": "human" + }, + "implement": { + "allowed_operations": [ + { + "type": "git", + "pattern": "push *", + "description": "Push code to remote" + }, + { + "type": "egg-contract", + "pattern": "add-commit *", + "description": "Link commits to tasks" + }, + { + "type": "egg-contract", + "pattern": "update-notes *", + "description": "Add implementation notes" + }, + { + "type": "egg-contract", + "pattern": "mark-task *", + "description": "Mark task status (reviewer only)" + }, + { + "type": "egg-contract", + "pattern": "mark-phase *", + "description": "Mark phase status (reviewer only)" + }, + { + "type": "egg-contract", + "pattern": "show *", + "description": "View contract state" + } + ], + "blocked_operations": [ + { + "type": "gh", + "pattern": "pr create*", + "description": "Cannot create PRs until implementation is complete" + } + ], + "exit_requires": "reviewer" + }, + "pr": { + "allowed_operations": [ + { + "type": "gh", + "pattern": "pr create*", + "description": "Create pull requests" + }, + { + "type": "gh", + "pattern": "pr edit *", + "description": "Edit pull requests" + }, + { + "type": "git", + "pattern": "push *", + "description": "Push code to remote" + }, + { + "type": "egg-contract", + "pattern": "show *", + "description": "View contract state" + } + ], + "blocked_operations": [], + "exit_requires": "human" + } + } +} diff --git a/.egg/schemas/phase-permissions.schema.json b/.egg/schemas/phase-permissions.schema.json new file mode 100644 index 0000000000..9a282115fc --- /dev/null +++ b/.egg/schemas/phase-permissions.schema.json @@ -0,0 +1,88 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/jwbron/egg/schemas/phase-permissions.schema.json", + "title": "SDLC Phase Permissions", + "description": "Defines operation permissions for each pipeline phase", + "type": "object", + "required": ["schemaVersion", "phases"], + "properties": { + "schemaVersion": { + "type": "string", + "description": "Schema version", + "pattern": "^[0-9]+\\.[0-9]+$", + "default": "1.0" + }, + "phases": { + "type": "object", + "description": "Permission definitions for each phase", + "additionalProperties": { + "$ref": "#/$defs/phasePermissions" + }, + "properties": { + "refine": { + "$ref": "#/$defs/phasePermissions" + }, + "plan": { + "$ref": "#/$defs/phasePermissions" + }, + "implement": { + "$ref": "#/$defs/phasePermissions" + }, + "pr": { + "$ref": "#/$defs/phasePermissions" + } + } + } + }, + "$defs": { + "phasePermissions": { + "type": "object", + "description": "Permission set for a single phase", + "required": ["allowed_operations", "blocked_operations", "exit_requires"], + "properties": { + "allowed_operations": { + "type": "array", + "description": "Operations allowed in this phase", + "items": { + "$ref": "#/$defs/operation" + } + }, + "blocked_operations": { + "type": "array", + "description": "Operations explicitly blocked in this phase", + "items": { + "$ref": "#/$defs/operation" + } + }, + "exit_requires": { + "type": "string", + "description": "Role required to exit this phase", + "enum": ["human", "reviewer", "implementer"] + } + }, + "additionalProperties": false + }, + "operation": { + "type": "object", + "description": "An operation that can be allowed or blocked", + "required": ["type", "pattern"], + "properties": { + "type": { + "type": "string", + "description": "Operation type", + "enum": ["git", "gh", "egg-contract"] + }, + "pattern": { + "type": "string", + "description": "Command pattern to match (supports wildcards)", + "minLength": 1 + }, + "description": { + "type": "string", + "description": "Human-readable description of the operation" + } + }, + "additionalProperties": false + } + } +} diff --git a/gateway/gateway.py b/gateway/gateway.py index fafda47556..173ca541e2 100644 --- a/gateway/gateway.py +++ b/gateway/gateway.py @@ -159,6 +159,16 @@ app.register_blueprint(contract_bp) +# Register phase API blueprint +try: + from .phase_api import phase_bp + + app.register_blueprint(phase_bp) +except ImportError: + from phase_api import phase_bp # type: ignore[import-not-found, no-redef] + + app.register_blueprint(phase_bp) + @app.errorhandler(Exception) def handle_unhandled_exception(e: Exception) -> tuple[Response, int]: diff --git a/gateway/phase_api.py b/gateway/phase_api.py new file mode 100644 index 0000000000..c8fafdf8ad --- /dev/null +++ b/gateway/phase_api.py @@ -0,0 +1,511 @@ +""" +Phase API endpoints for the gateway. + +Provides REST endpoints for phase transitions and operation filtering +in the SDLC pipeline. +""" + +import os +import sys +from pathlib import Path +from typing import Any + +from flask import Blueprint, Response, g, jsonify, request + +# Try relative imports first (module mode), fall back to absolute (script mode) +try: + from .auth import require_session_auth + from .phase_filter import ( + OperationType, + PipelinePhase, + filter_operation, + get_phase_filter, + ) + from .phase_transition import ( + TransitionRole, + can_transition_to, + get_next_phase, + ) +except ImportError: + from auth import require_session_auth # type: ignore[no-redef, import-not-found] + from phase_filter import ( # type: ignore[no-redef, import-not-found] + OperationType, + PipelinePhase, + filter_operation, + get_phase_filter, + ) + from phase_transition import ( # type: ignore[no-redef, import-not-found] + TransitionRole, + can_transition_to, + get_next_phase, + ) + +# Add shared directory to path for egg_contracts +_shared_path = Path(__file__).parent.parent / "shared" +if _shared_path.exists() and str(_shared_path) not in sys.path: + sys.path.insert(0, str(_shared_path)) + +from egg_contracts import ( + ContractNotFoundError, + ContractValidationError, + Role, + apply_mutation, + load_contract, + save_contract, +) + +# Import gateway logging +try: + from egg_logging import get_logger +except ImportError: + import logging + + def get_logger( # type: ignore[misc] + name: str, + level: int | str = logging.INFO, + component: str | None = None, + ) -> logging.Logger: + return logging.getLogger(name) + + +logger = get_logger("gateway.phase") + +# Blueprint for phase endpoints +phase_bp = Blueprint("phase", __name__, url_prefix="/api/v1/phase") + +# Allowed base paths for repository access +# In production, contracts should be in the worktree directory or /app +ALLOWED_REPO_BASES = [ + Path("/app"), + Path("/home/egg/repos"), + Path.home() / "repos", +] + + +def validate_repo_path(repo_path: Path) -> tuple[bool, str]: + """Validate that a repo path is within allowed directories. + + Args: + repo_path: The repository path to validate + + Returns: + Tuple of (is_valid, error_message) + """ + try: + resolved = repo_path.resolve() + except (OSError, ValueError) as e: + return False, f"Invalid path: {e}" + + # Allow current directory (.) + if repo_path == Path("."): + return True, "" + + # Check if path is within any allowed base + for base in ALLOWED_REPO_BASES: + try: + resolved_base = base.resolve() + if resolved_base.exists() and resolved.is_relative_to(resolved_base): + return True, "" + except (OSError, ValueError): + continue + + return False, f"Path '{repo_path}' is not within allowed directories" + + +def get_role_from_context() -> TransitionRole | None: + """Get the agent role from workflow context. + + Role source priority (highest to lowest): + 1. Session metadata (production path - set by launcher) + 2. X-Egg-Role header (for gateway testing only) + 3. EGG_AGENT_ROLE environment variable (development fallback) + + Returns: + The TransitionRole if valid, None otherwise + """ + # Production path: role from session metadata + if hasattr(g, "session") and g.session: + session_role = getattr(g.session, "agent_role", None) + if session_role: + try: + return TransitionRole(session_role.lower()) + except ValueError: + return None + + # Testing path: role from header (only when enabled) + if os.environ.get("EGG_ENABLE_TEST_ROLE_HEADER") == "1": + header_role = request.headers.get("X-Egg-Role") + if header_role: + try: + return TransitionRole(header_role.lower()) + except ValueError: + return None + + # Fallback: environment variable + env_role = os.environ.get("EGG_AGENT_ROLE") + if env_role: + try: + return TransitionRole(env_role.lower()) + except ValueError: + return None + + return None + + +def make_phase_error( + message: str, + status_code: int = 400, + details: dict[str, Any] | None = None, +) -> tuple[Response, int]: + """Create a phase error response.""" + response: dict[str, Any] = {"success": False, "message": message} + if details: + response["details"] = details + return jsonify(response), status_code + + +def make_phase_success( + message: str, + data: dict[str, Any] | None = None, +) -> tuple[Response, int]: + """Create a phase success response.""" + response: dict[str, Any] = {"success": True, "message": message} + if data: + response["data"] = data + return jsonify(response), 200 + + +@phase_bp.route("/advance", methods=["POST"]) +@require_session_auth +def advance_phase() -> tuple[Response, int]: + """ + Advance the pipeline to the next phase. + + Request body: + { + "issue_number": 123, + "repo_path": "/path/to/repo", // optional + "reason": "All tasks complete" // optional + } + + The role is determined from workflow context. + The current phase is read from the contract. + The next phase is determined by the transition graph. + + Returns: + Success: {"success": true, "message": "...", "data": {"from_phase": "...", "to_phase": "..."}} + Error: {"success": false, "message": "...", "details": {...}} + """ + data = request.get_json() + if not data: + return make_phase_error("Missing request body") + + issue_number = data.get("issue_number") + if not issue_number: + return make_phase_error("Missing issue_number") + + repo_path = Path(data.get("repo_path", ".")) + + # Validate repo_path to prevent path traversal + is_valid, error = validate_repo_path(repo_path) + if not is_valid: + return make_phase_error(error, status_code=400) + + reason = data.get("reason") + actor = data.get("actor", "agent") + + # Get role from context + role = get_role_from_context() + if not role: + return make_phase_error( + "Cannot determine agent role. Role must be set via workflow context.", + status_code=403, + details={"hint": "Set EGG_AGENT_ROLE via workflow inputs"}, + ) + + # Load the contract to get current phase + try: + contract = load_contract(issue_number, repo_path) + except ContractNotFoundError: + return make_phase_error( + f"Contract for issue #{issue_number} not found", + status_code=404, + ) + except ContractValidationError as e: + return make_phase_error( + f"Contract validation failed: {e}", + status_code=500, + ) + + # Get current phase and determine next phase + current_phase = PipelinePhase(contract.current_phase.value) + next_phase = get_next_phase(current_phase) + + if next_phase is None: + return make_phase_error( + f"Cannot advance from phase '{current_phase.value}': terminal state", + status_code=400, + details={"current_phase": current_phase.value}, + ) + + # Validate the transition + result = can_transition_to(current_phase, next_phase, role, actor) + + if not result.success: + logger.warning( + "Phase transition denied", + issue=issue_number, + role=role.value, + from_phase=current_phase.value, + to_phase=next_phase.value, + error=result.message, + ) + return make_phase_error( + result.message, + status_code=403, + details={ + "role": role.value, + "from_phase": current_phase.value, + "to_phase": next_phase.value, + }, + ) + + # Apply the phase transition to the contract + # Use the contract's mutation system for audit trail + mutation_result = apply_mutation( + contract=contract, + role=Role.HUMAN if role == TransitionRole.HUMAN else Role.REVIEWER, + actor=actor, + field_path="current_phase", + new_value=next_phase.value, + reason=reason or f"Transition from {current_phase.value} to {next_phase.value}", + ) + + if not mutation_result.success: + return make_phase_error( + f"Failed to update contract: {mutation_result.message}", + status_code=500, + ) + + # Save the updated contract + assert mutation_result.contract is not None + try: + save_contract(mutation_result.contract, repo_path) + except Exception as e: + logger.error( + "Failed to save contract after phase transition", + issue=issue_number, + error=str(e), + ) + return make_phase_error( + f"Failed to save contract: {e}", + status_code=500, + ) + + logger.info( + "Phase transition completed", + issue=issue_number, + role=role.value, + actor=actor, + from_phase=current_phase.value, + to_phase=next_phase.value, + ) + + return make_phase_success( + f"Advanced from '{current_phase.value}' to '{next_phase.value}'", + data={ + "from_phase": current_phase.value, + "to_phase": next_phase.value, + "transitioned_by": actor, + }, + ) + + +@phase_bp.route("/filter", methods=["POST"]) +@require_session_auth +def filter_phase_operation() -> tuple[Response, int]: + """ + Check if an operation is allowed in the current phase. + + Request body: + { + "issue_number": 123, + "repo_path": "/path/to/repo", // optional + "operation_type": "git", // "git", "gh", or "egg-contract" + "command": "push origin main" + } + + Returns: + Success: {"success": true, "message": "Operation allowed", "data": {"allowed": true}} + Blocked: {"success": false, "message": "...", "data": {"allowed": false, "reason": "..."}} + """ + data = request.get_json() + if not data: + return make_phase_error("Missing request body") + + issue_number = data.get("issue_number") + operation_type = data.get("operation_type") + command = data.get("command") + + if not issue_number: + return make_phase_error("Missing issue_number") + if not operation_type: + return make_phase_error("Missing operation_type") + if not command: + return make_phase_error("Missing command") + + repo_path = Path(data.get("repo_path", ".")) + + # Validate repo_path to prevent path traversal + is_valid, error = validate_repo_path(repo_path) + if not is_valid: + return make_phase_error(error, status_code=400) + + # Validate operation type + try: + op_type = OperationType(operation_type) + except ValueError: + return make_phase_error( + f"Invalid operation_type: {operation_type}", + details={"valid_types": [t.value for t in OperationType]}, + ) + + # Load contract to get current phase + try: + contract = load_contract(issue_number, repo_path) + except ContractNotFoundError: + return make_phase_error( + f"Contract for issue #{issue_number} not found", + status_code=404, + ) + except ContractValidationError as e: + return make_phase_error( + f"Contract validation failed: {e}", + status_code=500, + ) + + current_phase = PipelinePhase(contract.current_phase.value) + + # Filter the operation + result = filter_operation(current_phase, op_type, command) + + if result.allowed: + return make_phase_success( + result.message, + data={"allowed": True, "phase": current_phase.value}, + ) + else: + logger.warning( + "Operation blocked by phase filter", + issue=issue_number, + phase=current_phase.value, + operation_type=operation_type, + command=command, + reason=result.blocked_reason, + ) + return make_phase_error( + result.message, + status_code=403, + details={ + "allowed": False, + "phase": current_phase.value, + "operation_type": operation_type, + "reason": result.blocked_reason, + }, + ) + + +@phase_bp.route("/current/", methods=["GET"]) +@require_session_auth +def get_current_phase(issue_number: int) -> tuple[Response, int]: + """ + Get the current phase for an issue. + + URL params: + issue_number: GitHub issue number + + Query params: + repo_path: Path to the repository (optional) + + Returns: + {"success": true, "data": {"phase": "implement", "exit_requires": "reviewer"}} + """ + repo_path = Path(request.args.get("repo_path", ".")) + + # Validate repo_path to prevent path traversal + is_valid, error = validate_repo_path(repo_path) + if not is_valid: + return make_phase_error(error, status_code=400) + + try: + contract = load_contract(issue_number, repo_path) + except ContractNotFoundError: + return make_phase_error( + f"Contract for issue #{issue_number} not found", + status_code=404, + ) + except ContractValidationError as e: + return make_phase_error( + f"Contract validation failed: {e}", + status_code=500, + ) + + current_phase = PipelinePhase(contract.current_phase.value) + phase_filter = get_phase_filter() + exit_requires = phase_filter.get_exit_requirement(current_phase) + next_phase = get_next_phase(current_phase) + + return make_phase_success( + f"Current phase: {current_phase.value}", + data={ + "phase": current_phase.value, + "exit_requires": exit_requires, + "next_phase": next_phase.value if next_phase else None, + }, + ) + + +@phase_bp.route("/permissions/", methods=["GET"]) +@require_session_auth +def get_phase_permissions(phase: str) -> tuple[Response, int]: + """ + Get the permissions for a specific phase. + + URL params: + phase: Phase name (refine, plan, implement, pr) + + Returns: + {"success": true, "data": {"allowed": [...], "blocked": [...], "exit_requires": "..."}} + """ + try: + pipeline_phase = PipelinePhase(phase) + except ValueError: + return make_phase_error( + f"Invalid phase: {phase}", + details={"valid_phases": [p.value for p in PipelinePhase]}, + ) + + phase_filter = get_phase_filter() + permissions = phase_filter.get_permissions(pipeline_phase) + + if not permissions: + return make_phase_error( + f"No permissions configured for phase: {phase}", + status_code=404, + ) + + return make_phase_success( + f"Permissions for phase: {phase}", + data={ + "phase": phase, + "allowed_operations": [ + {"type": op.type.value, "pattern": op.pattern, "description": op.description} + for op in permissions.allowed_operations + ], + "blocked_operations": [ + {"type": op.type.value, "pattern": op.pattern, "description": op.description} + for op in permissions.blocked_operations + ], + "exit_requires": permissions.exit_requires, + }, + ) diff --git a/gateway/phase_filter.py b/gateway/phase_filter.py new file mode 100644 index 0000000000..1de9032db5 --- /dev/null +++ b/gateway/phase_filter.py @@ -0,0 +1,428 @@ +""" +Phase-based operation filtering for the SDLC pipeline. + +This module enforces phase-specific operation restrictions. Each pipeline phase +(refine, plan, implement, pr) has a defined set of permitted and blocked operations. +The gateway uses this module to filter operations against the current phase. +""" + +from __future__ import annotations + +import fnmatch +import json +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path +from typing import Any + + +class OperationType(StrEnum): + """Types of operations that can be filtered.""" + + GIT = "git" + GH = "gh" + EGG_CONTRACT = "egg-contract" + + +class PipelinePhase(StrEnum): + """Pipeline phases. + + Note: This duplicates egg_contracts.models.PipelinePhase to avoid import + complexity in the gateway module. Values must be kept in sync. + """ + + REFINE = "refine" + PLAN = "plan" + IMPLEMENT = "implement" + PR = "pr" + + +@dataclass +class Operation: + """An operation that can be allowed or blocked.""" + + type: OperationType + pattern: str + description: str = "" + + def matches(self, command: str) -> bool: + """Check if this operation pattern matches the given command. + + Args: + command: The command to match (e.g., "push origin main") + + Returns: + True if the pattern matches the command + """ + # Use fnmatch for wildcard matching + return fnmatch.fnmatch(command, self.pattern) + + +@dataclass +class PhasePermissions: + """Permission set for a single phase.""" + + allowed_operations: list[Operation] + blocked_operations: list[Operation] + exit_requires: str # "human", "reviewer", or "implementer" + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> PhasePermissions: + """Create PhasePermissions from a dictionary.""" + allowed = [ + Operation( + type=OperationType(op["type"]), + pattern=op["pattern"], + description=op.get("description", ""), + ) + for op in data.get("allowed_operations", []) + ] + blocked = [ + Operation( + type=OperationType(op["type"]), + pattern=op["pattern"], + description=op.get("description", ""), + ) + for op in data.get("blocked_operations", []) + ] + return cls( + allowed_operations=allowed, + blocked_operations=blocked, + exit_requires=data.get("exit_requires", "human"), + ) + + +@dataclass +class FilterResult: + """Result of an operation filter check.""" + + allowed: bool + message: str + operation_type: OperationType | None = None + phase: PipelinePhase | None = None + blocked_reason: str | None = None + + @classmethod + def allow(cls, message: str = "Operation allowed") -> FilterResult: + """Create an allowed result.""" + return cls(allowed=True, message=message) + + @classmethod + def block( + cls, + message: str, + operation_type: OperationType, + phase: PipelinePhase, + blocked_reason: str, + ) -> FilterResult: + """Create a blocked result.""" + return cls( + allowed=False, + message=message, + operation_type=operation_type, + phase=phase, + blocked_reason=blocked_reason, + ) + + +class PhaseFilter: + """Filter operations based on the current pipeline phase.""" + + def __init__(self, permissions_path: Path | None = None): + """Initialize the phase filter. + + Args: + permissions_path: Path to the phase-permissions.json file. + If None, uses the default path. + """ + self._permissions: dict[PipelinePhase, PhasePermissions] = {} + self._permissions_path = permissions_path + self._loaded = False + + def _get_default_permissions_path(self) -> Path: + """Get the default path to phase-permissions.json.""" + # In container: /app/.egg/phase-permissions.json + # On host: relative to this file + container_path = Path("/app/.egg/phase-permissions.json") + if container_path.exists(): + return container_path + + # Try relative to this file + relative_path = Path(__file__).parent.parent / ".egg" / "phase-permissions.json" + if relative_path.exists(): + return relative_path + + # Fall back to current directory + return Path(".egg/phase-permissions.json") + + def _load_permissions(self) -> None: + """Load permissions from the JSON file.""" + if self._loaded: + return + + path = self._permissions_path or self._get_default_permissions_path() + + if not path.exists(): + # Use default permissions if file doesn't exist + self._permissions = self._get_default_permissions() + self._loaded = True + return + + with path.open() as f: + data = json.load(f) + + phases_data = data.get("phases", {}) + for phase_name, phase_data in phases_data.items(): + try: + phase = PipelinePhase(phase_name) + self._permissions[phase] = PhasePermissions.from_dict(phase_data) + except ValueError: + # Skip unknown phases + pass + + self._loaded = True + + def _get_default_permissions(self) -> dict[PipelinePhase, PhasePermissions]: + """Get default permissions when no file is available.""" + return { + PipelinePhase.REFINE: PhasePermissions( + allowed_operations=[ + Operation(OperationType.GH, "issue comment *", "Comment on issues"), + Operation(OperationType.GH, "issue edit *", "Edit issues"), + Operation( + OperationType.EGG_CONTRACT, "add-decision *", "Create HITL decisions" + ), + Operation(OperationType.EGG_CONTRACT, "show *", "View contract state"), + ], + blocked_operations=[ + Operation(OperationType.GIT, "push *", "Cannot push during refine"), + Operation(OperationType.GH, "pr create*", "Cannot create PRs during refine"), + ], + exit_requires="human", + ), + PipelinePhase.PLAN: PhasePermissions( + allowed_operations=[ + Operation(OperationType.GH, "issue comment *", "Comment on issues"), + Operation(OperationType.GH, "issue edit *", "Edit issues"), + Operation( + OperationType.EGG_CONTRACT, "add-decision *", "Create HITL decisions" + ), + Operation(OperationType.EGG_CONTRACT, "show *", "View contract state"), + ], + blocked_operations=[ + Operation(OperationType.GIT, "push *", "Cannot push during plan"), + Operation(OperationType.GH, "pr create*", "Cannot create PRs during plan"), + ], + exit_requires="human", + ), + PipelinePhase.IMPLEMENT: PhasePermissions( + allowed_operations=[ + Operation(OperationType.GIT, "push *", "Push code"), + Operation(OperationType.EGG_CONTRACT, "add-commit *", "Link commits"), + Operation(OperationType.EGG_CONTRACT, "update-notes *", "Add notes"), + Operation(OperationType.EGG_CONTRACT, "mark-task *", "Mark task status"), + Operation(OperationType.EGG_CONTRACT, "mark-phase *", "Mark phase status"), + Operation(OperationType.EGG_CONTRACT, "show *", "View contract state"), + ], + blocked_operations=[ + Operation(OperationType.GH, "pr create*", "Cannot create PRs until complete"), + ], + exit_requires="reviewer", + ), + PipelinePhase.PR: PhasePermissions( + allowed_operations=[ + Operation(OperationType.GH, "pr create*", "Create PRs"), + Operation(OperationType.GH, "pr edit *", "Edit PRs"), + Operation(OperationType.GIT, "push *", "Push code"), + Operation(OperationType.EGG_CONTRACT, "show *", "View contract state"), + ], + blocked_operations=[], + exit_requires="human", + ), + } + + def get_permissions(self, phase: PipelinePhase) -> PhasePermissions | None: + """Get the permissions for a phase. + + Args: + phase: The pipeline phase + + Returns: + PhasePermissions for the phase, or None if not found + """ + self._load_permissions() + return self._permissions.get(phase) + + def filter_operation( + self, + phase: PipelinePhase, + operation_type: OperationType, + command: str, + ) -> FilterResult: + """Filter an operation against phase permissions. + + Args: + phase: Current pipeline phase + operation_type: Type of operation (git, gh, egg-contract) + command: The command being executed (e.g., "push origin main") + + Returns: + FilterResult indicating whether the operation is allowed + """ + self._load_permissions() + + permissions = self._permissions.get(phase) + if not permissions: + return FilterResult.allow("Phase permissions not configured, allowing by default") + + # Check blocked operations first (blocked takes precedence) + for blocked_op in permissions.blocked_operations: + if blocked_op.type == operation_type and blocked_op.matches(command): + return FilterResult.block( + message=self._format_blocked_message( + operation_type, command, phase, blocked_op.description + ), + operation_type=operation_type, + phase=phase, + blocked_reason=blocked_op.description, + ) + + # Check if operation is in allowed list (if list is non-empty) + if permissions.allowed_operations: + for allowed_op in permissions.allowed_operations: + if allowed_op.type == operation_type and allowed_op.matches(command): + return FilterResult.allow(f"Operation allowed: {allowed_op.description}") + + # If we have an allowed list and this operation isn't in it, + # we need to decide whether to block or allow by default + # For now, we only block explicitly blocked operations + return FilterResult.allow("Operation not explicitly blocked") + + return FilterResult.allow("No restrictions configured for this phase") + + def _format_blocked_message( + self, + operation_type: OperationType, + command: str, + phase: PipelinePhase, + reason: str, + ) -> str: + """Format a blocked operation message. + + Args: + operation_type: Type of operation + command: The command that was blocked + phase: Current pipeline phase + reason: Reason the operation was blocked + + Returns: + Formatted error message + """ + return ( + f"Operation blocked: {operation_type.value} {command}\n" + f"Phase '{phase.value}' does not permit this operation.\n" + f"Reason: {reason}\n" + f"To perform this operation, the pipeline must advance to a later phase." + ) + + def is_operation_blocked( + self, + phase: PipelinePhase, + operation_type: OperationType, + command: str, + ) -> bool: + """Check if an operation is blocked (convenience method). + + Args: + phase: Current pipeline phase + operation_type: Type of operation + command: The command being executed + + Returns: + True if the operation is blocked + """ + result = self.filter_operation(phase, operation_type, command) + return not result.allowed + + def get_exit_requirement(self, phase: PipelinePhase) -> str | None: + """Get the role required to exit a phase. + + Args: + phase: The pipeline phase + + Returns: + Role name required to exit, or None if not configured + """ + permissions = self.get_permissions(phase) + if permissions: + return permissions.exit_requires + return None + + +# Module-level instance for convenience +_filter: PhaseFilter | None = None + + +def get_phase_filter() -> PhaseFilter: + """Get the global PhaseFilter instance.""" + global _filter + if _filter is None: + _filter = PhaseFilter() + return _filter + + +def reset_phase_filter() -> None: + """Reset the global PhaseFilter instance. + + This clears the cached filter, causing the next call to get_phase_filter() + to create a fresh instance. Useful for testing and when configuration + files are updated. + """ + global _filter + _filter = None + + +def filter_operation( + phase: str | PipelinePhase, + operation_type: str | OperationType, + command: str, +) -> FilterResult: + """Filter an operation against phase permissions (convenience function). + + Args: + phase: Current pipeline phase (string or PipelinePhase enum) + operation_type: Type of operation (string or OperationType enum) + command: The command being executed + + Returns: + FilterResult indicating whether the operation is allowed + """ + if isinstance(phase, str): + phase = PipelinePhase(phase) + if isinstance(operation_type, str): + operation_type = OperationType(operation_type) + + return get_phase_filter().filter_operation(phase, operation_type, command) + + +def is_operation_blocked( + phase: str | PipelinePhase, + operation_type: str | OperationType, + command: str, +) -> bool: + """Check if an operation is blocked (convenience function). + + Args: + phase: Current pipeline phase + operation_type: Type of operation + command: The command being executed + + Returns: + True if the operation is blocked + """ + if isinstance(phase, str): + phase = PipelinePhase(phase) + if isinstance(operation_type, str): + operation_type = OperationType(operation_type) + + return get_phase_filter().is_operation_blocked(phase, operation_type, command) diff --git a/gateway/phase_transition.py b/gateway/phase_transition.py new file mode 100644 index 0000000000..24d339f9aa --- /dev/null +++ b/gateway/phase_transition.py @@ -0,0 +1,273 @@ +""" +Phase transition logic for the SDLC pipeline. + +This module handles phase transitions, validating that the caller has +the appropriate role to advance the pipeline from one phase to another. +""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass +from datetime import UTC, datetime +from enum import StrEnum +from pathlib import Path +from typing import Any + +# Add shared directory to path for egg_contracts +_shared_path = Path(__file__).parent.parent / "shared" +if _shared_path.exists() and str(_shared_path) not in sys.path: + sys.path.insert(0, str(_shared_path)) + +# Try relative imports first (module mode), fall back to absolute (script mode) +try: + from .phase_filter import PipelinePhase, get_phase_filter +except ImportError: + from phase_filter import ( # type: ignore[no-redef, import-not-found] + PipelinePhase, + get_phase_filter, + ) + + +class TransitionRole(StrEnum): + """Roles that can perform phase transitions.""" + + IMPLEMENTER = "implementer" + REVIEWER = "reviewer" + HUMAN = "human" + + +# Phase transition graph: defines which phases can transition to which +VALID_TRANSITIONS: dict[PipelinePhase, list[PipelinePhase]] = { + PipelinePhase.REFINE: [PipelinePhase.PLAN], + PipelinePhase.PLAN: [PipelinePhase.IMPLEMENT], + PipelinePhase.IMPLEMENT: [PipelinePhase.PR], + PipelinePhase.PR: [], # Terminal state - no automatic transitions +} + + +@dataclass +class TransitionResult: + """Result of a phase transition attempt.""" + + success: bool + message: str + from_phase: PipelinePhase | None = None + to_phase: PipelinePhase | None = None + transitioned_at: datetime | None = None + transitioned_by: str | None = None + + @classmethod + def allowed( + cls, + from_phase: PipelinePhase, + to_phase: PipelinePhase, + transitioned_by: str, + ) -> TransitionResult: + """Create a successful transition result.""" + return cls( + success=True, + message=f"Transition from '{from_phase.value}' to '{to_phase.value}' allowed", + from_phase=from_phase, + to_phase=to_phase, + transitioned_at=datetime.now(UTC), + transitioned_by=transitioned_by, + ) + + @classmethod + def denied( + cls, + message: str, + from_phase: PipelinePhase | None = None, + to_phase: PipelinePhase | None = None, + ) -> TransitionResult: + """Create a denied transition result.""" + return cls( + success=False, + message=message, + from_phase=from_phase, + to_phase=to_phase, + ) + + +@dataclass +class TransitionRequest: + """A request to transition between phases.""" + + from_phase: PipelinePhase + to_phase: PipelinePhase + role: TransitionRole + actor: str + reason: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> TransitionRequest: + """Create a TransitionRequest from a dictionary.""" + return cls( + from_phase=PipelinePhase(data["from_phase"]), + to_phase=PipelinePhase(data["to_phase"]), + role=TransitionRole(data["role"]), + actor=data.get("actor", "unknown"), + reason=data.get("reason"), + ) + + +def validate_transition(request: TransitionRequest) -> TransitionResult: + """Validate a phase transition request. + + Checks that: + 1. The transition is valid (from_phase can transition to to_phase) + 2. The caller's role meets the exit requirement for from_phase + + Args: + request: The transition request to validate + + Returns: + TransitionResult indicating whether the transition is allowed + """ + # Check if the transition is valid + valid_targets = VALID_TRANSITIONS.get(request.from_phase, []) + if request.to_phase not in valid_targets: + return TransitionResult.denied( + message=( + f"Invalid transition: Cannot transition from '{request.from_phase.value}' " + f"to '{request.to_phase.value}'. " + f"Valid targets: {[p.value for p in valid_targets] if valid_targets else 'none'}" + ), + from_phase=request.from_phase, + to_phase=request.to_phase, + ) + + # Get the exit requirement for the from_phase + phase_filter = get_phase_filter() + exit_requires = phase_filter.get_exit_requirement(request.from_phase) + + if exit_requires is None: + # No requirement configured, allow the transition + return TransitionResult.allowed( + from_phase=request.from_phase, + to_phase=request.to_phase, + transitioned_by=request.actor, + ) + + # Check if the role meets the exit requirement + if not _role_can_exit(request.role, exit_requires): + return TransitionResult.denied( + message=( + f"Transition denied: Role '{request.role.value}' cannot exit phase " + f"'{request.from_phase.value}'. Required role: '{exit_requires}'" + ), + from_phase=request.from_phase, + to_phase=request.to_phase, + ) + + return TransitionResult.allowed( + from_phase=request.from_phase, + to_phase=request.to_phase, + transitioned_by=request.actor, + ) + + +def _role_can_exit(role: TransitionRole, required: str) -> bool: + """Check if a role can satisfy an exit requirement. + + Role hierarchy: + - human can satisfy any requirement + - reviewer can satisfy reviewer and implementer requirements + - implementer can only satisfy implementer requirements + + Args: + role: The caller's role + required: The required role to exit the phase + + Returns: + True if the role can satisfy the requirement + """ + # Human can do anything + if role == TransitionRole.HUMAN: + return True + + # Reviewer can satisfy reviewer and implementer + if role == TransitionRole.REVIEWER: + return required in ("reviewer", "implementer") + + # Implementer can only satisfy implementer + if role == TransitionRole.IMPLEMENTER: + return required == "implementer" + + return False + + +def get_next_phase(current: PipelinePhase) -> PipelinePhase | None: + """Get the next phase in the pipeline. + + Args: + current: The current phase + + Returns: + The next phase, or None if at terminal state + """ + valid_targets = VALID_TRANSITIONS.get(current, []) + if valid_targets: + return valid_targets[0] + return None + + +def can_transition_to( + from_phase: str | PipelinePhase, + to_phase: str | PipelinePhase, + role: str | TransitionRole, + actor: str = "unknown", +) -> TransitionResult: + """Check if a transition is allowed (convenience function). + + Args: + from_phase: Current phase + to_phase: Target phase + role: Caller's role + actor: Actor performing the transition + + Returns: + TransitionResult indicating whether the transition is allowed + """ + if isinstance(from_phase, str): + from_phase = PipelinePhase(from_phase) + if isinstance(to_phase, str): + to_phase = PipelinePhase(to_phase) + if isinstance(role, str): + role = TransitionRole(role) + + request = TransitionRequest( + from_phase=from_phase, + to_phase=to_phase, + role=role, + actor=actor, + ) + return validate_transition(request) + + +def create_audit_entry( + result: TransitionResult, + role: TransitionRole, + reason: str | None = None, +) -> dict[str, Any]: + """Create an audit log entry for a phase transition. + + Args: + result: The transition result + role: The role that performed the transition + reason: Optional reason for the transition + + Returns: + Audit entry dictionary matching the contract schema + """ + return { + "timestamp": datetime.now(UTC).isoformat(), + "actor": result.transitioned_by or "unknown", + "role": role.value, + "action": "transition", + "field_path": "current_phase", + "old_value": result.from_phase.value if result.from_phase else None, + "new_value": result.to_phase.value if result.to_phase else None, + "reason": reason, + } diff --git a/gateway/tests/conftest.py b/gateway/tests/conftest.py index 0b02b4432d..dbb58e23d6 100644 --- a/gateway/tests/conftest.py +++ b/gateway/tests/conftest.py @@ -73,13 +73,16 @@ def _load_module_with_replaced_imports( module.__loader__ = None module.__package__ = "" + # Register the module BEFORE executing so that decorators like @dataclass + # can find the module in sys.modules when looking up the class's __module__ + sys.modules[name] = module + # Also register under gateway. prefix so package-style imports work with patches + sys.modules[f"gateway.{name}"] = module + # Execute the modified source code = compile(source, path, "exec") exec(code, module.__dict__) - sys.modules[name] = module - # Also register under gateway. prefix so package-style imports work with patches - sys.modules[f"gateway.{name}"] = module return module @@ -201,6 +204,33 @@ def _load_module_with_replaced_imports( GATEWAY_DIR / "contract_api.py", import_replacements={ "from .auth import": "from auth import", + "from .git_client import": "from git_client import", + }, +) + +# phase_filter has no relative imports to other gateway modules +phase_filter = _load_module_with_replaced_imports( + "phase_filter", + GATEWAY_DIR / "phase_filter.py", +) + +# phase_transition imports from phase_filter +phase_transition = _load_module_with_replaced_imports( + "phase_transition", + GATEWAY_DIR / "phase_transition.py", + import_replacements={ + "from .phase_filter import": "from phase_filter import", + }, +) + +# phase_api imports from auth, phase_filter, phase_transition +phase_api = _load_module_with_replaced_imports( + "phase_api", + GATEWAY_DIR / "phase_api.py", + import_replacements={ + "from .auth import": "from auth import", + "from .phase_filter import": "from phase_filter import", + "from .phase_transition import": "from phase_transition import", }, ) @@ -214,6 +244,7 @@ def _load_module_with_replaced_imports( "from .contract_api import": "from contract_api import", "from .git_client import": "from git_client import", "from .github_client import": "from github_client import", + "from .phase_api import": "from phase_api import", "from .policy import": "from policy import", "from .private_repo_policy import": "from private_repo_policy import", "from .repo_parser import": "from repo_parser import", diff --git a/gateway/tests/test_phase_api.py b/gateway/tests/test_phase_api.py new file mode 100644 index 0000000000..ad4b58588c --- /dev/null +++ b/gateway/tests/test_phase_api.py @@ -0,0 +1,626 @@ +""" +Tests for Phase API endpoints. + +Tests cover: +- POST /api/v1/phase/advance - Advance to next phase +- POST /api/v1/phase/filter - Check if operation is allowed +- GET /api/v1/phase/current/ - Get current phase +- GET /api/v1/phase/permissions/ - Get phase permissions +""" + +import os +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import auth +import phase_api +import pytest +import session_manager +from session_manager import SessionValidationResult + +import gateway + + +@pytest.fixture +def client(): + """Create test client for Flask app.""" + gateway.app.config["TESTING"] = True + with gateway.app.test_client() as client: + yield client + + +@pytest.fixture +def auth_headers(): + """Return valid session authentication headers with mocked session validation.""" + mock_session = MagicMock() + mock_session.mode = "public" + mock_session.container_id = "test-container" + mock_session.expires_at = None + mock_session.agent_role = "human" # Default to human for most tests + + mock_result = SessionValidationResult(valid=True, session=mock_session) + + from private_repo_policy import PrivateRepoPolicyResult + + mock_policy_result = PrivateRepoPolicyResult( + allowed=True, + reason="Test mode - access allowed", + visibility="public", + ) + + # Clear auth module's cached references + auth._session_manager = None + auth._rate_limiter = None + + if "gateway.auth" in sys.modules: + sys.modules["gateway.auth"]._session_manager = None + sys.modules["gateway.auth"]._rate_limiter = None + + current_session_manager = sys.modules.get("session_manager", session_manager) + + with ( + patch.object( + current_session_manager, "validate_session_for_request", return_value=mock_result + ), + patch.object(gateway, "check_private_repo_access", return_value=mock_policy_result), + ): + yield {"Authorization": "Bearer test-session-token"} + + +@pytest.fixture +def mock_contract(): + """Create a mock contract.""" + from egg_contracts.models import Contract, IssueInfo, PipelinePhase + + return Contract( + schemaVersion="1.0", + issue=IssueInfo( + number=123, + title="Test Issue", + url="https://github.com/test/repo/issues/123", + ), + current_phase=PipelinePhase.REFINE, + ) + + +@pytest.fixture +def mock_contract_implement(): + """Create a mock contract in implement phase.""" + from egg_contracts.models import Contract, IssueInfo, PipelinePhase + + return Contract( + schemaVersion="1.0", + issue=IssueInfo( + number=123, + title="Test Issue", + url="https://github.com/test/repo/issues/123", + ), + current_phase=PipelinePhase.IMPLEMENT, + ) + + +# --------------------------------------------------------------------------- +# GET /api/v1/phase/current/ tests +# --------------------------------------------------------------------------- + + +class TestGetCurrentPhase: + """Tests for GET /api/v1/phase/current/.""" + + def test_get_current_phase_success(self, client, auth_headers, mock_contract): + """Get current phase for an issue.""" + with patch("phase_api.load_contract", return_value=mock_contract): + response = client.get( + "/api/v1/phase/current/123", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = response.get_json() + assert data["success"] is True + assert data["data"]["phase"] == "refine" + assert data["data"]["exit_requires"] == "human" + assert data["data"]["next_phase"] == "plan" + + def test_get_current_phase_not_found(self, client, auth_headers): + """Get current phase for non-existent contract.""" + from egg_contracts import ContractNotFoundError + + with patch( + "phase_api.load_contract", + side_effect=ContractNotFoundError(123, Path(".")), + ): + response = client.get( + "/api/v1/phase/current/123", + headers=auth_headers, + ) + + assert response.status_code == 404 + data = response.get_json() + assert data["success"] is False + + +# --------------------------------------------------------------------------- +# GET /api/v1/phase/permissions/ tests +# --------------------------------------------------------------------------- + + +class TestGetPhasePermissions: + """Tests for GET /api/v1/phase/permissions/.""" + + def test_get_permissions_refine(self, client, auth_headers): + """Get permissions for refine phase.""" + response = client.get( + "/api/v1/phase/permissions/refine", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = response.get_json() + assert data["success"] is True + assert data["data"]["phase"] == "refine" + assert data["data"]["exit_requires"] == "human" + assert len(data["data"]["blocked_operations"]) > 0 + + def test_get_permissions_invalid_phase(self, client, auth_headers): + """Get permissions for invalid phase.""" + response = client.get( + "/api/v1/phase/permissions/invalid_phase", + headers=auth_headers, + ) + + assert response.status_code == 400 + data = response.get_json() + assert data["success"] is False + assert "valid_phases" in data.get("details", {}) + + +# --------------------------------------------------------------------------- +# POST /api/v1/phase/filter tests +# --------------------------------------------------------------------------- + + +class TestFilterPhaseOperation: + """Tests for POST /api/v1/phase/filter.""" + + def test_filter_allowed_operation(self, client, auth_headers, mock_contract_implement): + """Filter an allowed operation.""" + with patch("phase_api.load_contract", return_value=mock_contract_implement): + response = client.post( + "/api/v1/phase/filter", + headers=auth_headers, + json={ + "issue_number": 123, + "operation_type": "git", + "command": "push origin main", + }, + ) + + assert response.status_code == 200 + data = response.get_json() + assert data["success"] is True + assert data["data"]["allowed"] is True + + def test_filter_blocked_operation(self, client, auth_headers, mock_contract): + """Filter a blocked operation.""" + with patch("phase_api.load_contract", return_value=mock_contract): + response = client.post( + "/api/v1/phase/filter", + headers=auth_headers, + json={ + "issue_number": 123, + "operation_type": "git", + "command": "push origin main", + }, + ) + + assert response.status_code == 403 + data = response.get_json() + assert data["success"] is False + assert data["details"]["allowed"] is False + + def test_filter_missing_fields(self, client, auth_headers): + """Filter with missing fields.""" + response = client.post( + "/api/v1/phase/filter", + headers=auth_headers, + json={"issue_number": 123}, + ) + + assert response.status_code == 400 + data = response.get_json() + assert data["success"] is False + assert "Missing" in data["message"] + + def test_filter_invalid_operation_type(self, client, auth_headers, mock_contract): + """Filter with invalid operation type.""" + with patch("phase_api.load_contract", return_value=mock_contract): + response = client.post( + "/api/v1/phase/filter", + headers=auth_headers, + json={ + "issue_number": 123, + "operation_type": "invalid", + "command": "something", + }, + ) + + assert response.status_code == 400 + data = response.get_json() + assert "valid_types" in data.get("details", {}) + + +# --------------------------------------------------------------------------- +# POST /api/v1/phase/advance tests +# --------------------------------------------------------------------------- + + +class TestAdvancePhase: + """Tests for POST /api/v1/phase/advance.""" + + def test_advance_phase_success(self, client, auth_headers, mock_contract): + """Advance phase with proper authorization.""" + mock_mutation_result = MagicMock() + mock_mutation_result.success = True + mock_mutation_result.contract = mock_contract + + with ( + patch("phase_api.load_contract", return_value=mock_contract), + patch("phase_api.apply_mutation", return_value=mock_mutation_result), + patch("phase_api.save_contract"), + ): + response = client.post( + "/api/v1/phase/advance", + headers=auth_headers, + json={ + "issue_number": 123, + "reason": "Analysis complete", + }, + ) + + assert response.status_code == 200 + data = response.get_json() + assert data["success"] is True + assert data["data"]["from_phase"] == "refine" + assert data["data"]["to_phase"] == "plan" + + def test_advance_phase_unauthorized(self, client, mock_contract): + """Advance phase without proper role.""" + # Create a session with implementer role + mock_session = MagicMock() + mock_session.mode = "public" + mock_session.container_id = "test-container" + mock_session.expires_at = None + mock_session.agent_role = "implementer" + + mock_result = SessionValidationResult(valid=True, session=mock_session) + + from private_repo_policy import PrivateRepoPolicyResult + + mock_policy_result = PrivateRepoPolicyResult( + allowed=True, + reason="Test mode", + visibility="public", + ) + + auth._session_manager = None + auth._rate_limiter = None + + current_session_manager = sys.modules.get("session_manager", session_manager) + + with ( + patch.object( + current_session_manager, + "validate_session_for_request", + return_value=mock_result, + ), + patch.object(gateway, "check_private_repo_access", return_value=mock_policy_result), + patch("phase_api.load_contract", return_value=mock_contract), + ): + response = client.post( + "/api/v1/phase/advance", + headers={"Authorization": "Bearer test-token"}, + json={"issue_number": 123}, + ) + + assert response.status_code == 403 + data = response.get_json() + assert data["success"] is False + assert "cannot exit" in data["message"].lower() or "denied" in data["message"].lower() + + def test_advance_phase_terminal_state(self, client, auth_headers): + """Cannot advance from PR phase (terminal).""" + from egg_contracts.models import Contract, IssueInfo, PipelinePhase + + terminal_contract = Contract( + schemaVersion="1.0", + issue=IssueInfo( + number=123, + title="Test Issue", + url="https://github.com/test/repo/issues/123", + ), + current_phase=PipelinePhase.PR, + ) + + with patch("phase_api.load_contract", return_value=terminal_contract): + response = client.post( + "/api/v1/phase/advance", + headers=auth_headers, + json={"issue_number": 123}, + ) + + assert response.status_code == 400 + data = response.get_json() + assert data["success"] is False + assert "terminal" in data["message"].lower() + + def test_advance_phase_missing_issue(self, client, auth_headers): + """Advance phase without issue number.""" + response = client.post( + "/api/v1/phase/advance", + headers=auth_headers, + json={"reason": "test"}, # Provide some data but no issue_number + ) + + assert response.status_code == 400 + data = response.get_json() + assert "Missing issue_number" in data["message"] + + def test_advance_phase_contract_not_found(self, client, auth_headers): + """Advance phase for non-existent contract.""" + from egg_contracts import ContractNotFoundError + + with patch( + "phase_api.load_contract", + side_effect=ContractNotFoundError(123, Path(".")), + ): + response = client.post( + "/api/v1/phase/advance", + headers=auth_headers, + json={"issue_number": 123}, + ) + + assert response.status_code == 404 + + +# --------------------------------------------------------------------------- +# Role resolution tests +# --------------------------------------------------------------------------- + + +class TestGetRoleFromContext: + """Tests for get_role_from_context in phase_api.""" + + def test_role_from_session(self, client, auth_headers): + """Role resolved from session.""" + # auth_headers fixture sets agent_role to "human" + with client.application.test_request_context(): + from flask import g + + mock_session = MagicMock() + mock_session.agent_role = "reviewer" + g.session = mock_session + + role = phase_api.get_role_from_context() + + assert role is not None + assert role.value == "reviewer" + + def test_role_from_header_when_enabled(self, client, auth_headers): + """Role resolved from header when enabled.""" + with ( + client.application.test_request_context(headers={"X-Egg-Role": "implementer"}), + patch.dict(os.environ, {"EGG_ENABLE_TEST_ROLE_HEADER": "1"}, clear=False), + ): + from flask import g + + g.session = None + role = phase_api.get_role_from_context() + + assert role is not None + assert role.value == "implementer" + + def test_role_from_env(self, client, auth_headers): + """Role resolved from environment variable.""" + env = os.environ.copy() + env.pop("EGG_ENABLE_TEST_ROLE_HEADER", None) + env["EGG_AGENT_ROLE"] = "reviewer" + + with ( + client.application.test_request_context(), + patch.dict(os.environ, env, clear=True), + ): + from flask import g + + g.session = None + role = phase_api.get_role_from_context() + + assert role is not None + assert role.value == "reviewer" + + def test_invalid_role_returns_none(self, client, auth_headers): + """Invalid role returns None.""" + env = os.environ.copy() + env.pop("EGG_ENABLE_TEST_ROLE_HEADER", None) + env["EGG_AGENT_ROLE"] = "invalid_role" + + with ( + client.application.test_request_context(), + patch.dict(os.environ, env, clear=True), + ): + from flask import g + + g.session = None + role = phase_api.get_role_from_context() + + assert role is None + + +# --------------------------------------------------------------------------- +# Path traversal security tests +# --------------------------------------------------------------------------- + + +class TestPathTraversalProtection: + """Tests for path traversal attack prevention.""" + + def test_advance_phase_path_traversal_rejected(self, client, auth_headers): + """Path traversal in advance phase is rejected.""" + response = client.post( + "/api/v1/phase/advance", + headers=auth_headers, + json={ + "issue_number": 123, + "repo_path": "../../../etc", + }, + ) + assert response.status_code == 400 + data = response.get_json() + assert data["success"] is False + assert "not within allowed" in data["message"].lower() or "path" in data["message"].lower() + + def test_advance_phase_absolute_path_outside_allowed(self, client, auth_headers): + """Absolute path outside allowed directories is rejected.""" + response = client.post( + "/api/v1/phase/advance", + headers=auth_headers, + json={ + "issue_number": 123, + "repo_path": "/etc/passwd", + }, + ) + assert response.status_code == 400 + data = response.get_json() + assert data["success"] is False + + def test_filter_path_traversal_rejected(self, client, auth_headers): + """Path traversal in filter endpoint is rejected.""" + response = client.post( + "/api/v1/phase/filter", + headers=auth_headers, + json={ + "issue_number": 123, + "repo_path": "../../..", + "operation_type": "git", + "command": "push origin main", + }, + ) + assert response.status_code == 400 + data = response.get_json() + assert data["success"] is False + + def test_current_phase_path_traversal_rejected(self, client, auth_headers): + """Path traversal in current phase endpoint is rejected.""" + response = client.get( + "/api/v1/phase/current/123?repo_path=../../../etc", + headers=auth_headers, + ) + assert response.status_code == 400 + data = response.get_json() + assert data["success"] is False + + def test_allowed_repo_path_accepted(self, client, auth_headers, mock_contract): + """Allowed repo path (current directory) is accepted.""" + with patch("phase_api.load_contract", return_value=mock_contract): + response = client.get( + "/api/v1/phase/current/123?repo_path=.", + headers=auth_headers, + ) + assert response.status_code == 200 + + +# --------------------------------------------------------------------------- +# Integration test for reviewer advancing implement→PR +# --------------------------------------------------------------------------- + + +class TestReviewerPhaseTransitionIntegration: + """Integration tests for reviewer phase transitions. + + These tests use real contract mutations (not mocked) to verify + that reviewer can actually advance from implement to PR phase. + """ + + def test_reviewer_can_advance_implement_to_pr(self, client): + """Reviewer can advance from implement to PR phase with real mutation.""" + import tempfile + + from egg_contracts import save_contract + from egg_contracts.models import Contract, IssueInfo, PipelinePhase + + # Create a contract in implement phase + contract = Contract( + schemaVersion="1.0", + issue=IssueInfo( + number=999, + title="Test Issue", + url="https://github.com/test/repo/issues/999", + ), + current_phase=PipelinePhase.IMPLEMENT, + ) + + # Save to temp directory + with tempfile.TemporaryDirectory() as tmpdir: + tmppath = Path(tmpdir) + save_contract(contract, tmppath) + + # Create a session with reviewer role + mock_session = MagicMock() + mock_session.mode = "public" + mock_session.container_id = "test-container" + mock_session.expires_at = None + mock_session.agent_role = "reviewer" + + mock_result = SessionValidationResult(valid=True, session=mock_session) + + from private_repo_policy import PrivateRepoPolicyResult + + mock_policy_result = PrivateRepoPolicyResult( + allowed=True, + reason="Test mode", + visibility="public", + ) + + auth._session_manager = None + auth._rate_limiter = None + + current_session_manager = sys.modules.get("session_manager", session_manager) + + # Patch the allowed paths to include our temp directory + with ( + patch.object( + current_session_manager, + "validate_session_for_request", + return_value=mock_result, + ), + patch.object(gateway, "check_private_repo_access", return_value=mock_policy_result), + patch.object( + phase_api, + "ALLOWED_REPO_BASES", + [Path(tmpdir), Path("/app"), Path.home() / "repos"], + ), + ): + response = client.post( + "/api/v1/phase/advance", + headers={"Authorization": "Bearer test-token"}, + json={ + "issue_number": 999, + "repo_path": tmpdir, + "reason": "Implementation complete", + }, + ) + + assert response.status_code == 200, ( + f"Expected 200, got {response.status_code}: {response.get_json()}" + ) + data = response.get_json() + assert data["success"] is True + assert data["data"]["from_phase"] == "implement" + assert data["data"]["to_phase"] == "pr" + + # Verify the contract was actually updated + from egg_contracts import load_contract + + updated_contract = load_contract(999, tmppath) + assert updated_contract.current_phase == PipelinePhase.PR diff --git a/gateway/tests/test_phase_filter.py b/gateway/tests/test_phase_filter.py new file mode 100644 index 0000000000..71387fb0b3 --- /dev/null +++ b/gateway/tests/test_phase_filter.py @@ -0,0 +1,376 @@ +""" +Tests for Phase Filter module. + +Tests cover: +- Operation matching +- Phase permission loading +- Blocked/allowed operation filtering +- Exit requirements +""" + +import json +import tempfile +from pathlib import Path + +import phase_filter +import pytest +from phase_filter import ( + Operation, + OperationType, + PhaseFilter, + PhasePermissions, + PipelinePhase, + filter_operation, + is_operation_blocked, + reset_phase_filter, +) + + +class TestOperation: + """Tests for Operation class.""" + + def test_matches_exact(self): + """Exact pattern matches exact command.""" + op = Operation(OperationType.GIT, "push origin main") + assert op.matches("push origin main") is True + assert op.matches("push origin develop") is False + + def test_matches_wildcard(self): + """Wildcard pattern matches multiple commands.""" + op = Operation(OperationType.GIT, "push *") + assert op.matches("push origin main") is True + assert op.matches("push upstream develop") is True + assert op.matches("pull origin main") is False + + def test_matches_multiple_wildcards(self): + """Multiple wildcards work correctly.""" + op = Operation(OperationType.GH, "issue * *") + assert op.matches("issue comment 123") is True + assert op.matches("issue edit 456") is True + assert op.matches("pr comment 123") is False + + +class TestPhasePermissions: + """Tests for PhasePermissions class.""" + + def test_from_dict_basic(self): + """Create PhasePermissions from dictionary.""" + data = { + "allowed_operations": [ + {"type": "git", "pattern": "push *", "description": "Push code"} + ], + "blocked_operations": [ + {"type": "gh", "pattern": "pr create *", "description": "No PRs"} + ], + "exit_requires": "reviewer", + } + permissions = PhasePermissions.from_dict(data) + + assert len(permissions.allowed_operations) == 1 + assert len(permissions.blocked_operations) == 1 + assert permissions.exit_requires == "reviewer" + assert permissions.allowed_operations[0].type == OperationType.GIT + assert permissions.blocked_operations[0].pattern == "pr create *" + + def test_from_dict_empty_lists(self): + """Handle empty operation lists.""" + data = { + "allowed_operations": [], + "blocked_operations": [], + "exit_requires": "human", + } + permissions = PhasePermissions.from_dict(data) + + assert len(permissions.allowed_operations) == 0 + assert len(permissions.blocked_operations) == 0 + + +class TestPhaseFilter: + """Tests for PhaseFilter class.""" + + @pytest.fixture + def custom_permissions_file(self) -> Path: + """Create a temporary permissions file.""" + permissions = { + "schemaVersion": "1.0", + "phases": { + "refine": { + "allowed_operations": [ + {"type": "gh", "pattern": "issue comment *", "description": "Comment"}, + ], + "blocked_operations": [ + {"type": "git", "pattern": "push *", "description": "No push"}, + ], + "exit_requires": "human", + }, + "implement": { + "allowed_operations": [ + {"type": "git", "pattern": "push *", "description": "Push code"}, + ], + "blocked_operations": [ + {"type": "gh", "pattern": "pr create *", "description": "No PR yet"}, + ], + "exit_requires": "reviewer", + }, + }, + } + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump(permissions, f) + return Path(f.name) + + def test_load_from_file(self, custom_permissions_file: Path): + """Load permissions from a file.""" + pf = PhaseFilter(permissions_path=custom_permissions_file) + permissions = pf.get_permissions(PipelinePhase.REFINE) + + assert permissions is not None + assert permissions.exit_requires == "human" + assert len(permissions.blocked_operations) == 1 + + def test_default_permissions_when_no_file(self): + """Use default permissions when file doesn't exist.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent/path.json")) + permissions = pf.get_permissions(PipelinePhase.IMPLEMENT) + + assert permissions is not None + assert permissions.exit_requires == "reviewer" + + def test_filter_blocked_operation(self, custom_permissions_file: Path): + """Blocked operations are correctly identified.""" + pf = PhaseFilter(permissions_path=custom_permissions_file) + result = pf.filter_operation( + PipelinePhase.REFINE, + OperationType.GIT, + "push origin main", + ) + + assert result.allowed is False + assert result.phase == PipelinePhase.REFINE + assert result.operation_type == OperationType.GIT + assert "push" in result.message.lower() + + def test_filter_allowed_operation(self, custom_permissions_file: Path): + """Allowed operations are correctly identified.""" + pf = PhaseFilter(permissions_path=custom_permissions_file) + result = pf.filter_operation( + PipelinePhase.REFINE, + OperationType.GH, + "issue comment 123", + ) + + assert result.allowed is True + + def test_filter_not_explicitly_blocked(self, custom_permissions_file: Path): + """Operations not explicitly blocked are allowed.""" + pf = PhaseFilter(permissions_path=custom_permissions_file) + result = pf.filter_operation( + PipelinePhase.REFINE, + OperationType.EGG_CONTRACT, + "show", + ) + + # Not in blocked list, so allowed + assert result.allowed is True + + def test_is_operation_blocked_helper(self, custom_permissions_file: Path): + """is_operation_blocked helper works correctly.""" + pf = PhaseFilter(permissions_path=custom_permissions_file) + + assert ( + pf.is_operation_blocked(PipelinePhase.REFINE, OperationType.GIT, "push origin main") + is True + ) + assert ( + pf.is_operation_blocked(PipelinePhase.IMPLEMENT, OperationType.GIT, "push origin main") + is False + ) + + def test_get_exit_requirement(self, custom_permissions_file: Path): + """Get exit requirement for a phase.""" + pf = PhaseFilter(permissions_path=custom_permissions_file) + + assert pf.get_exit_requirement(PipelinePhase.REFINE) == "human" + assert pf.get_exit_requirement(PipelinePhase.IMPLEMENT) == "reviewer" + + +class TestFilterOperationFunction: + """Tests for the convenience filter_operation function.""" + + def test_filter_with_strings(self): + """filter_operation accepts strings.""" + # Reset global filter to use defaults + phase_filter._filter = None + + result = filter_operation("implement", "gh", "pr create") + + assert result.allowed is False + assert "pr create" in result.message.lower() or "pr" in str(result.blocked_reason).lower() + + def test_filter_with_enums(self): + """filter_operation accepts enums.""" + phase_filter._filter = None + + result = filter_operation( + PipelinePhase.IMPLEMENT, + OperationType.GIT, + "push origin main", + ) + + assert result.allowed is True + + +class TestIsOperationBlockedFunction: + """Tests for the convenience is_operation_blocked function.""" + + def test_blocked_during_refine(self): + """Git push is blocked during refine phase.""" + phase_filter._filter = None + + assert is_operation_blocked("refine", "git", "push origin main") is True + + def test_allowed_during_implement(self): + """Git push is allowed during implement phase.""" + phase_filter._filter = None + + assert is_operation_blocked("implement", "git", "push origin main") is False + + def test_pr_create_blocked_until_pr_phase(self): + """PR create is blocked until PR phase.""" + phase_filter._filter = None + + assert is_operation_blocked("refine", "gh", "pr create") is True + assert is_operation_blocked("plan", "gh", "pr create") is True + assert is_operation_blocked("implement", "gh", "pr create") is True + assert is_operation_blocked("pr", "gh", "pr create") is False + + +class TestDefaultPermissions: + """Tests for default permission configuration.""" + + @pytest.fixture(autouse=True) + def reset_filter(self): + """Reset the global filter before each test.""" + phase_filter._filter = None + yield + phase_filter._filter = None + + def test_refine_phase_blocks_push(self): + """Refine phase blocks git push.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + result = pf.filter_operation(PipelinePhase.REFINE, OperationType.GIT, "push origin main") + assert result.allowed is False + + def test_refine_phase_blocks_pr_create(self): + """Refine phase blocks PR creation.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + result = pf.filter_operation(PipelinePhase.REFINE, OperationType.GH, "pr create") + assert result.allowed is False + + def test_plan_phase_blocks_push(self): + """Plan phase blocks git push.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + result = pf.filter_operation(PipelinePhase.PLAN, OperationType.GIT, "push origin main") + assert result.allowed is False + + def test_implement_phase_allows_push(self): + """Implement phase allows git push.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + result = pf.filter_operation(PipelinePhase.IMPLEMENT, OperationType.GIT, "push origin main") + assert result.allowed is True + + def test_implement_phase_blocks_pr_create(self): + """Implement phase blocks PR creation.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + result = pf.filter_operation(PipelinePhase.IMPLEMENT, OperationType.GH, "pr create") + assert result.allowed is False + + def test_pr_phase_allows_pr_create(self): + """PR phase allows PR creation.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + result = pf.filter_operation(PipelinePhase.PR, OperationType.GH, "pr create") + assert result.allowed is True + + def test_pr_phase_allows_push(self): + """PR phase allows git push.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + result = pf.filter_operation(PipelinePhase.PR, OperationType.GIT, "push origin main") + assert result.allowed is True + + +class TestResetPhaseFilter: + """Tests for reset_phase_filter function.""" + + def test_reset_clears_cached_filter(self): + """reset_phase_filter clears the cached filter instance.""" + # Access the filter to cache it + from phase_filter import get_phase_filter + + _ = get_phase_filter() + assert phase_filter._filter is not None + + # Reset should clear it + reset_phase_filter() + assert phase_filter._filter is None + + def test_reset_allows_new_instance(self): + """After reset, get_phase_filter creates a new instance.""" + from phase_filter import get_phase_filter + + filter1 = get_phase_filter() + reset_phase_filter() + filter2 = get_phase_filter() + + assert filter1 is not filter2 + + +class TestPatternEdgeCases: + """Tests for pattern matching edge cases.""" + + @pytest.fixture(autouse=True) + def reset_filter(self): + """Reset the global filter before each test.""" + phase_filter._filter = None + yield + phase_filter._filter = None + + def test_pr_create_without_args_matches_pattern(self): + """'pr create' without arguments matches 'pr create*' pattern.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + # In refine phase, pr create should be blocked + result = pf.filter_operation(PipelinePhase.REFINE, OperationType.GH, "pr create") + assert result.allowed is False + + # In pr phase, pr create should be allowed + result = pf.filter_operation(PipelinePhase.PR, OperationType.GH, "pr create") + assert result.allowed is True + + def test_pr_create_with_args_matches_pattern(self): + """'pr create --title foo' matches 'pr create*' pattern.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + result = pf.filter_operation( + PipelinePhase.PR, OperationType.GH, "pr create --title 'Test PR'" + ) + assert result.allowed is True + + def test_partial_command_does_not_match_blocked_pattern(self): + """Commands that partially match blocked patterns should not be blocked.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + # 'push-status' should not match 'push *' pattern + result = pf.filter_operation(PipelinePhase.REFINE, OperationType.GIT, "push-status") + # push-status doesn't match "push *" because there's no space after push + assert result.allowed is True + + def test_git_push_without_remote_matches_pattern(self): + """'git push' without remote matches 'push *' pattern.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + # Note: 'push *' requires at least one character after 'push ' + # 'push' alone won't match 'push *', it would need 'push something' + result = pf.filter_operation(PipelinePhase.IMPLEMENT, OperationType.GIT, "push origin") + assert result.allowed is True + + def test_pr_creates_typo_not_blocked(self): + """'pr creates' (typo) should not be blocked by 'pr create*' pattern.""" + pf = PhaseFilter(permissions_path=Path("/nonexistent")) + # 'pr creates' matches 'pr create*' because * matches 's' + result = pf.filter_operation(PipelinePhase.REFINE, OperationType.GH, "pr creates") + assert result.allowed is False # It does match, so it's blocked in refine diff --git a/gateway/tests/test_phase_transition.py b/gateway/tests/test_phase_transition.py new file mode 100644 index 0000000000..384946adbc --- /dev/null +++ b/gateway/tests/test_phase_transition.py @@ -0,0 +1,328 @@ +""" +Tests for Phase Transition module. + +Tests cover: +- Valid and invalid transitions +- Role-based transition authorization +- Transition result creation +- Audit entry generation +""" + +from phase_filter import PipelinePhase +from phase_transition import ( + VALID_TRANSITIONS, + TransitionRequest, + TransitionResult, + TransitionRole, + can_transition_to, + create_audit_entry, + get_next_phase, + validate_transition, +) + + +class TestTransitionResult: + """Tests for TransitionResult class.""" + + def test_allowed_result(self): + """Create an allowed transition result.""" + result = TransitionResult.allowed( + from_phase=PipelinePhase.REFINE, + to_phase=PipelinePhase.PLAN, + transitioned_by="egg", + ) + + assert result.success is True + assert result.from_phase == PipelinePhase.REFINE + assert result.to_phase == PipelinePhase.PLAN + assert result.transitioned_by == "egg" + assert result.transitioned_at is not None + + def test_denied_result(self): + """Create a denied transition result.""" + result = TransitionResult.denied( + message="Role 'implementer' cannot exit phase 'refine'", + from_phase=PipelinePhase.REFINE, + to_phase=PipelinePhase.PLAN, + ) + + assert result.success is False + assert "implementer" in result.message + assert result.from_phase == PipelinePhase.REFINE + assert result.to_phase == PipelinePhase.PLAN + + +class TestTransitionRequest: + """Tests for TransitionRequest class.""" + + def test_from_dict(self): + """Create TransitionRequest from dictionary.""" + data = { + "from_phase": "refine", + "to_phase": "plan", + "role": "human", + "actor": "test-user", + "reason": "Analysis complete", + } + request = TransitionRequest.from_dict(data) + + assert request.from_phase == PipelinePhase.REFINE + assert request.to_phase == PipelinePhase.PLAN + assert request.role == TransitionRole.HUMAN + assert request.actor == "test-user" + assert request.reason == "Analysis complete" + + def test_from_dict_minimal(self): + """Create TransitionRequest with minimal fields.""" + data = { + "from_phase": "implement", + "to_phase": "pr", + "role": "reviewer", + } + request = TransitionRequest.from_dict(data) + + assert request.from_phase == PipelinePhase.IMPLEMENT + assert request.to_phase == PipelinePhase.PR + assert request.actor == "unknown" + assert request.reason is None + + +class TestValidTransitions: + """Tests for the valid transitions graph.""" + + def test_refine_to_plan(self): + """Refine can only transition to plan.""" + assert PipelinePhase.PLAN in VALID_TRANSITIONS[PipelinePhase.REFINE] + assert len(VALID_TRANSITIONS[PipelinePhase.REFINE]) == 1 + + def test_plan_to_implement(self): + """Plan can only transition to implement.""" + assert PipelinePhase.IMPLEMENT in VALID_TRANSITIONS[PipelinePhase.PLAN] + assert len(VALID_TRANSITIONS[PipelinePhase.PLAN]) == 1 + + def test_implement_to_pr(self): + """Implement can only transition to PR.""" + assert PipelinePhase.PR in VALID_TRANSITIONS[PipelinePhase.IMPLEMENT] + assert len(VALID_TRANSITIONS[PipelinePhase.IMPLEMENT]) == 1 + + def test_pr_is_terminal(self): + """PR phase has no outgoing transitions.""" + assert len(VALID_TRANSITIONS[PipelinePhase.PR]) == 0 + + +class TestValidateTransition: + """Tests for validate_transition function.""" + + def test_valid_transition_with_human(self): + """Human can transition between any valid phases.""" + request = TransitionRequest( + from_phase=PipelinePhase.REFINE, + to_phase=PipelinePhase.PLAN, + role=TransitionRole.HUMAN, + actor="test-human", + ) + result = validate_transition(request) + + assert result.success is True + assert result.from_phase == PipelinePhase.REFINE + assert result.to_phase == PipelinePhase.PLAN + + def test_invalid_transition_path(self): + """Cannot skip phases in the pipeline.""" + request = TransitionRequest( + from_phase=PipelinePhase.REFINE, + to_phase=PipelinePhase.IMPLEMENT, # Invalid - must go through plan + role=TransitionRole.HUMAN, + actor="test-human", + ) + result = validate_transition(request) + + assert result.success is False + assert "Invalid transition" in result.message + + def test_backwards_transition_blocked(self): + """Cannot transition backwards in the pipeline.""" + request = TransitionRequest( + from_phase=PipelinePhase.IMPLEMENT, + to_phase=PipelinePhase.PLAN, + role=TransitionRole.HUMAN, + actor="test-human", + ) + result = validate_transition(request) + + assert result.success is False + assert "Invalid transition" in result.message + + def test_implementer_cannot_exit_refine(self): + """Implementer cannot exit refine phase (requires human).""" + request = TransitionRequest( + from_phase=PipelinePhase.REFINE, + to_phase=PipelinePhase.PLAN, + role=TransitionRole.IMPLEMENTER, + actor="egg", + ) + result = validate_transition(request) + + assert result.success is False + assert "cannot exit" in result.message.lower() + + def test_implementer_cannot_exit_plan(self): + """Implementer cannot exit plan phase (requires human).""" + request = TransitionRequest( + from_phase=PipelinePhase.PLAN, + to_phase=PipelinePhase.IMPLEMENT, + role=TransitionRole.IMPLEMENTER, + actor="egg", + ) + result = validate_transition(request) + + assert result.success is False + + def test_reviewer_can_exit_implement(self): + """Reviewer can exit implement phase.""" + request = TransitionRequest( + from_phase=PipelinePhase.IMPLEMENT, + to_phase=PipelinePhase.PR, + role=TransitionRole.REVIEWER, + actor="reviewer-agent", + ) + result = validate_transition(request) + + assert result.success is True + + def test_implementer_cannot_exit_implement(self): + """Implementer cannot exit implement phase (requires reviewer).""" + request = TransitionRequest( + from_phase=PipelinePhase.IMPLEMENT, + to_phase=PipelinePhase.PR, + role=TransitionRole.IMPLEMENTER, + actor="egg", + ) + result = validate_transition(request) + + assert result.success is False + + def test_transition_from_terminal_phase(self): + """Cannot transition from PR phase (terminal).""" + request = TransitionRequest( + from_phase=PipelinePhase.PR, + to_phase=PipelinePhase.IMPLEMENT, # Trying to go back + role=TransitionRole.HUMAN, + actor="test-human", + ) + result = validate_transition(request) + + assert result.success is False + + +class TestRoleHierarchy: + """Tests for role hierarchy in transitions.""" + + def test_human_can_satisfy_any_requirement(self): + """Human role can satisfy any exit requirement.""" + # Human can exit refine (requires human) + result = can_transition_to(PipelinePhase.REFINE, PipelinePhase.PLAN, TransitionRole.HUMAN) + assert result.success is True + + # Human can exit implement (requires reviewer) + result = can_transition_to(PipelinePhase.IMPLEMENT, PipelinePhase.PR, TransitionRole.HUMAN) + assert result.success is True + + def test_reviewer_can_satisfy_reviewer_and_lower(self): + """Reviewer role can satisfy reviewer and implementer requirements.""" + # Reviewer can exit implement (requires reviewer) + result = can_transition_to( + PipelinePhase.IMPLEMENT, PipelinePhase.PR, TransitionRole.REVIEWER + ) + assert result.success is True + + def test_reviewer_cannot_satisfy_human_requirement(self): + """Reviewer cannot satisfy human requirement.""" + result = can_transition_to( + PipelinePhase.REFINE, PipelinePhase.PLAN, TransitionRole.REVIEWER + ) + assert result.success is False + + def test_implementer_limited_permissions(self): + """Implementer can only satisfy implementer requirement.""" + # No phase currently requires only implementer to exit + # But the logic should work if one existed + result = can_transition_to( + PipelinePhase.IMPLEMENT, PipelinePhase.PR, TransitionRole.IMPLEMENTER + ) + assert result.success is False + + +class TestGetNextPhase: + """Tests for get_next_phase function.""" + + def test_refine_next_is_plan(self): + """Next phase after refine is plan.""" + assert get_next_phase(PipelinePhase.REFINE) == PipelinePhase.PLAN + + def test_plan_next_is_implement(self): + """Next phase after plan is implement.""" + assert get_next_phase(PipelinePhase.PLAN) == PipelinePhase.IMPLEMENT + + def test_implement_next_is_pr(self): + """Next phase after implement is PR.""" + assert get_next_phase(PipelinePhase.IMPLEMENT) == PipelinePhase.PR + + def test_pr_next_is_none(self): + """PR has no next phase (terminal).""" + assert get_next_phase(PipelinePhase.PR) is None + + +class TestCanTransitionTo: + """Tests for can_transition_to convenience function.""" + + def test_with_strings(self): + """Function accepts string arguments.""" + result = can_transition_to("refine", "plan", "human", "test-actor") + + assert result.success is True + assert result.from_phase == PipelinePhase.REFINE + assert result.to_phase == PipelinePhase.PLAN + + def test_with_enums(self): + """Function accepts enum arguments.""" + result = can_transition_to( + PipelinePhase.IMPLEMENT, + PipelinePhase.PR, + TransitionRole.REVIEWER, + "reviewer-agent", + ) + + assert result.success is True + + +class TestCreateAuditEntry: + """Tests for create_audit_entry function.""" + + def test_creates_valid_entry(self): + """Create a valid audit entry from transition result.""" + result = TransitionResult.allowed( + from_phase=PipelinePhase.REFINE, + to_phase=PipelinePhase.PLAN, + transitioned_by="egg", + ) + entry = create_audit_entry(result, TransitionRole.HUMAN, "Analysis approved") + + assert entry["action"] == "transition" + assert entry["field_path"] == "current_phase" + assert entry["old_value"] == "refine" + assert entry["new_value"] == "plan" + assert entry["role"] == "human" + assert entry["actor"] == "egg" + assert entry["reason"] == "Analysis approved" + assert "timestamp" in entry + + def test_handles_none_values(self): + """Handle None values in result.""" + result = TransitionResult.denied("Test denial") + entry = create_audit_entry(result, TransitionRole.IMPLEMENTER) + + assert entry["old_value"] is None + assert entry["new_value"] is None + assert entry["actor"] == "unknown" + assert entry["reason"] is None diff --git a/shared/egg_contracts/roles.py b/shared/egg_contracts/roles.py index 46db0f8de2..2bf243b094 100644 --- a/shared/egg_contracts/roles.py +++ b/shared/egg_contracts/roles.py @@ -38,6 +38,8 @@ class Role(StrEnum): "phases.*.review_feedback.*": Role.REVIEWER, # Acceptance criteria owned by reviewer "acceptance_criteria.*.verified": Role.REVIEWER, + # Pipeline phase transitions owned by reviewer (allows implement→pr advancement) + "current_phase": Role.REVIEWER, # Decisions owned by human "decisions.*.resolved": Role.HUMAN, "decisions.*.resolution": Role.HUMAN, diff --git a/tests/shared/egg_contracts/test_roles.py b/tests/shared/egg_contracts/test_roles.py index 3e7592aa0f..09e3554891 100644 --- a/tests/shared/egg_contracts/test_roles.py +++ b/tests/shared/egg_contracts/test_roles.py @@ -59,7 +59,11 @@ def test_system_fields(self): """Test fields with default system ownership.""" assert get_field_owner("issue.number") == Role.SYSTEM assert get_field_owner("schemaVersion") == Role.SYSTEM - assert get_field_owner("current_phase") == Role.SYSTEM + + def test_current_phase_reviewer_owned(self): + """Test current_phase is owned by reviewer for phase transitions.""" + # current_phase is owned by reviewer to allow implement→pr advancement + assert get_field_owner("current_phase") == Role.REVIEWER class TestCanModify: