python/src/server/services/dev_ops_agent_service.py

import asyncio
import json
import re
import uuid
from typing import Any, cast

import aiofiles

from ..config.logfire_config import get_logger
from ..prompts.dev_ops_prompts import DEVBOT_TOOLS, get_devbot_analysis_prompt
from ..utils.code_modifier import CodeModifier
from .agent_tool_executor import AgentToolExecutor
from .credential_service import credential_service
from .llm_provider_service import get_llm_client


class DevOpsAgentService:
    def __init__(self, tool_executor: AgentToolExecutor):
        self.code_modifier = CodeModifier(base_path=".")
        self.tool_executor = tool_executor

    @property
    def mcp_client(self):
        return self.tool_executor.mcp_client

    async def _analyze_error_with_structured_output(
        self, command: str, stderr: str, agent_id: str
    ) -> dict[str, Any] | None:
        logger = get_logger(__name__)
        from ..services.search.rag_service import RAGService
        from ..services.token_usage_service import TokenUsageService

        sop_context = ""
        try:
            success, rag_res = await RAGService().perform_rag_query(query=f"SOP for {command}", match_count=2)
            if success:
                for r in rag_res.get("results", []):
                    sop_context += f"\n[SOP]: {r['content']}\n"
        except Exception:
            pass

        prompt = get_devbot_analysis_prompt(command, stderr)
        full_prompt = f"{prompt}\n\n### CONSTRAINTS ###\n{sop_context or 'Standard conventions.'}"
        messages = [{"role": "user", "content": full_prompt}]
        tools = DEVBOT_TOOLS if self.mcp_client else None
        request_id = f"fix-{uuid.uuid4().hex[:8]}"

        try:
            from ..config.model_ssot import SYSTEM_MODELS

            model = SYSTEM_MODELS["DEFAULT_PRO"]
            admin_api_key = await credential_service.get_credential(
                "GEMINI_API_KEY"
            ) or await credential_service.get_credential("GOOGLE_API_KEY")
            from .system.rate_limiter import GlobalThrottler

            await GlobalThrottler.wait_for_capacity(tier="pro")

            async with get_llm_client(api_key=admin_api_key) as client:
                response = await client.chat.completions.create(
                    model=model, messages=messages, tools=tools, tool_choice="auto" if tools else None, temperature=0.1
                )
                res_msg = response.choices[0].message

                if hasattr(response, "usage") and response.usage:
                    from .agent_registry import get_agent_uuid

                    agent_uuid = get_agent_uuid(agent_id)
                    asyncio.create_task(
                        TokenUsageService.log_usage(
                            request_id=f"{request_id}-r1",
                            user_id=agent_uuid,
                            model=model,
                            provider="google",
                            input_tokens=response.usage.prompt_tokens,
                            output_tokens=response.usage.completion_tokens,
                            context_type="self_healing",
                        )
                    )

                tool_calls = res_msg.tool_calls
                if tool_calls and self.mcp_client:
                    messages.append(res_msg)
                    tool_results = await self.tool_executor.handle_tool_calls(tool_calls, agent_id=agent_id)
                    messages.extend(tool_results)
                    final_response = await client.chat.completions.create(
                        model=model, messages=messages, tools=tools, response_format={"type": "json_object"}
                    )
                    content = final_response.choices[0].message.content.strip()
                else:
                    content = res_msg.content.strip()

                if not content:
                    return None
                try:
                    return cast(dict[str, Any], json.loads(content))
                except Exception:
                    match = re.search(r"\{.*\}", content, re.DOTALL)
                    return json.loads(match.group(0)) if match else None
        except Exception as e:
            logger.error(f"Analysis failed: {e}")
            return {"file_path": None, "reasoning": f"Analysis: Check syntax. Error: {e}"}

    async def run_command_with_self_healing(
        self, command: str, agent_id: str, task_id: str | None = None
    ) -> tuple[bool, str]:
        logger = get_logger(__name__)
        task_id = task_id or f"auto-{uuid.uuid4().hex[:8]}"
        logger.info(f"Executing command with self-healing (L2): {command}")

        process = await asyncio.create_subprocess_shell(
            command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
        )
        stdout, stderr = await process.communicate()
        if process.returncode == 0:
            logger.info(f"Command '{command}' succeeded initially.")
            return True, stdout.decode().strip()

        logger.warning(f"Command '{command}' failed. Starting Active Repair Loop.")
        fix_proposal = await self._analyze_error_with_structured_output(
            command, stderr.decode().strip()[-2000:], agent_id=agent_id
        )

        # --- Poisson Gate Check (1.5 Governance) ---
        # Self-healing involving file modification is classified as Level 2 (Moderate)
        is_trusted, curr_level = await self.tool_executor.check_poisson_gate(agent_id=agent_id, required_level=2)
        if not is_trusted:
            logger.warning(f"Poisson Gate: Insufficient credibility for Level 2 auto-repair. ({curr_level})")
            return (
                False,
                f"Poisson Security Block: Your current level is {curr_level}, but Level 2 is required for autonomous repair. Proposal: {fix_proposal.get('reasoning') if fix_proposal else 'Check logs'}",
            )

        # Test baseline compatibility: If no file path, it's an "Analysis only" path
        if not fix_proposal or not fix_proposal.get("file_path"):
            logger.warning("LLM could not propose a valid code fix.")
            reasoning = fix_proposal.get("reasoning", "No fix proposed") if fix_proposal else "No fix proposed"
            return False, f"Analysis: {reasoning}"

        original_branch = self.code_modifier.get_current_branch()
        sandbox_branch = self.code_modifier.create_sandbox_branch(task_id)
        try:
            self.code_modifier.apply_modification(fix_proposal["file_path"], fix_proposal["fixed_content"])
            logger.info(f"Applied fix to {fix_proposal['file_path']} on branch {sandbox_branch}")

            process_retry = await asyncio.create_subprocess_shell(
                command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
            )
            await process_retry.communicate()
            if process_retry.returncode == 0:
                lint_proc = await asyncio.create_subprocess_shell(
                    "make lint-be", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
                )
                await lint_proc.communicate()
                if lint_proc.returncode != 0:
                    return await self.run_command_with_self_healing("make lint-be", agent_id=agent_id, task_id=task_id)
                return True, f"Repair successful on {sandbox_branch}. Reasoning: {fix_proposal.get('reasoning')}"
            else:
                self.code_modifier.revert_sandbox(original_branch)
                return False, "Fix verification failed"
        except Exception as e:
            self.code_modifier.revert_sandbox(original_branch)
            return False, f"Repair error: {e}"

    async def diagnose_file_health(self, file_path: str) -> dict[str, Any]:
        """
        Performs a physical diagnostic of a file based on SOP metrics (1.7.1).
        Assigns L1 (Green), L2 (Yellow), or L3 (Red) severity.
        """
        try:
            async with aiofiles.open(file_path, encoding="utf-8") as f:
                content = await f.read()

            lines = content.splitlines()
            line_count = len(lines)

            # Detect anti-patterns: Direct SQL in API controllers
            has_direct_sql = bool(re.search(r"supabase\.table\(|execute\(\)", content))

            # 1.7.1 Logic: Severity Grading
            if line_count > 500 or has_direct_sql:
                severity = 3  # RED / Critical
                advice = "Critical Technical Debt: Exceeds 500 lines or contains direct SQL calls. Refactor to Service layer required."
            elif line_count > 300:
                severity = 2  # YELLOW / Moderate
                advice = "Moderate Technical Debt: Large file detected. Consider extracting helpers."
            else:
                severity = 1  # GREEN / Minor
                advice = "Healthy: File matches standard complexity guidelines."

            return {
                "file_path": file_path,
                "line_count": line_count,
                "has_direct_sql": has_direct_sql,
                "severity_level": severity,
                "advice": advice,
                "timestamp": uuid.uuid4().hex[:8],  # Trace ID
            }
        except Exception as e:
            return {"error": f"Failed to diagnose file: {e}"}