from __future__ import annotations from dataclasses import dataclass from typing import Any from api.task_status import ACTIVE_TASK_STATUSES, normalize_task_status, status_value from api.workplan_status import normalize_workplan_status TASK_STARTED_STATUS = "progress" TASK_NOT_STARTED_STATUS = "todo" TASK_ACTIVE_STATUSES = ACTIVE_TASK_STATUSES PARENT_ACTIVATION_STATUSES = {"proposed", "ready", "backlog"} # Legacy alias normalize_workstream_status = normalize_workplan_status @dataclass(frozen=True) class LifecycleTransitionResult: entity_type: str previous_status: str target_status: str changed: bool parent_activated: bool = False def should_activate_parent_for_task_start( *, previous_task_status: Any, new_task_status: Any, parent_workplan_status: Any = None, parent_workstream_status: Any = None, ) -> bool: """Return whether a task start should move its parent to active.""" parent_status = parent_workplan_status if parent_workplan_status is not None else parent_workstream_status return ( status_value(previous_task_status) == TASK_NOT_STARTED_STATUS and status_value(new_task_status) == TASK_STARTED_STATUS and normalize_workplan_status(parent_status) in PARENT_ACTIVATION_STATUSES ) def has_active_task_status(task_statuses: list[Any] | tuple[Any, ...]) -> bool: """Return whether any task status represents currently active work.""" return any(status_value(status) in TASK_ACTIVE_STATUSES for status in task_statuses) def should_activate_parent_for_active_tasks( *, parent_workplan_status: Any = None, parent_workstream_status: Any = None, task_statuses: list[Any] | tuple[Any, ...], ) -> bool: """Return whether existing task state implies an active parent workplan.""" parent_status = parent_workplan_status if parent_workplan_status is not None else parent_workstream_status return ( normalize_workplan_status(parent_status) in PARENT_ACTIVATION_STATUSES and has_active_task_status(task_statuses) ) def activate_parent_for_task_start( *, previous_task_status: Any, new_task_status: Any, parent_workplan: Any = None, parent_workstream: Any = None, ) -> bool: """Activate a planning-state parent workplan when real task work starts.""" parent = parent_workplan if parent_workplan is not None else parent_workstream if parent is None: return False if not should_activate_parent_for_task_start( previous_task_status=previous_task_status, new_task_status=new_task_status, parent_workplan_status=getattr(parent, "status", None), parent_workstream_status=getattr(parent, "status", None), ): return False parent.status = "active" return True def transition_workplan_status( workplan: Any, target_status: Any, ) -> LifecycleTransitionResult: """Apply a canonical workplan status transition.""" previous_status = normalize_workplan_status(getattr(workplan, "status", None)) normalised_target = normalize_workplan_status(target_status) workplan.status = normalised_target return LifecycleTransitionResult( entity_type="workplan", previous_status=previous_status, target_status=normalised_target, changed=previous_status != normalised_target, ) transition_workstream_status = transition_workplan_status def transition_task_status( task: Any, target_status: Any, *, parent_workplan: Any = None, parent_workstream: Any = None, previous_task_status: Any = None, status_coercer: Any = None, ) -> LifecycleTransitionResult: """Apply a task status transition and activate the parent when work starts.""" parent = parent_workplan if parent_workplan is not None else parent_workstream previous_status = status_value( getattr(task, "status", None) if previous_task_status is None else previous_task_status ) normalised_target = normalize_task_status(target_status) task.status = status_coercer(normalised_target) if status_coercer else normalised_target parent_activated = activate_parent_for_task_start( previous_task_status=previous_status, new_task_status=normalised_target, parent_workplan=parent, parent_workstream=parent, ) return LifecycleTransitionResult( entity_type="task", previous_status=previous_status, target_status=normalised_target, changed=previous_status != normalised_target, parent_activated=parent_activated, )