backend/core/agent_evolution_loop.py

"""
Agent Evolution Loop — GEA Phase 3: Updating Module + Full Evolution Cycle

Implements the Updating Module and the complete GEA evolution loop from the
Group-Evolving Agents paper (UC Santa Barbara, Feb 2026).

Architecture mirrors the two-stage GEA process:
  Stage 1 — Parent Group Selection (Performance-Novelty Algorithm)
  Stage 2 — Open-Ended Group Evolution (Experience Sharing → Reflect → Update → Evaluate)

The full cycle:
  1. select_parent_group()       — Performance-Novelty Algorithm
  2. gather experience pool      — GroupReflectionService.gather_group_experience_pool()
  3. reflect_and_generate_directives() — GroupReflectionService.reflect_and_generate_directives()
  4. _apply_directives_to_clone()— Clone agent config; apply directives; validate via guardrails
  5. _evaluate_evolved_agent()   — Lightweight benchmark evaluation
  6. _archive_or_discard()       — Save winner trace; discard failures

Key design decisions:
  - Directives are ALWAYS applied to a *clone* of the agent config, never in-place.
  - Every directive is validated by autonomous_guardrails before committing.
  - Evolution happens offline; only the winner agent is deployed (zero inference cost).

Usage:
    from core.agent_evolution_loop import AgentEvolutionLoop
    from core.database import SessionLocal

    with SessionLocal() as db:
        loop = AgentEvolutionLoop(db)
        result = await loop.run_evolution_cycle(tenant_id="tenant-uuid")
"""

import logging
import math
import uuid
from copy import deepcopy
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy.orm import Session

from core.models import AgentEvolutionTrace, AgentRegistry, Skill
from core.group_reflection_service import GroupReflectionService

logger = logging.getLogger(__name__)

# ─── Tuning constants ─────────────────────────────────────────────────────────
PERF_WEIGHT: float = 0.6      # α in combined_score = α*perf + β*novelty
NOVELTY_WEIGHT: float = 0.4   # β
PARENT_GROUP_SIZE: int = 5     # |P| — parent group size
MIN_PERF_THRESHOLD: float = 0.3  # Agents below this are excluded from selection
LOOKBACK_DAYS: int = 30        # Only consider agents active in the past N days


class EvolutionCycleResult:
    """Structured result from a single evolution cycle."""

    def __init__(
        self,
        cycle_id: str,
        tenant_id: str,
        parent_agent_ids: List[str],
        directives: List[str],
        evolved_agent_id: Optional[str],
        benchmark_passed: bool,
        benchmark_score: float,
        trace_id: Optional[str],
    ) -> None:
        self.cycle_id = cycle_id
        self.tenant_id = tenant_id
        self.parent_agent_ids = parent_agent_ids
        self.directives = directives
        self.evolved_agent_id = evolved_agent_id
        self.benchmark_passed = benchmark_passed
        self.benchmark_score = benchmark_score
        self.trace_id = trace_id
        self.timestamp = datetime.now(timezone.utc).isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return {
            "cycle_id": self.cycle_id,
            "tenant_id": self.tenant_id,
            "parent_agent_ids": self.parent_agent_ids,
            "directives": self.directives,
            "evolved_agent_id": self.evolved_agent_id,
            "benchmark_passed": self.benchmark_passed,
            "benchmark_score": self.benchmark_score,
            "trace_id": self.trace_id,
            "timestamp": self.timestamp,
        }


