orchestrata_ultimate.verification.session.verification_session

Reference › orchestrata_ultimate › verification › session › verification_session

VerificationSessionManager creates and tracks VerificationSession objects, with optional Redis persistence. It is used by the orchestrator's verification_gate. Sessions live in an in-memory dictionary. create_session builds a session from constraints with a maximum attempt count (default 3); create_session_from_intent first converts an intent into constraints with the IntentBrokerAdapter (REQ-035-US-009-001/002). get_session, list_sessions and session_count read the store; increment_attempt (returns the new count, or 0 for an unknown session), is_max_exceeded, update_results and update_status change it.

Every mutation calls _persist, which writes the session's to_dict as JSON to Redis under the key <prefix>:<session id> (default prefix orchestrata_ultimate:verification:session) when a client was supplied. A Redis failure is logged as a warning and does not raise. Only writes go to Redis; reads use the in-memory dictionary. The session type is in types.

At a glance

Class diagram

Dependency diagram

API

Classes

VerificationSessionManager

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

Manages verification sessions with optional Redis persistence.

Source

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

# mypy: ignore-errors
"""Verification session state management.

Wiring status (2026-07-06, W6.2): ``IntentBrokerAdapter`` (REQ-035-US-009-001/002)
was fully built and unit-tested but had zero callers. It is now wired into
:meth:`VerificationSessionManager.create_session_from_intent`, a new
additive method that converts an Intent Broker intent into constraints
before delegating to the existing (unchanged) :meth:`create_session`. This
is purely additive -- no existing method's behavior changes.
"""

# Spec: 035 - Verification Agent

from __future__ import annotations

import json
import logging
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.integrations.intent_broker import (
    IntentBrokerAdapter,
)
from orchestrata_ultimate.verification.types import (
    Constraint,
    SessionStatus,
    VerificationResult,
    VerificationSession,
)

if TYPE_CHECKING:
    from redis import Redis

logger = logging.getLogger(__name__)


@traces(
    spec="035",
    story="US-001",
    reqs=["REQ-035-US-001-010"],
)
class VerificationSessionManager:
    """Manages verification sessions with optional Redis persistence."""

    def __init__(
        self,
        redis: Redis | None = None,
        prefix: str = "orchestrata_ultimate:verification:session",
    ) -> None:
        self._redis = redis
        self._prefix = prefix
        self._sessions: dict[str, VerificationSession] = {}

    @traces(spec="035", story="US-001", reqs=["REQ-035-US-001-010"])
    def create_session(
        self, constraints: list[Constraint], max_attempts: int = 3
    ) -> VerificationSession:
        """Create a new verification session."""
        session = VerificationSession(
            constraints=constraints,
            max_attempts=max_attempts,
        )
        self._sessions[session.id] = session
        self._persist(session)
        return session

    @traces(
        spec="035",
        story="US-009",
        reqs=[
            "REQ-035-US-009-001",
            "REQ-035-US-009-002",
            "REQ-035-US-009-009",
            "REQ-035-US-009-010",
        ],
    )
    def create_session_from_intent(
        self,
        intent: dict[str, Any],
        broker: IntentBrokerAdapter | None = None,
        max_attempts: int = 3,
    ) -> VerificationSession:
        """Create a session by extracting constraints from an intent.

        Wires the Intent Broker integration adapter (REQ-035-US-009-001/002)
        into session creation: intents are converted to ``Constraint``
        objects via :class:`IntentBrokerAdapter` before delegating to
        :meth:`create_session`.
        """
        broker = broker or IntentBrokerAdapter()
        constraints = broker.intent_to_constraints(intent)
        return self.create_session(constraints, max_attempts=max_attempts)

    def get_session(self, session_id: str) -> VerificationSession | None:
        """Get a session by ID."""
        return self._sessions.get(session_id)

    @traces(
        spec="035", story="US-008", reqs=["REQ-035-US-008-001", "REQ-035-US-008-010"]
    )
    def increment_attempt(self, session_id: str) -> int:
        """Increment and return the attempt count."""
        session = self._sessions.get(session_id)
        if session:
            session.attempt_count += 1
            self._persist(session)
            return session.attempt_count
        return 0

    @traces(
        spec="035",
        story="US-008",
        reqs=["REQ-035-US-008-002", "REQ-035-US-008-011", "REQ-035-US-008-018"],
    )
    def is_max_exceeded(self, session_id: str) -> bool:
        """Check if max attempts exceeded."""
        session = self._sessions.get(session_id)
        return session.max_exceeded if session else False

    @traces(
        spec="035",
        story="US-003",
        reqs=[
            "REQ-035-US-003-010",
            "REQ-035-US-003-017",
            "REQ-035-US-004-008",
            "REQ-035-US-008-007",
            "REQ-035-US-009-006",
            "REQ-035-US-009-013",
        ],
    )
    def update_results(
        self, session_id: str, results: list[VerificationResult]
    ) -> None:
        """Update session with verification results."""
        session = self._sessions.get(session_id)
        if session:
            for r in results:
                session.results[r.constraint_id] = r
            self._persist(session)

    @traces(
        spec="035", story="US-008", reqs=["REQ-035-US-008-008", "REQ-035-US-008-016"]
    )
    def update_status(self, session_id: str, status: SessionStatus) -> None:
        """Update session status."""
        session = self._sessions.get(session_id)
        if session:
            session.status = status
            self._persist(session)

    def list_sessions(self) -> list[str]:
        """List all session IDs."""
        return list(self._sessions.keys())

    @property
    def session_count(self) -> int:
        return len(self._sessions)

    @traces(spec="035", story="US-001", reqs=["REQ-035-US-001-010"])
    def _persist(self, session: VerificationSession) -> None:
        """Persist session to Redis if available."""
        if self._redis:
            try:
                key = f"{self._prefix}:{session.id}"
                self._redis.set(key, json.dumps(session.to_dict()))
            except Exception as exc:
                logger.warning("Failed to persist session: %s", exc)