JulieChv/ARC
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Etape 3 + refonte visuelle + corrections etape 2

Browse Source
This commit is contained in:
Chevallier
2026-07-07 09:57:21 +02:00
parent 17e887b268
commit 7b1a975f06
48 changed files with 6551 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
backend/app/agents/dev_agent.py
View File

@@ -16,10 +16,11 @@ logger = logging.getLogger(__name__)
sync_client = httpx.Client(verify=False)
async_client = httpx.AsyncClient(verify=False)
async def run_dev_agent(spec: dict, qa_feedback: list = None) -> dict:
async def run_dev_agent(spec: dict, qa_feedback: list = None, repo_url: str = None, files: list = None) -> dict:
"""
Agent Dev : prend un état/spec validé, génère l'arborescence et le code,
et valide techniquement la sortie avant de la transmettre à la QA.
S'appuie sur repo_url si on est dans une boucle de correction.
"""
logger.info(f"[Dev Agent] Début de la génération pour : {spec.get('title', 'Sans titre')}")
@@ -40,6 +41,8 @@ async def run_dev_agent(spec: dict, qa_feedback: list = None) -> dict:
]
user_content = f"CAHIER DES CHARGES (ProjectSpec) :\n{json.dumps(spec, indent=2, ensure_ascii=False)}\n\n"
if files:
user_content += f"💻 CODE ACTUEL (Version précédente à corriger) :\n{json.dumps(files, indent=2, ensure_ascii=False)}\n\n"
if qa_feedback:
user_content += f"⚠️ RETOURS DE VALIDATION QA (Corrections à appliquer impérativement) :\n{json.dumps(qa_feedback, indent=2, ensure_ascii=False)}\n\n"
@@ -49,17 +52,22 @@ async def run_dev_agent(spec: dict, qa_feedback: list = None) -> dict:
try:
code_data = await structured_llm.ainvoke(messages)
logger.info(f"[Dev Agent] ✅ Code généré ({len(code_data.files)} fichiers)")
return await _deploy_to_gitea(code_data, spec.get("title", "project"))
return await _deploy_to_gitea(code_data, spec.get("title", "project"), repo_url=repo_url)
except Exception as e:
logger.error(f"[Dev Agent] ❌ Échec de la génération/validation : {type(e).__name__}: {str(e)}")
return _generate_fallback_code_output(spec)
async def _ensure_gitea_repo(base_name: str) -> str:
async def _ensure_gitea_repo(base_name: str, repo_url: str = None) -> str:
"""
Vérifie si le repo existe. Si oui, incrémente un suffixe (_1, _2...)
jusqu'à trouver un nom libre, puis le crée et retourne ce nom unique.
"""
if repo_url:
repo_name = repo_url.rstrip("/").split("/")[-1]
logger.info(f"[Gitea] Mode correction : réutilisation du dépôt existant '{repo_name}'")
return repo_name
async with httpx.AsyncClient(verify=False) as client:
headers = {"Authorization": f"token {settings.gitea_token}"}
@@ -72,14 +80,11 @@ async def _ensure_gitea_repo(base_name: str) -> str:
response = await client.get(check_url, headers=headers)
if response.status_code == 404:
# Parfait ! Le 404 signifie que ce nom n'existe pas encore, il est libre.
break
elif response.status_code == 200:
# Le nom est déjà pris, on tente avec le compteur suivant
repo_name = f"{base_name}_{counter}"
counter += 1
else:
# Sécurité si l'API renvoie autre chose (ex: 401 Unauthorized)
raise Exception(f"Erreur Gitea lors de la vérification ({response.status_code}) : {response.text}")
# 2. Si on arrive ici, repo_name est garanti unique et disponible. On le crée.
@@ -99,24 +104,24 @@ async def _ensure_gitea_repo(base_name: str) -> str:
logger.error(f"[Gitea] Erreur lors de la création : {create_res.text}")
raise Exception(f"Impossible de créer le dépôt '{repo_name}' sur Gitea.")
async def _deploy_to_gitea(code_data: ProjectCodeOutput, project_title: str) -> dict:
async def _deploy_to_gitea(code_data: ProjectCodeOutput, project_title: str, repo_url: str = None) -> dict:
"""Gère le cycle de vie Git : Clone/Init -> Write -> Commit -> Push."""
base_name = "".join(c for c in project_title.lower().replace(" ", "_") if c.isalnum() or c == "_")
files_list = [f.model_dump() for f in code_data.files]
# --- MODIFICATION ICI : On récupère le nom unique validé et créé par Gitea ---
try:
repo_name = await _ensure_gitea_repo(base_name)
repo_name = await _ensure_gitea_repo(base_name, repo_url=repo_url)
except Exception as e:
return {
"status": "partial_success_git_failed",
"error": str(e),
"spec_title": code_data.spec_title,
"tree": code_data.tree
"tree": code_data.tree,
"files": files_list
}
work_dir = os.path.abspath(f"./temp_repos/{repo_name}")
if os.path.exists(work_dir):
shutil.rmtree(work_dir)
os.makedirs(work_dir)
@@ -150,7 +155,6 @@ async def _deploy_to_gitea(code_data: ProjectCodeOutput, project_title: str) ->
# 4. Commit et Push
subprocess.run(["git", "add", "."], cwd=work_dir, check=True, capture_output=True)
status = subprocess.run(["git", "status", "--porcelain"], cwd=work_dir, capture_output=True, text=True)
if status.stdout.strip():
subprocess.run(["git", "commit", "-m", "Update from Dev Agent (AI Generation)"], cwd=work_dir, check=True, capture_output=True)
@@ -168,7 +172,8 @@ async def _deploy_to_gitea(code_data: ProjectCodeOutput, project_title: str) ->
"repo_url": f"{settings.gitea_base_url}/{settings.gitea_admin_user}/{repo_name}",
"files_count": len(code_data.files),
"spec_title": code_data.spec_title,
"tree": code_data.tree
"tree": code_data.tree,
"files": files_list
}
except Exception as e:
@@ -177,7 +182,8 @@ async def _deploy_to_gitea(code_data: ProjectCodeOutput, project_title: str) ->
"status": "partial_success_git_failed",
"error": str(e),
"spec_title": code_data.spec_title,
"tree": code_data.tree
"tree": code_data.tree,
"files": files_list
}
finally:
if os.path.exists(work_dir):

