from __future__ import annotations from app.modules.agent.code_qa_runtime import CodeQaRuntimeExecutor from app.modules.rag.contracts.enums import RagLayer from app.modules.rag.intent_router_v2 import RepoContext from tests.pipeline_setup_v2.core.models import ExecutionPayload, V2Case class CodeQaEvalAdapter: def __init__(self) -> None: self._executor = CodeQaRuntimeExecutor(llm=None) def execute(self, case: V2Case, rag_session_id: str | None) -> ExecutionPayload: if not rag_session_id: raise ValueError(f"Case '{case.case_id}' requires rag_session_id or repo_path") result = self._executor.execute(user_query=case.query, rag_session_id=rag_session_id) route_result = result.router_result route = result.diagnostics.router_result details = { "case_id": case.case_id, "query": case.query, "rag_session_id": rag_session_id, "actual": { "intent": route.get("intent"), "sub_intent": route.get("sub_intent"), "graph_id": route.get("graph_id"), "conversation_mode": route.get("conversation_mode"), "rag_count": len(result.retrieval_result.code_chunks) if result.retrieval_result else 0, "llm_answer": "", "answer_mode": result.answer_mode, "path_scope": tuple(route.get("path_scope") or []), "symbol_candidates": tuple(route_result.query_plan.symbol_candidates or []) if route_result else (), "layers": tuple(route.get("layers") or []), }, "timings_ms": result.diagnostics.timings_ms, } actual = dict(details["actual"]) return ExecutionPayload(actual=actual, details=details) def _repo_context() -> RepoContext: return RepoContext( languages=["python"], available_domains=["CODE", "DOCS"], available_layers=[ RagLayer.CODE_ENTRYPOINTS, RagLayer.CODE_SYMBOL_CATALOG, RagLayer.CODE_DEPENDENCY_GRAPH, RagLayer.CODE_SEMANTIC_ROLES, RagLayer.CODE_SOURCE_CHUNKS, RagLayer.DOCS_MODULE_CATALOG, RagLayer.DOCS_FACT_INDEX, RagLayer.DOCS_SECTION_INDEX, RagLayer.DOCS_POLICY_INDEX, ], )