python/src/server/services/system/meta_twin_service.py

"""
MetaTwinService - Dynamic internal monitoring and self-healing loop for AI Agents.
"""

from datetime import UTC, datetime, timedelta

from ...config.logfire_config import get_logger
from ...utils import get_supabase_client

logger = get_logger(__name__)


class MetaTwinService:
    """
    Self-healing orchestrator for the Archon Agent ecosystem.
    Audits agent parameters dynamically and executes fallback actions.
    """

    def __init__(self):
        self.supabase = get_supabase_client()

    async def run_telemetry_audit(self) -> dict:
        """
        Scans agent execution logs and statistics to identify failures (e.g., rate limits, loops).
        Returns a summary of diagnoses and corrective actions taken.
        """
        logger.info("MetaTwin: Starting system telemetry and agent-health audit...")
        one_hour_ago = (datetime.now(UTC) - timedelta(hours=1)).isoformat()

        # 1. Fetch recent agent errors
        try:
            logs_res = (
                self.supabase.table("archon_logs")
                .select("id, source, message, level, details, created_at")
                .gt("created_at", one_hour_ago)
                .execute()
            )
            logs = logs_res.data or []
        except Exception as e:
            logger.error(f"MetaTwin: Failed to query archon_logs: {e}")
            logs = []

        diagnoses = []
        corrections = []

        # Analyze errors per agent
        agent_errors: dict[str, list[dict]] = {}
        for log in logs:
            source = log.get("source", "system").lower()
            if log.get("level") == "ERROR" or "rate limit" in log.get("message", "").lower():
                agent_errors.setdefault(source, []).append(log)

        # 2. Diagnose & Heal
        for agent_name, errors in agent_errors.items():
            error_count = len(errors)
            rate_limit_hits = sum(1 for e in errors if "429" in e.get("message", "") or "rate limit" in e.get("message", "").lower())

            # Scenario A: Rate Limit Risk (Multiple 429 hits)
            if rate_limit_hits >= 3:
                diagnoses.append({
                    "agent": agent_name,
                    "issue": "RATE_LIMIT_RISK",
                    "reason": f"Detected {rate_limit_hits} rate limit errors in the last hour."
                })

                # Apply Actuator: Switch to fallback model
                fallback_success = await self.switch_model_to_fallback(agent_name, "models/gemini-3.1-flash-lite")
                if fallback_success:
                    corrections.append({
                        "agent": agent_name,
                        "action": "MODEL_SWAP",
                        "details": "Swapped DEFAULT_PRO model to DEFAULT_TEXT (Lite) fallback."
                    })

            # Scenario B: Stuck in loop / Infinite calls (High error count)
            elif error_count >= 5:
                diagnoses.append({
                    "agent": agent_name,
                    "issue": "STUCK_IN_LOOP",
                    "reason": f"Detected high error frequency ({error_count} errors) suggesting execution loop."
                })

                # Apply Actuator: Throttle concurrency / Cooldown
                throttle_success = await self.throttle_concurrency(agent_name, 1)
                if throttle_success:
                    corrections.append({
                        "agent": agent_name,
                        "action": "THROTTLE_CONCURRENCY",
                        "details": "Conformed execution rate limit to 1 concurrent request."
                    })

        # Log audit outcomes to archon_logs as system trace
        if corrections:
            try:
                self.supabase.table("archon_logs").insert({
                    "level": "INFO",
                    "source": "MetaTwinService",
                    "type": "system",
                    "message": f"Self-healing executed: {len(corrections)} corrections applied.",
                    "details": {"diagnoses": diagnoses, "corrections": corrections}
                }).execute()
            except Exception as e:
                logger.error(f"MetaTwin: Failed to insert audit outcome log: {e}")

        return {
            "status": "completed",
            "diagnoses_count": len(diagnoses),
            "corrections_count": len(corrections),
            "diagnoses": diagnoses,
            "corrections": corrections
        }

    async def switch_model_to_fallback(self, agent_name: str, fallback_model: str) -> bool:
        """
        Actuator: Swaps active system models dynamically to prevent further rate limits.
        """
        logger.info(f"MetaTwin: Swapping agent {agent_name} model to {fallback_model}")
        try:
            from ...config.model_ssot import SYSTEM_MODELS
            # Update physical models mapping in memory
            SYSTEM_MODELS["DEFAULT_PRO"] = fallback_model

            # Persist setting to DB so it propagates
            await self.supabase.table("archon_settings").upsert({
                "key": f"MODEL_OVERRIDE_{agent_name.upper()}",
                "value": fallback_model,
                "description": f"Auto-Healing Model Override by MetaTwin at {datetime.now(UTC).isoformat()}"
            }, on_conflict="key").execute()
            return True
        except Exception as e:
            logger.error(f"MetaTwin: Model override failed for {agent_name}: {e}")
            return False

    async def throttle_concurrency(self, agent_name: str, new_limit: int) -> bool:
        """
        Actuator: Dynamically scales down concurrency limits for the specified agent.
        """
        logger.info(f"MetaTwin: Throttling agent {agent_name} concurrency limit to {new_limit}")
        try:
            await self.supabase.table("archon_settings").upsert({
                "key": f"CRAWL_CONCURRENT_MAX_{agent_name.upper()}",
                "value": str(new_limit),
                "description": f"Auto-Healing Concurrency Limit by MetaTwin at {datetime.now(UTC).isoformat()}"
            }, on_conflict="key").execute()
            return True
        except Exception as e:
            logger.error(f"MetaTwin: Throttling failed for {agent_name}: {e}")
            return False


# Singleton Instance
meta_twin_service = MetaTwinService()