Skip to content

director — Cross-feature transition coordinator

Director

Director

Coordinator of transitions between features (scenes).

The Director is instantiated per incoming request (request-scoped) to capture abstract session and context identifiers. It facilitates the "Stateful Navigation, Stateless Logic" pattern.

Attributes:

Name Type Description
REDIRECT_KEY

Primary metadata key for-driven navigation.
MAX_REDIRECTS int

Maximum number of allowed redirects (loop prevention).
Source code in src/codex_bot/director/director.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class Director:
    """Coordinator of transitions between features (scenes).

    The Director is instantiated per incoming request (request-scoped) to capture
    abstract session and context identifiers. It facilitates the
    "Stateful Navigation, Stateless Logic" pattern.

    Attributes:
        REDIRECT_KEY: Primary metadata key for-driven navigation.
        MAX_REDIRECTS: Maximum number of allowed redirects (loop prevention).
    """

    REDIRECT_KEY = "__next_scene__"
    MAX_REDIRECTS: int = 5

    def __init__(
        self,
        container: ContainerProtocol,
        state: FSMContext | None = None,
        session_key: int | str | None = None,
        context_id: int | str | None = None,
        trigger_id: int | None = None,
    ) -> None:
        self.container = container
        self.state = state
        self.session_key = session_key
        self.context_id = context_id
        self.trigger_id = trigger_id

        self._redirect_count: int = 0

    async def resolve(self, data: Any) -> UnifiedViewDTO | Any:
        """Analyze incoming data for navigation instructions and resolve redirects.

        Args:
            data: Incoming request payload.

        Returns:
            A `UnifiedViewDTO` if a redirect was resolved, else business payload.
        """
        # Recursion protection
        if self._redirect_count >= self.MAX_REDIRECTS:
            log.error("Director | Loop detected in resolve transitions")
            return data

        # Safe parsing: Only dicts contain metadata for transitions
        if not isinstance(data, dict):
            return data

        # 1. Detect Navigation inside meta envelope
        meta = data.get("meta", {})
        if not isinstance(meta, dict):
            # Fallback if meta is not a dict
            return data

        next_feature = meta.get(self.REDIRECT_KEY)

        # 2. Extract Business Payload
        # If 'payload' key exists, it's the payload, otherwise it's the whole dict
        payload = data.get("payload", data)

        # 3. Handle redirect
        if next_feature and isinstance(next_feature, str):
            log.info(f"Director | Smart Resolve: Redirecting to '{next_feature}'")
            return await self.set_scene(feature=next_feature, payload=payload)

        return payload

    async def set_scene(self, feature: str, payload: Any = None) -> UnifiedViewDTO | Any:
        """Execute a cross-feature transition with Guard checks and Auto-Wrapping.

        Args:
            feature: Identifier of the target feature.
            payload: Ephemeral data for the transition.
        """
        # 0. Safety Invariants
        self._redirect_count += 1
        if self._redirect_count > self.MAX_REDIRECTS:
            log.error(f"Director | Redirect limit reached at '{feature}'")
            return UnifiedViewDTO(alert_text="Ошибка навигации: обнаружен цикл")

        orchestrator = self.container.features.get(feature)
        if orchestrator is None:
            log.error(f"Director | Unknown feature='{feature}'")
            return None

        # 1. Transition Guards (OCP Integration)
        for guard in self.container.transition_guards:
            result = await guard.check_access(self, feature=feature, orchestrator=orchestrator, payload=payload)
            if isinstance(result, UnifiedViewDTO):
                log.warning(f"Director | Guard {guard.__class__.__name__} blocked {feature}")
                return result

        # 2. Atomic FSM State Change
        expected_state = getattr(orchestrator, "expected_state", None)
        if self.state and expected_state:
            await self.state.set_state(expected_state)

        # 3. Handle Entry
        view = await orchestrator.handle_entry(director=self, payload=payload)

        # 4. Auto-Wrapping and Enrichment
        if not isinstance(view, UnifiedViewDTO):
            # If orchestrator returned a ViewResultDTO or raw content
            view = UnifiedViewDTO(content=view)

        # 5. Domain-Agnostic Session Enrichment
        view = view.model_copy(
            update={
                "chat_id": view.chat_id or self.context_id,
                "session_key": view.session_key or self.session_key,
                "trigger_message_id": view.trigger_message_id or self.trigger_id,
            }
        )

        return view

