core/auto_dev/base_engine.py

"""
Base Learning Engine

Abstract interface for self-improving agent modules. Both MementoEngine
(skill generation) and AlphaEvolverEngine (skill optimization) implement
this interface, enabling a unified lifecycle:

    analyze_episode → propose_code_change → validate_change
"""

import logging
from abc import ABC, abstractmethod
from typing import Any, Protocol, runtime_checkable

from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)


@runtime_checkable
class SandboxProtocol(Protocol):
    """
    Abstract sandbox interface for executing untrusted code.

    Upstream uses ContainerSandbox (Docker).
    SaaS implementations can inject SandboxExecutionService (Fly.io).
    """

    async def execute_raw_python(
        self,
        tenant_id: str,
        code: str,
        input_params: dict[str, Any],
        timeout: int = 60,
        safety_level: str = "MEDIUM_RISK",
        **kwargs,
    ) -> dict[str, Any]:
        """
        Execute raw Python code in an isolated sandbox.

        Returns:
            {
                "status": "success" | "failed",
                "output": str,
                "execution_seconds": float,
                "execution_id": str,
            }
        """
        ...


class BaseLearningEngine(ABC):
    """
    Unified interface for self-improving agent modules.

    Subclasses must implement three core lifecycle methods:
    1. analyze_episode — read and interpret execution data
    2. propose_code_change — generate a code modification
    3. validate_change — execute in sandbox and assess fitness
    """

    def __init__(
        self,
        db: Session,
        llm_service: Any | None = None,
        sandbox: SandboxProtocol | None = None,
    ):
        self.db = db
        self.llm = llm_service
        self.sandbox = sandbox

    @abstractmethod
    async def analyze_episode(self, episode_id: str, **kwargs) -> dict[str, Any]:
        """
        Read and interpret an episode's execution data.

        Returns a structured analysis dict containing:
        - task_description, error_trace, tool_calls (for failures)
        - latency, token_usage, edge_case_signals (for successes)
        """

    @abstractmethod
    async def propose_code_change(
        self, context: dict[str, Any], **kwargs
    ) -> str:
        """
        Generate a code modification proposal via LLM.

        Args:
            context: Analysis output from analyze_episode()

        Returns:
            Generated Python code string
        """

    @abstractmethod
    async def validate_change(
        self, code: str, test_inputs: list[dict[str, Any]], tenant_id: str, **kwargs
    ) -> dict[str, Any]:
        """
        Execute proposed code in sandbox and assess fitness.

        Returns:
            {
                "passed": bool,
                "proxy_signals": dict,
                "execution_result": dict,
            }
        """

    def _get_llm_service(self):
        """Get LLM service with graceful fallback."""
        if self.llm is not None:
            return self.llm

        try:
            from core.llm_service import get_llm_service

            self.llm = get_llm_service()
            return self.llm
        except Exception as e:
            logger.warning(
                f"LLM service unavailable — Auto-Dev features requiring LLM will be skipped: {e}"
            )
            return None

    def _get_sandbox(self):
        """Get sandbox with graceful fallback to ContainerSandbox."""
        if self.sandbox is not None:
            return self.sandbox

        try:
            from core.auto_dev.container_sandbox import ContainerSandbox

            self.sandbox = ContainerSandbox()
            return self.sandbox
        except Exception as e:
            logger.warning(f"Sandbox unavailable — validation will be skipped: {e}")
            return None

    def _strip_markdown_fences(self, code: str) -> str:
        """Strip markdown code fences from LLM output."""
        code = code.strip()
        if code.startswith("```python"):
            code = code[len("```python") :]
        elif code.startswith("```"):
            code = code[3:]
        if code.endswith("```"):
            code = code[:-3]
        return code.strip()