orchestrata_ultimate.session.interfaces

Reference › orchestrata_ultimate › session › interfaces

The interfaces of the dual-storage architecture. ISessionManager describes the lifecycle: open a session with goals and an optional portfolio, close it with a summary, get one by id, list active sessions, pause and resume. IFactEmitter has emit, get_facts and query_facts; the FactEmitter implements it. IRecoveryManager has create_checkpoint, recover and list_checkpoints; the RecoveryManager implements it. ICarryoverManager covers creating, getting, resolving and linking carryover tasks, and IPortfolioManager returns a portfolio's sessions and aggregated context. The data types they use are in types.

At a glance

Class diagram

Dependency diagram

API

Classes

ISessionManager (protocol)

Bases: Protocol · Decorators: traces(spec='006', story='US-001', reqs=['REQ-006-FR10-005']) · Line 19

Interface for session lifecycle management.

IFactEmitter (protocol)

Bases: Protocol · Decorators: traces(spec='006', story='US-004', reqs=['REQ-006-FR8-001', 'REQ-006-US-004-001']) · Line 54

Interface for session fact emission.

IRecoveryManager (protocol)

Bases: Protocol · Line 75

Interface for session recovery.

ICarryoverManager (protocol)

Bases: Protocol · Line 91

Interface for work carryover between sessions.

IPortfolioManager (protocol)

Bases: Protocol · Line 115

Interface for portfolio management.

Source

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

"""Session interfaces for dual-storage architecture.

Protocol definitions for session management components.
"""

# Spec: 006 - Session Management

from typing import Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.types import Checkpoint, Fact, FactFilter


@traces(
    spec="006",
    story="US-001",
    reqs=["REQ-006-FR10-005"],
)
class ISessionManager(Protocol):
    """Interface for session lifecycle management."""

    @traces(spec="006", story="US-001", reqs=["REQ-006-FR10-001", "REQ-006-US-001-001"])
    def open_session(self, goals: list[str], portfolio_id: str | None = None) -> dict:
        """Open a new session."""
        ...

    @traces(spec="006", story="US-002", reqs=["REQ-006-FR10-002", "REQ-006-US-002-001"])
    def close_session(self, session_id: str, summary: str = "") -> dict:
        """Close an active session."""
        ...

    @traces(spec="006", story="US-003", reqs=["REQ-006-FR10-003", "REQ-006-US-003-001"])
    def get_session(self, session_id: str) -> dict | None:
        """Retrieve session by ID."""
        ...

    @traces(spec="006", story="US-007", reqs=["REQ-006-FR4-003", "REQ-006-US-007-005"])
    def get_active_sessions(self) -> list[dict]:
        """List all active sessions."""
        ...

    @traces(spec="006", story="US-007", reqs=["REQ-006-US-007-004"])
    def pause_session(self, session_id: str) -> None:
        """Pause an active session."""
        ...

    @traces(spec="006", story="US-007", reqs=["REQ-006-US-007-004"])
    def resume_session(self, session_id: str) -> dict:
        """Resume a paused session."""
        ...


@traces(spec="006", story="US-004", reqs=["REQ-006-FR8-001", "REQ-006-US-004-001"])
class IFactEmitter(Protocol):
    """Interface for session fact emission."""

    @traces(spec="006", story="US-004", reqs=["REQ-006-FR8-001", "REQ-006-US-004-001"])
    def emit(self, session_id: str, fact: Fact) -> None:
        """Emit a single fact to storage."""
        ...

    @traces(spec="006", story="US-003", reqs=["REQ-006-US-003-002"])
    def get_facts(
        self, session_id: str, filter: FactFilter | None = None
    ) -> list[Fact]:
        """Get facts for a session, optionally filtered."""
        ...

    @traces(spec="006", story="FR-009", reqs=["REQ-006-FR9-001"])
    def query_facts(self, session_id: str, fact_type: str | None = None) -> list[Fact]:
        """Query facts by type."""
        ...


class IRecoveryManager(Protocol):
    """Interface for session recovery."""

    def create_checkpoint(self, session_id: str) -> Checkpoint:
        """Create a recovery checkpoint."""
        ...

    def recover(self, session_id: str) -> dict:
        """Recover session from latest checkpoint."""
        ...

    def list_checkpoints(self, session_id: str) -> list[Checkpoint]:
        """List all checkpoints for a session."""
        ...


class ICarryoverManager(Protocol):
    """Interface for work carryover between sessions."""

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-013"])
    def create_carryover(self, session_id: str, tasks: list[dict]) -> None:
        """Create carryover tasks from a session."""
        ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-013"])
    def get_carryover(self, session_id: str) -> list[dict]:
        """Get carryover tasks for a session."""
        ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-013"])
    def resolve_carryover(self, task_id: str) -> None:
        """Mark a carryover task as resolved."""
        ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-028"])
    def link_carryover(self, from_session: str, to_session: str) -> None:
        """Link carryover between sessions."""
        ...


class IPortfolioManager(Protocol):
    """Interface for portfolio management."""

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-012"])
    def get_portfolio_sessions(self, portfolio_id: str) -> list[dict]:
        """Get all sessions in a portfolio."""
        ...

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-012"])
    def get_portfolio_context(self, portfolio_id: str) -> dict:
        """Get aggregated portfolio context."""
        ...