src/agents/executor/service.py - master - azathd/mutiagent

src/agents/executor/service.py
94 строки3 KB
Your Name
init
15 май 2026, 17:47
15 май 2026, 17:47359c81e
"""Executor: consume approved actions, audit, publish executed."""

from __future__ import annotations

import asyncio
import uuid
from datetime import UTC, datetime

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from src.agents.executor.audit_store import append_audit, find_by_action_id
from src.agents.executor.handlers import ActionHandlers
from src.common.bus import MessageBus
from src.common.db import async_session_fact, init_engine
from src.common.logging import configure_logging, get_logger
from src.common.models import Envelope
from src.common.settings import settings

log = get_logger("executor")
ADVISORY_LOCK = 424242


class ExecutorService:
    def __init__(self) -> None:
        self.bus = MessageBus()
        self.handlers = ActionHandlers()
        self._session_factory = async_session_fact(init_engine())

    async def _try_lock(self, session: AsyncSession) -> bool:
        result = await session.execute(text("SELECT pg_try_advisory_lock(:k)"), {"k": ADVISORY_LOCK})
        return bool(result.scalar())

    async def start(self) -> None:
        configure_logging()
        await self.bus.connect()
        await self.bus.subscribe_durable("xray.actions.approved", "executor", self._on_approved)
        log.info("executor_started", dry_run=settings.executor_dry_run)

    async def _on_approved(self, env: Envelope) -> None:
        payload = env.payload
        action_id = str(payload.get("action_id", uuid.uuid4()))
        action_type = str(payload["action_type"])
        target = str(payload["target"])
        actor = str(payload.get("actor", "orchestrator"))

        async with self._session_factory() as session:
            if not await self._try_lock(session):
                log.warning("executor_lock_busy")
                return
            existing = await find_by_action_id(session, action_id)
            if existing:
                result = existing.result
            else:
                snap_key = await self.handlers.snapshot(action_id, payload)
                try:
                    result = await self.handlers.apply(action_type, target, payload.get("payload", {}))
                    result["snapshot"] = snap_key
                except Exception as exc:
                    result = {"status": "error", "error": str(exc)}
                    await append_audit(
                        session,
                        action_id=action_id,
                        action_type=action_type,
                        actor=actor,
                        target=target,
                        payload=payload,
                        result=result,
                    )
                    raise
                await append_audit(
                    session,
                    action_id=action_id,
                    action_type=action_type,
                    actor=actor,
                    target=target,
                    payload=payload,
                    result=result,
                )

        out = Envelope(
            occurred_at=datetime.now(tz=UTC),
            producer="executor",
            payload={"action_id": action_id, "result": result},
        )
        await self.bus.publish("xray.actions.executed", out)

    async def run_forever(self) -> None:
        await self.start()
        await asyncio.Future()


async def run() -> None:
    await ExecutorService().run_forever()