""" FSM Transitions for Task model. This module defines declarative transitions for the Task entity. Note: Most task state transitions (annotation_started, annotation_complete, completed) are triggered by Annotation changes, not Task field changes. Those are handled by annotation transitions via post_transition_hooks that update the parent task. """ from typing import Any, Dict, Optional from fsm.registry import register_state_transition from fsm.state_choices import TaskStateChoices from fsm.transitions import ModelChangeTransition, TransitionContext @register_state_transition('task', 'task_created', triggers_on_create=True, triggers_on_update=False) class TaskCreatedTransition(ModelChangeTransition): """ Transition when a new task is created. This is the initial state transition that occurs when a task is first saved to the database. Trigger: Automatically on creation (triggers_on_create=True) Note: Other task transitions (annotation_started, completed, etc.) are triggered by Annotation model changes, not Task field changes. """ def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TaskStateChoices.CREATED def get_reason(self, context: TransitionContext) -> str: """Return detailed reason for task creation.""" return 'Task created in the system' def transition(self, context: TransitionContext) -> Dict[str, Any]: """ Execute task creation transition. Args: context: Transition context containing task and user information Returns: Context data to store with the state record """ task = context.entity return { 'project_id': task.project_id, 'data_keys': list(task.data.keys()) if task.data else [], } # Note: Task state transitions (COMPLETED, IN_PROGRESS) are triggered by annotation changes # via post_transition_hooks in annotation transitions, not by direct task model changes. @register_state_transition('task', 'task_completed', triggers_on_create=False, triggers_on_update=False) class TaskCompletedTransition(ModelChangeTransition): """ Transition when task moves to COMPLETED state. Triggered when: First annotation is submitted on this task From: CREATED -> COMPLETED or IN_PROGRESS -> COMPLETED """ def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TaskStateChoices.COMPLETED def get_reason(self, context: TransitionContext) -> str: return 'Task completed - annotation submitted' def transition(self, context: TransitionContext) -> Dict[str, Any]: task = context.entity return { 'task_id': task.id, 'project_id': task.project_id, 'total_annotations': task.total_annotations, 'cancelled_annotations': task.cancelled_annotations, 'is_labeled': task.is_labeled, } @register_state_transition('task', 'task_in_progress', triggers_on_create=False, triggers_on_update=False) class TaskInProgressTransition(ModelChangeTransition): """ Transition when task moves to IN_PROGRESS state. Triggered when: All annotations are deleted from a completed task From: COMPLETED -> IN_PROGRESS """ def get_target_state(self, context: Optional[TransitionContext] = None) -> str: return TaskStateChoices.IN_PROGRESS def get_reason(self, context: TransitionContext) -> str: return 'Task moved to in progress - annotations deleted' def transition(self, context: TransitionContext) -> Dict[str, Any]: task = context.entity return { 'task_id': task.id, 'project_id': task.project_id, 'total_annotations': task.total_annotations, 'cancelled_annotations': task.cancelled_annotations, 'is_labeled': task.is_labeled, }