core/auto_dev/reflection_engine.py

"""
Reflection Engine

Monitors the event bus for task failures and identifies recurring failure
patterns that warrant automatic skill generation via MementoEngine.

Operates as a "pattern detector":
- Filters for agents at Student/Intern maturity level
- Batches failure events by task similarity
- Triggers MementoEngine when a pattern occurs ≥ threshold times
- Queries ReflectionService for past critiques to enrich context
"""

import logging
from collections import defaultdict
from typing import Any

from sqlalchemy.orm import Session

from core.auto_dev.event_hooks import TaskEvent, event_bus

logger = logging.getLogger(__name__)

# Minimum number of similar failures before triggering skill generation
DEFAULT_FAILURE_THRESHOLD = 2


class ReflectionEngine:
    """
    Monitors task failures and triggers Memento-Skills when patterns emerge.

    Usage:
        engine = ReflectionEngine(db)
        engine.register()  # Registers on event bus

        # Or manually:
        await engine.process_failure(event)
    """

    def __init__(
        self,
        db: Session,
        failure_threshold: int = DEFAULT_FAILURE_THRESHOLD,
    ):
        self.db = db
        self.failure_threshold = failure_threshold
        # In-memory failure pattern tracker: agent_id → [failure descriptions]
        self._failure_buffer: dict[str, list[dict[str, Any]]] = defaultdict(list)

    def register(self) -> None:
        """Register this engine on the global event bus."""
        event_bus.on_task_fail(self.process_failure)
        logger.info("ReflectionEngine registered on event bus")

    async def process_failure(self, event: TaskEvent) -> None:
        """
        Process a task failure event.

        Adds the failure to the pattern buffer for the agent. If the
        number of similar failures exceeds the threshold, triggers
        MementoEngine to generate a skill candidate.
        """
        agent_id = event.agent_id

        # Check if this agent's maturity allows Auto-Dev
        if not self._should_process_agent(agent_id, event.tenant_id):
            return

        # Add to buffer
        self._failure_buffer[agent_id].append(
            {
                "episode_id": event.episode_id,
                "task_description": event.task_description,
                "error_trace": event.error_trace,
                "tenant_id": event.tenant_id,
            }
        )

        # Check for recurring pattern
        similar_failures = self._find_similar_failures(agent_id, event.task_description)

        if len(similar_failures) >= self.failure_threshold:
            logger.info(
                f"ReflectionEngine: {len(similar_failures)} similar failures detected "
                f"for agent {agent_id}. Triggering Memento-Skills."
            )
            await self._trigger_memento(
                agent_id=agent_id,
                tenant_id=event.tenant_id,
                episode_id=event.episode_id,
                similar_failures=similar_failures,
            )

            # Clear the buffer for this pattern to avoid re-triggering
            self._clear_pattern(agent_id, similar_failures)

    async def _trigger_memento(
        self,
        agent_id: str,
        tenant_id: str,
        episode_id: str,
        similar_failures: list[dict[str, Any]],
    ) -> None:
        """Trigger MementoEngine to generate a skill candidate."""
        try:
            from core.auto_dev.memento_engine import MementoEngine

            engine = MementoEngine(db=self.db)
            candidate = await engine.generate_skill_candidate(
                tenant_id=tenant_id,
                agent_id=agent_id,
                episode_id=episode_id,
            )
            logger.info(
                f"ReflectionEngine triggered skill candidate: {candidate.skill_name}"
            )
        except Exception as e:
            logger.error(f"ReflectionEngine failed to trigger Memento: {e}")

    def _should_process_agent(self, agent_id: str, tenant_id: str) -> bool:
        """Check if the agent should be processed for Auto-Dev."""
        try:
            from core.auto_dev.capability_gate import AutoDevCapabilityService

            gate = AutoDevCapabilityService(self.db)

            # Get workspace settings for this tenant
            workspace_settings = self._get_workspace_settings(tenant_id)

            return gate.can_use(
                agent_id=agent_id,
                capability="auto_dev.memento_skills",
                workspace_settings=workspace_settings,
            )
        except Exception:
            # If graduation framework isn't available, skip
            return False

    def _get_workspace_settings(self, tenant_id: str) -> dict[str, Any]:
        """Retrieve workspace settings for a tenant."""
        try:
            from core.models import Workspace

            workspace = (
                self.db.query(Workspace)
                .filter(Workspace.tenant_id == tenant_id)
                .first()
            )
            if workspace and workspace.metadata_json:
                return workspace.metadata_json
        except Exception:
            pass
        return {}

    def _find_similar_failures(
        self, agent_id: str, task_description: str
    ) -> list[dict[str, Any]]:
        """Find failures with similar task descriptions for an agent."""
        buffer = self._failure_buffer.get(agent_id, [])
        # Simple word-overlap similarity
        task_words = set(task_description.lower().split())

        similar = []
        for failure in buffer:
            other_words = set(failure["task_description"].lower().split())
            if task_words and other_words:
                overlap = len(task_words & other_words) / max(
                    len(task_words), len(other_words)
                )
                if overlap >= 0.5:  # 50% word overlap threshold
                    similar.append(failure)

        return similar

    def _clear_pattern(
        self, agent_id: str, similar_failures: list[dict[str, Any]]
    ) -> None:
        """Remove processed failures from the buffer."""
        episode_ids = {f["episode_id"] for f in similar_failures}
        self._failure_buffer[agent_id] = [
            f
            for f in self._failure_buffer[agent_id]
            if f["episode_id"] not in episode_ids
        ]