voidea/app/agents/triggers.py

"""Agent triggers - automated execution mechanisms."""

import asyncio
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable

from app.agents.base import AgentResult, AgentTrigger
from app.agents.registry import registry


class TriggerManager:
    """Manages agent triggers and execution scheduling."""

    def __init__(self):
        self._cron_tasks: list[asyncio.Task] = []
        self._running = False

    async def trigger_pre_commit(self, files: list[str]) -> dict[str, AgentResult]:
        """Trigger agents on pre-commit hook.

        Args:
            files: List of changed files

        Returns:
            Dict of agent -> result
        """
        results = {}

        context = {
            "trigger": "pre_commit",
            "files": files,
        }

        audit_result = await registry.run_agent("audit_agent", {
            "action": "quick",
            "paths": self._get_affected_paths(files),
        })
        results["audit_agent"] = audit_result

        doc_result = await registry.run_agent("doc_agent", {
            "action": "update_readme",
            "files": files,
        })
        results["doc_agent"] = doc_result

        return results

    async def trigger_push(self, branch: str) -> dict[str, AgentResult]:
        """Trigger agents on git push.

        Args:
            branch: Branch name

        Returns:
            Dict of agent -> result
        """
        results = {}

        context = {
            "trigger": "push",
            "branch": branch,
        }

        if branch in ("main", "develop"):
            spec_result = await registry.run_agent("spec_agent", context)
            results["spec_agent"] = spec_result

            doc_result = await registry.run_agent("doc_agent", context)
            results["doc_agent"] = doc_result

        return results

    async def trigger_cron(self) -> dict[str, AgentResult]:
        """Trigger agents on scheduled cron.

        Returns:
            Dict of agent -> result
        """
        results = {}

        audit_result = await registry.run_agent("audit_agent", {
            "action": "full",
            "paths": ["app"],
        })
        results["audit_agent"] = audit_result

        observer_result = await registry.run_agent("observer_agent", {
            "action": "report",
            "period": "daily",
        })
        results["observer_agent"] = observer_result

        evolution_result = await registry.run_agent("evolution_agent", {
            "action": "analyze",
        })
        results["evolution_agent"] = evolution_result

        return results

    async def trigger_manual(
        self,
        agent_name: str,
        context: dict[str, Any] | None = None,
    ) -> AgentResult:
        """Trigger a specific agent manually.

        Args:
            agent_name: Agent name
            context: Optional context

        Returns:
            AgentResult
        """
        return await registry.run_agent(agent_name, context)

    def _get_affected_paths(self, files: list[str]) -> list[str]:
        """Get affected paths from file list.

        Args:
            files: List of file paths

        Returns:
            List of unique directory paths
        """
        paths = set()
        for file in files:
            parts = Path(file).parts
            if len(parts) > 1 and parts[0] == "app":
                paths.add(parts[1])
        return list(paths) if paths else ["app"]


class PreCommitHook:
    """Pre-commit hook integration."""

    @staticmethod
    def install() -> None:
        """Install pre-commit hook."""
        project_root = Path(__file__).parent.parent.parent
        hook_dir = project_root / ".git" / "hooks"

        if not hook_dir.exists():
            return

        hook_content = """#!/bin/sh
# VoIdea Pre-commit Hook

python -m app.agents.triggers.pre_commit
"""

        hook_path = hook_dir / "pre-commit"
        hook_path.write_text(hook_content, encoding="utf-8")

    @staticmethod
    async def run() -> dict[str, Any]:
        """Run pre-commit checks.

        Returns:
            Check results
        """
        manager = TriggerManager()
        project_root = Path(__file__).parent.parent.parent

        try:
            result = subprocess.run(
                ["git", "diff", "--cached", "--name-only"],
                capture_output=True,
                text=True,
                cwd=str(project_root),
            )
            files = result.stdout.strip().split("\n")
            files = [f for f in files if f]
        except Exception:
            files = []

        if not files:
            return {"status": "skipped", "message": "No files to check"}

        results = await manager.trigger_pre_commit(files)

        failed = [name for name, result in results.items() if not result.success]

        return {
            "status": "passed" if not failed else "failed",
            "failed_agents": failed,
            "results": {name: r.message for name, r in results.items()},
        }


async def run_agent_manually(
    agent_name: str,
    action: str | None = None,
    **kwargs: Any,
) -> AgentResult:
    """Convenience function to run an agent manually.

    Args:
        agent_name: Name of agent to run
        action: Optional action to perform
        **kwargs: Additional context

    Returns:
        AgentResult
    """
    context = kwargs if not action else {"action": action, **kwargs}
    return await registry.run_agent(agent_name, context)