orchestrata_ultimate.verification.agents.supervisor_agent

Reference › orchestrata_ultimate › verification › agents › supervisor_agent

SupervisorAgent (default id U:Sup) is the reviewing role of the verification loop (Spec 035 US-007). review_work records a review of another agent's checkpoint artifacts; the record it builds always carries approved true and an empty comment list. generate_proof builds a Proof deterministically, defaulting to the practical-explanation type. generate_proof_via_llm is the asynchronous variant: it calls the injected LLMModel's generate_proof and records the model name in approaches_tried, and it falls back to the deterministic proof when no model is configured or when the model call raises (REQ-035-US-005-006, a warning is logged). submit_proof returns a summary dictionary for validation by the verification agent. The model protocol is defined in base_model.

At a glance

Class diagram

Dependency diagram

API

Classes

SupervisorAgent

Decorators: traces(spec='035', story='US-007', reqs=['REQ-035-US-007-020']) · Line 28

Reviews work and generates proofs.

Source

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

"""Supervisor agent — reviews work, generates proofs.

Wiring status (2026-07-06, W6.2): the ``LLMModel`` protocol
(REQ-035-US-005-001/011) had zero callers anywhere in the Supervisor's proof
generation path -- ``generate_proof`` was a purely deterministic stub. An
optional ``llm_model`` constructor param plus the new
``generate_proof_via_llm`` method now let a real model generate proof
content, falling back to the existing deterministic ``generate_proof``
when no model is configured or the model call fails (REQ-035-US-005-006:
handle model unavailability gracefully). The default is ``None`` and
``generate_proof`` itself is untouched -- zero behavior change for
existing callers.
"""

# Spec: 035 - Verification Agent

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.models.base_model import LLMModel
from orchestrata_ultimate.verification.types import Proof, ProofType

logger = logging.getLogger(__name__)


@traces(spec="035", story="US-007", reqs=["REQ-035-US-007-020"])
class SupervisorAgent:
    """Reviews work and generates proofs.

    Oversees the execution agent and validates artifacts.
    """

    def __init__(
        self, agent_id: str = "U:Sup", llm_model: LLMModel | None = None
    ) -> None:
        self._agent_id = agent_id
        self._llm = llm_model
        self._reviews: list[dict[str, Any]] = []

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

    def review_work(
        self, agent_id: str, checkpoint: str, artifacts: dict[str, Any] | None = None
    ) -> dict[str, Any]:
        """Review work submitted by an agent."""
        review = {
            "reviewer": self._agent_id,
            "agent_id": agent_id,
            "checkpoint": checkpoint,
            "artifacts": artifacts or {},
            "approved": True,
            "comments": [],
        }
        self._reviews.append(review)
        return review

    @traces(spec="035", story="US-007", reqs=["REQ-035-US-007-020"])
    def generate_proof(
        self,
        constraint_id: str,
        proof_type: ProofType = ProofType.PRACTICAL_EXPLANATION,
        context: dict[str, Any] | None = None,
    ) -> Proof:
        """Generate a proof for a constraint."""
        ctx = context or {}
        return Proof(
            constraint_id=constraint_id,
            proof_type=proof_type,
            content=ctx,
            approaches_tried=["direct_verification"],
            formal_reasoning=f"Verified by {self._agent_id}",
        )

    @traces(
        spec="035",
        story="US-005",
        reqs=["REQ-035-US-005-001", "REQ-035-US-005-006", "REQ-035-US-005-011"],
    )
    async def generate_proof_via_llm(
        self,
        constraint_id: str,
        proof_type: ProofType = ProofType.PRACTICAL_EXPLANATION,
        context: dict[str, Any] | None = None,
    ) -> Proof:
        """Generate a proof via the configured LLM model.

        Falls back to the deterministic :meth:`generate_proof` when no
        model is wired in, or when the model call raises
        (REQ-035-US-005-006: handle model unavailability gracefully).
        """
        if self._llm is None:
            return self.generate_proof(constraint_id, proof_type, context)
        ctx = context or {}
        try:
            content = await self._llm.generate_proof({"id": constraint_id}, ctx)
        except Exception as exc:
            logger.warning("LLM proof generation failed: %s", exc)
            return self.generate_proof(constraint_id, proof_type, context)
        return Proof(
            constraint_id=constraint_id,
            proof_type=proof_type,
            content=content,
            approaches_tried=[f"llm:{self._llm.name}"],
            formal_reasoning=f"Verified by {self._agent_id} via {self._llm.name}",
        )

    @traces(spec="035", story="US-007", reqs=["REQ-035-US-007-020"])
    def submit_proof(self, proof: Proof) -> dict[str, Any]:
        """Submit a proof for validation."""
        return {
            "proof_id": proof.id,
            "constraint_id": proof.constraint_id,
            "proof_type": proof.proof_type.value,
            "submitted_by": self._agent_id,
        }

    @property
    def review_count(self) -> int:
        return len(self._reviews)