orchestrata_ultimate.verification.checkpoints.verifier

Reference › orchestrata_ultimate › verification › checkpoints › verifier

CheckpointVerifier checks one checkpoint at a time and advances the session (Spec 035 US-004). It is used by the concurrent evaluator and by the verification engine.

verify_checkpoint selects the constraints whose predicate is in the checkpoint's criteria list, evaluates them through a ConstraintEvaluator and returns a CheckpointResult that passes only when every relevant constraint is satisfied. A checkpoint with no relevant constraints passes with a note saying so. get_checkpoint_criteria returns the predicate names. transition_state updates a VerificationSession: on a pass it moves current_checkpoint to the next checkpoint, or sets the status to completed after the last; on a failure it increments attempt_count, which feeds the escalation handler.

At a glance

Class diagram

Dependency diagram

API

Classes

CheckpointVerifier

Decorators: traces(spec='035', story='US-004', reqs=['REQ-035-US-004-002', 'REQ-035-US-004-004', 'REQ-035-US-004-006', 'REQ-035-US-… · Line 40

Verifies checkpoint criteria and manages state transitions.

Source

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

"""Checkpoint verification and state transitions."""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.checkpoints.definitions import (
    get_checkpoint_predicates,
    get_next_checkpoint,
)
from orchestrata_ultimate.verification.constraints.evaluator import (
    ConstraintEvaluator,
)
from orchestrata_ultimate.verification.types import (
    CheckpointName,
    CheckpointResult,
    Constraint,
    VerificationSession,
)

logger = logging.getLogger(__name__)


@traces(
    spec="035",
    story="US-004",
    reqs=[
        "REQ-035-US-004-002",
        "REQ-035-US-004-004",
        "REQ-035-US-004-006",
        "REQ-035-US-004-007",
        "REQ-035-US-004-011",
        "REQ-035-US-004-012",
        "REQ-035-US-004-016",
        "REQ-035-US-004-017",
        "REQ-035-US-004-018",
        "REQ-035-US-004-019",
    ],
)
class CheckpointVerifier:
    """Verifies checkpoint criteria and manages state transitions."""

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

    @traces(
        spec="035", story="US-004", reqs=["REQ-035-US-004-002", "REQ-035-US-004-011"]
    )
    def verify_checkpoint(
        self,
        checkpoint: CheckpointName,
        constraints: list[Constraint],
        context: dict[str, Any] | None = None,
    ) -> CheckpointResult:
        """Verify all constraints for a checkpoint."""
        predicates = get_checkpoint_predicates(checkpoint)
        relevant = [c for c in constraints if c.predicate in predicates]

        if not relevant:
            return CheckpointResult(
                checkpoint=checkpoint,
                passed=True,
                total_constraints=0,
                satisfied_count=0,
                details="No constraints for this checkpoint",
            )

        results = self._evaluator.evaluate_all(relevant, context)
        satisfied = sum(1 for r in results if r.is_satisfied)

        return CheckpointResult(
            checkpoint=checkpoint,
            passed=satisfied == len(relevant),
            results=results,
            total_constraints=len(relevant),
            satisfied_count=satisfied,
        )

    @traces(
        spec="035", story="US-004", reqs=["REQ-035-US-004-006", "REQ-035-US-004-017"]
    )
    def get_checkpoint_criteria(self, checkpoint: CheckpointName) -> list[str]:
        """Get the predicate names required for a checkpoint."""
        return get_checkpoint_predicates(checkpoint)

    @traces(
        spec="035",
        story="US-004",
        reqs=[
            "REQ-035-US-004-004",
            "REQ-035-US-004-007",
            "REQ-035-US-004-012",
            "REQ-035-US-004-016",
            "REQ-035-US-004-018",
        ],
    )
    def transition_state(
        self,
        session: VerificationSession,
        checkpoint: CheckpointName,
        result: CheckpointResult,
    ) -> None:
        """Transition session state based on checkpoint result."""
        if result.passed:
            next_cp = get_next_checkpoint(checkpoint)
            if next_cp:
                session.current_checkpoint = next_cp
            else:
                from orchestrata_ultimate.verification.types import (
                    SessionStatus,
                )

                session.status = SessionStatus.COMPLETED
        else:
            session.attempt_count += 1