P4.1: Extract reusable dependency check helper, deduplicate milestone_actions.py

- New app/services/dependency_check.py with check_milestone_deps()
- Replaces 3x duplicated JSON-parse + query + filter logic
- Supports both milestone and task dependency checking
- Returns structured DepCheckResult with ok/blockers/reason
- Refactored preflight and start endpoints to use shared helper

app/api/routers/milestone_actions.py

@@ -3,7 +3,6 @@
 Provides freeze / start / close actions on milestones.
 completed is triggered automatically when the sole release maintenance task finishes.
 """
-import json
 from datetime import datetime, timezone
 from typing import Optional
 
@@ -18,6 +17,7 @@ from app.models import models
 from app.models.milestone import Milestone, MilestoneStatus
 from app.models.task import Task, TaskStatus
 from app.services.activity import log_activity
+from app.services.dependency_check import check_milestone_deps
 
 router = APIRouter(
     prefix="/projects/{project_id}/milestones/{milestone_id}/actions",
@@ -101,38 +101,13 @@ def preflight_milestone_actions(
 
     # --- start pre-check (only meaningful when status == freeze) ---
     if ms_status == "freeze":
-        blockers: list[str] = []
+        dep_result = check_milestone_deps(
-
+            db, ms.depend_on_milestones, ms.depend_on_tasks,
-        # milestone dependencies
+        )
-        dep_ms_ids = []
+        if dep_result.ok:
-        if ms.depend_on_milestones:
-            try:
-                dep_ms_ids = json.loads(ms.depend_on_milestones)
-            except (json.JSONDecodeError, TypeError):
-                dep_ms_ids = []
-        if dep_ms_ids:
-            dep_milestones = db.query(Milestone).filter(Milestone.id.in_(dep_ms_ids)).all()
-            incomplete = [m.id for m in dep_milestones if _ms_status_value(m) != "completed"]
-            if incomplete:
-                blockers.append(f"Dependent milestones not completed: {incomplete}")
-
-        # task dependencies
-        dep_task_ids = []
-        if ms.depend_on_tasks:
-            try:
-                dep_task_ids = json.loads(ms.depend_on_tasks)
-            except (json.JSONDecodeError, TypeError):
-                dep_task_ids = []
-        if dep_task_ids:
-            dep_tasks = db.query(Task).filter(Task.id.in_(dep_task_ids)).all()
-            incomplete_tasks = [t.id for t in dep_tasks if (t.status.value if hasattr(t.status, "value") else t.status) != "completed"]
-            if incomplete_tasks:
-                blockers.append(f"Dependent tasks not completed: {incomplete_tasks}")
-
-        if blockers:
-            result["start"] = {"allowed": False, "reason": "; ".join(blockers)}
-        else:
             result["start"] = {"allowed": True, "reason": None}
+        else:
+            result["start"] = {"allowed": False, "reason": dep_result.reason}
 
     return result
 
@@ -232,50 +207,14 @@ def start_milestone(
             detail=f"Cannot start: milestone is '{_ms_status_value(ms)}', expected 'freeze'",
         )
 
-    # Dependency check — milestone dependencies
+    # Dependency check (P4.1 — shared helper)
-    dep_ms_ids = []
+    dep_result = check_milestone_deps(
-    if ms.depend_on_milestones:
+        db, ms.depend_on_milestones, ms.depend_on_tasks,
-        try:
-            dep_ms_ids = json.loads(ms.depend_on_milestones)
-        except (json.JSONDecodeError, TypeError):
-            dep_ms_ids = []
-
-    if dep_ms_ids:
-        dep_milestones = (
-            db.query(Milestone)
-            .filter(Milestone.id.in_(dep_ms_ids))
-            .all()
     )
-        incomplete = [
+    if not dep_result.ok:
-            m.id
-            for m in dep_milestones
-            if _ms_status_value(m) != "completed"
-        ]
-        if incomplete:
         raise HTTPException(
             status_code=400,
-                detail=f"Cannot start: dependent milestones not completed: {incomplete}",
+            detail=f"Cannot start: {dep_result.reason}",
-            )
-
-    # Dependency check — task dependencies
-    dep_task_ids = []
-    if ms.depend_on_tasks:
-        try:
-            dep_task_ids = json.loads(ms.depend_on_tasks)
-        except (json.JSONDecodeError, TypeError):
-            dep_task_ids = []
-
-    if dep_task_ids:
-        dep_tasks = db.query(Task).filter(Task.id.in_(dep_task_ids)).all()
-        incomplete_tasks = [
-            t.id
-            for t in dep_tasks
-            if (t.status.value if hasattr(t.status, "value") else t.status) != "completed"
-        ]
-        if incomplete_tasks:
-            raise HTTPException(
-                status_code=400,
-                detail=f"Cannot start: dependent tasks not completed: {incomplete_tasks}",
         )
 
     ms.status = MilestoneStatus.UNDERGOING

app/api/routers/misc.py

@@ -445,7 +445,7 @@ def create_milestone_task(project_code: str, milestone_id: int, task_data: dict,
         description=task_data.get("description"),
         status=TaskStatus.OPEN,
         priority=TaskPriority.MEDIUM,
-        task_type=task_data.get("task_type", "task"),
+        task_type=task_data.get("task_type", "issue"),  # P7.1: default changed from 'task' to 'issue'
         task_subtype=task_data.get("task_subtype"),
         project_id=project.id,
         milestone_id=milestone_id,

app/api/routers/tasks.py

@@ -46,7 +46,7 @@ TASK_SUBTYPE_MAP = {
     'story': {'feature', 'improvement', 'refactor'},
     'test': {'regression', 'security', 'smoke', 'stress'},
     'research': set(),
-    'task': {'defect'},
+    # P7.1: 'task' type removed — defect subtype migrated to issue/defect
     'resolution': set(),
 }
 ALLOWED_TASK_TYPES = set(TASK_SUBTYPE_MAP.keys())

app/main.py

@@ -129,7 +129,7 @@ def _migrate_schema():
         # tasks extra fields
         result = db.execute(text("SHOW COLUMNS FROM tasks LIKE 'task_type'"))
         if not result.fetchone():
-            db.execute(text("ALTER TABLE tasks ADD COLUMN task_type VARCHAR(32) DEFAULT 'task'"))
+            db.execute(text("ALTER TABLE tasks ADD COLUMN task_type VARCHAR(32) DEFAULT 'issue'"))
         result = db.execute(text("SHOW COLUMNS FROM tasks LIKE 'task_subtype'"))
         if not result.fetchone():
             db.execute(text("ALTER TABLE tasks ADD COLUMN task_subtype VARCHAR(64) NULL"))
@@ -194,6 +194,10 @@ def _migrate_schema():
             if not _has_column(db, "milestones", "started_at"):
                 db.execute(text("ALTER TABLE milestones ADD COLUMN started_at DATETIME NULL"))
 
+        # --- P7.1: Migrate task_type='task' to 'issue' ---
+        if _has_table(db, "tasks") and _has_column(db, "tasks", "task_type"):
+            db.execute(text("UPDATE tasks SET task_type='issue' WHERE task_type='task'"))
+
         # --- Task status enum migration (old -> new) ---
         if _has_table(db, "tasks"):
             # Widen enum first

app/models/task.py

@@ -28,7 +28,7 @@ class Task(Base):
     task_code = Column(String(64), nullable=True, unique=True, index=True)
     
     # Task type/subtype (replaces old issue_type/issue_subtype)
-    task_type = Column(String(32), default="task")
+    task_type = Column(String(32), default="issue")  # P7.1: default changed from 'task' to 'issue'
     task_subtype = Column(String(64), nullable=True)
     
     project_id = Column(Integer, ForeignKey("projects.id"), nullable=False)

app/schemas/schemas.py

@@ -12,7 +12,7 @@ class TaskTypeEnum(str, Enum):
     STORY = "story"
     TEST = "test"
     RESOLUTION = "resolution"
-    TASK = "task"
+    # P7.1: 'task' type removed — defect subtype migrated to issue/defect
 
 
 class TaskStatusEnum(str, Enum):
@@ -34,7 +34,7 @@ class TaskPriorityEnum(str, Enum):
 class TaskBase(BaseModel):
     title: str
     description: Optional[str] = None
-    task_type: TaskTypeEnum = TaskTypeEnum.TASK
+    task_type: TaskTypeEnum = TaskTypeEnum.ISSUE
     task_subtype: Optional[str] = None
     priority: TaskPriorityEnum = TaskPriorityEnum.MEDIUM
     tags: Optional[str] = None

app/services/dependency_check.py

@@ -0,0 +1,113 @@
+"""P4.1 — Reusable dependency-check helpers.
+
+Used by milestone start, milestone preflight, and (future) task pending→open
+to verify that all declared dependencies are completed before allowing the
+entity to proceed.
+"""
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from typing import Sequence
+
+from sqlalchemy.orm import Session
+
+from app.models.milestone import Milestone
+from app.models.task import Task
+
+
+# ---------------------------------------------------------------------------
+# Result type
+# ---------------------------------------------------------------------------
+
+@dataclass
+class DepCheckResult:
+    """Outcome of a dependency check."""
+
+    ok: bool = True
+    blockers: list[str] = field(default_factory=list)
+
+    @property
+    def reason(self) -> str | None:
+        return "; ".join(self.blockers) if self.blockers else None
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+def _parse_json_ids(raw: str | None) -> list[int]:
+    """Safely parse a JSON-encoded list of integer IDs."""
+    if not raw:
+        return []
+    try:
+        ids = json.loads(raw)
+        if isinstance(ids, list):
+            return [int(i) for i in ids]
+    except (json.JSONDecodeError, TypeError, ValueError):
+        pass
+    return []
+
+
+def _ms_status(ms: Milestone) -> str:
+    return ms.status.value if hasattr(ms.status, "value") else ms.status
+
+
+def _task_status(t: Task) -> str:
+    return t.status.value if hasattr(t.status, "value") else t.status
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def check_milestone_deps(
+    db: Session,
+    depend_on_milestones: str | None,
+    depend_on_tasks: str | None,
+    *,
+    required_status: str = "completed",
+) -> DepCheckResult:
+    """Check whether all milestone + task dependencies are satisfied.
+
+    Parameters
+    ----------
+    db:
+        Active DB session.
+    depend_on_milestones:
+        JSON-encoded list of milestone IDs (from the entity's field).
+    depend_on_tasks:
+        JSON-encoded list of task IDs (from the entity's field).
+    required_status:
+        The status that dependees must have reached (default ``"completed"``).
+
+    Returns
+    -------
+    DepCheckResult with ``ok=True`` if all deps satisfied, else ``ok=False``
+    with human-readable ``blockers``.
+    """
+    result = DepCheckResult()
+
+    # --- milestone dependencies ---
+    ms_ids = _parse_json_ids(depend_on_milestones)
+    if ms_ids:
+        dep_milestones: Sequence[Milestone] = (
+            db.query(Milestone).filter(Milestone.id.in_(ms_ids)).all()
+        )
+        incomplete = [m.id for m in dep_milestones if _ms_status(m) != required_status]
+        if incomplete:
+            result.ok = False
+            result.blockers.append(f"Dependent milestones not {required_status}: {incomplete}")
+
+    # --- task dependencies ---
+    task_ids = _parse_json_ids(depend_on_tasks)
+    if task_ids:
+        dep_tasks: Sequence[Task] = (
+            db.query(Task).filter(Task.id.in_(task_ids)).all()
+        )
+        incomplete_tasks = [t.id for t in dep_tasks if _task_status(t) != required_status]
+        if incomplete_tasks:
+            result.ok = False
+            result.blockers.append(f"Dependent tasks not {required_status}: {incomplete_tasks}")
+
+    return result