import json
from typing import List, Dict, Any, Tuple, Optional, Callable
from sqlalchemy.orm import Session
PROMPT_TEMPLATE = """
### ЪДа Core Directives
### **Code Review Directives**
Your role is a specialized code review AI. Your primary task is to review a set of code changes and confirm they **fully and accurately address the user's original request**.
---
### **Critical Constraints**
Your review is strictly limited to **code content completeness**. Do not suggest or perform any file splits, moves, or large-scale refactoring.
Identify and resolve any missing logic, placeholders (like "unchanged," "same as original," "to-do"), or incomplete code.
Return exactly one JSON object:
{{
"reasoning": "A detailed explanation of why the decision was made.",
"decision": "Either 'complete' or 'modify'.",
"answer": "If 'complete', an empty string. If 'modify', the new execution plan instructions in JSON."
}}
Input:
- `original_question`: {original_question}
- `execution_plan`: {execution_plan}
- `final_code_changes`: {final_code_changes}
- `original_files`: {original_files}
"""
class CodeReviewer:
def __init__(self):
pass
async def forward(
self,
original_question: str,
execution_plan: str,
final_code_changes: List[Dict[str, Any]],
original_files: List[Dict[str, Any]],
llm_provider = None,
prompt_service = None,
db: Optional[Session] = None,
user_id: Optional[str] = None,
prompt_slug: str = "code-reviewer"
) -> Tuple[str, str, str]:
if not llm_provider:
raise ValueError("LLM Provider is required.")
final_code_changes_json = json.dumps(final_code_changes)
original_files_json = json.dumps(original_files)
template = PROMPT_TEMPLATE
if prompt_service and db and user_id:
db_prompt = prompt_service.get_prompt_by_slug(db, prompt_slug, user_id)
if db_prompt:
template = db_prompt.content
prompt = template.format(
original_question=original_question,
execution_plan=execution_plan,
final_code_changes=final_code_changes_json,
original_files=original_files_json
)
response = await llm_provider.acompletion(prompt=prompt, response_format={"type": "json_object"})
content = response.choices[0].message.content
try:
data = json.loads(content)
decision = data.get("decision", "modify")
reasoning = data.get("reasoning", "")
answer = data.get("answer", "")
if isinstance(answer, list):
answer = json.dumps(answer)
return decision, reasoning, answer
except json.JSONDecodeError:
return "modify", f"Failed to parse JSON: {content}", ""