Functions

resolve(data) async

Analyze incoming data for navigation instructions and resolve redirects.

Parameters:

Name Type Description Default
data Any

Incoming request payload.

 required

Returns:

Type Description
UnifiedViewDTO | Any

A UnifiedViewDTO if a redirect was resolved, else business payload.
Source code in src/codex_bot/director/director.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def resolve(self, data: Any) -> UnifiedViewDTO | Any:
    """Analyze incoming data for navigation instructions and resolve redirects.

    Args:
        data: Incoming request payload.

    Returns:
        A `UnifiedViewDTO` if a redirect was resolved, else business payload.
    """
    # Recursion protection
    if self._redirect_count >= self.MAX_REDIRECTS:
        log.error("Director | Loop detected in resolve transitions")
        return data

    # Safe parsing: Only dicts contain metadata for transitions
    if not isinstance(data, dict):
        return data

    # 1. Detect Navigation inside meta envelope
    meta = data.get("meta", {})
    if not isinstance(meta, dict):
        # Fallback if meta is not a dict
        return data

    next_feature = meta.get(self.REDIRECT_KEY)

    # 2. Extract Business Payload
    # If 'payload' key exists, it's the payload, otherwise it's the whole dict
    payload = data.get("payload", data)

    # 3. Handle redirect
    if next_feature and isinstance(next_feature, str):
        log.info(f"Director | Smart Resolve: Redirecting to '{next_feature}'")
        return await self.set_scene(feature=next_feature, payload=payload)

    return payload

set_scene(feature, payload=None) async

Execute a cross-feature transition with Guard checks and Auto-Wrapping.

Parameters:

Name Type Description Default
feature str

Identifier of the target feature.

 required
payload Any

Ephemeral data for the transition.

 None
Source code in src/codex_bot/director/director.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
async def set_scene(self, feature: str, payload: Any = None) -> UnifiedViewDTO | Any:
    """Execute a cross-feature transition with Guard checks and Auto-Wrapping.

    Args:
        feature: Identifier of the target feature.
        payload: Ephemeral data for the transition.
    """
    # 0. Safety Invariants
    self._redirect_count += 1
    if self._redirect_count > self.MAX_REDIRECTS:
        log.error(f"Director | Redirect limit reached at '{feature}'")
        return UnifiedViewDTO(alert_text="Ошибка навигации: обнаружен цикл")

    orchestrator = self.container.features.get(feature)
    if orchestrator is None:
        log.error(f"Director | Unknown feature='{feature}'")
        return None

    # 1. Transition Guards (OCP Integration)
    for guard in self.container.transition_guards:
        result = await guard.check_access(self, feature=feature, orchestrator=orchestrator, payload=payload)
        if isinstance(result, UnifiedViewDTO):
            log.warning(f"Director | Guard {guard.__class__.__name__} blocked {feature}")
            return result

    # 2. Atomic FSM State Change
    expected_state = getattr(orchestrator, "expected_state", None)
    if self.state and expected_state:
        await self.state.set_state(expected_state)

    # 3. Handle Entry
    view = await orchestrator.handle_entry(director=self, payload=payload)

    # 4. Auto-Wrapping and Enrichment
    if not isinstance(view, UnifiedViewDTO):
        # If orchestrator returned a ViewResultDTO or raw content
        view = UnifiedViewDTO(content=view)

    # 5. Domain-Agnostic Session Enrichment
    view = view.model_copy(
        update={
            "chat_id": view.chat_id or self.context_id,
            "session_key": view.session_key or self.session_key,
            "trigger_message_id": view.trigger_message_id or self.trigger_id,
        }
    )

    return view

