orchestrata_ultimate.verification.agents.execution_agent

Reference › orchestrata_ultimate › verification › agents › execution_agent

ExecutionAgent (default id U:Exec) is the implementing role of the three-agent verification loop (Spec 035 US-002). It keeps the constraints it was given with receive_constraints, records an artifact per checkpoint in execute_step, hands that artifact back from submit_for_verification, and stores feedback from the verifier with receive_feedback. The class docstring says that in the real system it drives the code-generation agent; here execute_step only records a dictionary with the checkpoint, agent id, constraint count and status "executed". It is stateful but touches no external service. The other two roles are supervisor_agent and verification_agent; the types come from types.

At a glance

Class diagram

Dependency diagram

API

Classes

ExecutionAgent

Decorators: traces(spec='035', story='US-002', reqs=['REQ-035-US-002-001']) · Line 18

Receives constraints, executes steps, submits for verification.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/agents/execution_agent.py

"""Execution agent — implements solutions."""

# Spec: 035 - Verification Agent

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.types import (
    CheckpointName,
    Constraint,
)

logger = logging.getLogger(__name__)


@traces(spec="035", story="US-002", reqs=["REQ-035-US-002-001"])
class ExecutionAgent:
    """Receives constraints, executes steps, submits for verification.

    In the real system, this drives the code generation agent.
    """

    def __init__(self, agent_id: str = "U:Exec") -> None:
        self._agent_id = agent_id
        self._constraints: list[Constraint] = []
        self._artifacts: dict[str, Any] = {}
        self._feedback_history: list[dict[str, Any]] = []

    @property
    def agent_id(self) -> str:
        return self._agent_id

    def receive_constraints(self, constraints: list[Constraint]) -> None:
        """Receive constraints to work against."""
        self._constraints = list(constraints)

    def execute_step(self, checkpoint: CheckpointName) -> dict[str, Any]:
        """Execute a step for the given checkpoint."""
        artifact = {
            "checkpoint": checkpoint.value,
            "agent_id": self._agent_id,
            "constraint_count": len(self._constraints),
            "status": "executed",
        }
        self._artifacts[checkpoint.value] = artifact
        return artifact

    def submit_for_verification(self, checkpoint: CheckpointName) -> dict[str, Any]:
        """Submit artifacts for verification."""
        return {
            "checkpoint": checkpoint.value,
            "artifacts": self._artifacts.get(checkpoint.value, {}),
            "agent_id": self._agent_id,
        }

    def receive_feedback(self, feedback: dict[str, Any]) -> None:
        """Receive and store feedback."""
        self._feedback_history.append(feedback)

    @property
    def constraint_count(self) -> int:
        return len(self._constraints)

    @property
    def artifact_count(self) -> int:
        return len(self._artifacts)

    @property
    def feedback_count(self) -> int:
        return len(self._feedback_history)