""" codex_gate_check.py - Codex G1 설계 게이트 검증 스크립트 Codex CLI를 통해 설계 문서와 영향받는 코드를 리뷰하고, 발견된 리스크 및 개선 제안을 JSON 형식으로 반환합니다. Codex 장애 시 마아트 폴백(규칙 기반 검증)으로 자동 전환합니다. """ import json import logging import os import re import subprocess from typing import Any logger = logging.getLogger("codex_gate_check") # .env.keys 환경변수 로드 (OPENAI_API_KEY 등 subprocess에 전달 보장) try: from utils.env_loader import load_env_keys # type: ignore[import-not-found] load_env_keys() except ImportError: logger.debug("utils.env_loader 미설치 — .env.keys 자동 로드 스킵") CODEX_COMPANION_PATH = "/home/jay/.claude/plugins/cache/openai-codex/codex/1.0.3/scripts/codex-companion.mjs" def _read_file_safe(path: str) -> str: """파일을 안전하게 읽어 내용을 반환. 실패 시 빈 문자열 반환.""" try: with open(path, "r", encoding="utf-8") as f: return f.read() except Exception as e: logger.warning("파일 읽기 실패: %s — %s", path, e) return "" def _extract_json_from_output(output: str) -> dict | None: """ Codex stdout에서 JSON 객체를 추출합니다. ```json ... ``` 블록 또는 직접 JSON 형식을 지원합니다. """ # ```json 코드 블록 우선 탐색 code_block_match = re.search(r"```json\s*([\s\S]+?)\s*```", output) if code_block_match: candidate = code_block_match.group(1).strip() try: return json.loads(candidate) except json.JSONDecodeError: pass # 직접 JSON 객체 탐색 (첫 번째 { ... } 블록) brace_match = re.search(r"\{[\s\S]+\}", output) if brace_match: candidate = brace_match.group(0).strip() try: return json.loads(candidate) except json.JSONDecodeError: pass return None def _maat_fallback_check( task_file: str, affected_files: list[str], workspace_root: str, ) -> dict[str, Any]: """ 마아트 독립 검증 폴백. 규칙 기반으로 파일 존재 여부를 확인하고 결과를 반환합니다. """ logger.warning("Codex 장애 → 마아트 폴백") risks: list[dict[str, str]] = [] suggestions: list[str] = [] # task_file 존재 및 비어있지 않은지 확인 if not os.path.isfile(task_file): risks.append( { "severity": "critical", "description": f"설계 문서(task_file)가 존재하지 않습니다: {task_file}", } ) else: content = _read_file_safe(task_file) if not content.strip(): risks.append( { "severity": "critical", "description": f"설계 문서(task_file)가 비어 있습니다: {task_file}", } ) # affected_files 각 파일 존재 여부 확인 for file_path in affected_files: resolved = file_path if os.path.isabs(file_path) else os.path.join(workspace_root, file_path) if not os.path.isfile(resolved): risks.append( { "severity": "critical", "description": f"영향받는 파일이 존재하지 않습니다: {file_path}", } ) # [신규] 변경 범위 크기 경고 if len(affected_files) > 5: risks.append({ "severity": "high", "description": f"변경 범위가 큽니다 ({len(affected_files)}개 파일). 리뷰를 신중하게 진행하세요.", }) # [신규] AST 의존성 분석 — 호출자가 많은 파일 경고 callers_ctx = _get_callers_context(affected_files, workspace_root) if callers_ctx: for line in callers_ctx.split("\n"): match = re.search(r"(\d+)곳에서 호출됨", line) if match and int(match.group(1)) >= 10: risks.append({ "severity": "high", "description": f"고영향 파일 감지: {line.strip()}", }) suggestions.append(f"AST 의존성 분석 결과:\n{callers_ctx}") # [신규] task_file 위험 키워드 감지 if os.path.isfile(task_file): content = _read_file_safe(task_file) danger_keywords = ["인증", "결제", "보안", "권한", "PII", "개인정보", "API키", "토큰"] found = [kw for kw in danger_keywords if kw in content] if found: risks.append({ "severity": "medium", "description": f"설계 문서에 보안 민감 키워드 발견: {', '.join(found)}. 보안 리뷰 권장.", }) has_critical = any(r["severity"] == "critical" for r in risks) if not risks: suggestions.append("마아트 폴백 검증 통과: 모든 파일이 존재합니다.") return { "pass": not has_critical, "risks": risks, "suggestions": suggestions, "source": "maat_fallback", "error": None, } def _run_codex_companion(prompt: str, workspace_root: str) -> dict | None: """codex-companion.mjs task 명령으로 Codex 호출. 실패 시 None 반환.""" if not os.path.isfile(CODEX_COMPANION_PATH): logger.warning("codex-companion.mjs not found: %s", CODEX_COMPANION_PATH) return None try: # stdin으로 prompt 전달 (Argument list too long 방지) result = subprocess.run( ["node", CODEX_COMPANION_PATH, "task", "-", "--json"], input=prompt, capture_output=True, text=True, timeout=120, cwd=workspace_root, ) if result.returncode != 0: logger.warning("codex-companion 비정상 종료: rc=%d, stderr=%s", result.returncode, result.stderr[:200]) return None # JSON 출력 파싱 parsed = _extract_json_from_output(result.stdout) if parsed and "rawOutput" in parsed: # companion은 {status, rawOutput, ...} 형태로 반환 raw = parsed.get("rawOutput", "") if not raw: logger.warning("codex-companion rawOutput 비어있음") return None return _extract_json_from_output(raw) return parsed except subprocess.TimeoutExpired: logger.warning("codex-companion 타임아웃 (120s)") return None except Exception as e: logger.warning("codex-companion 호출 예외: %s", e) return None def _get_callers_context(affected_files: list[str], workspace_root: str) -> str: """ AST 스크립트를 이용해 affected_files 각각의 callers 정보를 조회합니다. Args: affected_files: 영향받는 파일 경로 목록 workspace_root: 워크스페이스 루트 경로 Returns: callers 정보 문자열 (비어있으면 빈 문자열) """ if not affected_files: return "" ast_script = os.path.join(workspace_root, "scripts", "ast_dependency_map.py") if not os.path.isfile(ast_script): logger.warning("AST 스크립트를 찾을 수 없음: %s", ast_script) return "" lines: list[str] = [] for file_path in affected_files: try: result = subprocess.run( [ "python3", ast_script, "--root", workspace_root, "--files", file_path, "--json", ], capture_output=True, text=True, timeout=30, cwd=workspace_root, ) if result.returncode != 0: logger.warning( "AST 스크립트 비정상 종료 (file=%s, returncode=%d): %s", file_path, result.returncode, result.stderr, ) continue try: data = json.loads(result.stdout) except json.JSONDecodeError as e: logger.warning("AST 스크립트 JSON 파싱 실패 (file=%s): %s", file_path, e) continue blast_radius = data.get("blast_radius", {}) callers: list[str] = blast_radius.get("callers", []) if callers: # 상위 5개만 표시 top_callers = callers[:5] caller_str = ", ".join(top_callers) lines.append(f"- {file_path}: {len(callers)}곳에서 호출됨: {caller_str}") except subprocess.TimeoutExpired: logger.warning("AST 스크립트 타임아웃 (30s 초과, file=%s)", file_path) except Exception as e: logger.warning("AST 스크립트 호출 중 예외 발생 (file=%s): %s", file_path, e) if not lines: return "" return "함수 호출자 정보:\n" + "\n".join(lines) def codex_gate_check( task_file: str, affected_files: list[str], workspace_root: str = "/home/jay/workspace", ) -> dict[str, Any]: """ Codex G1 설계 게이트 검증 메인 함수. Args: task_file: 설계 문서(task 파일) 경로 affected_files: 영향받는 파일 경로 목록 workspace_root: 워크스페이스 루트 경로 (기본값: /home/jay/workspace) Returns: 검증 결과 dict (pass, risks, suggestions, source, error) """ logger.info("codex_gate_check 시작: task_file=%s, affected_files=%s", task_file, affected_files) if not os.environ.get("OPENAI_API_KEY"): logger.warning("OPENAI_API_KEY 환경변수가 설정되지 않았습니다. Codex 호출이 실패할 수 있습니다.") # 1. task_file 읽기 task_content = _read_file_safe(task_file) if not task_content: logger.warning("task_file 내용 없음 또는 읽기 실패: %s", task_file) # 2. affected_files 내용 읽기 code_parts: list[str] = [] for file_path in affected_files: resolved = file_path if os.path.isabs(file_path) else os.path.join(workspace_root, file_path) content = _read_file_safe(resolved) code_parts.append(f"--- {file_path} ---\n{content}") code_content = "\n\n".join(code_parts) if code_parts else "(영향받는 파일 없음)" # 3. AST callers 컨텍스트 추가 (task-1869_2.2+1) callers_ctx = _get_callers_context(affected_files, workspace_root) # 4. Codex 프롬프트 조합 callers_section = f"\n\n{callers_ctx}" if callers_ctx else "" prompt = f"""다음 설계 문서와 코드를 리뷰하여 JSON 형식으로 응답하세요. 응답 형식: {{"risks": [{{"severity": "critical|high|medium|low", "description": "설명"}}], "suggestions": ["제안1"]}} 설계 문서: {task_content} 영향받는 코드: {code_content}{callers_section} """ # 5. Codex 호출 (2단계 캐스케이드: companion → maat fallback) source = "codex" parsed = None # 5a. codex-companion.mjs 시도 logger.info("Codex companion 호출 시도...") parsed = _run_codex_companion(prompt, workspace_root) if parsed is not None: source = "codex_companion" logger.info("codex-companion 성공") # 5b. 마아트 폴백 if parsed is None: logger.info("Codex companion 실패 → 마아트 폴백") return _maat_fallback_check(task_file, affected_files, workspace_root) # 6. 파싱된 결과 처리 risks = parsed.get("risks", []) suggestions = parsed.get("suggestions", []) has_critical = any(r.get("severity") == "critical" for r in risks) logger.info("Codex 리뷰 완료 (source=%s): risks=%d, critical=%s", source, len(risks), has_critical) return { "pass": not has_critical, "risks": risks, "suggestions": suggestions, "source": source, "error": None, } if __name__ == "__main__": import argparse logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") parser = argparse.ArgumentParser(description="Codex G1 설계 게이트 검증") parser.add_argument("--task-id", default=None) parser.add_argument("--task-file", default=None) parser.add_argument("--affected-files", nargs="*", default=[]) parser.add_argument("--workspace", default="/home/jay/workspace") args = parser.parse_args() # task-id에서 task-file 자동 유추 task_file = args.task_file if task_file is None and args.task_id: task_file = os.path.join(args.workspace, "memory", "tasks", f"{args.task_id}.md") if task_file is None: parser.error("--task-file 또는 --task-id 중 하나가 필요합니다.") result = codex_gate_check(task_file, args.affected_files, args.workspace) print(json.dumps(result, ensure_ascii=False, indent=2))