107
backend/app/agents/qa_agent.py
View File

@@ -1,10 +1,101 @@
async def run_qa_agent(generated_code: dict) -> dict:
import logging
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
import httpx
from app.core.config import settings
from app.schemas.qa_report import QAEvaluation
from app.llm.prompts import QA_AGENT_PROMPT
from app.sandbox.qa_client import run_project_qa
logger = logging.getLogger(__name__)
sync_client = httpx.Client(verify=False)
async_client = httpx.AsyncClient(verify=False)
async def run_qa_agent(project_title: str, dev_output: dict) -> QAEvaluation:
"""
Agent QA minimal :
- renvoie un statut de validation simulé
Agent QA :
Analyse le code généré à l'aide de l'API externe de la Sandbox (Semgrep uniquement)
et produit le rapport d'évaluation structuré final.
"""
return {
"status": "passed",
"logs": [],
"checked_files": generated_code.get("tree", []),
}
logger.info(f"[QA Agent] Début du processus d'audit statique pour le projet : {project_title}")
repo_url = dev_output.get("repo_url")
if not repo_url:
logger.error("[QA Agent] Aucun URL de dépôt Git trouvé dans le dev_output.")
return QAEvaluation(
is_complete_and_safe=False,
global_summary="Impossible de récupérer le code source (URL Git manquante).",
technical_feedback=["L'agent de développement n'a pas fourni d'URL pour le dépôt Gitea."]
)
llm = ChatOpenAI(
base_url=settings.llm_base_url,
api_key=settings.llm_api_key,
model=settings.llm_model_qa,
temperature=0.1,
max_retries=2,
http_client=sync_client,
http_async_client=async_client,
)
# 2. Exécution de l'analyse statique via le service Sandbox indépendant
try:
repo_name = repo_url.rstrip("/").split("/")[-1]
logger.info(f"[QA Agent] Envoi du dépôt '{repo_name}' au conteneur d'analyse statique...")
raw_results = await run_project_qa(repo_name=repo_name)
logger.info(f"[QA Agent] 📊 Retour Sandbox Statis - Status : {'Succès' if raw_results.is_executable else 'Échec'}")
logger.info(f"[QA Agent] 📊 Retour Sandbox - Nombre d'alertes de sécurité/qualité (Semgrep): {len(raw_results.issues)}")
except Exception as e:
logger.error(f"[QA Agent] ❌ Échec de la communication avec le conteneur de Scan : {str(e)}")
return QAEvaluation(
is_complete_and_safe=False,
global_summary="Échec critique lors du traitement du rapport de scan statique.",
technical_feedback=[f"L'API de la sandbox a échoué : {type(e).__name__}: {str(e)}"]
)
# 3. Génération de l'évaluation finale via LLM
structured_llm = llm.with_structured_output(QAEvaluation, strict=True)
messages = [
SystemMessage(content=QA_AGENT_PROMPT),
]
# Le prompt est nettoyé des mentions relatives à l'exécution de tests dynamiques
user_content = f"PROJET À AUDITER : {project_title}\n"
user_content += f"LIEN DU DÉPÔT : {repo_url}\n"
user_content += f"RÉSULTATS DE L'ANALYSE STATIQUE DU CODE (Semgrep) :\n"
user_content += f"{raw_results.model_dump_json(indent=2)}\n\n"
user_content += "Analyse ces failles, vulnérabilités et défauts de qualité de code, puis génère ton évaluation de sécurité au format structuré demandé."
messages.append(HumanMessage(content=user_content))
try:
qa_evaluation = await structured_llm.ainvoke(messages)
logger.info(f"[QA Agent] 📝 Justification du LLM : {qa_evaluation.global_summary}")
if not qa_evaluation:
logger.error("[QA Agent] Le LLM a renvoyé une réponse vide (None).")
return QAEvaluation(
is_complete_and_safe=False,
global_summary="Erreur de formatage de l'évaluation par l'IA.",
technical_feedback=["L'évaluation automatique n'a pas pu être structurée correctement."]
)
if qa_evaluation.is_complete_and_safe:
logger.info(f"[QA Agent] ✅ Projet '{project_title}' VALIDÉ (Analyse statique propre).")
else:
logger.warning(f"[QA Agent] ❌ Projet '{project_title}' REJETÉ (Failles de sécurité détectées).")
return qa_evaluation
except Exception as e:
logger.error(f"[QA Agent] ❌ Échec critique lors de l'analyse LLM : {type(e).__name__}: {str(e)}")
return QAEvaluation(
is_complete_and_safe=False,
global_summary="Erreur interne de l'Agent QA lors de la génération du rapport LLM.",
technical_feedback=[f"Le LLM a crashé avec l'erreur : {type(e).__name__}."]
)