Protocols

OrchestratorProtocol

Bases: Protocol

Structural protocol for stateless feature orchestrators.

Orchestrators complying with this protocol must be stateless singletons. They are responsible for transforming incoming payloads into UI responses using the provided Director context.

Source code in src/codex_bot/director/protocols.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@runtime_checkable
class OrchestratorProtocol(Protocol):
    """Structural protocol for stateless feature orchestrators.

    Orchestrators complying with this protocol must be stateless singletons.
    They are responsible for transforming incoming payloads into UI responses
    using the provided `Director` context.
    """

    async def render(self, director: Any, payload: Any = None) -> Any:
        """Renders content for the passed payload."""
        ...

    async def handle_entry(
        self,
        director: Any,
        payload: Any = None,
    ) -> Any:
        """Entry point into the feature."""
        ...

Functions

handle_entry(director, payload=None) async

Entry point into the feature.

Source code in src/codex_bot/director/protocols.py 
79
80
81
82
83
84
85
async def handle_entry(
    self,
    director: Any,
    payload: Any = None,
) -> Any:
    """Entry point into the feature."""
    ...

render(director, payload=None) async

Renders content for the passed payload.

Source code in src/codex_bot/director/protocols.py 
75
76
77
async def render(self, director: Any, payload: Any = None) -> Any:
    """Renders content for the passed payload."""
    ...

ContainerProtocol

Bases: Protocol

Minimum contract for the project's DI container.

The Director only requires the features attribute — a dictionary of orchestrators. The specific BotContainer of the project must provide this attribute.

Attributes:

Name Type Description
features dict[str, OrchestratorProtocol]

Dictionary of {feature_key: orchestrator}.
Example 
class BotContainer:
    def __init__(self):
        self.features: dict[str, Any] = {}
Source code in src/codex_bot/director/protocols.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@runtime_checkable
class ContainerProtocol(Protocol):
    """Minimum contract for the project's DI container.

    The Director only requires the ``features`` attribute — a dictionary of orchestrators.
    The specific BotContainer of the project must provide this attribute.

    Attributes:
        features: Dictionary of ``{feature_key: orchestrator}``.

    Example:
        ```python
        class BotContainer:
            def __init__(self):
                self.features: dict[str, Any] = {}
        ```
    """

    features: dict[str, OrchestratorProtocol]

    @property
    def transition_guards(self) -> Sequence[BaseTransitionGuard]:
        """Sequence of guards to run before any feature transition."""
        ...

Attributes

transition_guards property

Sequence of guards to run before any feature transition.

SceneConfig

Bases: NamedTuple

Configuration schema for feature entry points.

Defines the mapping between a logical feature key and its associated Telegram FSM state. Used to drive the orchestration logic in the Director.

Attributes:

Name Type Description
fsm_state State

The aiogram State object to activate upon entry.
entry_service str

The lookup key for the orchestrator in the DI registry.
Example 
SCENE_ROUTES = {
    "main": SceneConfig(fsm_state=Menu.main, entry_service="main_menu")
}
Source code in src/codex_bot/director/protocols.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class SceneConfig(NamedTuple):
    """Configuration schema for feature entry points.

    Defines the mapping between a logical feature key and its associated
    Telegram FSM state. Used to drive the orchestration logic in the Director.

    Attributes:
        fsm_state: The `aiogram` State object to activate upon entry.
        entry_service: The lookup key for the orchestrator in the DI registry.

    Example:
        ```python
        SCENE_ROUTES = {
            "main": SceneConfig(fsm_state=Menu.main, entry_service="main_menu")
        }
        ```
    """

    fsm_state: State
    entry_service: str