variet-agent/core/gemini_caller.py

"""GeminiCaller — gemini CLI 호출.

두 가지 모드:
1. call() — 텍스트 입출력 (분류/리뷰/총평)
2. call_agent() — 프로젝트 디렉토리에서 에이전트 실행 (코딩)
   → Gemini가 직접 파일 읽기/쓰기/명령 실행
"""

import asyncio
import logging
import time
import sys
from pathlib import Path

logger = logging.getLogger("variet.gemini")

ROLE_PROMPTS_DIR = Path(__file__).parent.parent / "prompts"

# 동시 호출 제한 (Gemini AI Ultra 120RPM 고려)
_semaphore = asyncio.Semaphore(4)

# Windows에서 PS ExecutionPolicy 우회
_IS_WIN = sys.platform == "win32"


class GeminiCallError(Exception):
    """Gemini CLI 호출 실패."""
    pass


class GeminiCaller:
    """Gemini CLI 호출을 관리합니다."""

    def __init__(self, project_path: str = None):
        self.project_path = project_path
        self.call_count = 0
        self.last_call_time = 0.0

    def _build_cmd(self):
        """OS에 맞는 gemini 커맨드 빌드."""
        if _IS_WIN:
            return ["cmd", "/c", "gemini", "--approval-mode", "yolo"]
        return ["gemini", "--approval-mode", "yolo"]

    def _clean_output(self, raw: str) -> str:
        """Gemini 출력에서 노이즈 라인 제거."""
        lines = raw.splitlines()
        cleaned = []
        for line in lines:
            if any(noise in line for noise in [
                "YOLO mode", "Loaded cached", "Welcome to Gemini",
                "Type /help", "Gemini CLI",
            ]):
                continue
            cleaned.append(line)
        return "\n".join(cleaned).strip()

    # ──────────────────────────────────────────
    #  텍스트 모드 (분류/리뷰/총평)
    # ──────────────────────────────────────────

    async def call(self, role: str, context: str, timeout: int = 120) -> str:
        """역할별 프롬프트로 텍스트 생성.

        파일 접근 없이 텍스트만 주고받는 역할에 사용.
        (unified, reviewer, summarizer)
        """
        async with _semaphore:
            return await self._call_text(role, context, timeout)

    async def _call_text(self, role: str, context: str, timeout: int) -> str:
        """텍스트 전용 호출."""
        prompt_file = ROLE_PROMPTS_DIR / f"{role}.md"
        if prompt_file.exists():
            system_prompt = prompt_file.read_text(encoding="utf-8")
        else:
            system_prompt = f"You are a {role}. Respond in Korean."

        full_input = (
            f"=== SYSTEM INSTRUCTIONS ===\n"
            f"{system_prompt}\n\n"
            f"=== USER INPUT ===\n"
            f"{context}"
        )

        try:
            proc = await asyncio.create_subprocess_exec(
                *self._build_cmd(),
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(input=full_input.encode("utf-8")),
                timeout=timeout,
            )

            self.call_count += 1
            self.last_call_time = time.time()

            result = self._clean_output(stdout.decode("utf-8", errors="replace"))

            if not result and stderr:
                err = stderr.decode("utf-8", errors="replace").strip()
                logger.warning(f"Gemini [{role}] 빈 응답. stderr: {err[:200]}")

            logger.info(
                f"Gemini [{role}] text #{self.call_count} "
                f"-- 입력 {len(full_input)}자 -> 출력 {len(result)}자"
            )
            return result

        except asyncio.TimeoutError:
            raise GeminiCallError(f"Gemini timeout ({timeout}s) -- role={role}")
        except FileNotFoundError:
            raise GeminiCallError("gemini CLI를 찾을 수 없습니다.")
        except Exception as e:
            raise GeminiCallError(f"Gemini CLI 호출 실패: {e}")

    # ──────────────────────────────────────────
    #  에이전트 모드 (코딩 — 파일 직접 읽기/쓰기)
    # ──────────────────────────────────────────

    async def call_agent(
        self, role: str, context: str, cwd: str,
        timeout: int = 300,
    ) -> str:
        """에이전트 모드 — 프로젝트 디렉토리에서 실행.

        Gemini가 직접 파일을 읽고/쓰고/명령을 실행합니다.
        cwd를 프로젝트 경로로 설정하여 파일 접근을 허용합니다.

        Args:
            role: 프롬프트 역할 (coder)
            context: 작업 지시 (태스크 설명)
            cwd: 프로젝트 루트 경로 (여기서 Gemini 실행)
            timeout: 타임아웃 (에이전트는 더 길게 — 기본 5분)
        """
        async with _semaphore:
            return await self._call_agent_impl(role, context, cwd, timeout)

    async def _call_agent_impl(
        self, role: str, context: str, cwd: str, timeout: int,
    ) -> str:
        """에이전트 모드 구현."""
        prompt_file = ROLE_PROMPTS_DIR / f"{role}.md"
        if prompt_file.exists():
            system_prompt = prompt_file.read_text(encoding="utf-8")
        else:
            system_prompt = f"You are a {role}. Respond in Korean."

        full_input = (
            f"=== SYSTEM INSTRUCTIONS ===\n"
            f"{system_prompt}\n\n"
            f"=== TASK ===\n"
            f"{context}\n\n"
            f"=== IMPORTANT ===\n"
            f"프로젝트 루트: {cwd}\n"
            f"파일을 직접 생성/수정하세요. 코드블록으로 출력하지 말고, 실제 파일로 저장하세요.\n"
            f"작업 완료 후 변경한 파일 목록을 간단히 출력하세요."
        )

        try:
            proc = await asyncio.create_subprocess_exec(
                *self._build_cmd(),
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                cwd=cwd,  # ★ 핵심: 프로젝트 디렉토리에서 실행
            )
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(input=full_input.encode("utf-8")),
                timeout=timeout,
            )

            self.call_count += 1
            self.last_call_time = time.time()

            result = self._clean_output(stdout.decode("utf-8", errors="replace"))

            logger.info(
                f"Gemini [{role}] agent #{self.call_count} "
                f"-- cwd={cwd} 입력 {len(full_input)}자 -> 출력 {len(result)}자"
            )
            return result

        except asyncio.TimeoutError:
            raise GeminiCallError(
                f"Gemini agent timeout ({timeout}s) -- role={role}, cwd={cwd}"
            )
        except FileNotFoundError:
            raise GeminiCallError("gemini CLI를 찾을 수 없습니다.")
        except Exception as e:
            raise GeminiCallError(f"Gemini agent 호출 실패: {e}")

    async def call_simple(self, prompt: str, timeout: int = 60) -> str:
        """시스템 프롬프트 없이 단순 호출."""
        return await self.call("default", prompt, timeout)