orchestrata_ultimate.verification.verification.engine

Reference › orchestrata_ultimate › verification › verification › engine

VerificationEngine is the core of a verification pass over a VerificationSession. It composes a ConstraintEvaluator and a CheckpointVerifier, creating them when none are injected, and is used by the orchestrator's verification_gate.

• verify returns the stored results at once if the session is already complete. Otherwise it evaluates all constraints in dependency order, stores each result on the session, and sets the status to completed when every constraint is satisfied. If any fails it increments attempt_count and, once the maximum is exceeded, sets the status to escalated.
• verify_checkpoint verifies the session's current checkpoint and then calls the verifier's transition_state, which advances the checkpoint or counts a failed attempt.
• generate_feedback builds a Feedback with one Suggestion per unsatisfied result (severity "error" for error status, otherwise "warning"), a summary such as "3/5 constraints satisfied" and the pass rate.

The engine mutates the session object it is given; persistence is the job of the session manager.

At a glance

Class diagram

Dependency diagram

API

Classes

VerificationEngine

Decorators: traces(spec='035', story='US-003', reqs=['REQ-035-US-003-009', 'REQ-035-US-003-016', 'REQ-035-US-008-009', 'REQ-035-US-… · Line 38

Core verification engine orchestrating evaluation.

Source

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

"""Core verification engine."""

# Spec: 035 - Verification Agent

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.checkpoints.verifier import (
    CheckpointVerifier,
)
from orchestrata_ultimate.verification.constraints.evaluator import (
    ConstraintEvaluator,
)
from orchestrata_ultimate.verification.types import (
    CheckpointResult,
    Feedback,
    SessionStatus,
    Suggestion,
    VerificationResult,
    VerificationSession,
    VerificationStatus,
)

logger = logging.getLogger(__name__)


@traces(
    spec="035",
    story="US-003",
    reqs=[
        "REQ-035-US-003-009",
        "REQ-035-US-003-016",
        "REQ-035-US-008-009",
        "REQ-035-US-008-017",
    ],
)
class VerificationEngine:
    """Core verification engine orchestrating evaluation."""

    def __init__(
        self,
        evaluator: ConstraintEvaluator | None = None,
        checkpoint_verifier: CheckpointVerifier | None = None,
    ) -> None:
        self._evaluator = evaluator or ConstraintEvaluator()
        self._checkpoint = checkpoint_verifier or CheckpointVerifier(self._evaluator)

    @traces(
        spec="035",
        story="US-003",
        reqs=[
            "REQ-035-US-003-009",
            "REQ-035-US-003-016",
            "REQ-035-US-008-009",
            "REQ-035-US-008-017",
        ],
    )
    def verify(
        self, session: VerificationSession, context: dict[str, Any] | None = None
    ) -> list[VerificationResult]:
        """Run verification for a session."""
        if session.is_complete:
            return list(session.results.values())

        results = self._evaluator.evaluate_with_dependencies(
            session.constraints, context
        )
        for result in results:
            session.results[result.constraint_id] = result

        all_satisfied = all(r.is_satisfied for r in results)
        if all_satisfied:
            session.status = SessionStatus.COMPLETED
        else:
            session.attempt_count += 1
            if session.max_exceeded:
                session.status = SessionStatus.ESCALATED

        return results

    def verify_checkpoint(
        self, session: VerificationSession, context: dict[str, Any] | None = None
    ) -> CheckpointResult:
        """Verify the current checkpoint of a session."""
        result = self._checkpoint.verify_checkpoint(
            session.current_checkpoint, session.constraints, context
        )
        self._checkpoint.transition_state(session, session.current_checkpoint, result)
        return result

    def generate_feedback(self, results: list[VerificationResult]) -> Feedback:
        """Generate actionable feedback from verification results."""
        satisfied = sum(1 for r in results if r.is_satisfied)
        total = len(results) if results else 1

        suggestions = []
        for r in results:
            if not r.is_satisfied:
                suggestions.append(
                    Suggestion(
                        constraint_id=r.constraint_id,
                        message=f"Fix: {r.details}",
                        severity=(
                            "error"
                            if r.status == VerificationStatus.ERROR
                            else "warning"
                        ),
                    )
                )

        return Feedback(
            results=results,
            suggestions=suggestions,
            summary=f"{satisfied}/{total} constraints satisfied",
            pass_rate=satisfied / total,
        )