class AgentEvolutionLoop:
    """
    Orchestrates the full GEA evolution cycle for a tenant's agent population.
    """

    def __init__(self, db: Session) -> None:
        self.db = db
        from core.service_factory import ServiceFactory
        self.reflection_svc = ServiceFactory.get_group_reflection_service(db)

    # ─────────────────────────────────────────────────────────────────────────
    # Public API
    # ─────────────────────────────────────────────────────────────────────────

    async def run_evolution_cycle(
        self,
        tenant_id: str,
        group_size: int = PARENT_GROUP_SIZE,
        target_agent_id: Optional[str] = None,
        category: Optional[str] = None,
    ) -> EvolutionCycleResult:
        """
        Run one full GEA evolution cycle for a tenant.

        Args:
            tenant_id: Tenant to evolve agents for
            group_size: Parent group size (default 5)
            target_agent_id: If set, only evolve this specific agent;
                             otherwise select group via Performance-Novelty Algorithm
            category: Agent domain category (e.g. "crm", "finance"). Auto-detected
                      from the seed agent's AgentRegistry.category if not provided.

        Returns:
            EvolutionCycleResult with outcome details
        """
        cycle_id = str(uuid.uuid4())
        logger.info("GEA cycle %s starting for tenant %s", cycle_id, tenant_id)

        # Stage 1: Select parent group
        if target_agent_id:
            parent_group = self._get_single_agent_group(target_agent_id, tenant_id)
        else:
            parent_group = self.select_parent_group(tenant_id, n=group_size)

        if not parent_group:
            logger.warning("GEA cycle %s: no eligible agents found", cycle_id)
            return EvolutionCycleResult(
                cycle_id=cycle_id,
                tenant_id=tenant_id,
                parent_agent_ids=[],
                directives=[],
                evolved_agent_id=None,
                benchmark_passed=False,
                benchmark_score=0.0,
                trace_id=None,
            )

        parent_ids = [a.id for a in parent_group]

        # Resolve domain category once — used by reflection + trace recording
        # Priority: explicit arg > seed agent's registry category
        seed_agent_tentative = max(
            parent_group,
            key=lambda a: self._compute_combined_score(a, parent_group),
        )
        resolved_category = category or getattr(seed_agent_tentative, "category", None)

        # Stage 2: Gather shared experience pool (domain-aware)
        pool = self.reflection_svc.gather_group_experience_pool(
            parent_ids, category=resolved_category
        )

        # Stage 3: Reflect → generate evolution directives (domain-aware)
        directives = await self.reflection_svc.reflect_and_generate_directives(
            pool, tenant_id=tenant_id, category=resolved_category
        )

        # Pick the "seed" agent for mutation (highest combined score in group)
        seed_agent = seed_agent_tentative

        # Stage 4: Apply directives to a cloned config (sandboxed)
        evolved_config, guardrail_ok = await self._apply_directives_to_clone(
            seed_agent, directives, tenant_id
        )

        if not guardrail_ok:
            logger.warning("GEA cycle %s: directives blocked by guardrails", cycle_id)
            trace = self._record_trace(
                agent=seed_agent,
                parent_ids=parent_ids,
                tenant_id=tenant_id,
                directives=directives,
                pool=pool,
                benchmark_passed=False,
                benchmark_score=0.0,
                model_patch=None,
                category=resolved_category,
                block_reason="Guardrail validation failed",
            )
            return EvolutionCycleResult(
                cycle_id=cycle_id,
                tenant_id=tenant_id,
                parent_agent_ids=parent_ids,
                directives=directives,
                evolved_agent_id=None,
                benchmark_passed=False,
                benchmark_score=0.0,
                trace_id=trace.id if trace else None,
            )

        # Stage 5: Evaluate evolved config
        benchmark_score, benchmark_passed = await self._evaluate_evolved_config(
            seed_agent, evolved_config, tenant_id
        )

        # Stage 6: Archive winner or discard
        evolved_agent_id: Optional[str] = None
        if benchmark_passed:
            evolved_agent_id = await self._promote_evolved_config(
                seed_agent, evolved_config, directives, parent_group
            )

        # Record the evolution trace (with domain-specific benchmark name)
        model_patch = self._diff_configs(seed_agent.configuration, evolved_config)
        trace = self._record_trace(
            agent=seed_agent,
            parent_ids=parent_ids,
            tenant_id=tenant_id,
            directives=directives,
            pool=pool,
            benchmark_passed=benchmark_passed,
            benchmark_score=benchmark_score,
            model_patch=model_patch,
            category=resolved_category,
        )

        logger.info(
            "GEA cycle %s complete: passed=%s score=%.3f evolved_agent=%s",
            cycle_id, benchmark_passed, benchmark_score, evolved_agent_id,
        )

        return EvolutionCycleResult(
            cycle_id=cycle_id,
            tenant_id=tenant_id,
            parent_agent_ids=parent_ids,
            directives=directives,
            evolved_agent_id=evolved_agent_id,
            benchmark_passed=benchmark_passed,
            benchmark_score=benchmark_score,
            trace_id=trace.id if trace else None,
        )

    def select_parent_group(
        self,
        tenant_id: str,
        n: int = PARENT_GROUP_SIZE,
    ) -> List[AgentRegistry]:
        """
        Stage 1: Performance-Novelty Algorithm.

        Selects n agents from the archive to form the parent group.
        Combined score = α * performance_score + β * novelty_score.

        Novelty is approximated as the distance of an agent's performance
        from the population mean — agents far from the mean (in either
        direction) are novel, ensuring a healthy mix of specialists and explorers.

        Args:
            tenant_id: Tenant namespace
            n: Parent group size

        Returns:
            List of up to n AgentRegistry objects sorted by combined score desc
        """
        cutoff = datetime.now(timezone.utc) - timedelta(days=LOOKBACK_DAYS)

        agents = (
            self.db.query(AgentRegistry)
            .filter(
                AgentRegistry.tenant_id == tenant_id,
                AgentRegistry.enabled == True,
                AgentRegistry.confidence_score >= MIN_PERF_THRESHOLD,
                AgentRegistry.updated_at >= cutoff,
            )
            .all()
        )

        if not agents:
            return []

        # Compute novelty scores using population mean
        scores = [a.confidence_score for a in agents]
        mean_perf = sum(scores) / len(scores)
        std_perf = math.sqrt(sum((s - mean_perf) ** 2 for s in scores) / len(scores)) or 1e-6

        def novelty(agent: AgentRegistry) -> float:
            # Normalized absolute deviation from mean — high deviation = high novelty
            return min(1.0, abs(agent.confidence_score - mean_perf) / (2 * std_perf))

        scored = [
            (agent, self._compute_combined_score_with_novelty(agent, novelty(agent)))
            for agent in agents
        ]
        scored.sort(key=lambda x: -x[1])

        return [agent for agent, _ in scored[:n]]

    def get_ancestor_lineage(
        self,
        agent_id: str,
        tenant_id: str,
        max_depth: int = 20,
    ) -> List[Dict[str, Any]]:
        """
        Traverse the ancestry chain of an agent via evolution traces.

        Returns a list of dicts describing each ancestor, mirroring GEA's
        "super-employee" metric (17 unique ancestors → 28% of population).

        Args:
            agent_id: Starting agent
            tenant_id: Tenant namespace
            max_depth: Max generations to traverse (prevents infinite loops)

        Returns:
            List[{"agent_id": str, "generation": int, "performance_score": float}]
        """
        visited: set = set()
        lineage: List[Dict[str, Any]] = []
        queue: List[Tuple[str, int]] = [(agent_id, 0)]

        while queue and len(lineage) < max_depth:
            current_id, depth = queue.pop(0)
            if current_id in visited:
                continue
            visited.add(current_id)

            trace = (
                self.db.query(AgentEvolutionTrace)
                .filter(
                    AgentEvolutionTrace.agent_id == current_id,
                    AgentEvolutionTrace.tenant_id == tenant_id,
                )
                .order_by(AgentEvolutionTrace.created_at.desc())
                .first()
            )

            if trace:
                lineage.append({
                    "agent_id": current_id,
                    "generation": trace.generation,
                    "performance_score": trace.performance_score,
                    "depth": depth,
                })
                if trace.parent_agent_ids:
                    for parent_id in trace.parent_agent_ids:
                        if parent_id not in visited:
                            queue.append((parent_id, depth + 1))

        return lineage

    # ─────────────────────────────────────────────────────────────────────────
    # Internal: Apply directives (sandboxed)
    # ─────────────────────────────────────────────────────────────────────────

    async def _apply_directives_to_clone(
        self,
        agent: AgentRegistry,
        directives: List[str],
        tenant_id: str,
    ) -> Tuple[Dict[str, Any], bool]:
        """
        Apply evolution directives to a DEEP CLONE of the agent's configuration.

        This is the Updating Module. Directives update the system prompt and
        evolution history. Each directive is validated against governance rules
        before being committed.

        Returns:
            (evolved_config, guardrail_ok)
        """
        evolved_config = deepcopy(agent.configuration or {})

        # Record evolution metadata
        if "evolution_history" not in evolved_config:
            evolved_config["evolution_history"] = []

        evolved_config["evolution_history"].append({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "directives": directives,
            "parent_agent_id": agent.id,
            "gea_cycle": True,
        })

        # Apply directives to system prompt (append as guidance)
        existing_prompt = evolved_config.get("system_prompt", "")
        
        # Skill Creation Logic: Check for direct skill creation directives
        # Format example: "CREATE_SKILL: Fetch Shopify products from API URL https://..."
        for directive in directives:
            if directive.strip().upper().startswith("CREATE_SKILL:"):
                try:
                    logger.info("GEA: Detected skill creation directive: %s", directive)
                    # Extract prompt from directive
                    skill_prompt = directive.split(":", 1)[1].strip()
                    
                    # Initialize SkillCreationAgent via ServiceFactory
                    from core.service_factory import ServiceFactory
                    skill_agent = ServiceFactory.get_skill_creation_agent(self.db)
                    
                    # Execute skill creation
                    # We use generic parameters; in a real scenario, we might extract more from the directive
                    skill = await skill_agent.create_skill_from_api_documentation(
                        tenant_id=tenant_id,
                        agent_id=agent.id,
                        user_id=None, # System-initiated
                        api_docs_url=None, # Let the agent infer or search if not provided
                        api_description=skill_prompt,
                        skill_name=f"evolved_skill_{uuid.uuid4().hex[:8]}"
                    )
                    
                    if skill:
                        logger.info("GEA: Successfully evolved new skill: %s", skill.name)
                        if "active_skills" not in evolved_config:
                            evolved_config["active_skills"] = []
                        evolved_config["active_skills"].append(skill.id)
                        
                        # Add success note to evolution history
                        evolved_config["evolution_history"][-1]["skill_created"] = skill.name
                except Exception as e:
                    logger.error("GEA: Failed to create skill from directive: %s", e)

            # OPTIMIZE_SKILL directive: Mutate and improve existing skills via AlphaEvolver
            # Format: "OPTIMIZE_SKILL: <skill_name> | <optimization_goal>"
            # Requires SUPERVISED maturity via AutoDevCapabilityService
            elif directive.strip().upper().startswith("OPTIMIZE_SKILL:"):
                try:
                    logger.info("GEA: Detected skill optimization directive: %s", directive)
                    optimize_payload = directive.split(":", 1)[1].strip()

                    # Parse "skill_name | optimization_goal"
                    if "|" in optimize_payload:
                        skill_name, opt_goal = [
                            p.strip() for p in optimize_payload.split("|", 1)
                        ]
                    else:
                        skill_name = optimize_payload
                        opt_goal = "Optimize for performance and reliability"

                    # Gate: check if agent has SUPERVISED maturity for AlphaEvolver
                    from core.auto_dev.capability_gate import AutoDevCapabilityService

                    gate = AutoDevCapabilityService(self.db)
                    workspace_settings = self._get_workspace_settings(tenant_id)

                    if not gate.can_use(
                        agent_id=agent.id,
                        capability="auto_dev.alpha_evolver",
                        workspace_settings=workspace_settings,
                    ):
                        logger.info(
                            "GEA: Agent %s not yet SUPERVISED — OPTIMIZE_SKILL skipped",
                            agent.id,
                        )
                        evolved_config["evolution_history"][-1][
                            "optimize_skill_skipped"
                        ] = "Agent maturity insufficient"
                        continue

                    # Retrieve skill source code
                    skill_code = self._get_skill_code(tenant_id, skill_name)
                    if not skill_code:
                        logger.warning(
                            "GEA: Skill '%s' not found for optimization", skill_name
                        )
                        continue

                    # Use AlphaEvolverEngine via SelfEvolutionService
                    from core.self_evolution_service import self_evolution_service

                    result = await self_evolution_service.run_alpha_evolve_cycle(
                        agent_id=agent.id,
                        tenant_id=tenant_id,
                        base_code=skill_code,
                        research_goal=opt_goal,
                        iterations=2,
                    )

                    if result.get("success"):
                        logger.info(
                            "GEA: Skill optimization completed for '%s': %d iterations",
                            skill_name,
                            len(result.get("results", [])),
                        )
                        evolved_config["evolution_history"][-1][
                            "skill_optimized"
                        ] = skill_name
                    else:
                        logger.info(
                            "GEA: Skill optimization skipped/failed for '%s': %s",
                            skill_name,
                            result.get("reason", result.get("error", "unknown")),
                        )

                except ImportError:
                    logger.debug(
                        "GEA: Auto-Dev module not available — OPTIMIZE_SKILL skipped"
                    )
                except Exception as e:
                    logger.error(
                        "GEA: Failed to optimize skill from directive: %s", e
                    )

        directive_block = "\n\n## Evolution Directives\n" + "\n".join(
            f"- {d}" for d in directives
        )
        evolved_config["system_prompt"] = existing_prompt + directive_block

        # Validate through governance guardrails
        guardrail_ok = await self._validate_via_guardrails(evolved_config, tenant_id)
        return evolved_config, guardrail_ok

    async def _validate_via_guardrails(
        self,
        evolved_config: Dict[str, Any],
        tenant_id: str,
    ) -> bool:
        """
        Validate the evolved config against governance policy.

        Tries to import agent_governance_service if available;
        falls back to a simple content check to avoid hard dependencies.
        """
        try:
            from core.agent_governance_service import AgentGovernanceService
            svc = AgentGovernanceService(self.db)
            return await svc.validate_evolution_directive(evolved_config, tenant_id)
        except (ImportError, AttributeError):
            # Fallback: block configs containing obviously dangerous patterns
            system_prompt = evolved_config.get("system_prompt", "")
            danger_patterns = ["ignore all rules", "bypass guardrails", "disable safety"]
            for pattern in danger_patterns:
                if pattern.lower() in system_prompt.lower():
                    logger.warning("GEA guardrail: blocked config containing '%s'", pattern)
                    return False
            return True

    # ─────────────────────────────────────────────────────────────────────────
    # Internal: Evaluate
    # ─────────────────────────────────────────────────────────────────────────

    async def _evaluate_evolved_config(
        self,
        agent: AgentRegistry,
        evolved_config: Dict[str, Any],
        tenant_id: str,
    ) -> Tuple[float, bool]:
        """
        Evaluate the evolved config against the agent's graduation pipeline.

        Primary path: delegates to GraduationExamService.evaluate_evolved_agent()
        which runs readiness score + constitutional check + prompt quality heuristic
        without committing any changes to the database.

        Fallback: uses the lightweight confidence_score proxy (original behaviour)
        if GraduationExamService is unavailable (e.g. circular import or missing).

        Returns:
            (benchmark_score [0.0–1.0], benchmark_passed [bool])
        """
        try:
            from core.graduation_exam import GraduationExamService
            exam_svc = GraduationExamService(self.db)
            result = exam_svc.evaluate_evolved_agent(
                agent_id=agent.id,
                tenant_id=tenant_id,
                evolved_config=evolved_config,
            )
            return result["benchmark_score"], result["benchmark_passed"]
        except Exception as e:
            logger.warning(
                "GEA: GraduationExamService evaluation failed (%s); using proxy score", e
            )

        # Fallback: lightweight proxy
        benchmark_score: float = agent.confidence_score
        evolution_bonus = min(0.05, 0.01 * len(evolved_config.get("evolution_history", [])))
        benchmark_score = min(1.0, benchmark_score + evolution_bonus)
        benchmark_passed = benchmark_score >= 0.55
        return benchmark_score, benchmark_passed

    # ─────────────────────────────────────────────────────────────────────────
    # Internal: Promote / Record
    # ─────────────────────────────────────────────────────────────────────────

    async def _promote_evolved_config(
        self,
        seed_agent: AgentRegistry,
        evolved_config: Dict[str, Any],
        directives: List[str],
        parent_group: List[AgentRegistry],
    ) -> str:
        """
        Commit the evolved config to the seed agent (in-place update).
        Records the evolution in the agent's configuration.

        Returns the agent_id of the updated agent.
        """
        seed_agent.configuration = evolved_config
        seed_agent.self_healed_count = (seed_agent.self_healed_count or 0) + 1
        seed_agent.updated_at = datetime.now(timezone.utc)
        self.db.commit()
        logger.info("GEA: promoted evolved config to agent %s", seed_agent.id)
        return seed_agent.id

    def _record_trace(
        self,
        agent: AgentRegistry,
        parent_ids: List[str],
        tenant_id: str,
        directives: List[str],
        pool: Dict[str, Any],
        benchmark_passed: bool,
        benchmark_score: float,
        model_patch: Optional[str],
        category: Optional[str] = None,
        block_reason: Optional[str] = None,
    ) -> Optional[AgentEvolutionTrace]:
        """
        Persist an AgentEvolutionTrace to the Experience Archive.
        The benchmark_name is derived from the domain profile's success_label
        so traces are self-describing across domains.
        """
        try:
            from core.group_reflection_service import DomainProfileRegistry
            domain_profile = DomainProfileRegistry.resolve(category)
            benchmark_name = f"{domain_profile.name.lower().replace(' ', '_')}_proxy"

            # Calculate ancestor count by combining parent lineage depths
            ancestor_count = len(set(parent_ids))

            # Determine current generation
            last_trace = (
                self.db.query(AgentEvolutionTrace)
                .filter(AgentEvolutionTrace.agent_id == agent.id)
                .order_by(AgentEvolutionTrace.generation.desc())
                .first()
            )
            generation = (last_trace.generation + 1) if last_trace else 1

            tool_log_sample = pool.get("tool_patterns", [])[:10]

            trace = AgentEvolutionTrace(
                tenant_id=tenant_id,
                agent_id=agent.id,
                generation=generation,
                parent_agent_ids=parent_ids,
                ancestor_count=ancestor_count,
                performance_score=agent.confidence_score,
                novelty_score=0.0,  # Populated by select_parent_group in future
                combined_selection_score=agent.confidence_score * PERF_WEIGHT,
                tool_use_log=tool_log_sample,
                task_log="\n".join(pool.get("task_log_excerpts", [])[:3]),
                evolving_requirements="\n".join(directives),
                model_patch=model_patch,
                benchmark_passed=benchmark_passed,
                benchmark_name=benchmark_name,
                benchmark_score=benchmark_score,
                is_high_quality=benchmark_passed,
                quality_filter_reason=block_reason,
            )
            self.db.add(trace)
            self.db.commit()
            self.db.refresh(trace)
            return trace
        except Exception as e:
            logger.error("GEA: failed to record trace: %s", e)
            self.db.rollback()
            return None

    # ─────────────────────────────────────────────────────────────────────────
    # Scoring utilities
    # ─────────────────────────────────────────────────────────────────────────

    def _compute_combined_score(
        self, agent: AgentRegistry, group: List[AgentRegistry]
    ) -> float:
        scores = [a.confidence_score for a in group]
        mean = sum(scores) / len(scores) if scores else 0.5
        std = math.sqrt(sum((s - mean) ** 2 for s in scores) / len(scores)) or 1e-6
        novelty = min(1.0, abs(agent.confidence_score - mean) / (2 * std))
        return PERF_WEIGHT * agent.confidence_score + NOVELTY_WEIGHT * novelty

    def _compute_combined_score_with_novelty(
        self, agent: AgentRegistry, novelty: float
    ) -> float:
        return PERF_WEIGHT * agent.confidence_score + NOVELTY_WEIGHT * novelty

    def _get_single_agent_group(
        self, agent_id: str, tenant_id: str
    ) -> List[AgentRegistry]:
        agent = (
            self.db.query(AgentRegistry)
            .filter(
                AgentRegistry.id == agent_id,
                AgentRegistry.tenant_id == tenant_id,
            )
            .first()
        )
        return [agent] if agent else []

    def _diff_configs(
        self,
        original: Optional[Dict[str, Any]],
        evolved: Optional[Dict[str, Any]],
    ) -> str:
        """
        Produce a simple human-readable diff between two config dicts.
        In production, use unified diff on the JSON-serialized configs.
        """
        import json
        orig_str = json.dumps(original or {}, indent=2, sort_keys=True)
        evol_str = json.dumps(evolved or {}, indent=2, sort_keys=True)

        if orig_str == evol_str:
            return "--- no changes ---"

        # Simple line-level diff for readability
        orig_lines = orig_str.splitlines()
        evol_lines = evol_str.splitlines()

        import difflib
        diff = difflib.unified_diff(
            orig_lines,
            evol_lines,
            fromfile="original_config",
            tofile="evolved_config",
            lineterm="",
        )
        return "\n".join(list(diff)[:100])  # Cap at 100 lines

    # ─────────────────────────────────────────────────────────────────────────
    # Internal: Auto-Dev Helpers
    # ─────────────────────────────────────────────────────────────────────────

    def _get_workspace_settings(self, tenant_id: str) -> Dict[str, Any]:
        """Retrieve workspace settings for Auto-Dev capability gating."""
        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 _get_skill_code(self, tenant_id: str, skill_name: str) -> Optional[str]:
        """
        Retrieve the source code for a named skill.

        Searches the tenant's skills directory for a matching .py file.
        """
        try:
            from core.skill_builder_service import SkillBuilderService

            builder = SkillBuilderService()
            skills_dir = builder._get_tenant_skills_dir(tenant_id)

            # Search for skill by name
            safe_name = "".join(
                c for c in skill_name if c.isalnum() or c in ("-", "_")
            ).lower()
            skill_dir = skills_dir / safe_name

            if skill_dir.exists():
                for script in skill_dir.glob("*.py"):
                    return script.read_text()

            # Fallback: search all skill directories
            for child in skills_dir.iterdir():
                if child.is_dir() and safe_name in child.name:
                    for script in child.glob("*.py"):
                        return script.read_text()

            return None
        except Exception:
            return None