parthpetkar/metahackathon
0
1"""Simulated CI/CD pipeline runner for Hugging Face Spaces deployment.2 3Replaces Docker/subprocess-based execution with high-fidelity simulation that:4- Produces realistic logs matching real tool output (git, docker, uv pip, pytest, compose)5- Honors all 20 fault types with correct stage failures and error messages6- Detects fixes by inspecting actual workspace files (with partial fix detection)7- Validates Python syntax via real AST parsing (not pattern matching)8- Validates SQL syntax via real token-level parsing9- Replicates the real secret scan logic (walks files, same patterns as pipeline_runner.py)10- Replicates check_logs.py static validation (LOG_LEVEL, RotatingFileHandler, PII scan)11- Simulates health-check probing for deploy-stage verification12- Simulates multi-fault cascading: all active faults surface at their own stages13- Partial-fix detection: scores fix completeness and emits targeted warnings when incomplete14- Clone log reflects actual workspace git state (real SHA + commit message)15- Stage durations are failure-mode-aware (fast for early errors, full range for success)16- stage.status is always a _StatusWrapper so .value never raises AttributeError17- Runs entirely in pure Python (no Docker, no subprocess, no privileged operations)18- Maintains exact API compatibility with RealPipelineRunner19 20Target: Hugging Face Spaces CPU environment (no Docker-in-Docker support)21"""22 23from __future__ import annotations24 25import ast26import hashlib27import os28import random29import re30import socket31import subprocess32import time33from dataclasses import dataclass, field34from typing import Callable, Dict, List, Optional, Tuple, TYPE_CHECKING35 36if TYPE_CHECKING:37 from models import AdversarialCICDScenario38 39 40# ── Stage and Pipeline Status Enums (must match pipeline_runner.py) ────────41 42class StageStatus:43 PENDING = "pending"44 RUNNING = "running"45 PASSED = "passed"46 FAILED = "failed"47 SKIPPED = "skipped"48 49 50class PipelineStatus:51 PENDING = "pending"52 RUNNING = "running"53 PASSED = "passed"54 FAILED = "failed"55 56 57STAGE_ORDER = ["clone", "build", "test", "deploy"]58 59 60# ── Stage Weights for Pipeline Health ──────────────────────────────────────61 62STAGE_WEIGHTS = {63 "clone": 0.10,64 "build": 0.30,65 "test": 0.30,66 "deploy": 0.30,67}68 69 70# ── Fault → Stage Mapping ──────────────────────────────────────────────────71 72FAULT_STAGE_MAP: Dict[str, str] = {73 "merge_conflict": "build",74 "dependency_conflict": "build",75 "docker_order": "build",76 "flaky_test": "test",77 "missing_permission": "deploy",78 "secret_exposure": "build",79 "env_drift": "deploy",80 "log_pii_leak": "build",81 "log_disabled": "build",82 "bad_migration_sql": "build",83 "schema_drift": "deploy",84}85 86 87# ── _StatusWrapper ─────────────────────────────────────────────────────────88 89class _StatusWrapper:90 """Wraps a status string to provide a .value attribute (matches str-Enum API)."""91 92 def __init__(self, status: str):93 self._status = status94 self.value = status95 96 def __str__(self):97 return self._status98 99 def __repr__(self):100 return f"_StatusWrapper({self._status!r})"101 102 def __eq__(self, other):103 if isinstance(other, _StatusWrapper):104 return self._status == other._status105 return self._status == other106 107 def __hash__(self):108 return hash(self._status)109 110 111def _sw(status: str) -> _StatusWrapper:112 return _StatusWrapper(status)113 114 115# ── File helpers ───────────────────────────────────────────────────────────116 117def _read_file_safe(workspace: str, rel_path: str) -> str:118 try:119 with open(os.path.join(workspace, rel_path), "r", encoding="utf-8", errors="replace") as f:120 return f.read()121 except (OSError, FileNotFoundError):122 return ""123 124 125def _find_python_files(workspace: str, subdir: str = "") -> List[str]:126 """Walk a subdirectory and return relative paths of all .py files."""127 root = os.path.join(workspace, subdir) if subdir else workspace128 results: List[str] = []129 try:130 for dirpath, dirnames, filenames in os.walk(root):131 dirnames[:] = [d for d in dirnames if d not in (".git", "__pycache__", ".venv")]132 for fname in filenames:133 if fname.endswith(".py"):134 full = os.path.join(dirpath, fname)135 results.append(os.path.relpath(full, workspace).replace("\\", "/"))136 except OSError:137 pass138 return results139 140 141# ── Real Syntax / Semantic Validators ─────────────────────────────────────142 143def _validate_python_syntax(workspace: str, rel_path: str) -> Tuple[bool, str]:144 """Parse a file with the real CPython AST. Returns (ok, error_message)."""145 content = _read_file_safe(workspace, rel_path)146 if not content:147 return True, ""148 try:149 ast.parse(content, filename=rel_path)150 return True, ""151 except SyntaxError as exc:152 return False, (153 f' File "/app/{rel_path}", line {exc.lineno}\n'154 f" {exc.text or ''}\n"155 f" {'':>{max((exc.offset or 1) - 1, 0)}}^\n"156 f"SyntaxError: {exc.msg}"157 )158 159 160def _validate_sql_tokens(workspace: str, rel_path: str) -> Tuple[bool, str]:161 """Token-level SQL keyword check for common typos."""162 content = _read_file_safe(workspace, rel_path)163 if not content:164 return True, ""165 bad_keywords = {"CREAT", "INSER", "SELEC", "UPDAT", "DELET", "DROPT", "ALTERR"}166 for lineno, raw_line in enumerate(content.splitlines(), 1):167 for word in re.split(r"\s+", raw_line.strip()):168 token = word.upper().rstrip("(")169 if token in bad_keywords:170 col = raw_line.index(word) if word in raw_line else 0171 return False, (172 f'psycopg2.errors.SyntaxError: syntax error at or near "{word}"\n'173 f"LINE {lineno}: {raw_line.strip()}\n"174 f" {'':>{col}}^\n"175 f"DETAIL: Expected {token[:-1] if token.endswith('T') else token}, found {word}\n"176 f"CONTEXT: SQL statement in migration file: {rel_path}:{lineno}\n"177 f"ERROR: Database migration failed during build"178 )179 return True, ""180 181 182def _simulate_health_check(host: str = "localhost", port: int = 5000, path: str = "/health") -> Tuple[bool, str]:183 """Probe TCP reachability as a stand-in for HTTP health-check."""184 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)185 sock.settimeout(0.5)186 try:187 connected = sock.connect_ex((host, port)) == 0188 if connected:189 return True, f"Health check: GET http://{host}:{port}{path} -> 200 OK"190 return True, f"Health check: simulated GET http://{host}:{port}{path} -> 200 OK (no real server)"191 except OSError:192 return True, f"Health check: simulated GET http://{host}:{port}{path} -> 200 OK (no real server)"193 finally:194 try:195 sock.close()196 except OSError:197 pass198 199 200# ── Real Secret Scan (mirrors pipeline_runner.py _secret_scan exactly) ────201 202_TOKEN_PATTERNS = [203 "sk-live-", "sk-test-", "sk_live_", "sk_test_",204 "AKIA", "ghp_", "gho_", "github_pat_",205]206_ASSIGN_PATTERNS = [207 re.compile(r"API_KEY\s*=\s*['\"]"),208 re.compile(r"SECRET_KEY\s*=\s*['\"]"),209 re.compile(r"PASSWORD\s*=\s*['\"]"),210]211_SCAN_EXTENSIONS = {".py", ".yml", ".yaml", ".json", ".env", ".cfg"}212 213 214def _run_secret_scan(workspace: str) -> Tuple[int, str, str]:215 """Walk workspace files and detect hardcoded secrets (same logic as real runner)."""216 secrets_found: List[str] = []217 for root, dirs, files in os.walk(workspace):218 dirs[:] = [d for d in dirs if d not in (".git", ".venv", "__pycache__")]219 rel_root = os.path.relpath(root, workspace).replace("\\", "/")220 # Skip scripts/ (contains scanner patterns as literals) and .github/ (CI yml grep commands)221 if rel_root.startswith("scripts") or rel_root.startswith(".github"):222 continue223 for fname in files:224 if os.path.splitext(fname)[1] not in _SCAN_EXTENSIONS:225 continue226 filepath = os.path.join(root, fname)227 rel = os.path.relpath(filepath, workspace).replace("\\", "/")228 try:229 with open(filepath, "r", encoding="utf-8", errors="ignore") as f:230 content = f.read()231 for i, line in enumerate(content.splitlines(), 1):232 if any(p in line for p in _TOKEN_PATTERNS):233 secrets_found.append(f"{rel}:{i}: {line.strip()}")234 if fname.endswith(".py"):235 for pat in _ASSIGN_PATTERNS:236 for m in pat.finditer(content):237 line_num = content[: m.start()].count("\n") + 1238 line = content.splitlines()[line_num - 1].strip()239 entry = f"{rel}:{line_num}: {line}"240 if entry not in secrets_found:241 secrets_found.append(entry)242 except OSError:243 continue244 245 if secrets_found:246 stderr = "[SECURITY GATE] Secret scan FAILED\nHardcoded secrets detected:\n"247 stderr += "".join(f" ERROR: {f}\n" for f in secrets_found)248 stderr += "\nPolicy check failed: plaintext credential found in source code.\nBuild blocked: secret exposure policy violation"249 return 1, "", stderr250 return 0, "Secret scan passed: no hardcoded secrets detected.\n", ""251 252 253# ── Real Log Config Check (mirrors check_logs.py validate_config) ─────────254 255_SOURCE_PII_PATTERNS = [256 ("token_in_log_call",257 re.compile(r'_log\.\w+\s*\([^)]*(?:sk-live|sk-test|AKIA|Bearer\s+sk)[^)]*\)', re.DOTALL)),258 ("password_in_log_call",259 re.compile(r'_log\.\w+\s*\([^)]*(?i:password|passwd|secret)[^)]*[=:]\s*["\'][^"\']{6,}["\'][^)]*\)', re.DOTALL)),260]261_SILENCING_LEVELS = {"CRITICAL"}262_VALID_LEVELS = {"DEBUG", "INFO", "WARNING", "WARN", "ERROR", "CRITICAL"}263 264 265def _run_log_config_check(workspace: str) -> Tuple[int, str, str]:266 """Replicate check_logs.py --config-only inside the simulated build."""267 config_rel = "services/api/logging_config.py"268 routes_rel = "services/api/routes.py"269 270 config_src = _read_file_safe(workspace, config_rel)271 if not config_src:272 return 0, "Log config check skipped: scripts/check_logs.py not in workspace.\n", ""273 274 failures: List[str] = []275 276 # Parseable Python277 try:278 tree = ast.parse(config_src)279 except SyntaxError as exc:280 failures.append(f"SyntaxError in logging_config.py: {exc}")281 return _log_check_result(failures, config_rel)282 283 # Required variables284 assigned: set = set()285 for node in ast.walk(tree):286 if isinstance(node, ast.Assign):287 for t in node.targets:288 if isinstance(t, ast.Name):289 assigned.add(t.id)290 elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):291 assigned.add(node.target.id)292 for var in ("LOG_PATH", "LOG_LEVEL", "MAX_BYTES", "BACKUP_COUNT"):293 if var not in assigned:294 failures.append(f"logging_config.py is missing required variable: {var}")295 296 # RotatingFileHandler297 if "RotatingFileHandler(" not in config_src:298 failures.append(299 "logging_config.py does not use RotatingFileHandler — "300 "logs may grow unboundedly without rotation"301 )302 303 # LOG_LEVEL not silencing304 level_m = re.search(r'LOG_LEVEL\s*(?::\s*\w+)?\s*=\s*["\']([A-Z]+)["\']', config_src)305 if level_m:306 level = level_m.group(1)307 if level in _SILENCING_LEVELS:308 failures.append(309 f"LOG_LEVEL is hardcoded to {level!r} — "310 "this silences all log output below CRITICAL"311 )312 elif level not in _VALID_LEVELS:313 failures.append(f"LOG_LEVEL {level!r} is not a recognised logging level")314 315 # LOG_PATH not restricted316 path_m = re.search(r'LOG_PATH\s*(?::\s*\w+)?\s*=\s*["\']([^"\']+)["\']', config_src)317 if path_m:318 declared = path_m.group(1)319 if declared.startswith(("/var/log", "/root", "/sys", "/proc")):320 failures.append(321 f"LOG_PATH default {declared!r} points to a restricted system directory "322 "— the application process cannot write there"323 )324 325 # JSON formatter326 if "json.dumps" not in config_src:327 failures.append(328 "logging_config.py formatter does not call json.dumps() — "329 "log records will not be valid JSON"330 )331 332 # Required JSON fields333 for f_name in ("timestamp", "level", "message", "service"):334 if f'"{f_name}"' not in config_src and f"'{f_name}'" not in config_src:335 failures.append(336 f"JSON formatter appears to be missing required field: {f_name!r} — "337 "structured log records will be incomplete"338 )339 340 # Static PII scan of config341 for label, pattern in _SOURCE_PII_PATTERNS:342 if pattern.search(config_src):343 failures.append(344 f"logging_config.py contains a log call that may emit credential values "345 f"[pattern: {label}]"346 )347 348 # Static PII scan of routes.py349 routes_src = _read_file_safe(workspace, routes_rel)350 if routes_src:351 for label, pattern in _SOURCE_PII_PATTERNS:352 if pattern.search(routes_src):353 failures.append(354 f"routes.py contains a log call that may emit credential values "355 f"[pattern: {label}] — PII leak risk"356 )357 358 return _log_check_result(failures, config_rel)359 360 361def _log_check_result(failures: List[str], config_rel: str) -> Tuple[int, str, str]:362 if failures:363 stderr = "LOG CONFIG CHECK FAILED\n"364 for f in failures:365 stderr += f" FAIL: {f}\n"366 stderr += "\nBuild blocked: logging configuration violates observability requirements"367 return 1, "", stderr368 return 0, "Log config check passed: logging configuration is valid.\n", ""369 370 371# ── Partial Fix Detection ──────────────────────────────────────────────────372 373PARTIAL_FIX_CHECKS: Dict[str, List[Tuple[str, Callable[[str], bool]]]] = {374 "merge_conflict": [375 ("conflict markers removed from routes.py",376 lambda ws: "<<<<<<" not in _read_file_safe(ws, "services/api/routes.py")),377 ("routes.py has valid Python syntax",378 lambda ws: _validate_python_syntax(ws, "services/api/routes.py")[0]),379 ],380 "dependency_conflict": [381 ("urllib3==2.0 exact pin removed from requirements.txt",382 lambda ws: "urllib3==2.0" not in _read_file_safe(ws, "services/api/requirements.txt")),383 ("requests and urllib3 versions are mutually compatible",384 lambda ws: _no_version_drift(ws)),385 ],386 "docker_order": [387 ("COPY requirements.txt precedes RUN uv pip install",388 lambda ws: _dockerfile_copy_before_run(ws)),389 ],390 "flaky_test": [391 ("test_response_time test removed or threshold relaxed to >= 0.1s",392 lambda ws: _flaky_test_fixed(ws)),393 ],394 "missing_permission": [395 ("external: true removed from docker-compose.yml",396 lambda ws: "external: true" not in _read_file_safe(ws, "docker-compose.yml")),397 ],398 "secret_exposure": [399 ("secret scan passes (no hardcoded secrets found)",400 lambda ws: _run_secret_scan(ws)[0] == 0),401 ],402 "env_drift": [403 ("not-a-number port removed from docker-compose.yml",404 lambda ws: "not-a-number" not in _read_file_safe(ws, "docker-compose.yml")),405 ],406 "log_pii_leak": [407 ("log config check passes (no PII in log calls)",408 lambda ws: _run_log_config_check(ws)[0] == 0),409 ],410 "log_disabled": [411 ("log config check passes (LOG_LEVEL not CRITICAL)",412 lambda ws: _run_log_config_check(ws)[0] == 0),413 ],414 "bad_migration_sql": [415 ("CREAT TABLE typo corrected in 001_init.sql",416 lambda ws: "CREAT TABLE" not in _read_file_safe(ws, "db/migrations/001_init.sql")),417 ("SQL tokens valid",418 lambda ws: _validate_sql_tokens(ws, "db/migrations/001_init.sql")[0]),419 ],420 "schema_drift": [421 ("artifact_url removed from database.py CANONICAL_COLUMNS",422 lambda ws: "artifact_url" not in _read_file_safe(ws, "db/database.py")),423 ],424}425 426 427def _dockerfile_copy_before_run(workspace: str) -> bool:428 content = _read_file_safe(workspace, "Dockerfile")429 if not content:430 return False431 lines = [l.strip() for l in content.splitlines() if l.strip()]432 copy_idx, run_idx = -1, -1433 for i, line in enumerate(lines):434 if "COPY" in line and "requirements.txt" in line:435 copy_idx = i436 if "RUN" in line and ("pip install" in line or "uv pip install" in line):437 run_idx = i438 return copy_idx != -1 and run_idx != -1 and copy_idx < run_idx439 440 441def _flaky_test_fixed(workspace: str) -> bool:442 """True if the flaky timing test has been removed OR its threshold raised to >= 0.1s."""443 content = _read_file_safe(workspace, "tests/test_api.py")444 if "test_response_time" not in content:445 return True446 m = re.search(r"threshold\s*=\s*([0-9]*\.?[0-9]+)", content)447 return bool(m) and float(m.group(1)) >= 0.1448 449 450def _no_duplicate_ports(workspace: str) -> bool:451 content = _read_file_safe(workspace, "docker-compose.yml")452 ports = re.findall(r'"\s*(\d+):\d+"', content)453 return len(ports) == len(set(ports))454 455 456def _no_version_drift(workspace: str) -> bool:457 content = _read_file_safe(workspace, "services/api/requirements.txt")458 # Passes if no pinned version conflicts exist (simplistic: no ==x.y with known bad combos)459 has_bad_urllib3 = bool(re.search(r"urllib3==1\.", content))460 has_new_requests = bool(re.search(r"requests==2\.(3[0-9]|[4-9]\d)\.", content))461 return not (has_bad_urllib3 and has_new_requests)462 463 464FIX_DETECTION: Dict[str, Callable[[str], bool]] = {465 fault: (lambda ws, f=fault: _score_fix(ws, f)[0] == 1.0)466 for fault in PARTIAL_FIX_CHECKS467}468 469 470def _score_fix(workspace: str, fault_type: str) -> Tuple[float, List[str]]:471 """Return (fraction_complete, list_of_failing_check_descriptions)."""472 checks = PARTIAL_FIX_CHECKS.get(fault_type, [])473 if not checks:474 return 1.0, []475 passed = 0476 failing: List[str] = []477 for desc, fn in checks:478 try:479 ok = fn(workspace)480 except Exception:481 ok = False482 if ok:483 passed += 1484 else:485 failing.append(desc)486 return passed / len(checks), failing487 488 489# ── Stage Result and Pipeline Result ──────────────────────────────────────490 491@dataclass492class SimulatedStageResult:493 name: str494 status: _StatusWrapper = field(default_factory=lambda: _sw(StageStatus.PENDING))495 exit_code: int = -1496 stdout: str = ""497 stderr: str = ""498 duration_seconds: float = 0.0499 command: str = ""500 501 def __post_init__(self):502 if isinstance(self.status, str):503 self.status = _sw(self.status)504 505 506@dataclass507class SimulatedPipelineResult:508 pipeline_id: str = ""509 status: _StatusWrapper = field(default_factory=lambda: _sw(PipelineStatus.PENDING))510 stages: Dict[str, SimulatedStageResult] = field(default_factory=dict)511 failed_stage: str = ""512 total_duration_seconds: float = 0.0513 workspace_dir: str = ""514 image_tag: str = ""515 cache_tag: str = ""516 517 def __post_init__(self):518 if not self.stages:519 self.stages = {name: SimulatedStageResult(name=name) for name in STAGE_ORDER}520 if isinstance(self.status, str):521 self.status = _sw(self.status)522 523 def get_stage_logs(self, stage_name: str) -> str:524 stage = self.stages.get(stage_name)525 if not stage:526 return f"No logs available for stage '{stage_name}'"527 parts = []528 if stage.command:529 parts.append(f"$ {stage.command}")530 if stage.stdout:531 parts.append(stage.stdout)532 if stage.stderr:533 parts.append(stage.stderr)534 return "\n".join(parts) if parts else f"No output captured for stage '{stage_name}'"535 536 def get_stage_statuses(self) -> Dict[str, str]:537 return {name: str(stage.status) for name, stage in self.stages.items()}538 539 def get_stage_durations(self) -> Dict[str, float]:540 return {name: round(stage.duration_seconds, 2) for name, stage in self.stages.items()}541 542 543# ── Module-level partial-fix warning helper (importable by other runners) ─544 545def _partial_fix_warnings(546 active_faults: List[str],547 stage: str,548 fault_status: Dict[str, Tuple[bool, float, List[str]]],549) -> str:550 lines: List[str] = []551 for fault in active_faults:552 fully_fixed, score, failing = fault_status[fault]553 if not fully_fixed and score > 0 and FAULT_STAGE_MAP.get(fault) == stage:554 for check in failing:555 lines.append(f"Error: {check}")556 lines.append("Pipeline still failing — fix is incomplete.")557 return "\n".join(lines)558 559 560# ── Simulated Pipeline Runner ──────────────────────────────────────────────561 562class SimulatedPipelineRunner:563 """Simulated CI/CD pipeline runner that produces realistic logs without Docker.564 565 Maintains exact API compatibility with PipelineRunner from pipeline_runner.py.566 """567 568 def __init__(569 self,570 workspace_path: str,571 fault_type: Optional[str] = None,572 scenario: Optional["AdversarialCICDScenario"] = None,573 episode_id: Optional[str] = None,574 ):575 self.workspace_path = os.path.abspath(workspace_path)576 self.fault_type = fault_type577 self.scenario = scenario578 self.episode_id = episode_id or "sim-episode"579 580 # Collect all active faults (single fault_type OR scenario fault list)581 self._active_faults: List[str] = []582 if scenario and hasattr(scenario, 'steps'):583 # Extract fault_type from each IncidentStep in the scenario584 self._active_faults = [step.fault_type for step in scenario.steps]585 elif fault_type:586 self._active_faults = [fault_type]587 588 # Deterministic RNG seeded on episode + fault so reruns are identical589 seed_str = f"{self.episode_id}:{self.fault_type or 'none'}"590 seed_bytes = hashlib.sha256(seed_str.encode()).digest()591 self._seed = int.from_bytes(seed_bytes[:4], byteorder="big")592 self._rng = random.Random(self._seed)593 594 # ── Public run ─────────────────────────────────────────────────────────595 596 def run_stage(self, stage_name: str, workspace_dir: Optional[str] = None) -> SimulatedStageResult:597 """Execute a single pipeline stage and return its result.598 599 Useful for streaming stage-by-stage execution over WebSocket.600 Caller is responsible for tracking fault_status across calls.601 """602 ws_dir = workspace_dir or self.workspace_path603 604 fault_status: Dict[str, Tuple[bool, float, List[str]]] = {}605 for fault in self._active_faults:606 score, failing = _score_fix(ws_dir, fault)607 fault_status[fault] = (score == 1.0, score, failing)608 609 faults_by_stage: Dict[str, List[str]] = {s: [] for s in STAGE_ORDER}610 for fault in self._active_faults:611 fully_fixed, _, _ = fault_status[fault]612 if not fully_fixed:613 stage = FAULT_STAGE_MAP.get(fault, "build")614 faults_by_stage[stage].append(fault)615 616 syntax_errors_build: List[str] = []617 if stage_name == "build":618 for py_file in _find_python_files(ws_dir, "services") + _find_python_files(ws_dir, "tests"):619 ok, msg = _validate_python_syntax(ws_dir, py_file)620 if not ok:621 syntax_errors_build.append(msg)622 sql_ok, sql_msg = _validate_sql_tokens(ws_dir, "db/migrations/001_init.sql")623 if not sql_ok:624 syntax_errors_build.append(sql_msg)625 626 active_stage_faults = faults_by_stage.get(stage_name, [])627 stage = SimulatedStageResult(name=stage_name, status=_sw(StageStatus.RUNNING))628 629 if stage_name == "clone":630 exit_code, stdout, stderr = self._run_clone_stage(ws_dir)631 elif stage_name == "build":632 exit_code, stdout, stderr = self._run_build_stage(633 ws_dir, active_stage_faults, fault_status, syntax_errors_build634 )635 elif stage_name == "test":636 exit_code, stdout, stderr = self._run_test_stage(ws_dir, active_stage_faults, fault_status)637 elif stage_name == "deploy":638 exit_code, stdout, stderr = self._run_deploy_stage(ws_dir, active_stage_faults, fault_status)639 else:640 exit_code, stdout, stderr = 1, "", f"Unknown stage: {stage_name}"641 642 stage.duration_seconds = self._stage_duration(stage_name, exit_code, active_stage_faults)643 stage.exit_code = exit_code644 stage.stdout = stdout645 stage.stderr = stderr646 stage.command = self._stage_command(stage_name)647 stage.status = _sw(StageStatus.PASSED if exit_code == 0 else StageStatus.FAILED)648 return stage649 650 def run(self, workspace_dir: Optional[str] = None) -> SimulatedPipelineResult:651 ws_dir = workspace_dir or self.workspace_path652 653 result = SimulatedPipelineResult(654 pipeline_id=f"sim-{self.episode_id[:8]}",655 status=_sw(PipelineStatus.RUNNING),656 workspace_dir=ws_dir,657 image_tag=f"sample-app-sim-{self.episode_id[:8]}",658 cache_tag="sample-app-sim-cache",659 )660 661 start_time = time.time()662 663 # Score every active fault against current workspace state664 fault_status: Dict[str, Tuple[bool, float, List[str]]] = {}665 for fault in self._active_faults:666 score, failing = _score_fix(ws_dir, fault)667 fault_status[fault] = (score == 1.0, score, failing)668 669 # Group still-failing faults by their pipeline stage670 faults_by_stage: Dict[str, List[str]] = {s: [] for s in STAGE_ORDER}671 for fault in self._active_faults:672 fully_fixed, _, _ = fault_status[fault]673 if not fully_fixed:674 stage = FAULT_STAGE_MAP.get(fault, "build")675 faults_by_stage[stage].append(fault)676 677 # Always run real Python syntax check on all service + test files678 syntax_errors_build: List[str] = []679 for py_file in _find_python_files(ws_dir, "services") + _find_python_files(ws_dir, "tests"):680 ok, msg = _validate_python_syntax(ws_dir, py_file)681 if not ok:682 syntax_errors_build.append(msg)683 684 # Always run real SQL syntax check685 sql_ok, sql_msg = _validate_sql_tokens(ws_dir, "db/migrations/001_init.sql")686 if not sql_ok:687 syntax_errors_build.append(sql_msg)688 689 for stage_name in STAGE_ORDER:690 stage = result.stages[stage_name]691 stage.status = _sw(StageStatus.RUNNING)692 693 active_stage_faults = faults_by_stage.get(stage_name, [])694 695 if stage_name == "clone":696 exit_code, stdout, stderr = self._run_clone_stage(ws_dir)697 elif stage_name == "build":698 exit_code, stdout, stderr = self._run_build_stage(699 ws_dir, active_stage_faults, fault_status, syntax_errors_build700 )701 elif stage_name == "test":702 exit_code, stdout, stderr = self._run_test_stage(703 ws_dir, active_stage_faults, fault_status704 )705 elif stage_name == "deploy":706 exit_code, stdout, stderr = self._run_deploy_stage(707 ws_dir, active_stage_faults, fault_status708 )709 else:710 exit_code, stdout, stderr = 1, "", f"Unknown stage: {stage_name}"711 712 stage.duration_seconds = self._stage_duration(stage_name, exit_code, active_stage_faults)713 stage.exit_code = exit_code714 stage.stdout = stdout715 stage.stderr = stderr716 stage.command = self._stage_command(stage_name)717 718 if exit_code == 0:719 stage.status = _sw(StageStatus.PASSED)720 else:721 stage.status = _sw(StageStatus.FAILED)722 result.status = _sw(PipelineStatus.FAILED)723 result.failed_stage = stage_name724 725 for remaining in STAGE_ORDER[STAGE_ORDER.index(stage_name) + 1:]:726 secondary = faults_by_stage.get(remaining, [])727 note = f"Stage skipped due to upstream failure in {stage_name}."728 if secondary:729 note += (730 f" NOTE: the following fault(s) would also fail here "731 f"if reached: {', '.join(secondary)}"732 )733 result.stages[remaining].status = _sw(StageStatus.SKIPPED)734 result.stages[remaining].stdout = note735 break736 else:737 result.status = _sw(PipelineStatus.PASSED)738 739 result.total_duration_seconds = time.time() - start_time740 return result741 742 # ── Duration helpers ───────────────────────────────────────────────────743 744 # (success ranges, fast-failure range) per stage745 _DURATION_SUCCESS = {746 "clone": (0.8, 2.2),747 "build": (8.0, 14.0),748 "test": (2.5, 5.0),749 "deploy": (3.0, 6.0),750 }751 # Fault types that fail very early (< 1 s into the stage)752 _FAST_FAIL_FAULTS = {753 "dependency_conflict", "docker_order", "secret_exposure",754 "log_pii_leak", "log_disabled", "bad_migration_sql",755 "merge_conflict", # SyntaxError caught at pytest import, fast756 }757 758 def _stage_duration(self, stage_name: str, exit_code: int, active_faults: List[str]) -> float:759 if exit_code != 0:760 if any(f in self._FAST_FAIL_FAULTS for f in active_faults):761 return round(self._rng.uniform(0.1, 0.8), 2)762 return round(self._rng.uniform(0.5, 2.0), 2)763 lo, hi = self._DURATION_SUCCESS.get(stage_name, (1.0, 3.0))764 return round(self._rng.uniform(lo, hi), 2)765 766 def _stage_command(self, stage_name: str) -> str:767 return {768 "clone": "git clone /workspace/repo .",769 "build": "docker build -t sample-app:latest .",770 "test": "docker run --rm sample-app:latest python -m pytest tests/ -v",771 "deploy": "docker compose -f docker-compose.yml up -d",772 }.get(stage_name, "")773 774 # ── Stage simulators ───────────────────────────────────────────────────775 776 def _run_clone_stage(self, workspace: str) -> Tuple[int, str, str]:777 """Simulate git clone — reflect real workspace SHA if available."""778 sha, msg = self._real_git_head(workspace)779 stdout = (780 f"Cloning into 'sample-app'...\n"781 f"remote: Enumerating objects: 142, done.\n"782 f"remote: Counting objects: 100% (142/142), done.\n"783 f"remote: Compressing objects: 100% (89/89), done.\n"784 f"Receiving objects: 100% (142/142), 48.31 KiB | 2.1 MiB/s, done.\n"785 f"Resolving deltas: 100% (34/34), done.\n"786 f"HEAD is now at {sha} {msg}"787 )788 return 0, stdout, ""789 790 def _real_git_head(self, workspace: str) -> Tuple[str, str]:791 """Return (short_sha, subject) from the real workspace git log, or placeholders."""792 try:793 r = subprocess.run(794 ["git", "-C", workspace, "log", "-1", "--pretty=%h %s"],795 capture_output=True, text=True, timeout=5,796 )797 if r.returncode == 0 and r.stdout.strip():798 parts = r.stdout.strip().split(" ", 1)799 return parts[0], parts[1] if len(parts) > 1 else ""800 except Exception:801 pass802 return "a3f812c", "feat: initial sample-app scaffold"803 804 def _run_build_stage(805 self,806 workspace: str,807 active_faults: List[str],808 fault_status: Dict[str, Tuple[bool, float, List[str]]],809 syntax_errors: List[str],810 ) -> Tuple[int, str, str]:811 """Simulate docker build + real secret scan + real log config check."""812 extra_warnings = self._partial_fix_warnings("build", fault_status)813 814 # 1. Active fault log815 if active_faults:816 _, stdout, stderr = self._fault_log(active_faults[0])817 if extra_warnings:818 stderr = extra_warnings + "\n" + stderr819 return 1, stdout, stderr820 821 # 2. Real Python / SQL syntax gate822 if syntax_errors:823 stderr = "\n\n".join(syntax_errors)824 if extra_warnings:825 stderr = extra_warnings + "\n" + stderr826 return 1, "", stderr827 828 # 3. Real secret scan829 scan_exit, scan_out, scan_err = _run_secret_scan(workspace)830 if scan_exit != 0:831 return 1, scan_out, scan_err832 833 # 4. Real log config check834 log_exit, log_out, log_err = _run_log_config_check(workspace)835 if log_exit != 0:836 return 1, log_out, log_err837 838 stdout = (839 "Step 1/8 : FROM python:3.11-slim\n"840 " ---> 4b8a3f2e1c9d\n"841 "Step 2/8 : WORKDIR /app\n"842 " ---> 9f2c4a7b3d61\n"843 "Step 3/8 : COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /usr/local/bin/\n"844 " ---> 1a2b3c4d5e6f\n"845 "Step 4/8 : COPY services/api/requirements.txt /app/requirements.txt\n"846 " ---> 2b3c4d5e6f7a\n"847 "Step 5/8 : RUN uv pip install --system --no-cache -r requirements.txt\n"848 "Using Python 3.11 environment\n"849 "Resolved 5 packages in 140ms\n"850 "Downloading Flask-3.0.3-py3-none-any.whl (101 kB)\n"851 "Downloading requests-2.31.0-py3-none-any.whl (62 kB)\n"852 "Downloading urllib3-2.0.7-py3-none-any.whl (123 kB)\n"853 "Downloading gunicorn-21.2.0-py3-none-any.whl (79 kB)\n"854 "Downloading pytest-8.1.0-py3-none-any.whl (341 kB)\n"855 "Installed 5 packages in 380ms\n"856 " + flask==3.0.3\n"857 " + gunicorn==21.2.0\n"858 " + pytest==8.1.0\n"859 " + requests==2.31.0\n"860 " + urllib3==2.0.7\n"861 "Step 6/8 : COPY . /app/\n"862 " ---> 3c4d5e6f7a8b\n"863 "Step 7/8 : EXPOSE 5000\n"864 " ---> 4d5e6f7a8b9c\n"865 "Step 8/8 : CMD [\"python\", \"-m\", \"services.api.app\"]\n"866 " ---> 5e6f7a8b9c0d\n"867 "Successfully built 5e6f7a8b9c0d\n"868 "Successfully tagged sample-app:latest\n"869 + scan_out870 + log_out871 )872 return 0, stdout, ""873 874 def _run_test_stage(875 self,876 workspace: str,877 active_faults: List[str],878 fault_status: Dict[str, Tuple[bool, float, List[str]]],879 ) -> Tuple[int, str, str]:880 extra_warnings = self._partial_fix_warnings("test", fault_status)881 882 if active_faults:883 _, stdout, stderr = self._fault_log(active_faults[0])884 if extra_warnings:885 stderr = extra_warnings + "\n" + stderr886 return 1, stdout, stderr887 888 # Count real test functions in the workspace for a realistic test count889 n_tests = self._count_test_functions(workspace)890 stdout = self._pytest_success_log(n_tests)891 return 0, stdout, ""892 893 def _run_deploy_stage(894 self,895 workspace: str,896 active_faults: List[str],897 fault_status: Dict[str, Tuple[bool, float, List[str]]],898 ) -> Tuple[int, str, str]:899 extra_warnings = self._partial_fix_warnings("deploy", fault_status)900 901 if active_faults:902 _, stdout, stderr = self._fault_log(active_faults[0])903 if extra_warnings:904 stderr = extra_warnings + "\n" + stderr905 return 1, stdout, stderr906 907 _, health_msg = _simulate_health_check()908 stdout = (909 'Creating network "sample-app_default" with the default driver\n'910 "Pulling db (postgres:15-alpine)...\n"911 "Pulling api (sample-app:latest)...\n"912 "Creating sample-app_db_1 ... done\n"913 "Creating sample-app_api_1 ... done\n"914 "api-service | INFO: Started server process [1]\n"915 "api-service | INFO: Waiting for application startup.\n"916 "api-service | INFO: Application startup complete.\n"917 "api-service | INFO: Uvicorn running on http://0.0.0.0:5000 (Press CTRL+C to quit)\n"918 f"{health_msg}\n"919 "All services healthy. Deploy complete."920 )921 return 0, stdout, ""922 923 # ── Partial-fix warning block ──────────────────────────────────────────924 925 def _partial_fix_warnings(926 self,927 stage: str,928 fault_status: Dict[str, Tuple[bool, float, List[str]]],929 ) -> str:930 return _partial_fix_warnings(self._active_faults, stage, fault_status)931 932 # ── Test helpers ───────────────────────────────────────────────────────933 934 def _count_test_functions(self, workspace: str) -> int:935 """Count def test_* functions across the workspace tests/ directory."""936 count = 0937 for py_file in _find_python_files(workspace, "tests"):938 content = _read_file_safe(workspace, py_file)939 count += len(re.findall(r"^def test_", content, re.MULTILINE))940 return max(count, 4) # at least 4 so the log looks plausible941 942 def _pytest_success_log(self, n: int) -> str:943 files = ["tests/test_api.py", "tests/test_db.py", "tests/test_integration.py"]944 lines = [945 "============================= test session starts ==============================",946 "platform linux -- Python 3.11.7, pytest-8.1.0, pluggy-1.3.0",947 "rootdir: /workspace/sample-app",948 f"collected {n} items",949 "",950 ]951 per_file = max(1, n // len(files))952 collected = 0953 test_names = [954 "test_health_endpoint", "test_list_items", "test_get_item_exists",955 "test_get_item_not_found", "test_auth_flow", "test_response_schema",956 "test_migration_applies", "test_query_returns_rows", "test_schema_match",957 "test_end_to_end_flow", "test_concurrent_requests", "test_error_propagation",958 ]959 idx = 0960 for fpath in files:961 for _ in range(per_file):962 if idx >= len(test_names) or collected >= n:963 break964 pct = int(((collected + 1) / n) * 100)965 name = test_names[idx]966 pad = max(0, 40 - len(fpath) - len(name) - 9)967 lines.append(f"{fpath}::{name} PASSED{' ' * pad}[{pct:3d}%]")968 idx += 1969 collected += 1970 dur = round(self._rng.uniform(2.1, 5.8), 2)971 lines.append("")972 lines.append(f"============================== {n} passed in {dur}s ==============================")973 return "\n".join(lines)974 975 # ── Fault log templates ────────────────────────────────────────────────976 977 def _fault_log(self, fault: str) -> Tuple[int, str, str]:978 """Return (exit_code, stdout, stderr) for a given fault."""979 980 # ── build-stage faults ─────────────────────────────────────────────981 if fault == "dependency_conflict":982 return 1, "", (983 " × No solution found when resolving dependencies:\n"984 " ╰─▶ Because requests==2.28.0 requires urllib3>=1.21.1,<1.27\n"985 " and the requested urllib3==2.0.7 is incompatible,\n"986 " we can conclude that requests==2.28.0 cannot be installed.\n"987 " And because your project requires requests==2.28.0, we can\n"988 " conclude that your project's requirements are unsatisfiable.\n\n"989 "hint: Pre-releases are available for urllib3 in the requested range\n"990 " (e.g. 2.0.0a1), and pre-releases were not requested.\n"991 " Add `--prerelease=allow` to allow them.\n\n"992 "ERROR: ResolutionImpossible\n"993 "The command '/bin/sh -c uv pip install --system --no-cache -r requirements.txt'"994 " returned a non-zero code: 1"995 )996 997 if fault == "docker_order":998 return 1, "", (999 "Step 3/8 : RUN uv pip install --system --no-cache -r services/api/requirements.txt\n"1000 " ---> Running in 8f3a2b1e9c45\n"1001 "error: Failed to open file `services/api/requirements.txt`\n"1002 " Caused by: No such file or directory (os error 2)\n"1003 "The command '/bin/sh -c uv pip install --system --no-cache"1004 " -r services/api/requirements.txt' returned a non-zero code: 2"1005 )1006 1007 if fault == "secret_exposure":1008 # Use real scan result so logs match actual file state1009 _, _, stderr = _run_secret_scan(self.workspace_path)1010 if not stderr:1011 stderr = (1012 "[SECURITY GATE] Hardcoded secret detected in services/api/app.py:23\n"1013 ' ERROR: services/api/app.py:23: API_KEY = "sk-live-4f3c2a1b0e9d8c7f6a5b4e3d2c1a0f9e8d7c6b5a"\n'1014 ' ERROR: services/api/app.py:24: DATABASE_PASSWORD = "super_secret_db_password_2024"\n'1015 ' ERROR: services/api/app.py:25: WEBHOOK_SECRET = "whsec_a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6"\n\n'1016 "Policy check failed: plaintext credential found in source code.\n"1017 "Build blocked: secret exposure policy violation"1018 )1019 return 1, "", stderr1020 1021 if fault in ("log_pii_leak", "log_disabled"):1022 # Use real log config check so logs match actual file state1023 _, _, stderr = _run_log_config_check(self.workspace_path)1024 if not stderr:1025 _fallbacks = {1026 "log_pii_leak": (1027 "LOG CONFIG CHECK FAILED\n"1028 "[SECURITY SCAN] PII/credential leak detected: routes.py:89\n"1029 ' ERROR: services/api/routes.py:89: _log.warning("Auth token received:'1030 ' sk-live-4f3c2a1b0e9d8c7f6a5b4e3d2c1a0f9e8d7c6b5a")\n\n'1031 "Build blocked: PII policy violation"1032 ),1033 "log_disabled": (1034 "LOG CONFIG CHECK FAILED\n"1035 "WARNING: LOG_LEVEL=CRITICAL — all application logging suppressed.\n"1036 " ERROR: services/api/logging_config.py:12: LOG_LEVEL hardcoded to CRITICAL\n\n"1037 "Build blocked: logging configuration violates observability requirements"1038 ),1039 }1040 stderr = _fallbacks.get(fault, "LOG CONFIG CHECK FAILED\nBuild blocked.")1041 return 1, "", stderr1042 1043 if fault == "bad_migration_sql":1044 # Use real SQL validator for accurate line/column info1045 sql_ok, sql_msg = _validate_sql_tokens(self.workspace_path, "db/migrations/001_init.sql")1046 if not sql_ok:1047 return 1, "", sql_msg1048 return 1, "", (1049 'psycopg2.errors.SyntaxError: syntax error at or near "TABLE"\n'1050 "LINE 1: CREAT TABLE IF NOT EXISTS builds (\n"1051 " ^\n"1052 "DETAIL: Expected CREATE, found CREAT\n"1053 "CONTEXT: SQL statement in migration file: db/migrations/001_init.sql:1\n"1054 "ERROR: Database migration failed during build"1055 )1056 1057 # ── test-stage faults ──────────────────────────────────────────────1058 if fault == "merge_conflict":1059 ok, msg = _validate_python_syntax(self.workspace_path, "services/api/routes.py")1060 if not ok:1061 return 1, "", (1062 f"{msg}\n\n"1063 "During handling of the above exception, another exception occurred:\n\n"1064 "Traceback (most recent call last):\n"1065 ' File "/usr/local/lib/python3.11/site-packages/pytest/__main__.py", line 5, in <module>\n'1066 " from pytest import console_main\n"1067 ' File "/app/tests/test_api.py", line 3, in <module>\n'1068 " from services.api.routes import register_routes\n"1069 + msg + "\n"1070 "ERROR: InvocationError for command pytest tests/ -v (exited with code 1)"1071 )1072 return 1, "", (1073 ' File "/app/services/api/routes.py", line 47\n'1074 " <<<<<<< HEAD\n"1075 " ^\n"1076 "SyntaxError: invalid syntax\n\n"1077 "ERROR: InvocationError for command pytest tests/ -v (exited with code 1)"1078 )1079 1080 if fault == "flaky_test":1081 # If the agent already relaxed the threshold or deleted the test, don't fire.1082 if _flaky_test_fixed(self.workspace_path):1083 return 0, "", ""1084 # Read the actual threshold from the workspace file so the log reflects reality.1085 content = _read_file_safe(self.workspace_path, "tests/test_api.py")1086 m = re.search(r"threshold\s*=\s*([0-9]*\.?[0-9]+)", content)1087 live_threshold = float(m.group(1)) if m else 0.0011088 elapsed = round(self._rng.uniform(0.12, 0.19), 3)1089 return 1, "", (1090 "tests/test_api.py::test_response_time_health FAILED [ 50%]\n\n"1091 "================================== FAILURES ===================================\n"1092 "_________________________ test_response_time_health __________________________\n\n"1093 "client = <FlaskClient <Flask 'api'>>\n\n"1094 " def test_response_time_health(client):\n"1095 " import time\n"1096 " start = time.time()\n"1097 " time.sleep(0.1)\n"1098 " response = client.get(\"/health\")\n"1099 " elapsed = time.time() - start\n"1100 " assert response.status_code == 200\n"1101 f" threshold = {live_threshold}\n"1102 "> assert elapsed < threshold, (\n"1103 f' f"Health endpoint took {elapsed:.3f}s, expected < {{threshold}}s "\n'1104 ' f"(flaky: timing constraint too strict for this environment)."\n'1105 " )\n"1106 f"E AssertionError: Health endpoint took {elapsed:.3f}s, expected < {live_threshold}s"1107 " (flaky: timing constraint too strict for this environment).\n\n"1108 "tests/test_api.py:89: AssertionError\n"1109 "=========================== short test summary info ============================\n"1110 f"FAILED tests/test_api.py::test_response_time_health - AssertionError: response time {elapsed:.3f}s exceeds threshold {live_threshold}s\n"1111 "========================= 1 failed, 11 passed in 3.21s ========================="1112 )1113 1114 # ── deploy-stage faults ────────────────────────────────────────────1115 if fault == "missing_permission":1116 return 1, "", (1117 "ERROR: for api Cannot create container for service api: "1118 "network corp-internal-network-v2 declared as external, but could not be found\n"1119 "ERROR: Encountered errors while bringing up the project."1120 )1121 1122 if fault == "env_drift":1123 return 1, "", (1124 'ERROR: for api Cannot create container for service api: '1125 'invalid port specification: "not-a-number:5000"\n'1126 "ERROR: Encountered errors while bringing up the project."1127 )1128 1129 if fault == "schema_drift":1130 return 1, "", (1131 'sqlalchemy.exc.OperationalError: (psycopg2.errors.UndefinedColumn) '1132 'column "artifact_url" of relation "builds" does not exist\n'1133 "LINE 1: SELECT id, task_key, status, started_at, finished_at, exit_code, artifact_url\n"1134 " ^\n"1135 'HINT: Perhaps you meant to reference the column "builds.exit_code".\n'1136 "ERROR: Schema mismatch detected in database.py CANONICAL_COLUMNS\n"1137 "Hint: Either add a migration to CREATE the column, or remove it from CANONICAL_COLUMNS."1138 )1139 1140 # Generic fallback1141 return 1, "", f"ERROR: Pipeline fault '{fault}' triggered at this stage."1142 1143 # ── Pipeline health ────────────────────────────────────────────────────1144 1145 def _compute_pipeline_health(self, stage_results: Dict[str, SimulatedStageResult]) -> float:1146 health = 0.01147 for stage, weight in STAGE_WEIGHTS.items():1148 if str(stage_results[stage].status) == StageStatus.PASSED:1149 health += weight1150 return round(health, 3)1151 1152 1153# ── Compatibility Adapter ──────────────────────────────────────────────────1154 1155def cleanup_pipeline(result: SimulatedPipelineResult) -> None:1156 """No-op cleanup for simulated pipeline (no Docker resources to clean)."""1157 pass1158 1159 1160def cleanup_cache_image(cache_tag: str) -> None:1161 """No-op cleanup for simulated pipeline (no Docker images to remove)."""1162 pass1163 1164 1165# Type aliases for API compatibility with observation_builder and other consumers1166PipelineResult = SimulatedPipelineResult1167