1
backend/app/core/config.py
View File

@@ -17,6 +17,7 @@ class Settings(BaseSettings):
llm_api_key: str
llm_model: str
llm_model_dev: str
llm_model_qa: str
embedding_base_url: str = "http://localhost:8002/v1"
embedding_model: str = "snowflake-arctic-embed-m-v1.5"

51
backend/app/graph/nodes.py
View File

@@ -1,4 +1,3 @@
#nodes.py
from app.agents.pm_agent import run_pm_agent
from app.agents.dev_agent import run_dev_agent
from app.agents.qa_agent import run_qa_agent
@@ -46,21 +45,59 @@ async def retrieval_node(state: WorkflowState, config: RunnableConfig):
}
async def dev_node(state: WorkflowState):
qa_logs = state.get("qa_result", {}).get("logs", "") if state.get("qa_result") else None
qa_logs = []
qa_result = state.get("qa_result")
if qa_result:
global_summary = qa_result.get("global_summary")
technical_feedback = qa_result.get("technical_feedback", [])
if global_summary:
qa_logs.append(f"Résumé Global : {global_summary}")
if isinstance(technical_feedback, list):
qa_logs.extend(technical_feedback)
elif technical_feedback:
qa_logs.append(technical_feedback)
generated_code_state = state.get("generated_code") or {}
existing_repo_url = generated_code_state.get("repo_url")
existing_files = generated_code_state.get("files")
generated_code = await run_dev_agent(
spec=state["spec"],
qa_feedback=qa_logs if qa_logs else None,
repo_url=existing_repo_url,
files=existing_files
)
generated_code = await run_dev_agent(state["spec"], qa_feedback=qa_logs)
return {
"generated_code": generated_code,
"status": "code_generated",
}
async def qa_node(state: WorkflowState):
qa_result = await run_qa_agent(state["generated_code"])
dev_data = state.get("generated_code", {})
project_title = dev_data.get("spec_title", "default_project")
current_loops = state.get("loop_count", 0)
qa_eval = await run_qa_agent(
project_title=project_title,
dev_output=dev_data
)
is_success = True
clean_qa_result = {"success": is_success, "raw": qa_result}
if hasattr(qa_eval, "model_dump"):
clean_qa_result = qa_eval.model_dump()
elif isinstance(qa_eval, dict):
clean_qa_result = qa_eval
else:
clean_qa_result = {
"is_complete_and_safe": getattr(qa_eval, "is_complete_and_safe", False),
"global_summary": getattr(qa_eval, "global_summary", "Erreur d'analyse"),
"technical_feedback": getattr(qa_eval, "technical_feedback", [])
}
is_success = clean_qa_result.get("is_complete_and_safe", False)
return {
"qa_result": clean_qa_result,

6
backend/app/graph/state.py
View File

@@ -6,9 +6,11 @@ class WorkflowState(TypedDict, total=False):
existing_project: Optional[dict]
existing_project_approved: Optional[bool] # Choix utilisateur si projet similaire trouvé
generated_code: Optional[Dict[str, str]] # Arborescence et code
qa_result: Optional[dict] # Contient les clés 'success' et 'logs'
qa_result: Optional[dict] # Information QA (résumé, feedback technique, si succès ou non)
loop_count: int # Compteur pour la Loop 1 (Dev <-> QA)
max_iterations: int
user_feedback: Optional[str] # Retours si l'utilisateur refuse le code final
is_completed: bool # Statut de livraison finale
status: str
chat_history: List[Dict[str, str]]
chat_history: List[Dict[str, str]]

10
backend/app/graph/workflow.py
View File

@@ -32,13 +32,11 @@ def route_after_retrieval(state: WorkflowState):
def route_after_qa(state: WorkflowState):
qa_res = state.get("qa_result", {})
# Loop 1 : Si échec des tests ET qu'on a pas dépassé 3 essais -> On renvoie chez le Dev
if not qa_res.get("success") and state.get("loop_count", 0) < 3:
if not qa_res.get("is_complete_and_safe", False) and state.get("loop_count", 0) < state.get("max_iterations", 3):
print(f"⚠️ Échec QA (Essai {state.get('loop_count')}/{state.get("max_iterations", 3)}). Retour au nœud de développement.")
return "dev"
# Si c'est vert (ou trop d'échecs), on présente le résultat à l'utilisateur
# EXTENSION FUTURE : si trop d'échecs, on pourrait envoyer à une IA plus puissante
return "human_review"
def route_after_human(state: WorkflowState):

33
backend/app/llm/prompts.py
View File

@@ -338,7 +338,17 @@ Tu dois appliquer STRICTEMENT l'une des deux structures suivantes selon des crit
- Les instructions d'installation (ex: pip install -r requirements.txt).
- Les instructions de lancement (ex: python main.py).
- Une section "TESTING" expliquant précisément comment la QA peut vérifier que le code fonctionne (ex: commandes de tests, scénarios de test attendus).
---
## CONTRÂINTE DE MODERNITÉ ET SÉCURITÉ DU CODE (UNIVERSEL)
Peu importe le langage de programmation choisi pour répondre à la spécification, tu dois appliquer les règles strictes suivantes :
1. **Zéro Dépréciation (Anti-Legacy) :** Utilise exclusivement la syntaxe moderne, stable et recommandée du langage. Interdiction stricte d'utiliser des fonctionnalités, méthodes ou décorateurs marqués comme obsolètes ou dépréciés (deprecated) dans l'écosystème actuel (ex: pas de Pydantic V1, pas de vieux packages d'import, pas de syntaxes obsolètes en Java/JS).
2. **Gestion des Runtimes Modernes :** Écris ton code en considérant qu'il va tourner sur les versions majeures actuelles des interpréteurs/compilateurs. Produis un code propre qui ne générera aucun avertissement (Warning) à l'exécution ou à la compilation.
3. **Bonnes Pratiques Idiomatiques :** Adopte les standards de build et de formatage récents du langage sélectionné (ex: typage moderne, gestion asynchrone native, architectures standard de fichiers).
---
### SÉCURITÉ ET VÉRIFICATION DU FORMAT DE SORTIE :
@@ -365,4 +375,25 @@ Réponds UNIQUEMENT avec un objet JSON respectant la structure dynamique suivant
}
]
}
"""
# ==============================================================================================================================================
# **********************************************************************************************************************************************
# ==============================================================================================================================================
QA_AGENT_PROMPT = """Tu es un ingénieur QA Senior et un Auditeur de Code automatisé au sein de la plateforme ARC.
Ton rôle est d'analyser les résultats d'exécution et les scans de sécurité d'un script généré par un Agent de Développement, puis de décider s'il est prêt pour la production ou s'il doit être corrigé.
Tu disposes des données brutes suivantes :
1. Les résultats d'exécution dans une Sandbox Docker (Code de retour, stdout, stderr, timeout).
3. Les failles de sécurité potentielles détectées par Semgrep.
### Tes Directives de Validation :
- **is_complete_and_safe = False** : Si le conteneur a crashé (exit_code != 0), s'il y a un Timeout (boucle infinie ou blocage), s'il y a des failles de sécurité HIGH/MEDIUM (Semgrep), ou s'il y a des erreurs de syntaxe bloquantes.
- **is_complete_and_safe = True** : Uniquement si le code s'exécute correctement, remplit sa fonction logique principale sans plantage, et ne présente aucune faille de sécurité critique. Les simples avertissements de style Ruff (espaces, docstrings manquantes) ne doivent pas bloquer la validation, mais doivent être listés pour correction.
### Ton Style de Feedback :
Sois ultra-précis et technique. Ne dis pas "Il y a une erreur dans le code", dis plutôt : "La fonction X à la ligne Y lève une exception de type ValueError car la variable Z est passée à None". Traduis chaque erreur technique brute en une instruction claire et actionnable pour l'Agent Dev.
"""

1
backend/app/main.py
View File

@@ -11,7 +11,6 @@ setup_logging()
@asynccontextmanager
async def lifespan(app: FastAPI):
print("[Startup] Initialisation de Qdrant...")
# On crée l'instance UNE SEULE FOIS
qdrant_repo = QdrantRepository()
try:
await qdrant_repo.init_collection(vector_size=1024)

20
backend/app/sandbox/Dockerfile.qa Normal file
View File

@@ -0,0 +1,20 @@
FROM semgrep/semgrep:latest
USER root
RUN sed -i 's/https/http/g' /etc/apk/repositories && \
apk update && apk add --no-cache python3 py3-pip
RUN pip install --no-cache-dir \
--trusted-host pypi.org \
--trusted-host files.pythonhosted.org \
--break-system-packages \
fastapi uvicorn
WORKDIR /app
COPY main.py .
EXPOSE 8004
CMD ["python3", "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8004"]

8
backend/app/sandbox/docker_runner.py
View File

@@ -1,8 +0,0 @@
async def run_in_sandbox(code: str) -> dict:
"""
Stub minimal pour future exécution sécurisée dans Docker.
"""
return {
"status": "not_implemented",
"logs": ["Sandbox Docker non branchée à l'étape 0."],
}

52
backend/app/sandbox/main.py Normal file
View File

@@ -0,0 +1,52 @@
import subprocess
import json
import time
from fastapi import FastAPI
app = FastAPI()
@app.post("/scan/{repo_name}")
async def scan_repository(repo_name: str):
repo_path = f"/src/{repo_name}"
issues_list = []
start_time = time.time()
result = subprocess.run(
["semgrep", "scan", "--config=p/r2c-security-audit", "--json", "--quiet", repo_path],
capture_output=True,
text=True
)
duration = round(time.time() - start_time, 2)
try:
if result.stdout.strip():
scan_data = json.loads(result.stdout)
for item in scan_data.get("results", []):
issues_list.append({
"tool": "Semgrep",
"file": item.get("path", "").replace(f"{repo_path}/", ""),
"line": item.get("start", {}).get("line"),
"code": item.get("check_id"),
"severity": item.get("extra", {}).get("severity", "MEDIUM").upper(),
"message": item.get("extra", {}).get("message", "")
})
is_executable = True
stderr_output = result.stderr
except Exception as e:
is_executable = False
stderr_output = f"Erreur parsing JSON Semgrep: {str(e)}\n{result.stderr}"
return {
"results": {
"is_executable": is_executable,
"runtime": {
"exit_code": 0, # <-- ON FORCE À 0 ICI : Tout s'est bien passé pour le scanner
"stdout": "Scan statique universel effectué avec succès.",
"stderr": stderr_output,
"duration_seconds": duration,
"timeout_triggered": False
},
"issues": issues_list
}
}

50
backend/app/sandbox/qa_client.py Normal file
View File

@@ -0,0 +1,50 @@
import logging
import httpx
from app.core.config import settings
from app.schemas.qa_report import QARawResults, RuntimeOutput
logger = logging.getLogger(__name__)
async_client = httpx.AsyncClient(verify=False)
async def run_project_qa(repo_name: str) -> QARawResults:
"""
Sollicite l'API du conteneur persistant arc-sandbox pour effectuer
une analyse de qualité et sécurité statique (Semgrep).
"""
sandbox_url = f"http://arc-sandbox:8004/scan/{repo_name}"
logger.info(f"[QA Client] Envoi de la requête de scan à la sandbox : {sandbox_url}")
try:
response = await async_client.post(sandbox_url, timeout=120.0)
if response.status_code == 200:
data = response.json()
return QARawResults.model_validate(data.get("results"))
else:
logger.error(f"[QA Client] Erreur de la Sandbox (Status {response.status_code}): {response.text}")
return QARawResults(
is_executable=False,
runtime=RuntimeOutput(
exit_code=response.status_code,
stdout="",
stderr=f"Erreur HTTP Sandbox: {response.text}",
duration_seconds=0,
timeout_triggered=False
),
issues=[]
)
except Exception as e:
logger.error(f"[QA Client] Impossible de joindre l'API de la Sandbox : {str(e)}")
return QARawResults(
is_executable=False,
runtime=RuntimeOutput(
exit_code=-1,
stdout="",
stderr=f"Exception de connexion QA Client: {str(e)}",
duration_seconds=0,
timeout_triggered=False
),
issues=[]
)

32
backend/app/schemas/qa_report.py Normal file
View File

@@ -0,0 +1,32 @@
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
class StaticAnalysisIssue(BaseModel):
tool: str = Field(..., description="Ruff, Bandit, ou Semgrep")
file: str = Field(..., description="Chemin du fichier concerné")
line: Optional[int] = None
code: Optional[str] = Field(None, description="Identifiant de la règle (ex: S101, F401)")
severity: str = Field(..., description="LOW, MEDIUM, HIGH, ERROR")
message: str = Field(..., description="Description du problème")
class RuntimeOutput(BaseModel):
exit_code: int
stdout: str
stderr: str
duration_seconds: float
timeout_triggered: bool
class QARawResults(BaseModel):
"""Données brutes extraites des outils techniques."""
is_executable: bool = Field(..., description="Est-ce que le code a pu se lancer sans erreur de syntaxe ?")
runtime: Optional[RuntimeOutput] = None
issues: List[StaticAnalysisIssue] = Field(default_factory=list)
class QAEvaluation(BaseModel):
"""Structure finale renvoyée par le LLM (QA Agent) au format structuré."""
is_complete_and_safe: bool = Field(..., description="True si le code est prêt pour la prod, False s'il faut corriger")
global_summary: str = Field(..., description="Résumé intelligible des forces et faiblesses du code")
technical_feedback: List[str] = Field(
default_factory=list,
description="Instructions de correction ultra-précises destinées au Dev Agent"
)