orchestrata_ultimate.verification.checkpoints.concurrent

Reference › orchestrata_ultimate › verification › checkpoints › concurrent

ConcurrentCheckpointEvaluator runs several checkpoints through a CheckpointVerifier (Spec 035 US-004). evaluate_all is asynchronous: it bounds parallelism with an asyncio.Semaphore of max_concurrent (default 5), launches one task per requested checkpoint with asyncio.gather and returns a dictionary of CheckpointResult keyed by checkpoint. evaluate_sequential walks the checkpoints in the order defined in definitions and stops at the first failure, returning the results gathered so far. Both only read the constraints; they do not change a session.

At a glance

Class diagram

Dependency diagram

API

Classes

ConcurrentCheckpointEvaluator

Decorators: traces(spec='035', story='US-004', reqs=['REQ-035-US-004-009', 'REQ-035-US-004-014', 'REQ-035-US-004-020', 'REQ-035-US-… · Line 32

Evaluates multiple checkpoints concurrently.

Source

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

"""Concurrent checkpoint evaluation."""

# Spec: 035 - Verification Agent

import asyncio
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.types import (
    CheckpointName,
    CheckpointResult,
    Constraint,
)

logger = logging.getLogger(__name__)


@traces(
    spec="035",
    story="US-004",
    reqs=[
        "REQ-035-US-004-009",
        "REQ-035-US-004-014",
        "REQ-035-US-004-020",
        "REQ-035-US-004-021",
    ],
)
class ConcurrentCheckpointEvaluator:
    """Evaluates multiple checkpoints concurrently."""

    def __init__(
        self, verifier: CheckpointVerifier | None = None, max_concurrent: int = 5
    ) -> None:
        self._verifier = verifier or CheckpointVerifier()
        self._max_concurrent = max_concurrent

    @traces(
        spec="035",
        story="US-004",
        reqs=["REQ-035-US-004-009", "REQ-035-US-004-020", "REQ-035-US-004-021"],
    )
    async def evaluate_all(
        self,
        checkpoints: list[CheckpointName],
        constraints: list[Constraint],
        context: dict[str, Any] | None = None,
    ) -> dict[CheckpointName, CheckpointResult]:
        """Evaluate multiple checkpoints concurrently."""
        semaphore = asyncio.Semaphore(self._max_concurrent)
        results: dict[CheckpointName, CheckpointResult] = {}

        @traces(
            spec="035",
            story="US-004",
            reqs=["REQ-035-US-004-009", "REQ-035-US-004-020", "REQ-035-US-004-021"],
        )
        async def eval_one(cp: CheckpointName) -> None:
            async with semaphore:
                result = self._verifier.verify_checkpoint(cp, constraints, context)
                results[cp] = result

        tasks = [eval_one(cp) for cp in checkpoints]
        await asyncio.gather(*tasks)
        return results

    @traces(spec="035", story="US-004", reqs=["REQ-035-US-004-002"])
    async def evaluate_sequential(
        self,
        constraints: list[Constraint],
        context: dict[str, Any] | None = None,
    ) -> list[CheckpointResult]:
        """Evaluate checkpoints sequentially, stopping on failure."""
        from orchestrata_ultimate.verification.checkpoints.definitions import (
            CHECKPOINT_ORDER,
        )

        results = []
        for cp in CHECKPOINT_ORDER:
            result = self._verifier.verify_checkpoint(cp, constraints, context)
            results.append(result)
            if not result.passed:
                break
        return results