backend/modules/impact/risk_engine.py

"""
Change Risk Intelligence Engine: IBM Bob predicts blast radius of code changes.
"""

from __future__ import annotations

import json
import re

from core.watsonx import watsonx_generate
from utils.cache import repo_cache
from utils.context_builder import build_impact_context

RISK_SCORES = {"CRITICAL": 95, "HIGH": 75, "MEDIUM": 50, "LOW": 20}


async def analyze_change_impact(repo_id: str, target_file: str) -> dict:
    repo_data = repo_cache.get(repo_id)
    if not repo_data:
        raise ValueError("Repository not loaded")

    context = build_impact_context(repo_data, target_file)
    prompt = f"""You are an expert software risk analyst performing blast radius analysis.

REPOSITORY INTELLIGENCE:
{context}

TARGET FILE FOR CHANGE: {target_file}

Analyze what would break if this file is modified. Return ONLY valid JSON:
{{
  "risk_level": "CRITICAL|HIGH|MEDIUM|LOW",
  "risk_score": 0,
  "summary": "One sentence summary of the risk",
  "impacted_services": [
    {{
      "name": "Service/module name",
      "impact_type": "direct|indirect|potential",
      "description": "How it is impacted",
      "severity": "high|medium|low"
    }}
  ],
  "impacted_apis": ["List of API endpoints that could break"],
  "affected_workflows": ["User workflows that could be disrupted"],
  "failure_scenarios": [
    {{
      "scenario": "Failure scenario description",
      "probability": "high|medium|low",
      "user_impact": "How users would experience this failure"
    }}
  ],
  "recommended_tests": ["Specific test to run before deploying changes"],
  "deployment_checklist": ["Step to take before deploying"],
  "safe_to_modify": false,
  "modification_guidance": "Specific guidance on how to safely modify this file"
}}
###END###"""

    response = await watsonx_generate(
        prompt,
        max_tokens=2000,
        use_code_model=True,
        working_dir=repo_data.get("repo_path"),
    )
    json_match = re.search(r"\{.*\}", response, re.DOTALL)
    if json_match:
        try:
            result = json.loads(json_match.group())
            if not result.get("risk_score"):
                result["risk_score"] = RISK_SCORES.get(result.get("risk_level", "LOW"), 20)
            return result
        except json.JSONDecodeError:
            pass
    return {"error": "Risk analysis failed", "raw": response, "risk_level": "UNKNOWN", "risk_score": 0}