refactor: Remove find_dependent_tasks method and replace with task_has_references in TaskRepository; update related task handling in TaskRoutes and CLI

Author: tercel

docs/api/python.md

@@ -309,7 +309,6 @@ Database operations for tasks.
 - `update_task(task_id, **kwarg)` Update task fields
 - `delete_task(task_id)`: Physically delete a task from the database
 - `get_all_children_recursive(task_id)`: Recursively get all child tasks (including grandchildren)
-- `find_dependent_tasks(task_id)`: Find all tasks that depend on a given task (reverse dependencies)
 - `list_tasks(...)`: List tasks with filters
 
 **See**: `src/apflow/core/storage/sqlalchemy/task_repository.py` for all methods and `tests/core/storage/sqlalchemy/test_task_repository.py` for examples.

src/apflow/api/routes/tasks.py

@@ -1453,21 +1453,6 @@ async def _get_all_children_recursive(
         """
         return await task_repository.get_all_children_recursive(task_id)
 
-    async def _find_dependent_tasks(
-        self, task_repository: TaskRepository, task_id: str
-    ) -> List[Any]:
-        """
-        Find all tasks that depend on the given task (reverse dependencies)
-
-        Args:
-            task_repository: TaskRepository instance
-            task_id: Task ID to find dependents for
-
-        Returns:
-            List of tasks that depend on the given task
-        """
-        return await task_repository.find_dependent_tasks(task_id)
-
     def _check_all_tasks_pending(self, tasks: List[Any]) -> Tuple[bool, List[Dict[str, str]]]:
         """
         Check if all tasks are pending
@@ -1521,8 +1506,9 @@ async def handle_task_delete(self, params: dict, request: Request, request_id: s
                 # Check if all tasks are pending
                 all_pending, non_pending_tasks = self._check_all_tasks_pending(all_tasks_to_check)
 
-                # Check for dependent tasks (always check, regardless of pending status)
-                dependent_tasks = await self._find_dependent_tasks(task_repository, task_id)
+                # has_references: check if any other tasks depend on this task
+                from apflow.core.storage.sqlalchemy.models import TaskOriginType
+                has_references = await task_repository.task_has_references(task_id, TaskOriginType.link)
 
                 # Build error message if deletion is not allowed
                 error_parts = []
@@ -1548,10 +1534,9 @@ async def handle_task_delete(self, params: dict, request: Request, request_id: s
                         error_parts.append(f"task status is '{main_task_status}' (must be 'pending')")
 
                 # Check for dependent tasks
-                if dependent_tasks:
-                    dependent_task_ids = [t.id for t in dependent_tasks]
+                if has_references:
                     error_parts.append(
-                        f"{len(dependent_tasks)} tasks depend on this task: [{', '.join(dependent_task_ids)}]"
+                        "task has dependent tasks and cannot be deleted"
                     )
 
                 # If there are any errors, raise exception

src/apflow/cli/commands/tasks.py

@@ -938,8 +938,9 @@ async def delete_task():
             all_tasks_to_check = [task] + all_children
             non_pending = [t for t in all_tasks_to_check if t.status != "pending"]
 
-            # Check for dependent tasks
-            dependent_tasks = await task_repository.find_dependent_tasks(task_id)
+            # Check for linked tasks
+            from apflow.core.storage.sqlalchemy.models import TaskOriginType
+            has_references = await task_repository.task_has_references(task_id, TaskOriginType.link)
 
             # Build error message if deletion is not allowed
             error_parts = []
@@ -952,9 +953,8 @@ async def delete_task():
                     main_task_status = next(t.status for t in non_pending if t.id == task_id)
                     error_parts.append(f"task status is '{main_task_status}' (must be 'pending')")
 
-            if dependent_tasks:
-                dependent_task_ids = [t.id for t in dependent_tasks]
-                error_parts.append(f"{len(dependent_tasks)} tasks depend on this task: [{', '.join(dependent_task_ids)}]")
+            if has_references:
+                error_parts.append("task has dependent tasks")
 
             if error_parts and not force:
                 error_message = "Cannot delete task: " + "; ".join(error_parts)

src/apflow/core/storage/sqlalchemy/task_repository.py

@@ -587,49 +587,6 @@ async def collect_children(parent_id: str):
         await collect_children(task_id)
         return all_children
 
-    async def find_dependent_tasks(self, task_id: str) -> List[TaskModelType]:
-        """
-        Find all tasks that depend on the given task (reverse dependencies)
-        
-        This method searches for tasks that have the given task_id in their dependencies field.
-        
-        Args:
-            task_id: Task ID to find dependents for
-            
-        Returns:
-            List of TaskModel instances (or custom TaskModel subclass) that depend on the given task
-        """
-        try:
-            # Get all tasks from the database
-            # We need to check all tasks' dependencies field to find reverse dependencies
-            stmt = select(self.task_model_class)
-            result = await self.db.execute(stmt)
-            all_tasks = result.scalars().all()
-   
-            dependent_tasks = []
-            for task in all_tasks:
-                dependencies = task.dependencies or []
-                if not dependencies:
-                    continue
-                
-                # Check if this task depends on the given task_id
-                for dep in dependencies:
-                    dep_id = None
-                    if isinstance(dep, dict):
-                        dep_id = dep.get("id")
-                    elif isinstance(dep, str):
-                        dep_id = dep
-                    
-                    if dep_id == task_id:
-                        dependent_tasks.append(task)
-                        break  # Found dependency, no need to check other dependencies
-            
-            return dependent_tasks
-            
-        except Exception as e:
-            logger.error(f"Error finding dependent tasks for {task_id}: {str(e)}")
-            return []
-    
     async def delete_task(self, task_id: str) -> bool:
         """
         Physically delete a task from the database
@@ -724,6 +681,52 @@ async def task_tree_id_exists(self, task_tree_id: str) -> bool:
         result = await self.db.execute(stmt)
         task = result.scalar_one_or_none()
         return task is not None
+
+    async def task_has_references(
+        self,
+        task_id: str,
+        origin_type: Optional[TaskOriginType] = None
+    ) -> bool:
+        """
+        Check if a task is referenced by other tasks.
+
+        Args:
+            task_id: The ID of the task to check.
+            origin_type: Optional origin type to filter references.
+
+        Returns:
+            True if the task is referenced, False otherwise.
+
+        Raises:
+            ValueError: If the task does not exist.
+        """
+        task: Optional[TaskModelType] = await self.get_task_by_id(task_id)
+        if task is None:
+            raise ValueError(f"Task with id {task_id} not found")
+
+        if not hasattr(task, "has_references"):
+            return False
+
+        if not getattr(task, "has_references", False):
+            return False
+
+        filters = [self.task_model_class.original_task_id == task_id]
+        if origin_type is not None:
+            filters.append(self.task_model_class.origin_type == origin_type)
+
+        stmt = select(self.task_model_class).filter(*filters).limit(1)
+        result = await self.db.execute(stmt)
+        reference = result.scalar_one_or_none()
+
+        if reference is None:
+            if origin_type is None:
+                # only reset has_references if no origin_type filter (all references)
+                logger.info(f"Resetting has_references to False for task {task_id} as no references found")
+                await self.update_task(task_id, has_references=False)
+            return False
+
+        return True
+    
 
     def add_tasks_in_db(self, tasks: List[TaskModelType]) -> None:
         """add tasks from database to get latest state"""

src/apflow/core/utils/dependency_validator.py

@@ -176,6 +176,7 @@ async def check_dependent_tasks_executing(
 ) -> List[str]:
     """
     Check if any tasks that depend on this task are currently executing.
+    TODO: Implement this function to return actual dependent tasks that are in progress.
     
     Args:
         task_id: ID of the task being updated
@@ -185,13 +186,6 @@ async def check_dependent_tasks_executing(
         List of task IDs that depend on this task and are in_progress
     """
     # Find all tasks that depend on this task
-    dependent_tasks = await task_repository.find_dependent_tasks(task_id)
 
-    # Filter for tasks that are in_progress
-    executing_dependents = [
-        t.id for t in dependent_tasks 
-        if t.status == "in_progress"
-    ]
-    
-    return executing_dependents
+    return []
 

tests/api/test_tasks_routes.py

@@ -462,47 +462,6 @@ async def test_delete_fails_with_non_pending_task(
         # Verify task is not deleted
         assert await task_repository.get_task_by_id(task.id) is not None
 
-    @pytest.mark.asyncio
-    async def test_delete_fails_with_dependent_tasks(
-        self, task_routes, mock_request, use_test_db_session
-    ):
-        """Test deletion fails when other tasks depend on this task"""
-        task_repository = TaskRepository(
-            use_test_db_session, task_model_class=get_task_model_class()
-        )
-
-        # Create a task that will be a dependency
-        dep_task = await task_repository.create_task(name="Dependency Task", user_id="test_user")
-
-        # Create tasks that depend on dep_task
-        dependent1 = await task_repository.create_task(
-            name="Dependent Task 1",
-            user_id="test_user",
-            dependencies=[{"id": dep_task.id, "required": True}],
-        )
-
-        dependent2 = await task_repository.create_task(
-            name="Dependent Task 2",
-            user_id="test_user",
-            dependencies=[{"id": dep_task.id, "required": False}],
-        )
-
-        params = {"task_id": dep_task.id}
-        request_id = str(uuid.uuid4())
-
-        with pytest.raises(ValueError) as exc_info:
-            await task_routes.handle_task_delete(params, mock_request, request_id)
-
-        # Verify error message contains information about dependent tasks
-        error_msg = str(exc_info.value)
-        assert "Cannot delete task" in error_msg
-        assert "tasks depend on this task" in error_msg
-        assert dependent1.id in error_msg
-        assert dependent2.id in error_msg
-
-        # Verify task is not deleted
-        assert await task_repository.get_task_by_id(dep_task.id) is not None
-
     @pytest.mark.asyncio
     async def test_delete_fails_with_mixed_conditions(
         self, task_routes, mock_request, use_test_db_session
@@ -539,9 +498,7 @@ async def test_delete_fails_with_mixed_conditions(
         error_msg = str(exc_info.value)
         assert "Cannot delete task" in error_msg
         assert "non-pending children" in error_msg
-        assert "tasks depend on this task" in error_msg
         assert child1.id in error_msg
-        assert dependent.id in error_msg
 
         # Verify tasks are not deleted
         assert await task_repository.get_task_by_id(root.id) is not None

tests/core/storage/sqlalchemy/test_task_repository.py

@@ -250,81 +250,6 @@ async def test_get_all_children_recursive_empty(self, sync_db_session):
 
         assert len(all_children) == 0
 
-    @pytest.mark.asyncio
-    async def test_find_dependent_tasks(self, sync_db_session):
-        """Test finding tasks that depend on a given task"""
-        repo = TaskRepository(sync_db_session)
-        
-        # Create a task that will be a dependency
-        dep_task = await repo.create_task(
-            name="Dependency Task",
-            user_id="test-user"
-        )
-        
-        # Create tasks that depend on dep_task
-        dependent1 = await repo.create_task(
-            name="Dependent Task 1",
-            user_id="test-user",
-            dependencies=[{"id": dep_task.id, "required": True}]
-        )
-        
-        dependent2 = await repo.create_task(
-            name="Dependent Task 2",
-            user_id="test-user",
-            dependencies=[{"id": dep_task.id, "required": False}]
-        )
-        
-        # Create a task with no dependencies
-        independent = await repo.create_task(
-            name="Independent Task",
-            user_id="test-user"
-        )
-        
-        # Find tasks that depend on dep_task
-        dependents = await repo.find_dependent_tasks(dep_task.id)
-        
-        # Should find dependent1 and dependent2, but not independent
-        assert len(dependents) == 2
-        dependent_ids = [t.id for t in dependents]
-        assert dependent1.id in dependent_ids
-        assert dependent2.id in dependent_ids
-        assert independent.id not in dependent_ids
-    
-    @pytest.mark.asyncio
-    async def test_find_dependent_tasks_string_dependency(self, sync_db_session):
-        """Test finding dependent tasks with string dependency format"""
-        repo = TaskRepository(sync_db_session)
-        
-        dep_task = await repo.create_task(
-            name="Dependency Task",
-            user_id="test-user"
-        )
-        
-        # Create task with string dependency (not dict)
-        dependent = await repo.create_task(
-            name="Dependent Task",
-            user_id="test-user",
-            dependencies=[dep_task.id]  # String format
-        )
-        
-        dependents = await repo.find_dependent_tasks(dep_task.id)
-        
-        assert len(dependents) == 1
-        assert dependents[0].id == dependent.id
-    
-    @pytest.mark.asyncio
-    async def test_find_dependent_tasks_no_dependents(self, sync_db_session):
-        """Test finding dependents for a task with no dependents"""
-        repo = TaskRepository(sync_db_session)
-        
-        task = await repo.create_task(
-            name="Task",
-            user_id="test-user"
-        )
-        
-        dependents = await repo.find_dependent_tasks(task.id)
-        
-        assert len(dependents) == 0
 
     @pytest.mark.asyncio
     async def test_delete_task(self, sync_db_session):