diff --git a/.egg/schemas/contract.schema.json b/.egg/schemas/contract.schema.json new file mode 100644 index 0000000000..f3f7edba2a --- /dev/null +++ b/.egg/schemas/contract.schema.json @@ -0,0 +1,397 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/jwbron/egg/schemas/contract.schema.json", + "title": "SDLC Contract", + "description": "Contract schema for structurally enforced agent checkpoints and verification gates", + "type": "object", + "required": ["schemaVersion", "issue", "phases", "decisions", "audit_log"], + "properties": { + "schemaVersion": { + "type": "string", + "description": "Schema version for migrations", + "pattern": "^[0-9]+\\.[0-9]+$", + "default": "1.0" + }, + "issue": { + "type": "object", + "description": "Issue metadata", + "required": ["number", "title", "url"], + "properties": { + "number": { + "type": "integer", + "description": "GitHub issue number", + "minimum": 1 + }, + "title": { + "type": "string", + "description": "Issue title", + "minLength": 1 + }, + "url": { + "type": "string", + "description": "Issue URL", + "format": "uri" + } + }, + "additionalProperties": false + }, + "current_phase": { + "type": "string", + "description": "Current pipeline phase", + "enum": ["refine", "plan", "implement", "pr"], + "default": "refine" + }, + "acceptance_criteria": { + "type": "array", + "description": "Top-level acceptance criteria", + "items": { + "$ref": "#/$defs/acceptanceCriterion" + }, + "default": [] + }, + "phases": { + "type": "array", + "description": "Implementation phases with tasks", + "items": { + "$ref": "#/$defs/phase" + }, + "default": [] + }, + "decisions": { + "type": "array", + "description": "HITL decision points", + "items": { + "$ref": "#/$defs/decision" + }, + "default": [] + }, + "circuit_breaker": { + "$ref": "#/$defs/circuitBreaker" + }, + "audit_log": { + "type": "array", + "description": "Audit trail of all modifications", + "items": { + "$ref": "#/$defs/auditEntry" + }, + "default": [] + } + }, + "additionalProperties": false, + "$defs": { + "acceptanceCriterion": { + "type": "object", + "required": ["id", "description", "verified"], + "properties": { + "id": { + "type": "string", + "description": "Unique identifier (e.g., ac-1)", + "pattern": "^ac-[0-9]+$" + }, + "description": { + "type": "string", + "description": "Human-readable description", + "minLength": 1 + }, + "verified": { + "type": "boolean", + "description": "Whether this criterion has been verified", + "default": false, + "x-role-owner": "reviewer" + } + }, + "additionalProperties": false + }, + "phase": { + "type": "object", + "required": ["id", "name", "status", "tasks"], + "properties": { + "id": { + "type": "string", + "description": "Unique phase identifier (e.g., phase-1)", + "pattern": "^phase-[0-9]+$" + }, + "name": { + "type": "string", + "description": "Human-readable phase name", + "minLength": 1 + }, + "status": { + "type": "string", + "description": "Phase status", + "enum": ["pending", "in_progress", "complete", "blocked"], + "default": "pending", + "x-role-owner": "reviewer" + }, + "review_cycles": { + "type": "integer", + "description": "Number of implement->review cycles", + "minimum": 0, + "default": 0 + }, + "max_cycles": { + "type": "integer", + "description": "Maximum cycles before escalation", + "minimum": 1, + "default": 3 + }, + "escalated": { + "type": "boolean", + "description": "Whether this phase has been escalated", + "default": false + }, + "escalation_reason": { + "type": ["string", "null"], + "description": "Reason for escalation", + "default": null + }, + "tasks": { + "type": "array", + "description": "Tasks in this phase", + "items": { + "$ref": "#/$defs/task" + }, + "default": [] + }, + "review_feedback": { + "type": "array", + "description": "Feedback from reviewer", + "items": { + "$ref": "#/$defs/reviewFeedback" + }, + "default": [], + "x-role-owner": "reviewer" + } + }, + "additionalProperties": false + }, + "task": { + "type": "object", + "required": ["id", "description", "status"], + "properties": { + "id": { + "type": "string", + "description": "Unique task identifier (e.g., task-1)", + "pattern": "^task-[0-9]+$" + }, + "description": { + "type": "string", + "description": "Task description", + "minLength": 1 + }, + "status": { + "type": "string", + "description": "Task status", + "enum": ["pending", "in_progress", "complete", "incomplete", "blocked"], + "default": "pending", + "x-role-owner": "reviewer" + }, + "commit": { + "type": ["string", "null"], + "description": "Git commit SHA linked to this task", + "pattern": "^[a-f0-9]{7,40}$", + "default": null, + "x-role-owner": "implementer" + }, + "notes": { + "type": "string", + "description": "Implementation notes", + "default": "", + "x-role-owner": "implementer" + }, + "acceptance_criteria": { + "type": "string", + "description": "Acceptance criteria for this task", + "default": "" + }, + "files_affected": { + "type": "array", + "description": "Files affected by this task", + "items": { + "type": "string" + }, + "default": [] + }, + "review_cycles": { + "type": "integer", + "description": "Number of review cycles for this task", + "minimum": 0, + "default": 0 + }, + "max_cycles": { + "type": "integer", + "description": "Maximum cycles before escalation", + "minimum": 1, + "default": 3 + }, + "escalated": { + "type": "boolean", + "description": "Whether this task has been escalated", + "default": false + } + }, + "additionalProperties": false + }, + "reviewFeedback": { + "type": "object", + "required": ["timestamp", "task_id", "feedback"], + "properties": { + "timestamp": { + "type": "string", + "description": "ISO 8601 timestamp", + "format": "date-time" + }, + "task_id": { + "type": "string", + "description": "Task this feedback applies to" + }, + "feedback": { + "type": "string", + "description": "Reviewer feedback", + "minLength": 1 + }, + "status": { + "type": "string", + "description": "Status assigned by reviewer", + "enum": ["complete", "incomplete"] + } + }, + "additionalProperties": false + }, + "decision": { + "type": "object", + "required": ["id", "question", "type", "resolved"], + "properties": { + "id": { + "type": "string", + "description": "Unique decision identifier (e.g., decision-1)", + "pattern": "^decision-[0-9]+$" + }, + "question": { + "type": "string", + "description": "The decision question", + "minLength": 1 + }, + "type": { + "type": "string", + "description": "Decision type", + "enum": ["hitl", "auto"] + }, + "options": { + "type": "array", + "description": "Available options for the decision", + "items": { + "type": "object", + "required": ["id", "label"], + "properties": { + "id": { + "type": "string" + }, + "label": { + "type": "string" + }, + "description": { + "type": "string" + } + } + }, + "default": [] + }, + "resolved": { + "type": "boolean", + "description": "Whether this decision has been resolved", + "default": false, + "x-role-owner": "human" + }, + "resolution": { + "type": ["string", "null"], + "description": "The selected resolution", + "default": null, + "x-role-owner": "human" + }, + "resolved_by": { + "type": ["string", "null"], + "description": "Who resolved this decision", + "default": null + }, + "resolved_at": { + "type": ["string", "null"], + "description": "When this decision was resolved", + "format": "date-time", + "default": null + }, + "debounce_until": { + "type": ["string", "null"], + "description": "Debounce expiration timestamp", + "format": "date-time", + "default": null + } + }, + "additionalProperties": false + }, + "circuitBreaker": { + "type": "object", + "properties": { + "total_cycles": { + "type": "integer", + "description": "Total pipeline cycles", + "minimum": 0, + "default": 0 + }, + "max_total_cycles": { + "type": "integer", + "description": "Maximum total cycles before escalation", + "minimum": 1, + "default": 10 + }, + "status": { + "type": "string", + "description": "Circuit breaker status", + "enum": ["closed", "open"], + "default": "closed" + } + }, + "additionalProperties": false + }, + "auditEntry": { + "type": "object", + "required": ["timestamp", "actor", "role", "action", "field_path"], + "properties": { + "timestamp": { + "type": "string", + "description": "ISO 8601 timestamp", + "format": "date-time" + }, + "actor": { + "type": "string", + "description": "Who performed the action" + }, + "role": { + "type": "string", + "description": "Role of the actor", + "enum": ["implementer", "reviewer", "human", "system"] + }, + "action": { + "type": "string", + "description": "Action performed", + "enum": ["create", "update", "delete", "transition"] + }, + "field_path": { + "type": "string", + "description": "JSON path of the modified field" + }, + "old_value": { + "description": "Previous value (if applicable)" + }, + "new_value": { + "description": "New value" + }, + "reason": { + "type": "string", + "description": "Reason for the change" + } + }, + "additionalProperties": false + } + } +} diff --git a/gateway/Dockerfile b/gateway/Dockerfile index afe413f7d9..33506ef79a 100644 --- a/gateway/Dockerfile +++ b/gateway/Dockerfile @@ -54,9 +54,10 @@ WORKDIR /app # Copy gateway code COPY gateway/*.py ./ -# Copy shared modules (egg_logging for logging, egg_config for config parsing) +# Copy shared modules (egg_logging for logging, egg_config for config parsing, egg_contracts for contract API) COPY shared/egg_logging/ ./egg_logging/ COPY shared/egg_config/ ./egg_config/ +COPY shared/egg_contracts/ ./egg_contracts/ # Copy config module (repo_config needed by github_client for user mode) COPY config/repo_config.py /config/ @@ -75,7 +76,8 @@ COPY gateway/config_validator.py ./config_validator.py # Install dependencies # httpx: HTTP client with streaming support for Anthropic API proxy # pre-commit: Required to run pre-commit hooks when gateway executes git commit on repos -RUN pip install --no-cache-dir flask waitress pyyaml requests PyJWT cryptography httpx pre-commit +# pydantic: Required for egg_contracts module validation +RUN pip install --no-cache-dir flask waitress pyyaml requests PyJWT cryptography httpx pre-commit pydantic ENV PYTHONPATH="/app" # Expose both gateway API port and Squid proxy port diff --git a/gateway/auth.py b/gateway/auth.py new file mode 100644 index 0000000000..02bc341f87 --- /dev/null +++ b/gateway/auth.py @@ -0,0 +1,144 @@ +""" +Authentication decorators and utilities for gateway endpoints. + +This module is separate from gateway.py to avoid circular imports when +contract_api.py needs the require_session_auth decorator. +""" + +import functools +import logging +import sys +import types +from collections.abc import Callable +from pathlib import Path +from typing import Any, TypeVar + +from flask import Response, g, jsonify, request + +F = TypeVar("F", bound=Callable[..., Any]) + +# Set up logging - use egg_logging if available, otherwise standard logging +_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: + from egg_logging import get_logger + + _logger = get_logger("gateway.auth") +except ImportError: + _logger = logging.getLogger("gateway.auth") # type: ignore[assignment] + +# Import session validation lazily to avoid circular imports at module load time +# These are imported as modules so that tests can patch session_manager.validate_session_for_request +_session_manager: types.ModuleType | None = None +_rate_limiter: types.ModuleType | None = None + + +def _get_session_manager() -> types.ModuleType: + """Lazy import of session_manager module. + + Always checks sys.modules first to ensure tests can patch the module. + Checks both 'session_manager' (conftest.py loader) and 'gateway.session_manager' + (package import) naming conventions. + """ + global _session_manager + # Always prefer the module from sys.modules if available, to allow test patching + # Check both naming conventions used in different test contexts + if "session_manager" in sys.modules: + return sys.modules["session_manager"] + if "gateway.session_manager" in sys.modules: + return sys.modules["gateway.session_manager"] + if _session_manager is None: + try: + from . import session_manager as sm + + _session_manager = sm + except ImportError: + import session_manager as sm # type: ignore[no-redef, import-not-found] + + _session_manager = sm + return _session_manager + + +def _get_rate_limiter() -> types.ModuleType: + """Lazy import of rate_limiter module. + + Always checks sys.modules first to ensure tests can patch the module. + Checks both 'rate_limiter' (conftest.py loader) and 'gateway.rate_limiter' + (package import) naming conventions. + """ + global _rate_limiter + # Always prefer the module from sys.modules if available, to allow test patching + # Check both naming conventions used in different test contexts + if "rate_limiter" in sys.modules: + return sys.modules["rate_limiter"] + if "gateway.rate_limiter" in sys.modules: + return sys.modules["gateway.rate_limiter"] + if _rate_limiter is None: + try: + from . import rate_limiter as rl + + _rate_limiter = rl + except ImportError: + import rate_limiter as rl # type: ignore[no-redef, import-not-found] + + _rate_limiter = rl + return _rate_limiter + + +def make_auth_error(message: str, status_code: int = 401) -> tuple[Response, int]: + """Create an authentication error response.""" + return jsonify({"success": False, "message": message}), status_code + + +def require_session_auth(f: F) -> F: + """ + Decorator that validates session tokens in request handlers. + + - Extracts session token from Authorization header + - Validates token via session_manager + - Stores validated session and mode in Flask's g object for handler use + - Returns 401 on validation failure + + All containers must have a valid session. There is no legacy fallback. + """ + + @functools.wraps(f) + def decorated(*args: Any, **kwargs: Any) -> Any: + auth_header = request.headers.get("Authorization", "") + if not auth_header.startswith("Bearer "): + _logger.warning( + "Session auth failed - missing Authorization header", + endpoint=request.path, + source_ip=request.remote_addr, + ) + return make_auth_error("Missing or invalid Authorization header", status_code=401) + + token = auth_header[7:] # Remove "Bearer " prefix + source_ip = request.remote_addr + + # Validate session via session_manager (call via module to allow patching in tests) + session_manager = _get_session_manager() + result = session_manager.validate_session_for_request(token, source_ip) + if not result.valid: + # Record failed lookup for rate limiting + rate_limiter = _get_rate_limiter() + rate_limiter.record_failed_lookup(source_ip or "") + _logger.warning( + "Session auth failed - invalid token", + endpoint=request.path, + source_ip=source_ip, + error=result.error, + ) + return make_auth_error( + result.error or "Invalid or expired session token", status_code=401 + ) + + # Set session context from validation result + g.session = result.session + g.session_mode = result.session.mode if result.session else None + + return f(*args, **kwargs) + + return decorated # type: ignore[return-value] diff --git a/gateway/contract_api.py b/gateway/contract_api.py new file mode 100644 index 0000000000..0985b725fd --- /dev/null +++ b/gateway/contract_api.py @@ -0,0 +1,412 @@ +""" +Contract API endpoints for the gateway. + +Provides REST endpoints for contract mutations with role-based enforcement. +Role is determined from GitHub Actions workflow context, not agent environment. +""" + +import os +import sys +from pathlib import Path +from typing import Any + +from flask import Blueprint, Response, g, jsonify, request + +# Import gateway authentication - try relative import first (module mode), +# fall back to absolute import (standalone script mode in container) +try: + from .auth import require_session_auth + from .git_client import validate_repo_path +except ImportError: + from auth import require_session_auth # type: ignore[no-redef, import-not-found] + from git_client import validate_repo_path # type: ignore[no-redef, import-not-found] + +# 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, + contract_exists, + export_contract, + load_contract, + save_contract, + validate_mutation, +) + +# 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.contract") + +# Blueprint for contract endpoints +contract_bp = Blueprint("contract", __name__, url_prefix="/api/v1/contract") + + +def get_role_from_context() -> Role | None: + """ + Get the agent role from workflow context. + + Role source priority (highest to lowest): + 1. Session metadata (production path - set by launcher, most secure) + 2. X-Egg-Role header (for gateway testing only) + 3. EGG_AGENT_ROLE environment variable (development fallback) + + In GitHub Actions, the role is passed via workflow inputs and set + in the session metadata by the launcher, NOT by the agent. + This prevents privilege escalation. + + Returns: + The Role if valid, None otherwise + """ + # Production path: role comes from workflow context via session metadata + # The session is set by the launcher when starting the container + # This takes precedence to prevent header-based privilege escalation + if hasattr(g, "session") and g.session: + session_role = getattr(g.session, "agent_role", None) + if session_role: + try: + return Role(session_role.lower()) + except ValueError: + return None + + # Testing path: role can be passed in request header for gateway testing + # SECURITY: Only enabled when EGG_ENABLE_TEST_ROLE_HEADER=1 to prevent + # privilege escalation in production when sessions don't have agent_role set + if os.environ.get("EGG_ENABLE_TEST_ROLE_HEADER") == "1": + header_role = request.headers.get("X-Egg-Role") + if header_role: + try: + return Role(header_role.lower()) + except ValueError: + return None + + # Fallback: check environment (least secure, used in development only) + env_role = os.environ.get("EGG_AGENT_ROLE") + if env_role: + try: + return Role(env_role.lower()) + except ValueError: + return None + + return None + + +def get_repo_path_from_request(from_query: bool = False) -> tuple[Path | None, str | None]: + """Get the repository path from the request with validation. + + Args: + from_query: If True, look for repo_path in query parameters (for GET requests). + If False, look in JSON body (for POST requests). + + Returns: + Tuple of (path, error_message). If error_message is set, path validation failed. + """ + if from_query: + # For GET requests, use query parameters + repo_path = request.args.get("repo_path") + else: + # For POST requests, use JSON body + data = request.get_json() or {} + repo_path = data.get("repo_path") + + if repo_path: + # Validate path to prevent path traversal attacks + is_valid, error = validate_repo_path(repo_path) + if not is_valid: + return None, error + return Path(repo_path), None + + # Try to get from session + if hasattr(g, "session") and g.session: + session_repo = getattr(g.session, "repo_path", None) + if session_repo: + # Session paths are trusted (set by launcher), but validate anyway + is_valid, error = validate_repo_path(session_repo) + if not is_valid: + return None, error + return Path(session_repo), None + + return None, None + + +def make_contract_error( + message: str, + status_code: int = 400, + details: dict[str, Any] | None = None, +) -> tuple[Response, int]: + """Create a contract error response.""" + response: dict[str, Any] = {"success": False, "message": message} + if details: + response["details"] = details + return jsonify(response), status_code + + +def make_contract_success( + message: str, + data: dict[str, Any] | None = None, +) -> tuple[Response, int]: + """Create a contract success response.""" + response: dict[str, Any] = {"success": True, "message": message} + if data: + response["data"] = data + return jsonify(response), 200 + + +@contract_bp.route("/", methods=["GET"]) +@require_session_auth +def get_contract(issue_number: int) -> tuple[Response, int]: + """ + Get contract state for an issue. + + URL params: + issue_number: GitHub issue number + + Query params: + repo_path: Path to the repository (optional) + include_audit_log: Whether to include audit log (default: false) + """ + repo_path, path_error = get_repo_path_from_request(from_query=True) + if path_error: + return make_contract_error(path_error, status_code=400) + if not repo_path: + repo_path = Path.cwd() + + include_audit = request.args.get("include_audit_log", "false").lower() == "true" + + try: + contract = load_contract(issue_number, repo_path) + data = export_contract(contract, include_audit_log=include_audit) + return make_contract_success("Contract retrieved", data=data) + except ContractNotFoundError: + return make_contract_error( + f"Contract for issue #{issue_number} not found", + status_code=404, + ) + except ContractValidationError as e: + return make_contract_error( + f"Contract validation failed: {e}", + status_code=500, + ) + + +@contract_bp.route("/mutate", methods=["POST"]) +@require_session_auth +def mutate_contract() -> tuple[Response, int]: + """ + Apply a mutation to a contract. + + Request body: + { + "issue_number": 123, + "repo_path": "/path/to/repo", // optional + "field_path": "phases.0.tasks.0.commit", + "new_value": "abc1234", + "actor": "egg", // optional, defaults to "agent" + "reason": "Implementation complete" // optional + } + + The role is determined from workflow context, not the request body. + This prevents agents from escalating their privileges. + + Returns: + Success: {"success": true, "message": "...", "data": {"contract": {...}}} + Error: {"success": false, "message": "...", "details": {...}} + """ + data = request.get_json() + if not data: + return make_contract_error("Missing request body") + + # Required fields + issue_number = data.get("issue_number") + field_path = data.get("field_path") + new_value = data.get("new_value") + + if not issue_number: + return make_contract_error("Missing issue_number") + if not field_path: + return make_contract_error("Missing field_path") + if new_value is None: + return make_contract_error("Missing new_value") + + # Optional fields + if data.get("repo_path"): + is_valid, error = validate_repo_path(data["repo_path"]) + if not is_valid: + return make_contract_error(error, status_code=400) + repo_path = Path(data["repo_path"]) + else: + repo_path = Path.cwd() + actor = data.get("actor", "agent") + reason = data.get("reason") + + # Get role from context (NOT from request body) + role = get_role_from_context() + if not role: + return make_contract_error( + "Cannot determine agent role. Role must be set via workflow context.", + status_code=403, + details={"hint": "Set EGG_AGENT_ROLE via workflow inputs, not agent env vars"}, + ) + + # Load the contract + try: + contract = load_contract(issue_number, repo_path) + except ContractNotFoundError: + return make_contract_error( + f"Contract for issue #{issue_number} not found", + status_code=404, + ) + except ContractValidationError as e: + return make_contract_error( + f"Contract validation failed: {e}", + status_code=500, + ) + + # Apply the mutation + result = apply_mutation( + contract=contract, + role=role, + actor=actor, + field_path=field_path, + new_value=new_value, + reason=reason, + ) + + if not result.success: + logger.warning( + "Contract mutation rejected", + issue=issue_number, + role=role.value, + field_path=field_path, + error=result.message, + ) + return make_contract_error( + result.message, + status_code=403, + details={ + "role": role.value, + "field_path": field_path, + }, + ) + + # Save the updated contract + # Type assertion: contract is always set when success is True + assert result.contract is not None + try: + save_contract(result.contract, repo_path) + except Exception as e: + logger.error( + "Failed to save contract", + issue=issue_number, + error=str(e), + ) + return make_contract_error( + f"Failed to save contract: {e}", + status_code=500, + ) + + logger.info( + "Contract mutation applied", + issue=issue_number, + role=role.value, + actor=actor, + field_path=field_path, + ) + + return make_contract_success( + "Mutation applied successfully", + data={"contract": export_contract(result.contract, include_audit_log=False)}, + ) + + +@contract_bp.route("/validate", methods=["POST"]) +@require_session_auth +def validate_contract_mutation() -> tuple[Response, int]: + """ + Validate a mutation without applying it. + + Request body: + { + "field_path": "phases.0.tasks.0.status", + "new_value": "complete" + } + + The role is determined from workflow context. + + Returns: + {"success": true, "message": "Mutation allowed"} + or + {"success": false, "message": "...", "details": {...}} + """ + data = request.get_json() + if not data: + return make_contract_error("Missing request body") + + field_path = data.get("field_path") + new_value = data.get("new_value") + + if not field_path: + return make_contract_error("Missing field_path") + if new_value is None: + return make_contract_error("Missing new_value") + + # Get role from context + role = get_role_from_context() + if not role: + return make_contract_error( + "Cannot determine agent role", + status_code=403, + ) + + # Validate the mutation + result = validate_mutation(role, field_path, new_value) + + if result.valid: + return make_contract_success("Mutation allowed") + else: + return make_contract_error( + result.message, + status_code=403, + details={ + "role": role.value, + "field_path": result.field_path, + "required_role": result.required_role, + }, + ) + + +@contract_bp.route("/exists/", methods=["GET"]) +@require_session_auth +def check_contract_exists(issue_number: int) -> tuple[Response, int]: + """Check if a contract exists for an issue. + + Query params: + repo_path: Path to the repository (optional) + """ + repo_path, path_error = get_repo_path_from_request(from_query=True) + if path_error: + return make_contract_error(path_error, status_code=400) + if not repo_path: + repo_path = Path.cwd() + + exists = contract_exists(issue_number, repo_path) + return make_contract_success( + "Contract exists" if exists else "Contract does not exist", + data={"exists": exists}, + ) diff --git a/gateway/gateway.py b/gateway/gateway.py index 02beaceb0f..dedfb967bc 100644 --- a/gateway/gateway.py +++ b/gateway/gateway.py @@ -149,6 +149,16 @@ app = Flask(__name__) +# Register contract API blueprint +try: + from .contract_api import contract_bp + + app.register_blueprint(contract_bp) +except ImportError: + from contract_api import contract_bp # type: ignore[import-not-found, no-redef] + + app.register_blueprint(contract_bp) + @app.errorhandler(Exception) def handle_unhandled_exception(e: Exception) -> tuple[Response, int]: @@ -214,52 +224,11 @@ def translate_to_host_path(container_path: str) -> str: return container_path -def require_session_auth(f: F) -> F: - """ - Decorator that validates session tokens in request handlers. - - - Extracts session token from Authorization header - - Validates token via session_manager - - Stores validated session and mode in Flask's g object for handler use - - Returns 401 on validation failure - - All containers must have a valid session. There is no legacy fallback. - """ - - @functools.wraps(f) - def decorated(*args: Any, **kwargs: Any) -> Any: - auth_header = request.headers.get("Authorization", "") - if not auth_header.startswith("Bearer "): - logger.warning( - "Session auth failed - missing Authorization header", - endpoint=request.path, - source_ip=request.remote_addr, - ) - return make_error("Missing or invalid Authorization header", status_code=401) - - token = auth_header[7:] # Remove "Bearer " prefix - source_ip = request.remote_addr - - # Validate session via session_manager - result = validate_session_for_request(token, source_ip) - if not result.valid: - # Record failed lookup for rate limiting - record_failed_lookup(source_ip or "") - logger.warning( - "Session auth failed - invalid token", - endpoint=request.path, - source_ip=source_ip, - error=result.error, - ) - return make_error(result.error or "Invalid or expired session token", status_code=401) - - # Set session context from validation result - g.session = result.session - g.session_mode = result.session.mode if result.session else None - - return f(*args, **kwargs) - - return decorated # type: ignore[return-value] +# Import session auth decorator from auth module to avoid circular imports +try: + from .auth import require_session_auth +except ImportError: + from auth import require_session_auth # type: ignore[no-redef, import-not-found] # Launcher secret for session management and worktree operations diff --git a/gateway/session_manager.py b/gateway/session_manager.py index 3ea15452e7..35dde3566e 100644 --- a/gateway/session_manager.py +++ b/gateway/session_manager.py @@ -84,6 +84,7 @@ class Session: created_at: Session creation timestamp last_seen: Last request timestamp (for heartbeat) expires_at: Session expiry timestamp + agent_role: Role set by workflow context for contract operations """ session_token: str | None # Raw token, only in memory @@ -94,6 +95,7 @@ class Session: created_at: datetime last_seen: datetime expires_at: datetime + agent_role: str | None = None # Role set by workflow context def is_expired(self) -> bool: """Check if session has expired.""" @@ -106,7 +108,7 @@ def extend_ttl(self, hours: int = DEFAULT_SESSION_TTL_HOURS) -> None: def to_dict_for_persistence(self) -> dict[str, Any]: """Convert to dictionary for persistence (excludes raw token).""" - return { + result = { "session_token_hash": self.session_token_hash, "container_id": self.container_id, "container_ip": self.container_ip, @@ -115,6 +117,9 @@ def to_dict_for_persistence(self) -> dict[str, Any]: "last_seen": self.last_seen.isoformat(), "expires_at": self.expires_at.isoformat(), } + if self.agent_role is not None: + result["agent_role"] = self.agent_role + return result @classmethod def from_persistence(cls, data: dict[str, Any]) -> "Session": @@ -128,6 +133,7 @@ def from_persistence(cls, data: dict[str, Any]) -> "Session": created_at=datetime.fromisoformat(data["created_at"]), last_seen=datetime.fromisoformat(data["last_seen"]), expires_at=datetime.fromisoformat(data["expires_at"]), + agent_role=data.get("agent_role"), ) diff --git a/gateway/tests/conftest.py b/gateway/tests/conftest.py index ceea1bac98..0b02b4432d 100644 --- a/gateway/tests/conftest.py +++ b/gateway/tests/conftest.py @@ -78,6 +78,8 @@ def _load_module_with_replaced_imports( 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 @@ -180,12 +182,36 @@ def _load_module_with_replaced_imports( GATEWAY_DIR / "config_validator.py", ) +# auth imports from session_manager and rate_limiter +auth = _load_module_with_replaced_imports( + "auth", + GATEWAY_DIR / "auth.py", + import_replacements={ + "from .rate_limiter import": "from rate_limiter import", + "from .session_manager import": "from session_manager import", + }, +) +# Reset auth module's cached module references to ensure it uses our loaded modules +auth._session_manager = None +auth._rate_limiter = None + +# contract_api imports from auth and egg_contracts +contract_api = _load_module_with_replaced_imports( + "contract_api", + GATEWAY_DIR / "contract_api.py", + import_replacements={ + "from .auth import": "from auth import", + }, +) + # gateway imports from all gateway = _load_module_with_replaced_imports( "gateway", GATEWAY_DIR / "gateway.py", import_replacements={ "from .anthropic_credentials import": "from anthropic_credentials import", + "from .auth import": "from auth import", + "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 .policy import": "from policy import", diff --git a/gateway/tests/test_contract_api.py b/gateway/tests/test_contract_api.py new file mode 100644 index 0000000000..3e09651254 --- /dev/null +++ b/gateway/tests/test_contract_api.py @@ -0,0 +1,656 @@ +""" +Tests for Contract API endpoints. + +Tests cover: +- get_role_from_context() role resolution +- GET /api/v1/contract/ - Get contract state +- GET /api/v1/contract/exists/ - Check contract existence +- POST /api/v1/contract/validate - Validate mutation +- POST /api/v1/contract/mutate - Apply mutation +""" + +import json +import os +import sys +from unittest.mock import MagicMock, patch + +import auth +import contract_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. + + Note: We patch sys.modules entries directly to handle cases where other tests + may have loaded different module instances into sys.modules. + """ + mock_session = MagicMock() + mock_session.mode = "public" + mock_session.container_id = "test-container" + mock_session.expires_at = None + + 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 so it picks up our patched module + auth._session_manager = None + auth._rate_limiter = None + + # Also clear any package-style cached references + if "gateway.auth" in sys.modules: + sys.modules["gateway.auth"]._session_manager = None + sys.modules["gateway.auth"]._rate_limiter = None + + # Patch the module that's currently in sys.modules + 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"} + + +# --------------------------------------------------------------------------- +# get_role_from_context tests +# --------------------------------------------------------------------------- + + +class TestGetRoleFromContext: + """Tests for get_role_from_context() helper function.""" + + def test_role_from_session_agent_role(self, client, auth_headers): + """Role is resolved from g.session.agent_role when present.""" + mock_session = MagicMock() + mock_session.agent_role = "implementer" + + with client.application.test_request_context(): + from flask import g + + g.session = mock_session + role = contract_api.get_role_from_context() + + assert role is not None + assert role.value == "implementer" + + def test_role_from_x_egg_role_header_when_enabled(self, client, auth_headers): + """Role is resolved from X-Egg-Role header when EGG_ENABLE_TEST_ROLE_HEADER=1.""" + with ( + client.application.test_request_context(headers={"X-Egg-Role": "reviewer"}), + patch.dict(os.environ, {"EGG_ENABLE_TEST_ROLE_HEADER": "1"}, clear=False), + ): + from flask import g + + g.session = None + role = contract_api.get_role_from_context() + + assert role is not None + assert role.value == "reviewer" + + def test_role_from_x_egg_role_header_blocked_when_env_not_set(self, client, auth_headers): + """X-Egg-Role header is ignored when EGG_ENABLE_TEST_ROLE_HEADER is not set.""" + env = os.environ.copy() + env.pop("EGG_ENABLE_TEST_ROLE_HEADER", None) + env.pop("EGG_AGENT_ROLE", None) + + with ( + client.application.test_request_context(headers={"X-Egg-Role": "reviewer"}), + patch.dict(os.environ, env, clear=True), + ): + from flask import g + + g.session = None + role = contract_api.get_role_from_context() + + assert role is None + + def test_role_from_env_var(self, client, auth_headers): + """Role is resolved from EGG_AGENT_ROLE env var as fallback.""" + env = os.environ.copy() + env.pop("EGG_ENABLE_TEST_ROLE_HEADER", None) + env["EGG_AGENT_ROLE"] = "human" + + with ( + client.application.test_request_context(), + patch.dict(os.environ, env, clear=True), + ): + from flask import g + + g.session = None + role = contract_api.get_role_from_context() + + assert role is not None + assert role.value == "human" + + def test_invalid_role_returns_none(self, client, auth_headers): + """Invalid role string returns None.""" + env = os.environ.copy() + env.pop("EGG_ENABLE_TEST_ROLE_HEADER", None) + env["EGG_AGENT_ROLE"] = "superadmin" + + with ( + client.application.test_request_context(), + patch.dict(os.environ, env, clear=True), + ): + from flask import g + + g.session = None + role = contract_api.get_role_from_context() + + assert role is None + + def test_no_role_set_returns_none(self, client, auth_headers): + """Returns None when no role source is available.""" + env = os.environ.copy() + env.pop("EGG_ENABLE_TEST_ROLE_HEADER", None) + env.pop("EGG_AGENT_ROLE", None) + + with ( + client.application.test_request_context(), + patch.dict(os.environ, env, clear=True), + ): + from flask import g + + g.session = None + role = contract_api.get_role_from_context() + + assert role is None + + +# --------------------------------------------------------------------------- +# GET /api/v1/contract/ tests +# --------------------------------------------------------------------------- + + +class TestGetContract: + """Tests for GET /api/v1/contract/ endpoint.""" + + def test_get_contract_success(self, client, auth_headers): + """Successfully retrieves a contract.""" + mock_contract = MagicMock() + mock_exported = {"issue": 42, "phases": []} + + with ( + patch.object(contract_api, "load_contract", return_value=mock_contract) as mock_load, + patch.object( + contract_api, "export_contract", return_value=mock_exported + ) as mock_export, + ): + response = client.get( + "/api/v1/contract/42?repo_path=/home/egg/repos/test", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = json.loads(response.data) + assert data["success"] is True + assert data["message"] == "Contract retrieved" + assert data["data"]["issue"] == 42 + mock_load.assert_called_once() + mock_export.assert_called_once_with(mock_contract, include_audit_log=False) + + def test_get_contract_not_found(self, client, auth_headers): + """Returns 404 when contract is not found.""" + from pathlib import Path + + from egg_contracts import ContractNotFoundError + + with patch.object( + contract_api, + "load_contract", + side_effect=ContractNotFoundError(999, Path("/home/egg/repos/test")), + ): + response = client.get( + "/api/v1/contract/999?repo_path=/home/egg/repos/test", + headers=auth_headers, + ) + + assert response.status_code == 404 + data = json.loads(response.data) + assert data["success"] is False + assert "not found" in data["message"].lower() + + def test_get_contract_validation_error(self, client, auth_headers): + """Returns 500 when contract validation fails.""" + from egg_contracts import ContractValidationError + + with patch.object( + contract_api, + "load_contract", + side_effect=ContractValidationError(42, ["Bad schema"]), + ): + response = client.get( + "/api/v1/contract/42?repo_path=/home/egg/repos/test", + headers=auth_headers, + ) + + assert response.status_code == 500 + data = json.loads(response.data) + assert data["success"] is False + assert "validation failed" in data["message"].lower() + + def test_get_contract_with_audit_log(self, client, auth_headers): + """Passes include_audit_log=True when query param is set.""" + mock_contract = MagicMock() + mock_exported = {"issue": 42, "phases": [], "audit_log": []} + + with ( + patch.object(contract_api, "load_contract", return_value=mock_contract), + patch.object( + contract_api, "export_contract", return_value=mock_exported + ) as mock_export, + ): + response = client.get( + "/api/v1/contract/42?repo_path=/home/egg/repos/test&include_audit_log=true", + headers=auth_headers, + ) + + assert response.status_code == 200 + mock_export.assert_called_once_with(mock_contract, include_audit_log=True) + + +# --------------------------------------------------------------------------- +# GET /api/v1/contract/exists/ tests +# --------------------------------------------------------------------------- + + +class TestCheckContractExists: + """Tests for GET /api/v1/contract/exists/ endpoint.""" + + def test_contract_exists(self, client, auth_headers): + """Returns exists=True when contract exists.""" + with patch.object(contract_api, "contract_exists", return_value=True): + response = client.get( + "/api/v1/contract/exists/42?repo_path=/home/egg/repos/test", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = json.loads(response.data) + assert data["success"] is True + assert data["data"]["exists"] is True + assert "exists" in data["message"].lower() + + def test_contract_does_not_exist(self, client, auth_headers): + """Returns exists=False when contract does not exist.""" + with patch.object(contract_api, "contract_exists", return_value=False): + response = client.get( + "/api/v1/contract/exists/999?repo_path=/home/egg/repos/test", + headers=auth_headers, + ) + + assert response.status_code == 200 + data = json.loads(response.data) + assert data["success"] is True + assert data["data"]["exists"] is False + assert "does not exist" in data["message"].lower() + + +# --------------------------------------------------------------------------- +# POST /api/v1/contract/validate tests +# --------------------------------------------------------------------------- + + +class TestValidateContractMutation: + """Tests for POST /api/v1/contract/validate endpoint.""" + + def test_missing_body_returns_400(self, client, auth_headers): + """Returns 400 when request body is empty JSON.""" + response = client.post( + "/api/v1/contract/validate", + headers=auth_headers, + data=json.dumps(None), + content_type="application/json", + ) + + assert response.status_code == 400 + data = json.loads(response.data) + assert data["success"] is False + assert "Missing request body" in data["message"] + + def test_missing_field_path_returns_400(self, client, auth_headers): + """Returns 400 when field_path is missing.""" + response = client.post( + "/api/v1/contract/validate", + headers=auth_headers, + data=json.dumps({"new_value": "complete"}), + content_type="application/json", + ) + + assert response.status_code == 400 + data = json.loads(response.data) + assert data["success"] is False + assert "field_path" in data["message"] + + def test_missing_new_value_returns_400(self, client, auth_headers): + """Returns 400 when new_value is missing.""" + response = client.post( + "/api/v1/contract/validate", + headers=auth_headers, + data=json.dumps({"field_path": "phases.0.tasks.0.status"}), + content_type="application/json", + ) + + assert response.status_code == 400 + data = json.loads(response.data) + assert data["success"] is False + assert "new_value" in data["message"] + + def test_no_role_returns_403(self, client, auth_headers): + """Returns 403 when agent role cannot be determined.""" + with patch.object(contract_api, "get_role_from_context", return_value=None): + response = client.post( + "/api/v1/contract/validate", + headers=auth_headers, + data=json.dumps( + { + "field_path": "phases.0.tasks.0.status", + "new_value": "complete", + } + ), + content_type="application/json", + ) + + assert response.status_code == 403 + data = json.loads(response.data) + assert data["success"] is False + assert "role" in data["message"].lower() + + def test_valid_mutation(self, client, auth_headers): + """Returns success when mutation is valid.""" + from egg_contracts import Role, ValidationResult + + mock_result = ValidationResult(valid=True, message="Mutation allowed") + + with ( + patch.object(contract_api, "get_role_from_context", return_value=Role.IMPLEMENTER), + patch.object(contract_api, "validate_mutation", return_value=mock_result), + ): + response = client.post( + "/api/v1/contract/validate", + headers=auth_headers, + data=json.dumps( + { + "field_path": "phases.0.tasks.0.commit", + "new_value": "abc1234", + } + ), + content_type="application/json", + ) + + assert response.status_code == 200 + data = json.loads(response.data) + assert data["success"] is True + assert "allowed" in data["message"].lower() + + def test_invalid_mutation_returns_403(self, client, auth_headers): + """Returns 403 when mutation is not allowed for the role.""" + from egg_contracts import Role, ValidationResult + + mock_result = ValidationResult( + valid=False, + message="Cannot modify field 'phases.*.tasks.*.status'.", + field_path="phases.0.tasks.0.status", + required_role="reviewer", + ) + + with ( + patch.object(contract_api, "get_role_from_context", return_value=Role.IMPLEMENTER), + patch.object(contract_api, "validate_mutation", return_value=mock_result), + ): + response = client.post( + "/api/v1/contract/validate", + headers=auth_headers, + data=json.dumps( + { + "field_path": "phases.0.tasks.0.status", + "new_value": "complete", + } + ), + content_type="application/json", + ) + + assert response.status_code == 403 + data = json.loads(response.data) + assert data["success"] is False + assert "details" in data + assert data["details"]["role"] == "implementer" + assert data["details"]["required_role"] == "reviewer" + + +# --------------------------------------------------------------------------- +# POST /api/v1/contract/mutate tests +# --------------------------------------------------------------------------- + + +class TestMutateContract: + """Tests for POST /api/v1/contract/mutate endpoint.""" + + def test_missing_body_returns_400(self, client, auth_headers): + """Returns 400 when request body is empty JSON.""" + response = client.post( + "/api/v1/contract/mutate", + headers=auth_headers, + data=json.dumps(None), + content_type="application/json", + ) + + assert response.status_code == 400 + data = json.loads(response.data) + assert data["success"] is False + assert "Missing request body" in data["message"] + + def test_missing_issue_number_returns_400(self, client, auth_headers): + """Returns 400 when issue_number is missing.""" + response = client.post( + "/api/v1/contract/mutate", + headers=auth_headers, + data=json.dumps( + { + "field_path": "phases.0.tasks.0.commit", + "new_value": "abc1234", + } + ), + content_type="application/json", + ) + + assert response.status_code == 400 + data = json.loads(response.data) + assert data["success"] is False + assert "issue_number" in data["message"] + + def test_missing_field_path_returns_400(self, client, auth_headers): + """Returns 400 when field_path is missing.""" + response = client.post( + "/api/v1/contract/mutate", + headers=auth_headers, + data=json.dumps( + { + "issue_number": 42, + "new_value": "abc1234", + } + ), + content_type="application/json", + ) + + assert response.status_code == 400 + data = json.loads(response.data) + assert data["success"] is False + assert "field_path" in data["message"] + + def test_missing_new_value_returns_400(self, client, auth_headers): + """Returns 400 when new_value is missing.""" + response = client.post( + "/api/v1/contract/mutate", + headers=auth_headers, + data=json.dumps( + { + "issue_number": 42, + "field_path": "phases.0.tasks.0.commit", + } + ), + content_type="application/json", + ) + + assert response.status_code == 400 + data = json.loads(response.data) + assert data["success"] is False + assert "new_value" in data["message"] + + def test_no_role_returns_403(self, client, auth_headers): + """Returns 403 when agent role cannot be determined.""" + with patch.object(contract_api, "get_role_from_context", return_value=None): + response = client.post( + "/api/v1/contract/mutate", + headers=auth_headers, + data=json.dumps( + { + "issue_number": 42, + "field_path": "phases.0.tasks.0.commit", + "new_value": "abc1234", + "repo_path": "/home/egg/repos/test", + } + ), + content_type="application/json", + ) + + assert response.status_code == 403 + data = json.loads(response.data) + assert data["success"] is False + assert "role" in data["message"].lower() + + def test_contract_not_found_returns_404(self, client, auth_headers): + """Returns 404 when contract is not found.""" + from pathlib import Path + + from egg_contracts import ContractNotFoundError, Role + + with ( + patch.object(contract_api, "get_role_from_context", return_value=Role.IMPLEMENTER), + patch.object( + contract_api, + "load_contract", + side_effect=ContractNotFoundError(999, Path("/home/egg/repos/test")), + ), + ): + response = client.post( + "/api/v1/contract/mutate", + headers=auth_headers, + data=json.dumps( + { + "issue_number": 999, + "field_path": "phases.0.tasks.0.commit", + "new_value": "abc1234", + "repo_path": "/home/egg/repos/test", + } + ), + content_type="application/json", + ) + + assert response.status_code == 404 + data = json.loads(response.data) + assert data["success"] is False + assert "not found" in data["message"].lower() + + def test_mutation_denied_returns_403(self, client, auth_headers): + """Returns 403 when mutation is denied by role-based enforcement.""" + from egg_contracts import MutationResult, Role + + mock_contract = MagicMock() + mock_mutation_result = MutationResult( + success=False, + message="Cannot modify field 'phases.*.tasks.*.status'. " + "Role 'implementer' is not authorized.", + ) + + with ( + patch.object(contract_api, "get_role_from_context", return_value=Role.IMPLEMENTER), + patch.object(contract_api, "load_contract", return_value=mock_contract), + patch.object(contract_api, "apply_mutation", return_value=mock_mutation_result), + ): + response = client.post( + "/api/v1/contract/mutate", + headers=auth_headers, + data=json.dumps( + { + "issue_number": 42, + "field_path": "phases.0.tasks.0.status", + "new_value": "complete", + "repo_path": "/home/egg/repos/test", + } + ), + content_type="application/json", + ) + + assert response.status_code == 403 + data = json.loads(response.data) + assert data["success"] is False + assert "details" in data + assert data["details"]["role"] == "implementer" + + def test_mutate_success(self, client, auth_headers): + """Successfully applies a mutation and saves the contract.""" + from egg_contracts import MutationResult, Role + + mock_contract = MagicMock() + mock_updated_contract = MagicMock() + mock_exported = {"issue": 42, "phases": [{"tasks": [{"commit": "abc1234"}]}]} + + mock_mutation_result = MutationResult( + success=True, + message="Mutation applied successfully", + contract=mock_updated_contract, + ) + + with ( + patch.object(contract_api, "get_role_from_context", return_value=Role.IMPLEMENTER), + patch.object(contract_api, "load_contract", return_value=mock_contract), + patch.object(contract_api, "apply_mutation", return_value=mock_mutation_result), + patch.object(contract_api, "save_contract") as mock_save, + patch.object(contract_api, "export_contract", return_value=mock_exported), + ): + response = client.post( + "/api/v1/contract/mutate", + headers=auth_headers, + data=json.dumps( + { + "issue_number": 42, + "field_path": "phases.0.tasks.0.commit", + "new_value": "abc1234", + "actor": "egg", + "reason": "Implementation complete", + "repo_path": "/home/egg/repos/test", + } + ), + content_type="application/json", + ) + + assert response.status_code == 200 + data = json.loads(response.data) + assert data["success"] is True + assert "applied" in data["message"].lower() + assert data["data"]["contract"]["issue"] == 42 + mock_save.assert_called_once() diff --git a/gateway/tests/test_gateway.py b/gateway/tests/test_gateway.py index e9eea6afd9..d43eacec01 100644 --- a/gateway/tests/test_gateway.py +++ b/gateway/tests/test_gateway.py @@ -21,6 +21,8 @@ # Import the test secrets and modules (loaded by conftest.py) TEST_LAUNCHER_SECRET = os.environ.get("EGG_LAUNCHER_SECRET", "test-launcher-secret-12345") + +import session_manager from policy import PolicyResult from session_manager import SessionValidationResult @@ -47,7 +49,14 @@ def auth_headers(): Session-protected endpoints require valid session tokens. This fixture mocks session validation and private repo policy to allow tests to proceed. + + Note: We patch sys.modules entries directly to handle cases where other tests + may have loaded different module instances into sys.modules. """ + import sys + + import auth + mock_session = MagicMock() mock_session.mode = "public" mock_session.container_id = "test-container" @@ -64,8 +73,23 @@ def auth_headers(): visibility="public", ) + # Clear auth module's cached references so it picks up our patched module + auth._session_manager = None + auth._rate_limiter = None + + # Also clear any package-style cached references + if "gateway.auth" in sys.modules: + sys.modules["gateway.auth"]._session_manager = None + sys.modules["gateway.auth"]._rate_limiter = None + + # Patch the module that's currently in sys.modules, not the one we imported at module load time. + # This handles cases where other tests may have loaded different instances. + current_session_manager = sys.modules.get("session_manager", session_manager) + with ( - patch.object(gateway, "validate_session_for_request", return_value=mock_result), + 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"} @@ -840,6 +864,10 @@ class TestGhExecutePrivateMode: @pytest.fixture def private_mode_auth_headers(self): """Auth headers with private mode session.""" + import sys + + import auth + mock_session = MagicMock() mock_session.mode = "private" # Private mode session mock_session.container_id = "test-container" @@ -847,7 +875,21 @@ def private_mode_auth_headers(self): mock_result = SessionValidationResult(valid=True, session=mock_session) - with patch.object(gateway, "validate_session_for_request", return_value=mock_result): + # Clear auth module's cached references so it picks up our patched module + auth._session_manager = None + auth._rate_limiter = None + + # Also clear any package-style cached references + if "gateway.auth" in sys.modules: + sys.modules["gateway.auth"]._session_manager = None + sys.modules["gateway.auth"]._rate_limiter = None + + # Patch the module that's currently in sys.modules, not the one we imported at module load time. + current_session_manager = sys.modules.get("session_manager", session_manager) + + with patch.object( + current_session_manager, "validate_session_for_request", return_value=mock_result + ): yield {"Authorization": "Bearer test-session-token"} def test_search_blocked_in_private_mode(self, client, private_mode_auth_headers): @@ -890,7 +932,7 @@ def test_search_allowed_in_public_mode(self, client, auth_headers): def test_gh_repo_view_public_blocked_in_private_mode(self, client, private_mode_auth_headers): """gh repo view of public repo blocked in private mode (full integration).""" - with patch.object(gateway, "get_repo_visibility", return_value="public"): + with patch("private_repo_policy.get_repo_visibility", return_value="public"): response = client.post( "/api/v1/gh/execute", headers=private_mode_auth_headers, @@ -904,7 +946,7 @@ def test_gh_repo_view_public_blocked_in_private_mode(self, client, private_mode_ def test_gh_api_repos_path_blocked_in_private_mode(self, client, private_mode_auth_headers): """gh api /repos/owner/repo/... blocked for public repos in private mode.""" - with patch.object(gateway, "get_repo_visibility", return_value="public"): + with patch("private_repo_policy.get_repo_visibility", return_value="public"): response = client.post( "/api/v1/gh/execute", headers=private_mode_auth_headers, diff --git a/gateway/tests/test_gateway_integration.py b/gateway/tests/test_gateway_integration.py index 29189ff1bf..53804d94c4 100644 --- a/gateway/tests/test_gateway_integration.py +++ b/gateway/tests/test_gateway_integration.py @@ -92,7 +92,7 @@ def test_requires_auth(self, client): ) assert response.status_code == 401 - @patch("gateway.validate_session_for_request", side_effect=_mock_session_validation) + @patch("session_manager.validate_session_for_request", side_effect=_mock_session_validation) def test_missing_body(self, mock_session, client): """Missing request body should return error.""" response = client.post( @@ -102,7 +102,7 @@ def test_missing_body(self, mock_session, client): ) assert response.status_code == 400 - @patch("gateway.validate_session_for_request", side_effect=_mock_session_validation) + @patch("session_manager.validate_session_for_request", side_effect=_mock_session_validation) def test_missing_repo_path(self, mock_session, client): """Missing repo_path should return error.""" response = client.post( @@ -114,7 +114,7 @@ def test_missing_repo_path(self, mock_session, client): data = json.loads(response.data) assert "repo_path" in data["message"].lower() - @patch("gateway.validate_session_for_request", side_effect=_mock_session_validation) + @patch("session_manager.validate_session_for_request", side_effect=_mock_session_validation) def test_missing_operation(self, mock_session, client): """Missing operation should return error.""" response = client.post( @@ -126,7 +126,7 @@ def test_missing_operation(self, mock_session, client): data = json.loads(response.data) assert "operation" in data["message"].lower() - @patch("gateway.validate_session_for_request", side_effect=_mock_session_validation) + @patch("session_manager.validate_session_for_request", side_effect=_mock_session_validation) @patch("gateway.validate_repo_path", return_value=(True, "")) @patch("subprocess.run") def test_status_command_executed(self, mock_run, mock_validate, mock_session, client): @@ -151,7 +151,9 @@ def test_status_command_executed(self, mock_run, mock_validate, mock_session, cl data = json.loads(response.data) assert data["success"] is True - @patch("gateway.validate_session_for_request", side_effect=_mock_session_validation_public) + @patch( + "session_manager.validate_session_for_request", side_effect=_mock_session_validation_public + ) @patch("gateway.validate_repo_path", return_value=(True, "")) @patch("subprocess.run") def test_status_command_with_public_mode(self, mock_run, mock_validate, mock_session, client): @@ -175,7 +177,7 @@ def test_status_command_with_public_mode(self, mock_run, mock_validate, mock_ses data = json.loads(response.data) assert data["success"] is True - @patch("gateway.validate_session_for_request", side_effect=_mock_session_validation) + @patch("session_manager.validate_session_for_request", side_effect=_mock_session_validation) def test_disallowed_operation_rejected(self, mock_session, client): """Operations not in allowlist should be rejected.""" response = client.post( @@ -191,7 +193,7 @@ def test_disallowed_operation_rejected(self, mock_session, client): data = json.loads(response.data) assert "not allowed" in data["message"].lower() - @patch("gateway.validate_session_for_request", side_effect=_mock_session_validation) + @patch("session_manager.validate_session_for_request", side_effect=_mock_session_validation) def test_network_ops_redirected(self, mock_session, client): """Network operations should be redirected to dedicated endpoints.""" for op in ["push", "fetch", "ls-remote"]: @@ -305,7 +307,7 @@ def test_returns_worktrees(self, mock_manager, client, launcher_auth_headers): class TestPathValidation: """Tests for path validation in endpoints.""" - @patch("gateway.validate_session_for_request", side_effect=_mock_session_validation) + @patch("session_manager.validate_session_for_request", side_effect=_mock_session_validation) def test_path_traversal_blocked(self, mock_session, client): """Path traversal attempts should be blocked.""" response = client.post( @@ -321,7 +323,7 @@ def test_path_traversal_blocked(self, mock_session, client): data = json.loads(response.data) assert "allowed directories" in data["message"].lower() - @patch("gateway.validate_session_for_request", side_effect=_mock_session_validation) + @patch("session_manager.validate_session_for_request", side_effect=_mock_session_validation) def test_repos_parent_directory_rejected(self, mock_session, client): """Git operations from repos parent directory should fail with clear error. diff --git a/pyproject.toml b/pyproject.toml index 91215e977d..5940edba79 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [ "PyJWT>=2.8.0,<3.0.0", "cryptography>=41.0.0,<44.0.0", "httpx>=0.27.0,<1.0.0", + "pydantic>=2.0.0,<3.0.0", ] [project.optional-dependencies] @@ -87,6 +88,11 @@ module = ["egg_logging", "egg_logging.*"] ignore_missing_imports = true follow_untyped_imports = true +[[tool.mypy.overrides]] +module = ["egg_contracts", "egg_contracts.*"] +ignore_missing_imports = true +follow_untyped_imports = true + [[tool.mypy.overrides]] module = [ "repo_config", @@ -122,7 +128,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["gateway", "shared/egg_config", "shared/egg_container", "shared/egg_logging", "shared/egg_git", "sandbox/egg_lib"] +packages = ["gateway", "shared/egg_config", "shared/egg_container", "shared/egg_contracts", "shared/egg_logging", "shared/egg_git", "sandbox/egg_lib"] [tool.hatch.build.targets.sdist] include = [ diff --git a/shared/egg_contracts/__init__.py b/shared/egg_contracts/__init__.py new file mode 100644 index 0000000000..ada7006f21 --- /dev/null +++ b/shared/egg_contracts/__init__.py @@ -0,0 +1,142 @@ +""" +Egg Contracts Library. + +This module provides the core functionality for managing SDLC contracts +that enforce structurally-verified agent checkpoints and verification gates. + +Key concepts: +- Contract: JSON document tracking issue progress through SDLC phases +- Roles: Implementer, Reviewer, Human - each with specific permissions +- Mutations: All contract changes are validated against role permissions +- Audit Log: All modifications are tracked for accountability + +Usage: + from egg_contracts import Contract, Role, load_contract, save_contract + from egg_contracts import validate_mutation, apply_mutation + + # Load a contract + contract = load_contract(issue_number=133, repo_root=Path("/path/to/repo")) + + # Validate a mutation + result = validate_mutation( + role=Role.IMPLEMENTER, + field_path="phases.0.tasks.0.commit", + new_value="abc1234", + ) + + # Apply a mutation + result = apply_mutation( + contract=contract, + role=Role.IMPLEMENTER, + actor="egg", + field_path="phases.0.tasks.0.commit", + new_value="abc1234", + ) +""" + +from .audit import ( + create_audit_entry, + create_transition_entry, + create_update_entry, + format_audit_log, +) +from .loader import ( + ContractNotFoundError, + ContractValidationError, + contract_exists, + create_contract, + delete_contract, + export_contract, + get_contract_path, + list_contracts, + load_contract, + load_contract_from_branch, + save_contract, +) +from .models import ( + AcceptanceCriterion, + AuditAction, + AuditEntry, + AuditRole, + CircuitBreaker, + CircuitBreakerStatus, + Contract, + Decision, + DecisionOption, + DecisionType, + IssueInfo, + Phase, + PhaseStatus, + PipelinePhase, + ReviewFeedback, + Task, + TaskStatus, +) +from .roles import ( + FIELD_OWNERSHIP, + Role, + can_modify, + get_field_owner, + get_role_permissions, + normalize_path, +) +from .validator import ( + MutationResult, + ValidationResult, + apply_mutation, + validate_mutation, + validate_phase_mutation, + validate_task_mutation, +) + +__all__ = [ + # Models + "AcceptanceCriterion", + "AuditAction", + "AuditEntry", + "AuditRole", + "CircuitBreaker", + "CircuitBreakerStatus", + "Contract", + "ContractNotFoundError", + "ContractValidationError", + "Decision", + "DecisionOption", + "DecisionType", + # Roles + "FIELD_OWNERSHIP", + "IssueInfo", + "MutationResult", + "Phase", + "PhaseStatus", + "PipelinePhase", + "ReviewFeedback", + "Role", + "Task", + "TaskStatus", + "ValidationResult", + "apply_mutation", + "can_modify", + "contract_exists", + # Loader + "create_contract", + "create_audit_entry", + "create_transition_entry", + # Audit + "create_update_entry", + "delete_contract", + "export_contract", + "format_audit_log", + "get_contract_path", + "get_field_owner", + "get_role_permissions", + "list_contracts", + "load_contract", + "load_contract_from_branch", + "normalize_path", + "save_contract", + "validate_mutation", + "validate_phase_mutation", + # Validator + "validate_task_mutation", +] diff --git a/shared/egg_contracts/audit.py b/shared/egg_contracts/audit.py new file mode 100644 index 0000000000..59e0dec7ee --- /dev/null +++ b/shared/egg_contracts/audit.py @@ -0,0 +1,140 @@ +""" +Audit log management for contract modifications. + +This module provides functions for creating and managing audit log entries +that track all modifications to contracts. +""" + +from datetime import UTC, datetime +from typing import Any + +from .models import AuditAction, AuditEntry, AuditRole + + +def create_audit_entry( + actor: str, + role: AuditRole, + action: AuditAction, + field_path: str, + old_value: Any = None, + new_value: Any = None, + reason: str | None = None, +) -> AuditEntry: + """ + Create a new audit log entry. + + Args: + actor: Identifier of who performed the action + role: Role of the actor + action: Type of action performed + field_path: JSON path of the modified field + old_value: Previous value (if applicable) + new_value: New value + reason: Optional reason for the change + + Returns: + A new AuditEntry + """ + return AuditEntry( + timestamp=datetime.now(UTC), + actor=actor, + role=role, + action=action, + field_path=field_path, + old_value=old_value, + new_value=new_value, + reason=reason, + ) + + +def create_update_entry( + actor: str, + role: AuditRole, + field_path: str, + old_value: Any, + new_value: Any, + reason: str | None = None, +) -> AuditEntry: + """ + Create an audit entry for an update operation. + + Args: + actor: Identifier of who performed the action + role: Role of the actor + field_path: JSON path of the modified field + old_value: Previous value + new_value: New value + reason: Optional reason for the change + + Returns: + A new AuditEntry for an update action + """ + return create_audit_entry( + actor=actor, + role=role, + action=AuditAction.UPDATE, + field_path=field_path, + old_value=old_value, + new_value=new_value, + reason=reason, + ) + + +def create_transition_entry( + actor: str, + role: AuditRole, + from_phase: str, + to_phase: str, + reason: str | None = None, +) -> AuditEntry: + """ + Create an audit entry for a phase transition. + + Args: + actor: Identifier of who performed the action + role: Role of the actor + from_phase: Previous phase + to_phase: New phase + reason: Optional reason for the transition + + Returns: + A new AuditEntry for a transition action + """ + return create_audit_entry( + actor=actor, + role=role, + action=AuditAction.TRANSITION, + field_path="current_phase", + old_value=from_phase, + new_value=to_phase, + reason=reason, + ) + + +def format_audit_log(entries: list[AuditEntry], limit: int | None = None) -> str: + """ + Format audit log entries as human-readable text. + + Args: + entries: List of audit entries to format + limit: Optional limit on number of entries to show + + Returns: + Formatted string representation of the audit log + """ + if limit: + entries = entries[-limit:] + + lines = ["Audit Log:"] + for entry in entries: + timestamp = entry.timestamp.strftime("%Y-%m-%d %H:%M:%S") + line = f" [{timestamp}] {entry.role.value}:{entry.actor} {entry.action.value} {entry.field_path}" + if entry.old_value is not None and entry.new_value is not None: + line += f" ({entry.old_value} -> {entry.new_value})" + elif entry.new_value is not None: + line += f" = {entry.new_value}" + if entry.reason: + line += f" - {entry.reason}" + lines.append(line) + + return "\n".join(lines) diff --git a/shared/egg_contracts/loader.py b/shared/egg_contracts/loader.py new file mode 100644 index 0000000000..e9a1a7cf80 --- /dev/null +++ b/shared/egg_contracts/loader.py @@ -0,0 +1,289 @@ +""" +Contract loader and persistence. + +This module handles loading, saving, and initializing contracts from +the .egg-state/contracts/ directory. + +Note: The .egg/ directory (containing schemas/) holds the contract library — +shared schema definitions committed to main. The .egg-state/ directory holds +contract instances — per-issue runtime state committed only to feature branches. +""" + +import json +import os +import tempfile +from pathlib import Path +from typing import Any + +from .models import Contract, IssueInfo, PipelinePhase + +# Default contracts directory relative to repo root +# Uses .egg-state/ to distinguish contract instances (per-branch runtime state) +# from .egg/schemas/ which holds the contract schema library +DEFAULT_CONTRACTS_DIR = ".egg-state/contracts" + + +class ContractNotFoundError(Exception): + """Raised when a contract doesn't exist.""" + + def __init__(self, issue_number: int, path: Path) -> None: + self.issue_number = issue_number + self.path = path + super().__init__(f"Contract for issue #{issue_number} not found at {path}") + + +class ContractValidationError(Exception): + """Raised when a contract fails validation.""" + + def __init__(self, issue_number: int, errors: list[str]) -> None: + self.issue_number = issue_number + self.errors = errors + super().__init__(f"Contract for issue #{issue_number} is invalid: {'; '.join(errors)}") + + +def get_contract_path(issue_number: int, repo_root: Path | None = None) -> Path: + """ + Get the path to a contract file. + + Args: + issue_number: The GitHub issue number + repo_root: Optional repository root path. Defaults to current directory. + + Returns: + Path to the contract JSON file + """ + if repo_root is None: + repo_root = Path.cwd() + return repo_root / DEFAULT_CONTRACTS_DIR / f"{issue_number}.json" + + +def load_contract(issue_number: int, repo_root: Path | None = None) -> Contract: + """ + Load a contract from disk. + + Args: + issue_number: The GitHub issue number + repo_root: Optional repository root path + + Returns: + The loaded Contract + + Raises: + ContractNotFoundError: If the contract doesn't exist + ContractValidationError: If the contract is invalid + """ + path = get_contract_path(issue_number, repo_root) + + if not path.exists(): + raise ContractNotFoundError(issue_number, path) + + try: + with open(path) as f: + data = json.load(f) + return Contract.model_validate(data) + except json.JSONDecodeError as e: + raise ContractValidationError(issue_number, [f"Invalid JSON: {e}"]) from e + except Exception as e: + raise ContractValidationError(issue_number, [str(e)]) from e + + +def save_contract(contract: Contract, repo_root: Path | None = None) -> Path: + """ + Save a contract to disk atomically. + + Uses a write-to-temp-then-rename pattern to prevent corruption if the + process crashes mid-write. + + Args: + contract: The contract to save + repo_root: Optional repository root path + + Returns: + Path where the contract was saved + """ + path = get_contract_path(contract.issue.number, repo_root) + + # Ensure directory exists + path.parent.mkdir(parents=True, exist_ok=True) + + # Write to temp file first, then atomically rename + # Using dir=path.parent ensures the temp file is on the same filesystem + # so os.rename() is atomic + fd, temp_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp") + try: + with os.fdopen(fd, "w") as f: + json.dump(contract.model_dump(mode="json"), f, indent=2, default=str) + f.write("\n") # Trailing newline + # Atomic rename + os.rename(temp_path, path) + except Exception: + # Clean up temp file on failure + try: + os.unlink(temp_path) + except OSError: + pass + raise + + return path + + +def contract_exists(issue_number: int, repo_root: Path | None = None) -> bool: + """ + Check if a contract exists. + + Args: + issue_number: The GitHub issue number + repo_root: Optional repository root path + + Returns: + True if the contract exists + """ + path = get_contract_path(issue_number, repo_root) + return path.exists() + + +def create_contract( + issue_number: int, + title: str, + url: str, + repo_root: Path | None = None, + initial_phase: PipelinePhase = PipelinePhase.REFINE, +) -> Contract: + """ + Create a new contract for an issue. + + Args: + issue_number: The GitHub issue number + title: Issue title + url: Issue URL + repo_root: Optional repository root path + initial_phase: Initial pipeline phase + + Returns: + The newly created Contract + """ + contract = Contract( + issue=IssueInfo( + number=issue_number, + title=title, + url=url, + ), + current_phase=initial_phase, + ) + + save_contract(contract, repo_root) + return contract + + +def load_contract_from_branch( + issue_number: int, + repo_path: Path, + branch: str | None = None, +) -> Contract: + """ + Load a contract from a specific git branch. + + This is useful when the gateway needs to load a contract from the + agent's working branch rather than the current checkout. + + Args: + issue_number: The GitHub issue number + repo_path: Path to the repository + branch: Optional branch name. If None, uses current checkout. + + Returns: + The loaded Contract + + Note: + If branch is specified, this function shells out to git to read + the file contents from that branch. + """ + if branch is None: + return load_contract(issue_number, repo_path) + + # Read file from specific branch using git show + import subprocess + + contract_rel_path = f"{DEFAULT_CONTRACTS_DIR}/{issue_number}.json" + + try: + result = subprocess.run( + ["git", "show", f"{branch}:{contract_rel_path}"], + cwd=repo_path, + capture_output=True, + text=True, + check=True, + ) + data = json.loads(result.stdout) + return Contract.model_validate(data) + except subprocess.CalledProcessError as e: + raise ContractNotFoundError(issue_number, repo_path / contract_rel_path) from e + except json.JSONDecodeError as e: + raise ContractValidationError(issue_number, [f"Invalid JSON: {e}"]) from e + + +def list_contracts(repo_root: Path | None = None) -> list[int]: + """ + List all contract issue numbers in the repository. + + Args: + repo_root: Optional repository root path + + Returns: + List of issue numbers with contracts + """ + if repo_root is None: + repo_root = Path.cwd() + + contracts_dir = repo_root / DEFAULT_CONTRACTS_DIR + if not contracts_dir.exists(): + return [] + + issue_numbers = [] + for path in contracts_dir.glob("*.json"): + try: + issue_num = int(path.stem) + issue_numbers.append(issue_num) + except ValueError: + continue + + return sorted(issue_numbers) + + +def delete_contract(issue_number: int, repo_root: Path | None = None) -> bool: + """ + Delete a contract from disk. + + Args: + issue_number: The GitHub issue number + repo_root: Optional repository root path + + Returns: + True if the contract was deleted, False if it didn't exist + """ + path = get_contract_path(issue_number, repo_root) + + if path.exists(): + path.unlink() + return True + return False + + +def export_contract( + contract: Contract, + include_audit_log: bool = True, +) -> dict[str, Any]: + """ + Export a contract as a dictionary for API responses. + + Args: + contract: The contract to export + include_audit_log: Whether to include the audit log + + Returns: + Dictionary representation of the contract + """ + data = contract.model_dump(mode="json") + if not include_audit_log: + data.pop("audit_log", None) + return data diff --git a/shared/egg_contracts/models.py b/shared/egg_contracts/models.py new file mode 100644 index 0000000000..ffd66b6018 --- /dev/null +++ b/shared/egg_contracts/models.py @@ -0,0 +1,228 @@ +""" +Pydantic models for SDLC contract schema. + +These models match the JSON schema defined in .egg/schemas/contract.schema.json +and provide validation and type safety for contract operations. +""" + +from datetime import datetime +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, Field, field_validator + + +class TaskStatus(StrEnum): + """Status values for tasks.""" + + PENDING = "pending" + IN_PROGRESS = "in_progress" + COMPLETE = "complete" + INCOMPLETE = "incomplete" + BLOCKED = "blocked" + + +class PhaseStatus(StrEnum): + """Status values for phases.""" + + PENDING = "pending" + IN_PROGRESS = "in_progress" + COMPLETE = "complete" + BLOCKED = "blocked" + + +class PipelinePhase(StrEnum): + """Current pipeline phase.""" + + REFINE = "refine" + PLAN = "plan" + IMPLEMENT = "implement" + PR = "pr" + + +class DecisionType(StrEnum): + """Types of decisions.""" + + HITL = "hitl" + AUTO = "auto" + + +class CircuitBreakerStatus(StrEnum): + """Circuit breaker status values.""" + + CLOSED = "closed" + OPEN = "open" + + +class AuditAction(StrEnum): + """Types of audit actions.""" + + CREATE = "create" + UPDATE = "update" + DELETE = "delete" + TRANSITION = "transition" + + +class AuditRole(StrEnum): + """Roles for audit entries.""" + + IMPLEMENTER = "implementer" + REVIEWER = "reviewer" + HUMAN = "human" + SYSTEM = "system" + + +class IssueInfo(BaseModel): + """Issue metadata.""" + + number: int = Field(..., ge=1, description="GitHub issue number") + title: str = Field(..., min_length=1, description="Issue title") + url: str = Field(..., description="Issue URL") + + +class AcceptanceCriterion(BaseModel): + """Top-level acceptance criterion.""" + + id: str = Field(..., pattern=r"^ac-[0-9]+$", description="Unique identifier") + description: str = Field(..., min_length=1, description="Human-readable description") + verified: bool = Field(default=False, description="Whether verified by reviewer") + + +class ReviewFeedback(BaseModel): + """Feedback from reviewer on a task.""" + + timestamp: datetime = Field(..., description="When feedback was given") + task_id: str = Field(..., description="Task this feedback applies to") + feedback: str = Field(..., min_length=1, description="Reviewer feedback") + status: TaskStatus | None = Field(default=None, description="Status assigned by reviewer") + + +class Task(BaseModel): + """A task within a phase.""" + + id: str = Field(..., pattern=r"^task-[0-9]+$", description="Unique task identifier") + description: str = Field(..., min_length=1, description="Task description") + status: TaskStatus = Field(default=TaskStatus.PENDING, description="Task status") + commit: str | None = Field( + default=None, + pattern=r"^[a-f0-9]{7,40}$", + description="Git commit SHA", + ) + notes: str = Field(default="", description="Implementation notes") + acceptance_criteria: str = Field(default="", description="Acceptance criteria") + files_affected: list[str] = Field(default_factory=list, description="Files affected") + review_cycles: int = Field(default=0, ge=0, description="Number of review cycles") + max_cycles: int = Field(default=3, ge=1, description="Max cycles before escalation") + escalated: bool = Field(default=False, description="Whether escalated") + + @field_validator("commit", mode="before") + @classmethod + def validate_commit(cls, v: Any) -> str | None: + if v is None or v == "": + return None + return str(v) + + +class Phase(BaseModel): + """An implementation phase containing tasks.""" + + id: str = Field(..., pattern=r"^phase-[0-9]+$", description="Unique phase identifier") + name: str = Field(..., min_length=1, description="Human-readable phase name") + status: PhaseStatus = Field(default=PhaseStatus.PENDING, description="Phase status") + review_cycles: int = Field(default=0, ge=0, description="Number of review cycles") + max_cycles: int = Field(default=3, ge=1, description="Max cycles before escalation") + escalated: bool = Field(default=False, description="Whether escalated") + escalation_reason: str | None = Field(default=None, description="Reason for escalation") + tasks: list[Task] = Field(default_factory=list, description="Tasks in this phase") + review_feedback: list[ReviewFeedback] = Field( + default_factory=list, description="Feedback from reviewer" + ) + + +class DecisionOption(BaseModel): + """An option for a decision.""" + + id: str = Field(..., description="Option identifier") + label: str = Field(..., description="Option label") + description: str | None = Field(default=None, description="Option description") + + +class Decision(BaseModel): + """A HITL decision point.""" + + id: str = Field(..., pattern=r"^decision-[0-9]+$", description="Unique decision identifier") + question: str = Field(..., min_length=1, description="The decision question") + type: DecisionType = Field(..., description="Decision type") + options: list[DecisionOption] = Field(default_factory=list, description="Available options") + resolved: bool = Field(default=False, description="Whether resolved") + resolution: str | None = Field(default=None, description="Selected resolution") + resolved_by: str | None = Field(default=None, description="Who resolved") + resolved_at: datetime | None = Field(default=None, description="When resolved") + debounce_until: datetime | None = Field(default=None, description="Debounce expiration") + + +class CircuitBreaker(BaseModel): + """Circuit breaker state for pipeline.""" + + total_cycles: int = Field(default=0, ge=0, description="Total pipeline cycles") + max_total_cycles: int = Field(default=10, ge=1, description="Max cycles before escalation") + status: CircuitBreakerStatus = Field( + default=CircuitBreakerStatus.CLOSED, description="Circuit breaker status" + ) + + +class AuditEntry(BaseModel): + """Audit log entry for contract modifications.""" + + timestamp: datetime = Field(..., description="When the action occurred") + actor: str = Field(..., description="Who performed the action") + role: AuditRole = Field(..., description="Role of the actor") + action: AuditAction = Field(..., description="Action performed") + field_path: str = Field(..., description="JSON path of modified field") + old_value: Any = Field(default=None, description="Previous value") + new_value: Any = Field(default=None, description="New value") + reason: str | None = Field(default=None, description="Reason for change") + + +class Contract(BaseModel): + """The complete SDLC contract.""" + + schemaVersion: str = Field( # noqa: N815 + default="1.0", pattern=r"^[0-9]+\.[0-9]+$", description="Schema version" + ) + issue: IssueInfo = Field(..., description="Issue metadata") + current_phase: PipelinePhase = Field( + default=PipelinePhase.REFINE, description="Current pipeline phase" + ) + acceptance_criteria: list[AcceptanceCriterion] = Field( + default_factory=list, description="Top-level acceptance criteria" + ) + phases: list[Phase] = Field(default_factory=list, description="Implementation phases") + decisions: list[Decision] = Field(default_factory=list, description="HITL decisions") + circuit_breaker: CircuitBreaker = Field( + default_factory=CircuitBreaker, description="Circuit breaker state" + ) + audit_log: list[AuditEntry] = Field(default_factory=list, description="Audit trail") + + def get_task(self, phase_id: str, task_id: str) -> Task | None: + """Get a specific task by phase and task ID.""" + for phase in self.phases: + if phase.id == phase_id: + for task in phase.tasks: + if task.id == task_id: + return task + return None + + def get_phase(self, phase_id: str) -> Phase | None: + """Get a specific phase by ID.""" + for phase in self.phases: + if phase.id == phase_id: + return phase + return None + + def get_decision(self, decision_id: str) -> Decision | None: + """Get a specific decision by ID.""" + for decision in self.decisions: + if decision.id == decision_id: + return decision + return None diff --git a/shared/egg_contracts/pyproject.toml b/shared/egg_contracts/pyproject.toml new file mode 100644 index 0000000000..7dbb1c4b57 --- /dev/null +++ b/shared/egg_contracts/pyproject.toml @@ -0,0 +1,12 @@ +[project] +name = "egg-contracts" +version = "0.1.0" +description = "Contract library for structurally enforced agent checkpoints" +requires-python = ">=3.11" +dependencies = [ + "pydantic>=2.0.0", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" diff --git a/shared/egg_contracts/roles.py b/shared/egg_contracts/roles.py new file mode 100644 index 0000000000..46db0f8de2 --- /dev/null +++ b/shared/egg_contracts/roles.py @@ -0,0 +1,153 @@ +""" +Role definitions and field ownership mapping for contract mutations. + +This module defines the roles that can interact with contracts and +maps which fields each role is authorized to modify. +""" + +from enum import StrEnum + + +class Role(StrEnum): + """Roles that can interact with contracts. + + Uses StrEnum for consistency with AuditRole in models.py, making + conversions and comparisons simpler. + """ + + IMPLEMENTER = "implementer" + REVIEWER = "reviewer" + HUMAN = "human" + SYSTEM = "system" + + +# Field ownership mapping: maps JSON paths to the role that owns them +# Paths use dot notation (e.g., "phases.*.tasks.*.status") +# Wildcard (*) matches any array index +FIELD_OWNERSHIP: dict[str, Role] = { + # Task fields owned by implementer + "phases.*.tasks.*.commit": Role.IMPLEMENTER, + "phases.*.tasks.*.notes": Role.IMPLEMENTER, + "phases.*.tasks.*.files_affected": Role.IMPLEMENTER, + "phases.*.tasks.*.files_affected.*": Role.IMPLEMENTER, + # Task fields owned by reviewer + "phases.*.tasks.*.status": Role.REVIEWER, + # Phase fields owned by reviewer + "phases.*.status": Role.REVIEWER, + "phases.*.review_feedback": Role.REVIEWER, + "phases.*.review_feedback.*": Role.REVIEWER, + # Acceptance criteria owned by reviewer + "acceptance_criteria.*.verified": Role.REVIEWER, + # Decisions owned by human + "decisions.*.resolved": Role.HUMAN, + "decisions.*.resolution": Role.HUMAN, + "decisions.*.resolved_by": Role.HUMAN, + "decisions.*.resolved_at": Role.HUMAN, +} + +# Fields that any role can read but only the owner can write +# All other fields default to SYSTEM ownership (created at contract init) +DEFAULT_OWNER = Role.SYSTEM + + +def normalize_path(path: str) -> str: + """ + Normalize a JSON path by replacing numeric indices with wildcards. + + Args: + path: JSON path like "phases.0.tasks.1.status" + + Returns: + Normalized path like "phases.*.tasks.*.status" + """ + parts = path.split(".") + normalized = [] + for part in parts: + if part.isdigit(): + normalized.append("*") + else: + normalized.append(part) + return ".".join(normalized) + + +def get_field_owner(path: str) -> Role: + """ + Get the role that owns a specific field path. + + Args: + path: JSON path to the field (e.g., "phases.0.tasks.1.status") + + Returns: + The role that owns this field + """ + normalized = normalize_path(path) + + # Try exact match first + if normalized in FIELD_OWNERSHIP: + return FIELD_OWNERSHIP[normalized] + + # Try prefix match for nested paths (e.g., review_feedback.* matches review_feedback.0.feedback) + for pattern, owner in FIELD_OWNERSHIP.items(): + if pattern.endswith(".*"): + prefix = pattern[:-2] + if normalized.startswith(prefix + "."): + return owner + + return DEFAULT_OWNER + + +def can_modify(role: Role, path: str) -> bool: + """ + Check if a role can modify a specific field. + + Args: + role: The role attempting the modification + path: JSON path to the field + + Returns: + True if the role can modify the field + """ + # Human can modify everything + if role == Role.HUMAN: + return True + + # System can create/initialize contracts but not modify owned fields + if role == Role.SYSTEM: + owner = get_field_owner(path) + # System can only modify fields it owns + return owner == Role.SYSTEM + + # Check if the role owns this field + owner = get_field_owner(path) + return owner == role + + +def get_role_permissions(role: Role) -> dict[str, list[str]]: + """ + Get a summary of what fields a role can modify. + + Args: + role: The role to get permissions for + + Returns: + Dictionary with 'can_modify' and 'cannot_modify' lists + """ + if role == Role.HUMAN: + return { + "can_modify": ["*"], + "cannot_modify": [], + } + + can_mod = [] + cannot_mod = [] + + for path, owner in FIELD_OWNERSHIP.items(): + if owner == role: + can_mod.append(path) + else: + cannot_mod.append(path) + + return { + "can_modify": can_mod, + "cannot_modify": cannot_mod, + } diff --git a/shared/egg_contracts/validator.py b/shared/egg_contracts/validator.py new file mode 100644 index 0000000000..9416597868 --- /dev/null +++ b/shared/egg_contracts/validator.py @@ -0,0 +1,241 @@ +""" +Contract mutation validator. + +This module validates mutations against role permissions, ensuring that +only authorized roles can modify specific fields. +""" + +from dataclasses import dataclass +from typing import Any + +from .audit import create_update_entry +from .models import AuditEntry, AuditRole, Contract +from .roles import Role, can_modify, get_field_owner, normalize_path + + +@dataclass +class ValidationResult: + """Result of a mutation validation.""" + + valid: bool + message: str + field_path: str | None = None + required_role: str | None = None + + +@dataclass +class MutationResult: + """Result of applying a mutation.""" + + success: bool + message: str + contract: Contract | None = None + audit_entry: AuditEntry | None = None + + +def validate_mutation( + role: Role, + field_path: str, + new_value: Any, + contract: Contract | None = None, +) -> ValidationResult: + """ + Validate whether a role can make a specific mutation. + + Args: + role: The role attempting the mutation + field_path: JSON path to the field being modified + new_value: The new value being set + contract: Optional contract for additional context validation + + Returns: + ValidationResult indicating if the mutation is allowed + """ + # Check role authorization + if not can_modify(role, field_path): + owner = get_field_owner(field_path) + normalized = normalize_path(field_path) + return ValidationResult( + valid=False, + message=f"Cannot modify field '{normalized}'. " + f"Role '{role.value}' is not authorized to modify this field. " + f"This field can only be modified by role '{owner.value}'.", + field_path=field_path, + required_role=owner.value, + ) + + return ValidationResult(valid=True, message="Mutation allowed") + + +def apply_mutation( + contract: Contract, + role: Role, + actor: str, + field_path: str, + new_value: Any, + reason: str | None = None, +) -> MutationResult: + """ + Apply a mutation to the contract after validation. + + Args: + contract: The contract to mutate + role: The role attempting the mutation + actor: Identifier of who is making the change + field_path: JSON path to the field being modified + new_value: The new value to set + reason: Optional reason for the change + + Returns: + MutationResult with the updated contract or error + """ + # Validate the mutation + validation = validate_mutation(role, field_path, new_value, contract) + if not validation.valid: + return MutationResult( + success=False, + message=validation.message, + ) + + # Get the old value and apply the mutation + try: + old_value = _get_value(contract, field_path) + _set_value(contract, field_path, new_value) + except (KeyError, IndexError, AttributeError) as e: + return MutationResult( + success=False, + message=f"Failed to apply mutation: {e}", + ) + + # Create audit entry + audit_role = AuditRole(role.value) + audit_entry = create_update_entry( + actor=actor, + role=audit_role, + field_path=field_path, + old_value=old_value, + new_value=new_value, + reason=reason, + ) + contract.audit_log.append(audit_entry) + + return MutationResult( + success=True, + message="Mutation applied successfully", + contract=contract, + audit_entry=audit_entry, + ) + + +def _get_value(obj: Any, path: str) -> Any: + """ + Get a value from an object using a dot-notation path. + + Args: + obj: The object to traverse + path: Dot-notation path (e.g., "phases.0.tasks.1.status") + + Returns: + The value at the path + + Raises: + KeyError: If path doesn't exist + IndexError: If array index is out of bounds + """ + parts = path.split(".") + current = obj + + for part in parts: + if isinstance(current, list): + idx = int(part) + current = current[idx] + elif hasattr(current, part): + current = getattr(current, part) + elif isinstance(current, dict): + current = current[part] + else: + raise KeyError(f"Cannot access '{part}' on {type(current)}") + + return current + + +def _set_value(obj: Any, path: str, value: Any) -> None: + """ + Set a value on an object using a dot-notation path. + + Args: + obj: The object to modify + path: Dot-notation path (e.g., "phases.0.tasks.1.status") + value: The value to set + + Raises: + KeyError: If path doesn't exist + IndexError: If array index is out of bounds + """ + parts = path.split(".") + current = obj + + # Navigate to parent of the target + for part in parts[:-1]: + if isinstance(current, list): + idx = int(part) + current = current[idx] + elif hasattr(current, part): + current = getattr(current, part) + elif isinstance(current, dict): + current = current[part] + else: + raise KeyError(f"Cannot access '{part}' on {type(current)}") + + # Set the final value + final_part = parts[-1] + if isinstance(current, list): + idx = int(final_part) + current[idx] = value + elif hasattr(current, final_part): + setattr(current, final_part, value) + elif isinstance(current, dict): + current[final_part] = value + else: + raise KeyError(f"Cannot set '{final_part}' on {type(current)}") + + +def validate_task_mutation( + role: Role, + field: str, + new_value: Any, +) -> ValidationResult: + """ + Convenience function to validate a task field mutation. + + Args: + role: The role attempting the mutation + field: The field name (e.g., "status", "commit", "notes") + new_value: The new value + + Returns: + ValidationResult + """ + # Build the full path - we use placeholder indices since we normalize anyway + field_path = f"phases.*.tasks.*.{field}" + return validate_mutation(role, field_path, new_value) + + +def validate_phase_mutation( + role: Role, + field: str, + new_value: Any, +) -> ValidationResult: + """ + Convenience function to validate a phase field mutation. + + Args: + role: The role attempting the mutation + field: The field name (e.g., "status") + new_value: The new value + + Returns: + ValidationResult + """ + field_path = f"phases.*.{field}" + return validate_mutation(role, field_path, new_value) diff --git a/tests/shared/egg_contracts/__init__.py b/tests/shared/egg_contracts/__init__.py new file mode 100644 index 0000000000..80e96936d7 --- /dev/null +++ b/tests/shared/egg_contracts/__init__.py @@ -0,0 +1 @@ +"""Tests for egg_contracts library.""" diff --git a/tests/shared/egg_contracts/test_audit.py b/tests/shared/egg_contracts/test_audit.py new file mode 100644 index 0000000000..af70efddaf --- /dev/null +++ b/tests/shared/egg_contracts/test_audit.py @@ -0,0 +1,420 @@ +"""Tests for egg_contracts.audit module.""" + +from datetime import UTC, datetime + +from egg_contracts.audit import ( + create_audit_entry, + create_transition_entry, + create_update_entry, + format_audit_log, +) +from egg_contracts.models import AuditAction, AuditEntry, AuditRole + + +class TestCreateAuditEntry: + """Tests for create_audit_entry function.""" + + def test_all_fields_populated(self): + """Test creating an entry with all fields set.""" + entry = create_audit_entry( + actor="egg", + role=AuditRole.IMPLEMENTER, + action=AuditAction.UPDATE, + field_path="phases.0.tasks.0.commit", + old_value=None, + new_value="abc1234", + reason="Implementation complete", + ) + + assert entry.actor == "egg" + assert entry.role == AuditRole.IMPLEMENTER + assert entry.action == AuditAction.UPDATE + assert entry.field_path == "phases.0.tasks.0.commit" + assert entry.old_value is None + assert entry.new_value == "abc1234" + assert entry.reason == "Implementation complete" + assert isinstance(entry.timestamp, datetime) + + def test_timestamp_is_utc(self): + """Test that the timestamp uses UTC timezone.""" + before = datetime.now(UTC) + entry = create_audit_entry( + actor="egg", + role=AuditRole.SYSTEM, + action=AuditAction.CREATE, + field_path="issue", + ) + after = datetime.now(UTC) + + assert before <= entry.timestamp <= after + + def test_optional_fields_default_to_none(self): + """Test that old_value, new_value, and reason default to None.""" + entry = create_audit_entry( + actor="system", + role=AuditRole.SYSTEM, + action=AuditAction.CREATE, + field_path="contract", + ) + + assert entry.old_value is None + assert entry.new_value is None + assert entry.reason is None + + def test_returns_audit_entry_model(self): + """Test that the return type is AuditEntry.""" + entry = create_audit_entry( + actor="egg", + role=AuditRole.IMPLEMENTER, + action=AuditAction.UPDATE, + field_path="phases.0.tasks.0.notes", + ) + assert isinstance(entry, AuditEntry) + + def test_all_roles(self): + """Test creating entries with each role.""" + for role in AuditRole: + entry = create_audit_entry( + actor="test-actor", + role=role, + action=AuditAction.UPDATE, + field_path="test.path", + ) + assert entry.role == role + + def test_all_actions(self): + """Test creating entries with each action type.""" + for action in AuditAction: + entry = create_audit_entry( + actor="test-actor", + role=AuditRole.SYSTEM, + action=action, + field_path="test.path", + ) + assert entry.action == action + + +class TestCreateUpdateEntry: + """Tests for create_update_entry function.""" + + def test_creates_update_action(self): + """Test that the entry has UPDATE action.""" + entry = create_update_entry( + actor="egg", + role=AuditRole.IMPLEMENTER, + field_path="phases.0.tasks.0.notes", + old_value="old notes", + new_value="new notes", + ) + + assert entry.action == AuditAction.UPDATE + + def test_captures_old_and_new_values(self): + """Test that old and new values are recorded.""" + entry = create_update_entry( + actor="reviewer-bot", + role=AuditRole.REVIEWER, + field_path="phases.0.tasks.0.status", + old_value="pending", + new_value="complete", + ) + + assert entry.old_value == "pending" + assert entry.new_value == "complete" + + def test_with_reason(self): + """Test update entry with a reason.""" + entry = create_update_entry( + actor="egg", + role=AuditRole.IMPLEMENTER, + field_path="phases.0.tasks.0.commit", + old_value=None, + new_value="def5678", + reason="Fixed failing test", + ) + + assert entry.reason == "Fixed failing test" + + def test_without_reason(self): + """Test update entry without a reason defaults to None.""" + entry = create_update_entry( + actor="egg", + role=AuditRole.IMPLEMENTER, + field_path="phases.0.tasks.0.notes", + old_value="", + new_value="Added notes", + ) + + assert entry.reason is None + + def test_field_path_preserved(self): + """Test that the field_path is set correctly.""" + entry = create_update_entry( + actor="egg", + role=AuditRole.IMPLEMENTER, + field_path="acceptance_criteria.0.verified", + old_value=False, + new_value=True, + ) + + assert entry.field_path == "acceptance_criteria.0.verified" + + +class TestCreateTransitionEntry: + """Tests for create_transition_entry function.""" + + def test_creates_transition_action(self): + """Test that the entry has TRANSITION action.""" + entry = create_transition_entry( + actor="system", + role=AuditRole.SYSTEM, + from_phase="refine", + to_phase="implement", + ) + + assert entry.action == AuditAction.TRANSITION + + def test_field_path_is_current_phase(self): + """Test that field_path is always 'current_phase'.""" + entry = create_transition_entry( + actor="egg", + role=AuditRole.IMPLEMENTER, + from_phase="implement", + to_phase="pr", + ) + + assert entry.field_path == "current_phase" + + def test_phases_stored_as_old_and_new_values(self): + """Test that from_phase and to_phase map to old_value and new_value.""" + entry = create_transition_entry( + actor="system", + role=AuditRole.SYSTEM, + from_phase="refine", + to_phase="plan", + ) + + assert entry.old_value == "refine" + assert entry.new_value == "plan" + + def test_with_reason(self): + """Test transition entry with a reason.""" + entry = create_transition_entry( + actor="reviewer-bot", + role=AuditRole.REVIEWER, + from_phase="implement", + to_phase="pr", + reason="All tasks complete", + ) + + assert entry.reason == "All tasks complete" + + def test_without_reason(self): + """Test transition entry without a reason.""" + entry = create_transition_entry( + actor="system", + role=AuditRole.SYSTEM, + from_phase="refine", + to_phase="implement", + ) + + assert entry.reason is None + + +class TestFormatAuditLog: + """Tests for format_audit_log function.""" + + def test_empty_entries(self): + """Test formatting an empty list of entries.""" + result = format_audit_log([]) + assert result == "Audit Log:" + + def test_single_update_entry_with_old_and_new(self): + """Test formatting an update entry with both old and new values.""" + entry = AuditEntry( + timestamp=datetime(2025, 1, 15, 10, 30, 0, tzinfo=UTC), + actor="egg", + role=AuditRole.IMPLEMENTER, + action=AuditAction.UPDATE, + field_path="phases.0.tasks.0.notes", + old_value="old", + new_value="new", + ) + + result = format_audit_log([entry]) + lines = result.split("\n") + + assert lines[0] == "Audit Log:" + assert "[2025-01-15 10:30:00]" in lines[1] + assert "implementer:egg" in lines[1] + assert "update" in lines[1] + assert "phases.0.tasks.0.notes" in lines[1] + assert "(old -> new)" in lines[1] + + def test_entry_with_new_value_only(self): + """Test formatting an entry where only new_value is set (old_value is None).""" + entry = AuditEntry( + timestamp=datetime(2025, 1, 15, 10, 30, 0, tzinfo=UTC), + actor="system", + role=AuditRole.SYSTEM, + action=AuditAction.CREATE, + field_path="issue", + old_value=None, + new_value="created", + ) + + result = format_audit_log([entry]) + lines = result.split("\n") + + assert "= created" in lines[1] + # Should NOT have the arrow format + assert "->" not in lines[1] + + def test_entry_with_reason(self): + """Test formatting an entry that includes a reason.""" + entry = AuditEntry( + timestamp=datetime(2025, 1, 15, 10, 30, 0, tzinfo=UTC), + actor="egg", + role=AuditRole.IMPLEMENTER, + action=AuditAction.UPDATE, + field_path="phases.0.tasks.0.commit", + old_value=None, + new_value="abc1234", + reason="Implementation done", + ) + + result = format_audit_log([entry]) + assert "- Implementation done" in result + + def test_entry_without_values(self): + """Test formatting an entry with no old or new value.""" + entry = AuditEntry( + timestamp=datetime(2025, 1, 15, 10, 30, 0, tzinfo=UTC), + actor="egg", + role=AuditRole.IMPLEMENTER, + action=AuditAction.DELETE, + field_path="phases.0.tasks.0", + old_value=None, + new_value=None, + ) + + result = format_audit_log([entry]) + lines = result.split("\n") + + assert "delete" in lines[1] + # Should not have value indicators + assert "->" not in lines[1] + assert "= " not in lines[1] + + def test_multiple_entries(self): + """Test formatting multiple entries.""" + entries = [ + AuditEntry( + timestamp=datetime(2025, 1, 15, 10, 0, 0, tzinfo=UTC), + actor="system", + role=AuditRole.SYSTEM, + action=AuditAction.CREATE, + field_path="contract", + new_value="initialized", + ), + AuditEntry( + timestamp=datetime(2025, 1, 15, 11, 0, 0, tzinfo=UTC), + actor="egg", + role=AuditRole.IMPLEMENTER, + action=AuditAction.UPDATE, + field_path="phases.0.tasks.0.commit", + old_value=None, + new_value="abc1234", + ), + AuditEntry( + timestamp=datetime(2025, 1, 15, 12, 0, 0, tzinfo=UTC), + actor="reviewer", + role=AuditRole.REVIEWER, + action=AuditAction.TRANSITION, + field_path="current_phase", + old_value="implement", + new_value="pr", + ), + ] + + result = format_audit_log(entries) + lines = result.split("\n") + + assert len(lines) == 4 # header + 3 entries + assert lines[0] == "Audit Log:" + assert "system:system" in lines[1] + assert "implementer:egg" in lines[2] + assert "reviewer:reviewer" in lines[3] + + def test_limit_shows_last_n_entries(self): + """Test that limit parameter shows only the last N entries.""" + entries = [ + AuditEntry( + timestamp=datetime(2025, 1, 15, 10, 0, 0, tzinfo=UTC), + actor="first", + role=AuditRole.SYSTEM, + action=AuditAction.CREATE, + field_path="contract", + ), + AuditEntry( + timestamp=datetime(2025, 1, 15, 11, 0, 0, tzinfo=UTC), + actor="second", + role=AuditRole.IMPLEMENTER, + action=AuditAction.UPDATE, + field_path="phases.0.tasks.0.notes", + ), + AuditEntry( + timestamp=datetime(2025, 1, 15, 12, 0, 0, tzinfo=UTC), + actor="third", + role=AuditRole.REVIEWER, + action=AuditAction.UPDATE, + field_path="phases.0.status", + ), + ] + + result = format_audit_log(entries, limit=2) + lines = result.split("\n") + + # Header + 2 entries (the last two) + assert len(lines) == 3 + assert "second" in lines[1] + assert "third" in lines[2] + assert "first" not in result + + def test_limit_none_shows_all(self): + """Test that limit=None shows all entries.""" + entries = [ + AuditEntry( + timestamp=datetime(2025, 1, 15, i, 0, 0, tzinfo=UTC), + actor=f"actor-{i}", + role=AuditRole.SYSTEM, + action=AuditAction.UPDATE, + field_path="test", + ) + for i in range(5) + ] + + result = format_audit_log(entries, limit=None) + lines = result.split("\n") + assert len(lines) == 6 # header + 5 entries + + def test_transition_entry_format(self): + """Test formatting a transition entry with phase change.""" + entry = AuditEntry( + timestamp=datetime(2025, 1, 15, 10, 30, 0, tzinfo=UTC), + actor="system", + role=AuditRole.SYSTEM, + action=AuditAction.TRANSITION, + field_path="current_phase", + old_value="refine", + new_value="implement", + reason="Refinement complete", + ) + + result = format_audit_log([entry]) + lines = result.split("\n") + + assert "transition" in lines[1] + assert "current_phase" in lines[1] + assert "(refine -> implement)" in lines[1] + assert "- Refinement complete" in lines[1] diff --git a/tests/shared/egg_contracts/test_loader.py b/tests/shared/egg_contracts/test_loader.py new file mode 100644 index 0000000000..2ca62de3c4 --- /dev/null +++ b/tests/shared/egg_contracts/test_loader.py @@ -0,0 +1,412 @@ +"""Tests for egg_contracts.loader module.""" + +import json +import subprocess +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from egg_contracts.loader import ( + ContractNotFoundError, + ContractValidationError, + contract_exists, + create_contract, + delete_contract, + export_contract, + get_contract_path, + list_contracts, + load_contract, + load_contract_from_branch, + save_contract, +) +from egg_contracts.models import ( + Contract, + IssueInfo, + PipelinePhase, +) + + +def _make_contract( + issue_number: int = 42, + title: str = "Test issue", + url: str = "https://github.com/owner/repo/issues/42", + phase: PipelinePhase = PipelinePhase.REFINE, +) -> Contract: + """Helper to create a minimal contract for testing.""" + return Contract( + issue=IssueInfo(number=issue_number, title=title, url=url), + current_phase=phase, + ) + + +class TestContractNotFoundError: + """Tests for ContractNotFoundError exception.""" + + def test_attributes(self): + """Test that issue_number and path are stored.""" + path = Path("/repo/.egg-state/contracts/99.json") + err = ContractNotFoundError(99, path) + assert err.issue_number == 99 + assert err.path == path + + def test_message_format(self): + """Test the error message contains issue number and path.""" + path = Path("/repo/.egg-state/contracts/5.json") + err = ContractNotFoundError(5, path) + assert "#5" in str(err) + assert str(path) in str(err) + + +class TestContractValidationError: + """Tests for ContractValidationError exception.""" + + def test_attributes(self): + """Test that issue_number and errors are stored.""" + err = ContractValidationError(10, ["bad field", "missing value"]) + assert err.issue_number == 10 + assert err.errors == ["bad field", "missing value"] + + def test_message_format(self): + """Test the error message contains issue number and joined errors.""" + err = ContractValidationError(10, ["error A", "error B"]) + msg = str(err) + assert "#10" in msg + assert "error A" in msg + assert "error B" in msg + + +class TestGetContractPath: + """Tests for get_contract_path function.""" + + def test_with_repo_root(self, tmp_path): + """Test path construction with explicit repo_root.""" + path = get_contract_path(42, repo_root=tmp_path) + assert path == tmp_path / ".egg-state" / "contracts" / "42.json" + + def test_without_repo_root_uses_cwd(self): + """Test path defaults to cwd when repo_root is None.""" + path = get_contract_path(7) + expected = Path.cwd() / ".egg-state" / "contracts" / "7.json" + assert path == expected + + def test_different_issue_numbers(self, tmp_path): + """Test that different issue numbers produce different filenames.""" + path_1 = get_contract_path(1, repo_root=tmp_path) + path_2 = get_contract_path(999, repo_root=tmp_path) + assert path_1.name == "1.json" + assert path_2.name == "999.json" + assert path_1 != path_2 + + +class TestLoadContract: + """Tests for load_contract function.""" + + def test_load_valid_contract(self, tmp_path): + """Test loading a valid contract from disk.""" + contract = _make_contract() + contracts_dir = tmp_path / ".egg-state" / "contracts" + contracts_dir.mkdir(parents=True) + contract_path = contracts_dir / "42.json" + contract_path.write_text(json.dumps(contract.model_dump(mode="json"), indent=2)) + + loaded = load_contract(42, repo_root=tmp_path) + assert loaded.issue.number == 42 + assert loaded.issue.title == "Test issue" + assert loaded.current_phase == PipelinePhase.REFINE + + def test_load_not_found_raises(self, tmp_path): + """Test that loading a missing contract raises ContractNotFoundError.""" + with pytest.raises(ContractNotFoundError) as exc_info: + load_contract(999, repo_root=tmp_path) + assert exc_info.value.issue_number == 999 + + def test_load_invalid_json_raises(self, tmp_path): + """Test that invalid JSON raises ContractValidationError.""" + contracts_dir = tmp_path / ".egg-state" / "contracts" + contracts_dir.mkdir(parents=True) + contract_path = contracts_dir / "42.json" + contract_path.write_text("{not valid json!!!") + + with pytest.raises(ContractValidationError) as exc_info: + load_contract(42, repo_root=tmp_path) + assert exc_info.value.issue_number == 42 + assert any("Invalid JSON" in e for e in exc_info.value.errors) + + def test_load_validation_error_raises(self, tmp_path): + """Test that valid JSON with invalid schema raises ContractValidationError.""" + contracts_dir = tmp_path / ".egg-state" / "contracts" + contracts_dir.mkdir(parents=True) + contract_path = contracts_dir / "42.json" + # Valid JSON but missing required 'issue' field + contract_path.write_text(json.dumps({"schemaVersion": "1.0"})) + + with pytest.raises(ContractValidationError) as exc_info: + load_contract(42, repo_root=tmp_path) + assert exc_info.value.issue_number == 42 + + +class TestSaveContract: + """Tests for save_contract function.""" + + def test_save_creates_file(self, tmp_path): + """Test that save_contract creates a valid JSON file.""" + contract = _make_contract() + path = save_contract(contract, repo_root=tmp_path) + + assert path.exists() + assert path.name == "42.json" + + # Verify contents are valid JSON that round-trips + data = json.loads(path.read_text()) + loaded = Contract.model_validate(data) + assert loaded.issue.number == 42 + + def test_save_creates_directories(self, tmp_path): + """Test that save_contract creates parent directories if missing.""" + contract = _make_contract() + contracts_dir = tmp_path / ".egg-state" / "contracts" + assert not contracts_dir.exists() + + save_contract(contract, repo_root=tmp_path) + assert contracts_dir.exists() + + def test_save_overwrites_existing(self, tmp_path): + """Test that saving to the same path overwrites the old file.""" + contract_v1 = _make_contract(title="Version 1") + save_contract(contract_v1, repo_root=tmp_path) + + contract_v2 = _make_contract(title="Version 2") + save_contract(contract_v2, repo_root=tmp_path) + + loaded = load_contract(42, repo_root=tmp_path) + assert loaded.issue.title == "Version 2" + + def test_save_returns_correct_path(self, tmp_path): + """Test that save_contract returns the expected path.""" + contract = _make_contract(issue_number=77) + path = save_contract(contract, repo_root=tmp_path) + expected = tmp_path / ".egg-state" / "contracts" / "77.json" + assert path == expected + + def test_save_file_has_trailing_newline(self, tmp_path): + """Test that saved file ends with a newline.""" + contract = _make_contract() + path = save_contract(contract, repo_root=tmp_path) + content = path.read_text() + assert content.endswith("\n") + + def test_save_atomic_write_cleans_up_on_failure(self, tmp_path): + """Test that temp file is cleaned up if write fails.""" + contract = _make_contract() + contracts_dir = tmp_path / ".egg-state" / "contracts" + contracts_dir.mkdir(parents=True) + + with patch("egg_contracts.loader.json.dump", side_effect=RuntimeError("boom")): + with pytest.raises(RuntimeError, match="boom"): + save_contract(contract, repo_root=tmp_path) + + # No temp files should remain + tmp_files = list(contracts_dir.glob("*.tmp")) + assert tmp_files == [] + + +class TestContractExists: + """Tests for contract_exists function.""" + + def test_exists_returns_true(self, tmp_path): + """Test contract_exists returns True when contract file is present.""" + contract = _make_contract() + save_contract(contract, repo_root=tmp_path) + assert contract_exists(42, repo_root=tmp_path) is True + + def test_exists_returns_false(self, tmp_path): + """Test contract_exists returns False when file is missing.""" + assert contract_exists(999, repo_root=tmp_path) is False + + +class TestCreateContract: + """Tests for create_contract function.""" + + def test_creates_and_saves_contract(self, tmp_path): + """Test that create_contract creates a contract and persists it.""" + contract = create_contract( + issue_number=55, + title="New feature", + url="https://github.com/owner/repo/issues/55", + repo_root=tmp_path, + ) + + assert contract.issue.number == 55 + assert contract.issue.title == "New feature" + assert contract.current_phase == PipelinePhase.REFINE + + # Verify it was saved to disk + assert contract_exists(55, repo_root=tmp_path) + loaded = load_contract(55, repo_root=tmp_path) + assert loaded.issue.number == 55 + + def test_create_with_custom_phase(self, tmp_path): + """Test creating a contract with a non-default initial phase.""" + contract = create_contract( + issue_number=56, + title="Urgent fix", + url="https://github.com/owner/repo/issues/56", + repo_root=tmp_path, + initial_phase=PipelinePhase.IMPLEMENT, + ) + assert contract.current_phase == PipelinePhase.IMPLEMENT + + def test_create_default_phase_is_refine(self, tmp_path): + """Test that the default initial phase is REFINE.""" + contract = create_contract( + issue_number=57, + title="Test", + url="https://github.com/owner/repo/issues/57", + repo_root=tmp_path, + ) + assert contract.current_phase == PipelinePhase.REFINE + + +class TestLoadContractFromBranch: + """Tests for load_contract_from_branch function.""" + + def test_none_branch_falls_back_to_load_contract(self, tmp_path): + """Test that branch=None delegates to load_contract.""" + contract = _make_contract() + save_contract(contract, repo_root=tmp_path) + + loaded = load_contract_from_branch(42, tmp_path, branch=None) + assert loaded.issue.number == 42 + + def test_with_branch_uses_git_show(self, tmp_path): + """Test that specifying a branch calls git show.""" + contract = _make_contract() + contract_json = json.dumps(contract.model_dump(mode="json")) + + mock_result = MagicMock() + mock_result.stdout = contract_json + + with patch("subprocess.run", return_value=mock_result) as mock_run: + loaded = load_contract_from_branch(42, tmp_path, branch="feature/test") + + mock_run.assert_called_once_with( + ["git", "show", "feature/test:.egg-state/contracts/42.json"], + cwd=tmp_path, + capture_output=True, + text=True, + check=True, + ) + assert loaded.issue.number == 42 + + def test_git_error_raises_contract_not_found(self, tmp_path): + """Test that a git error raises ContractNotFoundError.""" + with patch( + "subprocess.run", + side_effect=subprocess.CalledProcessError(1, "git show"), + ): + with pytest.raises(ContractNotFoundError) as exc_info: + load_contract_from_branch(42, tmp_path, branch="nonexistent") + assert exc_info.value.issue_number == 42 + + def test_invalid_json_from_branch_raises_validation_error(self, tmp_path): + """Test that invalid JSON from git show raises ContractValidationError.""" + mock_result = MagicMock() + mock_result.stdout = "not valid json" + + with patch("subprocess.run", return_value=mock_result): + with pytest.raises(ContractValidationError) as exc_info: + load_contract_from_branch(42, tmp_path, branch="bad-branch") + assert exc_info.value.issue_number == 42 + assert any("Invalid JSON" in e for e in exc_info.value.errors) + + +class TestListContracts: + """Tests for list_contracts function.""" + + def test_no_contracts_dir(self, tmp_path): + """Test list_contracts returns empty list when directory doesn't exist.""" + result = list_contracts(repo_root=tmp_path) + assert result == [] + + def test_empty_contracts_dir(self, tmp_path): + """Test list_contracts returns empty list when directory is empty.""" + contracts_dir = tmp_path / ".egg-state" / "contracts" + contracts_dir.mkdir(parents=True) + result = list_contracts(repo_root=tmp_path) + assert result == [] + + def test_lists_contract_numbers_sorted(self, tmp_path): + """Test that contracts are listed as sorted issue numbers.""" + create_contract(100, "Issue 100", "https://example.com/100", repo_root=tmp_path) + create_contract(5, "Issue 5", "https://example.com/5", repo_root=tmp_path) + create_contract(42, "Issue 42", "https://example.com/42", repo_root=tmp_path) + + result = list_contracts(repo_root=tmp_path) + assert result == [5, 42, 100] + + def test_skips_non_numeric_filenames(self, tmp_path): + """Test that non-numeric JSON files are ignored.""" + contracts_dir = tmp_path / ".egg-state" / "contracts" + contracts_dir.mkdir(parents=True) + + # Create a valid contract + create_contract(10, "Issue 10", "https://example.com/10", repo_root=tmp_path) + + # Create non-numeric JSON files that should be skipped + (contracts_dir / "notes.json").write_text("{}") + (contracts_dir / "README.json").write_text("{}") + + result = list_contracts(repo_root=tmp_path) + assert result == [10] + + +class TestDeleteContract: + """Tests for delete_contract function.""" + + def test_delete_existing_contract(self, tmp_path): + """Test deleting an existing contract returns True.""" + create_contract(42, "Test", "https://example.com/42", repo_root=tmp_path) + assert contract_exists(42, repo_root=tmp_path) + + result = delete_contract(42, repo_root=tmp_path) + assert result is True + assert not contract_exists(42, repo_root=tmp_path) + + def test_delete_nonexistent_contract(self, tmp_path): + """Test deleting a missing contract returns False.""" + result = delete_contract(999, repo_root=tmp_path) + assert result is False + + +class TestExportContract: + """Tests for export_contract function.""" + + def test_export_with_audit_log(self): + """Test export includes audit_log by default.""" + contract = _make_contract() + data = export_contract(contract) + + assert isinstance(data, dict) + assert "audit_log" in data + assert data["issue"]["number"] == 42 + + def test_export_without_audit_log(self): + """Test export excludes audit_log when include_audit_log=False.""" + contract = _make_contract() + data = export_contract(contract, include_audit_log=False) + + assert "audit_log" not in data + assert data["issue"]["number"] == 42 + + def test_export_preserves_all_fields(self): + """Test that export includes all contract fields.""" + contract = _make_contract() + data = export_contract(contract) + + assert "schemaVersion" in data + assert "issue" in data + assert "current_phase" in data + assert "phases" in data + assert "decisions" in data + assert "circuit_breaker" in data + assert "acceptance_criteria" in data diff --git a/tests/shared/egg_contracts/test_models.py b/tests/shared/egg_contracts/test_models.py new file mode 100644 index 0000000000..d5fc45a35c --- /dev/null +++ b/tests/shared/egg_contracts/test_models.py @@ -0,0 +1,341 @@ +"""Tests for egg_contracts.models module.""" + +from datetime import UTC, datetime + +import pytest +from egg_contracts.models import ( + AuditAction, + AuditEntry, + AuditRole, + CircuitBreaker, + CircuitBreakerStatus, + Contract, + Decision, + DecisionType, + IssueInfo, + Phase, + PhaseStatus, + PipelinePhase, + Task, + TaskStatus, +) +from pydantic import ValidationError + + +class TestIssueInfo: + """Tests for IssueInfo model.""" + + def test_valid_issue(self): + """Test creating a valid issue info.""" + issue = IssueInfo( + number=123, + title="Test issue", + url="https://github.com/owner/repo/issues/123", + ) + assert issue.number == 123 + assert issue.title == "Test issue" + + def test_invalid_issue_number(self): + """Test that issue number must be positive.""" + with pytest.raises(ValidationError): + IssueInfo( + number=0, + title="Test", + url="https://github.com/owner/repo/issues/0", + ) + + def test_empty_title_rejected(self): + """Test that empty title is rejected.""" + with pytest.raises(ValidationError): + IssueInfo( + number=1, + title="", + url="https://github.com/owner/repo/issues/1", + ) + + +class TestTask: + """Tests for Task model.""" + + def test_valid_task(self): + """Test creating a valid task.""" + task = Task( + id="task-1", + description="Implement feature", + ) + assert task.id == "task-1" + assert task.status == TaskStatus.PENDING + assert task.commit is None + assert task.notes == "" + + def test_task_with_commit(self): + """Test task with commit SHA.""" + task = Task( + id="task-1", + description="Test", + commit="abc1234", + ) + assert task.commit == "abc1234" + + def test_invalid_task_id_pattern(self): + """Test that task ID must match pattern.""" + with pytest.raises(ValidationError): + Task( + id="invalid-id", + description="Test", + ) + + def test_invalid_commit_pattern(self): + """Test that commit must be valid SHA pattern.""" + with pytest.raises(ValidationError): + Task( + id="task-1", + description="Test", + commit="not-a-sha", + ) + + def test_all_task_statuses(self): + """Test all valid task statuses.""" + for status in TaskStatus: + task = Task( + id="task-1", + description="Test", + status=status, + ) + assert task.status == status + + +class TestPhase: + """Tests for Phase model.""" + + def test_valid_phase(self): + """Test creating a valid phase.""" + phase = Phase( + id="phase-1", + name="Setup", + ) + assert phase.id == "phase-1" + assert phase.name == "Setup" + assert phase.status == PhaseStatus.PENDING + assert phase.tasks == [] + + def test_phase_with_tasks(self): + """Test phase with nested tasks.""" + phase = Phase( + id="phase-1", + name="Setup", + tasks=[ + Task(id="task-1", description="First task"), + Task(id="task-2", description="Second task"), + ], + ) + assert len(phase.tasks) == 2 + assert phase.tasks[0].id == "task-1" + + def test_invalid_phase_id(self): + """Test that phase ID must match pattern.""" + with pytest.raises(ValidationError): + Phase( + id="invalid", + name="Test", + ) + + +class TestDecision: + """Tests for Decision model.""" + + def test_valid_hitl_decision(self): + """Test creating a HITL decision.""" + decision = Decision( + id="decision-1", + question="Approve plan?", + type=DecisionType.HITL, + ) + assert decision.id == "decision-1" + assert decision.type == DecisionType.HITL + assert decision.resolved is False + assert decision.resolution is None + + def test_resolved_decision(self): + """Test a resolved decision.""" + decision = Decision( + id="decision-1", + question="Approve plan?", + type=DecisionType.HITL, + resolved=True, + resolution="approved", + resolved_by="human@example.com", + resolved_at=datetime.now(UTC), + ) + assert decision.resolved is True + assert decision.resolution == "approved" + + +class TestCircuitBreaker: + """Tests for CircuitBreaker model.""" + + def test_default_circuit_breaker(self): + """Test default circuit breaker state.""" + cb = CircuitBreaker() + assert cb.total_cycles == 0 + assert cb.max_total_cycles == 10 + assert cb.status == CircuitBreakerStatus.CLOSED + + def test_open_circuit_breaker(self): + """Test open circuit breaker.""" + cb = CircuitBreaker( + total_cycles=5, + status=CircuitBreakerStatus.OPEN, + ) + assert cb.status == CircuitBreakerStatus.OPEN + + +class TestAuditEntry: + """Tests for AuditEntry model.""" + + def test_valid_audit_entry(self): + """Test creating a valid audit entry.""" + entry = AuditEntry( + timestamp=datetime.now(UTC), + actor="egg", + role=AuditRole.IMPLEMENTER, + action=AuditAction.UPDATE, + field_path="phases.0.tasks.0.commit", + old_value=None, + new_value="abc1234", + ) + assert entry.actor == "egg" + assert entry.role == AuditRole.IMPLEMENTER + assert entry.action == AuditAction.UPDATE + + +class TestContract: + """Tests for Contract model.""" + + def test_minimal_contract(self): + """Test creating a minimal contract.""" + contract = Contract( + issue=IssueInfo( + number=133, + title="Test issue", + url="https://github.com/owner/repo/issues/133", + ), + ) + assert contract.schemaVersion == "1.0" + assert contract.issue.number == 133 + assert contract.current_phase == PipelinePhase.REFINE + assert contract.phases == [] + assert contract.decisions == [] + + def test_full_contract(self): + """Test creating a contract with all fields.""" + contract = Contract( + issue=IssueInfo( + number=133, + title="Test issue", + url="https://github.com/owner/repo/issues/133", + ), + current_phase=PipelinePhase.IMPLEMENT, + phases=[ + Phase( + id="phase-1", + name="Setup", + tasks=[ + Task(id="task-1", description="First task"), + ], + ), + ], + decisions=[ + Decision( + id="decision-1", + question="Approve?", + type=DecisionType.HITL, + ), + ], + ) + assert contract.current_phase == PipelinePhase.IMPLEMENT + assert len(contract.phases) == 1 + assert len(contract.decisions) == 1 + + def test_get_task(self): + """Test get_task helper method.""" + contract = Contract( + issue=IssueInfo(number=1, title="Test", url="https://example.com"), + phases=[ + Phase( + id="phase-1", + name="Setup", + tasks=[ + Task(id="task-1", description="First"), + Task(id="task-2", description="Second"), + ], + ), + ], + ) + task = contract.get_task("phase-1", "task-2") + assert task is not None + assert task.description == "Second" + + # Non-existent task + assert contract.get_task("phase-1", "task-99") is None + assert contract.get_task("phase-99", "task-1") is None + + def test_get_phase(self): + """Test get_phase helper method.""" + contract = Contract( + issue=IssueInfo(number=1, title="Test", url="https://example.com"), + phases=[ + Phase(id="phase-1", name="First"), + Phase(id="phase-2", name="Second"), + ], + ) + phase = contract.get_phase("phase-2") + assert phase is not None + assert phase.name == "Second" + + assert contract.get_phase("phase-99") is None + + def test_get_decision(self): + """Test get_decision helper method.""" + contract = Contract( + issue=IssueInfo(number=1, title="Test", url="https://example.com"), + decisions=[ + Decision(id="decision-1", question="First?", type=DecisionType.HITL), + ], + ) + decision = contract.get_decision("decision-1") + assert decision is not None + assert decision.question == "First?" + + assert contract.get_decision("decision-99") is None + + +class TestContractSerialization: + """Tests for contract serialization.""" + + def test_json_roundtrip(self): + """Test that contract can be serialized and deserialized.""" + original = Contract( + issue=IssueInfo( + number=133, + title="Test", + url="https://example.com", + ), + phases=[ + Phase( + id="phase-1", + name="Setup", + tasks=[Task(id="task-1", description="Test")], + ), + ], + ) + + # Serialize + data = original.model_dump(mode="json") + assert isinstance(data, dict) + + # Deserialize + restored = Contract.model_validate(data) + assert restored.issue.number == original.issue.number + assert len(restored.phases) == 1 + assert restored.phases[0].tasks[0].id == "task-1" diff --git a/tests/shared/egg_contracts/test_roles.py b/tests/shared/egg_contracts/test_roles.py new file mode 100644 index 0000000000..3e7592aa0f --- /dev/null +++ b/tests/shared/egg_contracts/test_roles.py @@ -0,0 +1,160 @@ +"""Tests for egg_contracts.roles module.""" + +from egg_contracts.roles import ( + FIELD_OWNERSHIP, + Role, + can_modify, + get_field_owner, + get_role_permissions, + normalize_path, +) + + +class TestNormalizePath: + """Tests for normalize_path function.""" + + def test_numeric_indices_replaced(self): + """Test that numeric indices are replaced with wildcards.""" + assert normalize_path("phases.0.tasks.1.status") == "phases.*.tasks.*.status" + + def test_no_indices(self): + """Test paths without indices unchanged.""" + assert normalize_path("issue.number") == "issue.number" + + def test_mixed_path(self): + """Test path with mixed indices and names.""" + assert normalize_path("phases.0.name") == "phases.*.name" + assert normalize_path("decisions.5.resolved") == "decisions.*.resolved" + + def test_empty_path(self): + """Test empty path.""" + assert normalize_path("") == "" + + def test_single_component(self): + """Test single component path.""" + assert normalize_path("schemaVersion") == "schemaVersion" + + +class TestGetFieldOwner: + """Tests for get_field_owner function.""" + + def test_implementer_fields(self): + """Test fields owned by implementer.""" + assert get_field_owner("phases.0.tasks.0.commit") == Role.IMPLEMENTER + assert get_field_owner("phases.1.tasks.5.notes") == Role.IMPLEMENTER + + def test_reviewer_fields(self): + """Test fields owned by reviewer.""" + assert get_field_owner("phases.0.tasks.0.status") == Role.REVIEWER + assert get_field_owner("phases.0.status") == Role.REVIEWER + assert get_field_owner("acceptance_criteria.0.verified") == Role.REVIEWER + + def test_human_fields(self): + """Test fields owned by human.""" + assert get_field_owner("decisions.0.resolved") == Role.HUMAN + assert get_field_owner("decisions.0.resolution") == Role.HUMAN + assert get_field_owner("decisions.0.resolved_by") == Role.HUMAN + + 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 + + +class TestCanModify: + """Tests for can_modify function.""" + + def test_implementer_can_modify_own_fields(self): + """Test implementer can modify implementer fields.""" + assert can_modify(Role.IMPLEMENTER, "phases.0.tasks.0.commit") is True + assert can_modify(Role.IMPLEMENTER, "phases.0.tasks.0.notes") is True + + def test_implementer_cannot_modify_reviewer_fields(self): + """Test implementer cannot modify reviewer fields.""" + assert can_modify(Role.IMPLEMENTER, "phases.0.tasks.0.status") is False + assert can_modify(Role.IMPLEMENTER, "phases.0.status") is False + + def test_implementer_cannot_modify_human_fields(self): + """Test implementer cannot modify human fields.""" + assert can_modify(Role.IMPLEMENTER, "decisions.0.resolved") is False + + def test_reviewer_can_modify_own_fields(self): + """Test reviewer can modify reviewer fields.""" + assert can_modify(Role.REVIEWER, "phases.0.tasks.0.status") is True + assert can_modify(Role.REVIEWER, "phases.0.status") is True + assert can_modify(Role.REVIEWER, "acceptance_criteria.0.verified") is True + + def test_reviewer_cannot_modify_implementer_fields(self): + """Test reviewer cannot modify implementer fields.""" + assert can_modify(Role.REVIEWER, "phases.0.tasks.0.commit") is False + assert can_modify(Role.REVIEWER, "phases.0.tasks.0.notes") is False + + def test_reviewer_cannot_modify_human_fields(self): + """Test reviewer cannot modify human fields.""" + assert can_modify(Role.REVIEWER, "decisions.0.resolved") is False + + def test_human_can_modify_everything(self): + """Test human can modify all fields.""" + assert can_modify(Role.HUMAN, "phases.0.tasks.0.commit") is True + assert can_modify(Role.HUMAN, "phases.0.tasks.0.status") is True + assert can_modify(Role.HUMAN, "decisions.0.resolved") is True + assert can_modify(Role.HUMAN, "issue.number") is True + + def test_system_can_only_modify_system_fields(self): + """Test system can only modify system-owned fields.""" + assert can_modify(Role.SYSTEM, "issue.number") is True + assert can_modify(Role.SYSTEM, "schemaVersion") is True + assert can_modify(Role.SYSTEM, "phases.0.tasks.0.status") is False + assert can_modify(Role.SYSTEM, "phases.0.tasks.0.commit") is False + + +class TestGetRolePermissions: + """Tests for get_role_permissions function.""" + + def test_implementer_permissions(self): + """Test implementer permission summary.""" + perms = get_role_permissions(Role.IMPLEMENTER) + assert "phases.*.tasks.*.commit" in perms["can_modify"] + assert "phases.*.tasks.*.notes" in perms["can_modify"] + assert "phases.*.tasks.*.status" in perms["cannot_modify"] + + def test_reviewer_permissions(self): + """Test reviewer permission summary.""" + perms = get_role_permissions(Role.REVIEWER) + assert "phases.*.tasks.*.status" in perms["can_modify"] + assert "phases.*.status" in perms["can_modify"] + assert "phases.*.tasks.*.commit" in perms["cannot_modify"] + + def test_human_permissions(self): + """Test human permission summary.""" + perms = get_role_permissions(Role.HUMAN) + assert perms["can_modify"] == ["*"] + assert perms["cannot_modify"] == [] + + +class TestFieldOwnershipConfiguration: + """Tests for FIELD_OWNERSHIP configuration.""" + + def test_all_ownership_entries_valid(self): + """Test that all field ownership entries use valid roles.""" + for _path, role in FIELD_OWNERSHIP.items(): + assert isinstance(role, Role) + + def test_implementer_ownership_patterns(self): + """Test expected implementer ownership patterns exist.""" + implementer_paths = [p for p, r in FIELD_OWNERSHIP.items() if r == Role.IMPLEMENTER] + assert "phases.*.tasks.*.commit" in implementer_paths + assert "phases.*.tasks.*.notes" in implementer_paths + + def test_reviewer_ownership_patterns(self): + """Test expected reviewer ownership patterns exist.""" + reviewer_paths = [p for p, r in FIELD_OWNERSHIP.items() if r == Role.REVIEWER] + assert "phases.*.tasks.*.status" in reviewer_paths + assert "phases.*.status" in reviewer_paths + + def test_human_ownership_patterns(self): + """Test expected human ownership patterns exist.""" + human_paths = [p for p, r in FIELD_OWNERSHIP.items() if r == Role.HUMAN] + assert "decisions.*.resolved" in human_paths + assert "decisions.*.resolution" in human_paths diff --git a/tests/shared/egg_contracts/test_validator.py b/tests/shared/egg_contracts/test_validator.py new file mode 100644 index 0000000000..52420f3524 --- /dev/null +++ b/tests/shared/egg_contracts/test_validator.py @@ -0,0 +1,264 @@ +"""Tests for egg_contracts.validator module.""" + +import pytest +from egg_contracts.models import ( + Contract, + IssueInfo, + Phase, + Task, + TaskStatus, +) +from egg_contracts.roles import Role +from egg_contracts.validator import ( + apply_mutation, + validate_mutation, + validate_phase_mutation, + validate_task_mutation, +) + + +class TestValidateMutation: + """Tests for validate_mutation function.""" + + def test_valid_implementer_mutation(self): + """Test that implementer can modify allowed fields.""" + result = validate_mutation( + role=Role.IMPLEMENTER, + field_path="phases.0.tasks.0.commit", + new_value="abc1234", + ) + assert result.valid is True + assert result.message == "Mutation allowed" + + def test_invalid_implementer_mutation(self): + """Test that implementer cannot modify status.""" + result = validate_mutation( + role=Role.IMPLEMENTER, + field_path="phases.0.tasks.0.status", + new_value="complete", + ) + assert result.valid is False + assert "implementer" in result.message.lower() + assert "reviewer" in result.message.lower() + assert result.required_role == "reviewer" + + def test_valid_reviewer_mutation(self): + """Test that reviewer can modify status.""" + result = validate_mutation( + role=Role.REVIEWER, + field_path="phases.0.tasks.0.status", + new_value="complete", + ) + assert result.valid is True + + def test_invalid_reviewer_mutation(self): + """Test that reviewer cannot modify commit.""" + result = validate_mutation( + role=Role.REVIEWER, + field_path="phases.0.tasks.0.commit", + new_value="abc1234", + ) + assert result.valid is False + assert result.required_role == "implementer" + + def test_human_can_modify_anything(self): + """Test that human role can modify any field.""" + # Implementer field + result = validate_mutation( + role=Role.HUMAN, + field_path="phases.0.tasks.0.commit", + new_value="abc1234", + ) + assert result.valid is True + + # Reviewer field + result = validate_mutation( + role=Role.HUMAN, + field_path="phases.0.tasks.0.status", + new_value="complete", + ) + assert result.valid is True + + # Human field + result = validate_mutation( + role=Role.HUMAN, + field_path="decisions.0.resolved", + new_value=True, + ) + assert result.valid is True + + +class TestApplyMutation: + """Tests for apply_mutation function.""" + + @pytest.fixture + def sample_contract(self): + """Create a sample contract for testing.""" + return Contract( + issue=IssueInfo( + number=133, + title="Test", + url="https://example.com", + ), + phases=[ + Phase( + id="phase-1", + name="Setup", + tasks=[ + Task( + id="task-1", + description="First task", + status=TaskStatus.PENDING, + ), + ], + ), + ], + ) + + def test_apply_valid_mutation(self, sample_contract): + """Test applying a valid mutation.""" + result = apply_mutation( + contract=sample_contract, + role=Role.IMPLEMENTER, + actor="egg", + field_path="phases.0.tasks.0.commit", + new_value="abc1234", + reason="Implementation complete", + ) + assert result.success is True + assert result.contract is not None + assert result.contract.phases[0].tasks[0].commit == "abc1234" + assert result.audit_entry is not None + assert result.audit_entry.actor == "egg" + assert result.audit_entry.reason == "Implementation complete" + + def test_apply_invalid_mutation_rejected(self, sample_contract): + """Test that invalid mutations are rejected.""" + result = apply_mutation( + contract=sample_contract, + role=Role.IMPLEMENTER, + actor="egg", + field_path="phases.0.tasks.0.status", + new_value="complete", + ) + assert result.success is False + assert "implementer" in result.message.lower() + assert result.contract is None + assert result.audit_entry is None + + def test_apply_reviewer_mutation(self, sample_contract): + """Test reviewer can mark task complete.""" + result = apply_mutation( + contract=sample_contract, + role=Role.REVIEWER, + actor="reviewer-agent", + field_path="phases.0.tasks.0.status", + new_value=TaskStatus.COMPLETE, + ) + assert result.success is True + assert result.contract.phases[0].tasks[0].status == TaskStatus.COMPLETE + + def test_audit_log_appended(self, sample_contract): + """Test that audit log entry is appended.""" + initial_log_len = len(sample_contract.audit_log) + + result = apply_mutation( + contract=sample_contract, + role=Role.IMPLEMENTER, + actor="egg", + field_path="phases.0.tasks.0.notes", + new_value="Added implementation notes", + ) + + assert result.success is True + assert len(result.contract.audit_log) == initial_log_len + 1 + assert result.contract.audit_log[-1].field_path == "phases.0.tasks.0.notes" + + def test_old_value_captured(self, sample_contract): + """Test that old value is captured in audit log.""" + # First set a value + sample_contract.phases[0].tasks[0].notes = "Original notes" + + result = apply_mutation( + contract=sample_contract, + role=Role.IMPLEMENTER, + actor="egg", + field_path="phases.0.tasks.0.notes", + new_value="Updated notes", + ) + + assert result.success is True + assert result.audit_entry.old_value == "Original notes" + assert result.audit_entry.new_value == "Updated notes" + + +class TestValidateTaskMutation: + """Tests for validate_task_mutation helper.""" + + def test_implementer_commit_allowed(self): + """Test implementer can set commit.""" + result = validate_task_mutation( + role=Role.IMPLEMENTER, + field="commit", + new_value="abc1234", + ) + assert result.valid is True + + def test_implementer_status_denied(self): + """Test implementer cannot set status.""" + result = validate_task_mutation( + role=Role.IMPLEMENTER, + field="status", + new_value="complete", + ) + assert result.valid is False + assert result.required_role == "reviewer" + + +class TestValidatePhaseMutation: + """Tests for validate_phase_mutation helper.""" + + def test_reviewer_status_allowed(self): + """Test reviewer can set phase status.""" + result = validate_phase_mutation( + role=Role.REVIEWER, + field="status", + new_value="complete", + ) + assert result.valid is True + + def test_implementer_status_denied(self): + """Test implementer cannot set phase status.""" + result = validate_phase_mutation( + role=Role.IMPLEMENTER, + field="status", + new_value="complete", + ) + assert result.valid is False + + +class TestErrorMessages: + """Tests for error message formatting.""" + + def test_clear_error_message_format(self): + """Test that error messages are clear and helpful.""" + result = validate_mutation( + role=Role.IMPLEMENTER, + field_path="phases.0.tasks.0.status", + new_value="complete", + ) + assert result.valid is False + # Check error message contains key information + assert "phases.*.tasks.*.status" in result.message + assert "implementer" in result.message.lower() + assert "reviewer" in result.message.lower() + + def test_decision_field_error_message(self): + """Test error message for decision field.""" + result = validate_mutation( + role=Role.REVIEWER, + field_path="decisions.0.resolved", + new_value=True, + ) + assert result.valid is False + assert "human" in result.message.lower()