"""Challenge engine — analyzes user-submitted flaky tests via pattern matching and optional LLM.""" from __future__ import annotations import ast import json import logging import os import re from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from .api_models import ChallengeAnalysis try: from dotenv import load_dotenv load_dotenv(Path(__file__).resolve().parents[1] / ".env", override=False) except ImportError: pass try: from models import ROOT_CAUSE_TYPES except ImportError: try: from FlakeForge.models import ROOT_CAUSE_TYPES # type: ignore except ImportError: ROOT_CAUSE_TYPES = [ # minimal fallback; keep in sync with models.py "async_wait", "concurrency", "test_order_dependency", "resource_leak", "shared_state", "network", "platform_dependency", "nondeterminism", "import_side_effect", "module_cache_pollution", "fixture_scope_leak", "mock_residue", "unknown", ] logger = logging.getLogger(__name__) def _get_hf_api_token() -> str: """Token for Hugging Face Inference Router. Tries several common names (Space secrets / .env).""" for key in ( "HF_TOKEN", "HUGGING_FACE_TOKEN", "HUGGINGFACE_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACEHUB_API_TOKEN", # huggingface_hub "HF_HUB_TOKEN", ): v = os.environ.get(key, "").strip() if v: return v return "" def has_challenge_llm_token() -> bool: return bool(_get_hf_api_token()) # Pattern detectors: each returns (category, confidence, explanation, suggested_fix) _PATTERN_DETECTORS: List = [] def _register(fn): _PATTERN_DETECTORS.append(fn) return fn @_register def _detect_timing_race(code: str, test_code: str) -> Optional[Tuple[str, float, str, str]]: indicators = ["threading.Thread", "thread", "Thread(", "Lock(", "global "] race_signals = ["global ", "temp =", "counter", "+= 1", "counter ="] thread_count = sum(1 for i in indicators if i in code or i in test_code) race_count = sum(1 for s in race_signals if s in code) if thread_count >= 2 and race_count >= 2: return ( "concurrency", 0.92, "Non-atomic read-modify-write detected in threaded context. " "Multiple threads access shared state without synchronization.", "Wrap the critical section with threading.Lock() to make the operation atomic.", ) if thread_count >= 1 and race_count >= 1: return ( "concurrency", 0.75, "Shared mutable state accessed from threads without explicit locking.", "Add threading.Lock() around shared state access.", ) return None @_register def _detect_async_wait(code: str, test_code: str) -> Optional[Tuple[str, float, str, str]]: combined = code + test_code if "asyncio" in combined and ("timeout" in combined or "wait_for" in combined): return ( "async_wait", 0.88, "Async operation with tight timeout detected. Under load, the event loop " "may not schedule the coroutine in time.", "Increase timeout or use asyncio.Lock() for proper async synchronization.", ) if "await" in combined and ("gather" in combined or "create_task" in combined): if "session" in combined.lower() or "lock" in combined.lower(): return ( "async_wait", 0.82, "Concurrent async tasks sharing a session or resource without async locking.", "Use asyncio.Lock() to serialize access, or create separate sessions per task.", ) return None @_register def _detect_db_commit(code: str, test_code: str) -> Optional[Tuple[str, float, str, str]]: combined = code + test_code has_db = any(k in combined for k in ["sqlite3", "connect(", "execute(", "cursor"]) has_insert = "INSERT" in combined or "insert" in combined missing_commit = "commit()" not in combined if has_db and has_insert and missing_commit: return ( "resource_leak", 0.95, "Database write without explicit commit(). Transaction may not be flushed " "before the read query, causing intermittent data loss.", "Add conn.commit() after the INSERT to ensure data is persisted before reading.", ) return None @_register def _detect_external_dep(code: str, test_code: str) -> Optional[Tuple[str, float, str, str]]: combined = code + test_code external_signals = ["requests.post", "requests.get", "httpx", "urllib", "sandbox", "api."] hits = sum(1 for s in external_signals if s in combined) if hits >= 1: return ( "network", 0.85, "Test depends on an external HTTP endpoint. Network latency, DNS resolution, " "and endpoint availability introduce non-determinism.", "Mock the external dependency using unittest.mock.patch or responses library.", ) return None @_register def _detect_shared_state(code: str, test_code: str) -> Optional[Tuple[str, float, str, str]]: combined = code + test_code global_count = combined.count("global ") class_var_pattern = re.findall(r"class\s+\w+.*?:\s*\n\s+\w+\s*=", combined, re.DOTALL) if global_count >= 2 or len(class_var_pattern) >= 1: if "clear()" not in combined and "reset" not in combined.lower(): return ( "shared_state", 0.78, "Mutable global or class-level state is shared across test runs without cleanup.", "Reset shared state in a fixture teardown or use test-local copies.", ) return None @_register def _detect_nondeterminism(code: str, test_code: str) -> Optional[Tuple[str, float, str, str]]: combined = code + test_code nd_signals = ["random.", "time.time()", "datetime.now()", "uuid.", "shuffle("] hits = sum(1 for s in nd_signals if s in combined) if hits >= 1: return ( "nondeterminism", 0.80, "Test relies on non-deterministic values (random, time, UUID) without seeding.", "Seed random generators or mock time/uuid to produce deterministic results.", ) return None @_register def _detect_fixture_scope(code: str, test_code: str) -> Optional[Tuple[str, float, str, str]]: combined = code + test_code if "scope=" in combined and ("session" in combined or "module" in combined): if "yield" in combined: return ( "fixture_scope_leak", 0.82, "Session/module-scoped fixture yields mutable state that may leak across tests.", "Use function-scoped fixtures or deep-copy the yielded value.", ) return None @_register def _detect_mock_residue(code: str, test_code: str) -> Optional[Tuple[str, float, str, str]]: combined = code + test_code has_patch = "patch(" in combined or "monkeypatch" in combined has_cleanup = "stop()" in combined or "with " in combined if has_patch and not has_cleanup: return ( "mock_residue", 0.80, "Mock/monkeypatch applied without proper teardown. Patched state leaks to subsequent tests.", "Use context manager (with patch(...)) or ensure .stop() is called in teardown.", ) return None def _extract_function_name(code: str) -> str: """Extract the first function name from code.""" try: tree = ast.parse(code) for node in ast.walk(tree): if isinstance(node, ast.FunctionDef): return node.name except Exception: pass m = re.search(r"def\s+(\w+)", code) return m.group(1) if m else "" def _extract_file_hint(code: str) -> str: """Guess a filename from imports or class names.""" m = re.search(r"from\s+([\w.]+)\s+import", code) if m: parts = m.group(1).split(".") return parts[-1] + ".py" m = re.search(r"import\s+([\w.]+)", code) if m: parts = m.group(1).split(".") return parts[-1] + ".py" return "source.py" def _build_causal_chain(code: str, category: str) -> List[str]: """Build a simple causal chain from the code and detected category.""" chain = [] func = _extract_function_name(code) file_hint = _extract_file_hint(code) chain.append(f"test entry -> {func or 'target function'}") if category == "concurrency": chain.append(f"{func} -> Thread/Process spawn") chain.append("Thread -> shared state access (non-atomic)") chain.append("shared state -> RACE CONDITION") elif category == "async_wait": chain.append(f"{func} -> async task/gather") chain.append("async task -> shared resource contention") chain.append("contention -> TIMEOUT/DEADLOCK") elif category == "resource_leak": chain.append(f"{func} -> resource open (db/file/socket)") chain.append("resource -> missing cleanup/commit") chain.append("missing cleanup -> STALE STATE") elif category == "network": chain.append(f"{func} -> external HTTP call") chain.append("HTTP call -> network/endpoint variability") chain.append("variability -> NON-DETERMINISTIC RESPONSE") elif category == "shared_state": chain.append(f"{func} -> global/class state mutation") chain.append("mutation -> cross-test contamination") chain.append("contamination -> ORDER-DEPENDENT FAILURE") else: chain.append(f"{func} -> non-determinism source") chain.append("non-determinism -> FLAKY OUTCOME") return chain def _generate_patch_diff(code: str, category: str, func_name: str) -> str: """Generate a representative patch diff for the detected issue.""" if category == "concurrency": return ( f"--- {_extract_file_hint(code)}\n" "<<<<<<< SEARCH\n" f" global counter\n" "=======\n" " _lock = threading.Lock()\n" " with _lock:\n" ">>>>>>> REPLACE" ) elif category == "async_wait": return ( f"--- {_extract_file_hint(code)}\n" "<<<<<<< SEARCH\n" " timeout=0.5\n" "=======\n" " timeout=5.0\n" ">>>>>>> REPLACE" ) elif category == "resource_leak": return ( f"--- {_extract_file_hint(code)}\n" "<<<<<<< SEARCH\n" " conn.execute('INSERT INTO t VALUES (42)')\n" "=======\n" " conn.execute('INSERT INTO t VALUES (42)')\n" " conn.commit()\n" ">>>>>>> REPLACE" ) elif category == "network": return ( f"--- {_extract_file_hint(code)}\n" "<<<<<<< SEARCH\n" f" r = requests.post(url, json=payload)\n" "=======\n" " from unittest.mock import patch, MagicMock\n" " mock_resp = MagicMock(status_code=200)\n" " with patch('requests.post', return_value=mock_resp):\n" f" r = requests.post(url, json=payload)\n" ">>>>>>> REPLACE" ) return "" def _extract_json_object(text: str) -> Dict[str, Any]: """Parse a JSON object from model output (handles optional ```json fences).""" text = text.strip() if not text: raise ValueError("empty response") if "```" in text: for part in text.split("```"): part = part.strip() if part.lower().startswith("json"): part = part[4:].lstrip() if part.startswith("{"): return json.loads(part) start, end = text.find("{"), text.rfind("}") if start >= 0 and end > start: return json.loads(text[start : end + 1]) raise ValueError("no JSON object in model output") def _llm_dict_to_analysis(data: Dict[str, Any]) -> ChallengeAnalysis: """Map LLM JSON into ChallengeAnalysis with safe bounds.""" cat = str(data.get("detected_category") or "unknown").strip() or "unknown" if cat not in ROOT_CAUSE_TYPES: cat = "unknown" try: conf = float(data.get("confidence", 0)) except (TypeError, ValueError): conf = 0.0 conf = max(0.0, min(1.0, conf)) chain = data.get("causal_chain") if not isinstance(chain, list): chain = [] chain = [str(x) for x in chain if x is not None][:32] try: est = float(data.get("estimated_reward", 0)) except (TypeError, ValueError): est = 0.0 est = max(0.0, min(20.0, est)) infra = bool(data.get("infrastructure_sensitive", False)) return ChallengeAnalysis( detected_category=cat, confidence=round(conf, 2), root_cause_file=str(data.get("root_cause_file") or "")[:500], root_cause_function=str(data.get("root_cause_function") or "")[:200], causal_chain=chain, infrastructure_sensitive=infra, suggested_fix=str(data.get("suggested_fix") or "")[:8000], patch_diff=str(data.get("patch_diff") or "")[:8000], explanation=str(data.get("explanation") or "")[:8000], estimated_reward=round(est, 2), ) def _llm_analyze_with_token(code: str, test_code: str, preset: str) -> ChallengeAnalysis: """Call Hugging Face Inference Router. Caller must only invoke when a Hub token is set. Always returns a ChallengeAnalysis (model output, non-JSON raw text, or explicit error). Never heuristics. """ token = _get_hf_api_token() if not token: return ChallengeAnalysis( detected_category="unknown", confidence=0.0, explanation="No API token in environment (HUGGING_FACE_TOKEN / HF_TOKEN).", estimated_reward=0.0, ) try: from openai import OpenAI except ImportError: return ChallengeAnalysis( detected_category="unknown", confidence=0.0, explanation="The `openai` package is not installed on the server. Add it to your image requirements.", estimated_reward=0.0, ) model = os.environ.get( "FF_CHALLENGE_MODEL", "Qwen/Qwen2.5-Coder-7B:featherless-ai", ) try: client = OpenAI( base_url="https://router.huggingface.co/v1", api_key=token, ) allowed = ", ".join(ROOT_CAUSE_TYPES) system = ( "You are FlakeForge, an assistant that diagnoses likely flaky Python tests. " "Reply with ONE JSON object only (no markdown fences). Schema keys: " "detected_category (string, must be one of: " + allowed + "), " "confidence (0-1), root_cause_file, root_cause_function, " "causal_chain (array of short strings), infrastructure_sensitive (boolean), " "suggested_fix, patch_diff (string, optional), explanation, estimated_reward (0-10). " "If input is not code, set detected_category to unknown and explain briefly." ) user_msg = f"preset: {preset or 'none'}\n\n--- code ---\n{code}\n\n--- test_code ---\n{test_code or '(none)'}\n" logger.info("Challenge LLM: calling model=%s", model) completion = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": system}, {"role": "user", "content": user_msg}, ], temperature=0.2, max_tokens=2048, ) raw = (completion.choices[0].message.content or "").strip() except Exception as exc: logger.exception("LLM challenge call failed: %s", exc) return ChallengeAnalysis( detected_category="unknown", confidence=0.0, explanation=f"LLM request failed: {exc}", estimated_reward=0.0, ) if not raw: return ChallengeAnalysis( detected_category="unknown", confidence=0.0, explanation="The model returned an empty response. Try again or check FF_CHALLENGE_MODEL.", estimated_reward=0.0, ) try: payload = _extract_json_object(raw) except (json.JSONDecodeError, ValueError) as e: logger.warning("LLM response JSON parse failed: %s", e) return ChallengeAnalysis( detected_category="unknown", confidence=0.0, explanation=raw[:8000], suggested_fix="", estimated_reward=0.0, ) return _llm_dict_to_analysis(payload) def _heuristic_analyze(code: str, test_code: str) -> ChallengeAnalysis: """Pattern-matching and AST heuristics (no LLM).""" combined_code = code combined_test = test_code best_match: Optional[Tuple[str, float, str, str]] = None for detector in _PATTERN_DETECTORS: result = detector(combined_code, combined_test) if result is not None: if best_match is None or result[1] > best_match[1]: best_match = result if best_match is None: return ChallengeAnalysis( detected_category="unknown", confidence=0.3, explanation="No strong flakiness pattern detected. The code may have subtle " "non-determinism not covered by static analysis. Consider running " "the full FlakeForge episode with chaos probes for deeper analysis.", ) category, confidence, explanation, suggested_fix = best_match func_name = _extract_function_name(code) file_hint = _extract_file_hint(code) causal_chain = _build_causal_chain(code, category) patch_diff = _generate_patch_diff(code, category, func_name) infra_sensitive = category in ("concurrency", "async_wait", "network") estimated_reward = round(confidence * 7.5, 1) return ChallengeAnalysis( detected_category=category, confidence=round(confidence, 2), root_cause_file=file_hint, root_cause_function=func_name, causal_chain=causal_chain, infrastructure_sensitive=infra_sensitive, suggested_fix=suggested_fix, patch_diff=patch_diff, explanation=explanation, estimated_reward=estimated_reward, ) def analyze_challenge(code: str, test_code: str = "", preset: str = "") -> ChallengeAnalysis: """Analyze user-submitted code. If HUGGING_FACE_TOKEN / HF_TOKEN is set: **only** the Hugging Face router LLM is used (no silent fallback to pattern heuristics). On failure, the response explains the error. If no token: pattern heuristics (unless FF_CHALLENGE_ALLOW_HEURISTIC=0). """ if not code.strip() and not test_code.strip(): return ChallengeAnalysis( detected_category="unknown", confidence=0.0, explanation="No code provided for analysis.", ) if _get_hf_api_token(): return _llm_analyze_with_token(code, test_code, preset) allow_heuristic = os.environ.get("FF_CHALLENGE_ALLOW_HEURISTIC", "1").strip().lower() in ( "1", "true", "yes", ) if allow_heuristic: return _heuristic_analyze(code, test_code) return ChallengeAnalysis( detected_category="unknown", confidence=0.0, explanation="Set HUGGING_FACE_TOKEN or HF_TOKEN (Space secret or .env) to enable LLM analysis. " "Heuristic-only mode is disabled (FF_CHALLENGE_ALLOW_HEURISTIC=0).", estimated_reward=0.0, )