feat(F): F-PROACTIVE consolidate proactive AI — trigger_dispatcher dispatches agents on context/UI events

app/plugins/builtins/tasks/services.py

@@ -13,6 +13,14 @@ from app.core.visibility import apply_visibility_filter
 from app.plugins.builtins.tasks.models import Task
 
 
+# Lifecycle statuses in display order (Kanban columns)
+STATUS_ORDER = ["open", "in_progress", "review", "blocked", "done", "cancelled"]
+# Statuses that count as "done" for progress aggregation
+DONE_STATUSES = {"done"}
+# Statuses that count as "in progress" for progress aggregation
+ACTIVE_STATUSES = {"in_progress", "review"}
+
+
 def _task_to_dict(task: Task) -> dict[str, Any]:
     """Serialize a Task model to a dict."""
     return {
@@ -24,12 +32,157 @@ def _task_to_dict(task: Task) -> dict[str, Any]:
         "due_date": task.due_date.isoformat() if task.due_date else None,
         "assigned_to": str(task.assigned_to) if task.assigned_to else None,
         "contact_id": str(task.contact_id) if task.contact_id else None,
+        "assignee_type": task.assignee_type,
+        "assignee_id": str(task.assignee_id) if task.assignee_id else None,
+        "entity_type": task.entity_type,
+        "entity_id": str(task.entity_id) if task.entity_id else None,
+        "creator_type": task.creator_type,
+        "creator_id": str(task.creator_id) if task.creator_id else None,
+        "parent_task_id": str(task.parent_task_id) if task.parent_task_id else None,
+        "depends_on": [str(d) for d in (task.depends_on or [])],
+        "task_type": task.task_type,
+        "success_criteria": task.success_criteria,
+        "target_date": task.target_date.isoformat() if task.target_date else None,
+        "progress": task.progress or 0,
         "created_by": str(task.created_by) if task.created_by else None,
         "created_at": task.created_at.isoformat() if task.created_at else None,
         "updated_at": task.updated_at.isoformat() if task.updated_at else None,
     }
 
 
+async def _get_task(db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID) -> Task | None:
+    """Fetch a non-deleted task by id within a tenant."""
+    result = await db.execute(
+        select(Task).where(
+            Task.id == task_id,
+            Task.tenant_id == tenant_id,
+            Task.deleted_at.is_(None),
+        )
+    )
+    return result.scalar_one_or_none()
+
+
+async def _recompute_progress(db: AsyncSession, tenant_id: uuid.UUID, task: Task) -> None:
+    """Recompute a parent task's progress from its child tasks.
+
+    Progress = percentage of non-cancelled children that are done. If there are
+    no children, the task's own progress field is left untouched.
+    """
+    result = await db.execute(
+        select(Task).where(
+            Task.tenant_id == tenant_id,
+            Task.parent_task_id == task.id,
+            Task.deleted_at.is_(None),
+        )
+    )
+    children = result.scalars().all()
+    if not children:
+        return
+    active = [c for c in children if c.status != "cancelled"]
+    if not active:
+        task.progress = 100
+        return
+    done_count = sum(1 for c in active if c.status in DONE_STATUSES)
+    task.progress = round(done_count / len(active) * 100)
+
+
+async def _propagate_parent_status(db: AsyncSession, tenant_id: uuid.UUID, task: Task) -> None:
+    """Propagate status changes to the parent task.
+
+    When all children of a parent are done, the parent is moved to ``review``
+    (unless it is already done/cancelled). When any child is blocked, the
+    parent is moved to ``blocked``. Otherwise the parent is set to
+    ``in_progress`` if it was ``open``.
+    """
+    if not task.parent_task_id:
+        return
+    parent = await _get_task(db, tenant_id, task.parent_task_id)
+    if parent is None or parent.status in DONE_STATUSES or parent.status == "cancelled":
+        return
+
+    result = await db.execute(
+        select(Task).where(
+            Task.tenant_id == tenant_id,
+            Task.parent_task_id == parent.id,
+            Task.deleted_at.is_(None),
+        )
+    )
+    children = result.scalars().all()
+    if not children:
+        return
+
+    active = [c for c in children if c.status != "cancelled"]
+    if not active:
+        parent.status = "review"
+    elif any(c.status == "blocked" for c in active):
+        parent.status = "blocked"
+    elif all(c.status in DONE_STATUSES for c in active):
+        parent.status = "review"
+    elif parent.status == "open":
+        parent.status = "in_progress"
+
+    await _recompute_progress(db, tenant_id, parent)
+
+
+async def _evaluate_success_criteria(task: Task) -> bool:
+    """Evaluate structured success criteria for a goal.
+
+    Supported criteria shapes (JSONB):
+    - ``{"all_done": true}`` — all child tasks done
+    - ``{"criteria": [{"field": "status", "op": "eq", "value": "done"}]}``
+      — every criterion must match the task's own fields
+    - ``{"criteria": [{"field": "progress", "op": "gte", "value": 100}]}``
+    - ``{"criteria": [{"field": "target_date", "op": "lte", "value": "<iso>"}]}``
+
+    Returns True when the criteria are met (or when no criteria are set).
+    """
+    criteria = task.success_criteria
+    if not criteria:
+        return False
+
+    if criteria.get("all_done"):
+        return task.progress >= 100
+
+    raw = criteria.get("criteria") or []
+    if not raw:
+        return False
+
+    def _field_value(field: str) -> Any:
+        if field == "status":
+            return task.status
+        if field == "progress":
+            return task.progress or 0
+        if field == "target_date":
+            return task.target_date
+        if field == "due_date":
+            return task.due_date
+        return getattr(task, field, None)
+
+    for item in raw:
+        field = item.get("field")
+        op = item.get("op", "eq")
+        value = item.get("value")
+        actual = _field_value(field)
+        if op == "eq":
+            if actual != value:
+                return False
+        elif op == "neq":
+            if actual == value:
+                return False
+        elif op == "gte":
+            if actual is None or actual < value:
+                return False
+        elif op == "lte":
+            if actual is None or actual > value:
+                return False
+        elif op == "contains":
+            if value not in (actual or []):
+                return False
+        else:
+            return False
+    return True
+
+
 async def list_tasks(
     db: AsyncSession,
     tenant_id: uuid.UUID,
@@ -41,6 +194,12 @@ async def list_tasks(
     assigned_to: str | None = None,
     contact_id: str | None = None,
     search: str | None = None,
+    entity_type: str | None = None,
+    entity_id: str | None = None,
+    assignee_type: str | None = None,
+    assignee_id: str | None = None,
+    parent_task_id: str | None = None,
+    task_type: str | None = None,
     user_id: uuid.UUID | None = None,
     is_system_admin: bool = False,
 ) -> dict[str, Any]:
@@ -60,6 +219,18 @@ async def list_tasks(
         query = query.where(Task.assigned_to == uuid.UUID(assigned_to))
     if contact_id:
         query = query.where(Task.contact_id == uuid.UUID(contact_id))
+    if entity_type:
+        query = query.where(Task.entity_type == entity_type)
+    if entity_id:
+        query = query.where(Task.entity_id == uuid.UUID(entity_id))
+    if assignee_type:
+        query = query.where(Task.assignee_type == assignee_type)
+    if assignee_id:
+        query = query.where(Task.assignee_id == uuid.UUID(assignee_id))
+    if parent_task_id:
+        query = query.where(Task.parent_task_id == uuid.UUID(parent_task_id))
+    if task_type:
+        query = query.where(Task.task_type == task_type)
     if search:
         query = query.where(Task.title.ilike(f"%{search}%"))
 
@@ -85,10 +256,7 @@ async def list_tasks(
 
 async def get_task(db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID) -> dict[str, Any] | None:
     """Get a single task by ID."""
-    result = await db.execute(
-        select(Task).where(Task.id == task_id, Task.tenant_id == tenant_id, Task.deleted_at.is_(None))
-    )
-    task = result.scalar_one_or_none()
+    task = await _get_task(db, tenant_id, task_id)
     if task is None:
         return None
     return _task_to_dict(task)
@@ -103,6 +271,32 @@ async def create_task(
     """Create a new task."""
     from app.core.hooks import do_action
     await do_action("task.before_create", data, db=db, tenant_id=tenant_id, user_id=user_id)
+
+    # Resolve polymorphic assignee, mirroring legacy fields for compatibility.
+    assignee_type = data.get("assignee_type", "user")
+    assignee_id = data.get("assignee_id")
+    assigned_to = data.get("assigned_to")
+    if assignee_type == "user" and assignee_id and not assigned_to:
+        assigned_to = assignee_id
+    if assignee_type == "user" and assigned_to and not assignee_id:
+        assignee_id = assigned_to
+
+    # Resolve polymorphic entity link, mirroring legacy contact_id.
+    entity_type = data.get("entity_type")
+    entity_id = data.get("entity_id")
+    contact_id = data.get("contact_id")
+    if entity_type == "contact" and entity_id and not contact_id:
+        contact_id = entity_id
+    if not entity_type and contact_id:
+        entity_type = "contact"
+        entity_id = contact_id
+
+    # Resolve polymorphic creator.
+    creator_type = data.get("creator_type", "user")
+    creator_id = data.get("creator_id")
+    if creator_type == "user" and not creator_id:
+        creator_id = user_id
+
     task = Task(
         tenant_id=tenant_id,
         title=data["title"],
@@ -110,13 +304,33 @@ async def create_task(
         status=data.get("status", "open"),
         priority=data.get("priority", "medium"),
         due_date=data.get("due_date"),
-        assigned_to=uuid.UUID(data["assigned_to"]) if data.get("assigned_to") else None,
-        contact_id=uuid.UUID(data["contact_id"]) if data.get("contact_id") else None,
+        assigned_to=uuid.UUID(assigned_to) if assigned_to else None,
+        contact_id=uuid.UUID(contact_id) if contact_id else None,
+        assignee_type=assignee_type,
+        assignee_id=uuid.UUID(assignee_id) if assignee_id else None,
+        entity_type=entity_type,
+        entity_id=uuid.UUID(entity_id) if entity_id else None,
+        creator_type=creator_type,
+        creator_id=uuid.UUID(creator_id) if creator_id else None,
+        parent_task_id=uuid.UUID(data["parent_task_id"]) if data.get("parent_task_id") else None,
+        depends_on=[str(d) for d in (data.get("depends_on") or [])],
+        task_type=data.get("task_type", "todo"),
+        success_criteria=data.get("success_criteria"),
+        target_date=data.get("target_date"),
+        progress=data.get("progress", 0),
         created_by=user_id,
         owner_id=user_id,
     )
     db.add(task)
     await db.flush()
+
+    # Recompute parent progress/status when creating a subtask.
+    if task.parent_task_id:
+        parent = await _get_task(db, tenant_id, task.parent_task_id)
+        if parent is not None:
+            await _recompute_progress(db, tenant_id, parent)
+            await _propagate_parent_status(db, tenant_id, task)
+
     snapshot = _task_to_dict(task)
 
     # Record history (D-PLUG)
@@ -135,6 +349,9 @@ async def create_task(
         'user_id': str(user_id),
         'title': task.title,
         'assigned_to': str(task.assigned_to) if task.assigned_to else None,
+        'assignee_type': task.assignee_type,
+        'assignee_id': str(task.assignee_id) if task.assignee_id else None,
+        'task_type': task.task_type,
     })
 
     return _task_to_dict(task)
@@ -147,10 +364,7 @@ async def update_task(
     data: dict[str, Any],
 ) -> dict[str, Any] | None:
     """Update a task."""
-    result = await db.execute(
-        select(Task).where(Task.id == task_id, Task.tenant_id == tenant_id, Task.deleted_at.is_(None))
-    )
-    task = result.scalar_one_or_none()
+    task = await _get_task(db, tenant_id, task_id)
     if task is None:
         return None
 
@@ -172,9 +386,42 @@ async def update_task(
         task.assigned_to = uuid.UUID(data["assigned_to"]) if data["assigned_to"] else None
     if "contact_id" in data:
         task.contact_id = uuid.UUID(data["contact_id"]) if data["contact_id"] else None
+    # F.14 polymorphic fields
+    if "assignee_type" in data and data["assignee_type"] is not None:
+        task.assignee_type = data["assignee_type"]
+    if "assignee_id" in data:
+        task.assignee_id = uuid.UUID(data["assignee_id"]) if data["assignee_id"] else None
+    if "entity_type" in data:
+        task.entity_type = data["entity_type"]
+    if "entity_id" in data:
+        task.entity_id = uuid.UUID(data["entity_id"]) if data["entity_id"] else None
+    if "creator_type" in data and data["creator_type"] is not None:
+        task.creator_type = data["creator_type"]
+    if "creator_id" in data:
+        task.creator_id = uuid.UUID(data["creator_id"]) if data["creator_id"] else None
+    if "parent_task_id" in data:
+        task.parent_task_id = uuid.UUID(data["parent_task_id"]) if data["parent_task_id"] else None
+    if "depends_on" in data:
+        task.depends_on = [str(d) for d in (data["depends_on"] or [])]
+    if "task_type" in data and data["task_type"] is not None:
+        task.task_type = data["task_type"]
+    if "success_criteria" in data:
+        task.success_criteria = data["success_criteria"]
+    if "target_date" in data:
+        task.target_date = data["target_date"]
+    if "progress" in data and data["progress"] is not None:
+        task.progress = data["progress"]
 
     await db.flush()
     await db.refresh(task)
+
+    # Recompute parent progress/status after a child update.
+    if task.parent_task_id:
+        parent = await _get_task(db, tenant_id, task.parent_task_id)
+        if parent is not None:
+            await _recompute_progress(db, tenant_id, parent)
+            await _propagate_parent_status(db, tenant_id, task)
+
     snapshot_after = _task_to_dict(task)
     # Compute changes diff (D-PLUG)
     changes: dict = {}
@@ -193,10 +440,7 @@ async def update_task(
 
 async def delete_task(db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID) -> bool:
     """Soft-delete a task."""
-    result = await db.execute(
-        select(Task).where(Task.id == task_id, Task.tenant_id == tenant_id, Task.deleted_at.is_(None))
-    )
-    task = result.scalar_one_or_none()
+    task = await _get_task(db, tenant_id, task_id)
     if task is None:
         return False
     from app.core.hooks import do_action
@@ -217,16 +461,21 @@ async def assign_task(
     db: AsyncSession,
     tenant_id: uuid.UUID,
     task_id: uuid.UUID,
-    assigned_to: uuid.UUID,
+    assigned_to: uuid.UUID | None = None,
+    *,
+    assignee_type: str = "user",
+    assignee_id: uuid.UUID | None = None,
 ) -> dict[str, Any] | None:
-    """Assign a task to a user."""
-    result = await db.execute(
-        select(Task).where(Task.id == task_id, Task.tenant_id == tenant_id, Task.deleted_at.is_(None))
-    )
-    task = result.scalar_one_or_none()
+    """Assign a task to a user, agent, or group (polymorphic)."""
+    task = await _get_task(db, tenant_id, task_id)
     if task is None:
         return None
-    task.assigned_to = assigned_to
+    if assignee_id is None and assigned_to is not None:
+        assignee_id = assigned_to
+    task.assignee_type = assignee_type
+    task.assignee_id = assignee_id
+    # Mirror legacy field for user assignment.
+    task.assigned_to = assignee_id if assignee_type == "user" else None
     await db.flush()
     return _task_to_dict(task)
 
@@ -237,21 +486,127 @@ async def update_task_status(
     task_id: uuid.UUID,
     new_status: str,
 ) -> dict[str, Any] | None:
-    """Update task status."""
-    result = await db.execute(
-        select(Task).where(Task.id == task_id, Task.tenant_id == tenant_id, Task.deleted_at.is_(None))
-    )
-    task = result.scalar_one_or_none()
+    """Update task status and propagate to parent + evaluate goal criteria."""
+    task = await _get_task(db, tenant_id, task_id)
     if task is None:
         return None
     task.status = new_status
     await db.flush()
+
+    # Recompute parent progress/status.
+    if task.parent_task_id:
+        parent = await _get_task(db, tenant_id, task.parent_task_id)
+        if parent is not None:
+            await _recompute_progress(db, tenant_id, parent)
+            await _propagate_parent_status(db, tenant_id, task)
+
+    # Goal completion: when all success criteria are met, mark the goal done.
+    if task.task_type == "goal" and new_status != "done":
+        if await _evaluate_success_criteria(task):
+            task.status = "done"
+            await db.flush()
+
     if new_status == "done":
         from app.core.outbox import enqueue_outbox_event
         await enqueue_outbox_event(db, tenant_id, "task.completed", {"task_id": str(task.id), "tenant_id": str(tenant_id), "title": task.title, "assigned_to": str(task.assigned_to) if task.assigned_to else None}, aggregate_type="task", aggregate_id=task.id)
     return _task_to_dict(task)
 
 
+async def create_subtask(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    parent_task_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any] | None:
+    """Create a subtask under a parent task."""
+    parent = await _get_task(db, tenant_id, parent_task_id)
+    if parent is None:
+        return None
+    data = dict(data)
+    data["parent_task_id"] = str(parent_task_id)
+    return await create_task(db, tenant_id, user_id, data)
+
+
+async def list_subtasks(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    parent_task_id: uuid.UUID,
+) -> list[dict[str, Any]]:
+    """List subtasks for a parent task."""
+    result = await db.execute(
+        select(Task).where(
+            Task.tenant_id == tenant_id,
+            Task.parent_task_id == parent_task_id,
+            Task.deleted_at.is_(None),
+        ).order_by(Task.created_at.asc())
+    )
+    return [_task_to_dict(t) for t in result.scalars().all()]
+
+
+async def add_dependency(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    task_id: uuid.UUID,
+    depends_on: uuid.UUID,
+) -> dict[str, Any] | None:
+    """Add a dependency (this task depends on another task)."""
+    task = await _get_task(db, tenant_id, task_id)
+    if task is None:
+        return None
+    dep = await _get_task(db, tenant_id, depends_on)
+    if dep is None:
+        return None
+    current = [str(d) for d in (task.depends_on or [])]
+    if str(depends_on) not in current:
+        current.append(str(depends_on))
+        task.depends_on = current
+        await db.flush()
+    return _task_to_dict(task)
+
+
+async def remove_dependency(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    task_id: uuid.UUID,
+    depends_on: uuid.UUID,
+) -> dict[str, Any] | None:
+    """Remove a dependency from a task."""
+    task = await _get_task(db, tenant_id, task_id)
+    if task is None:
+        return None
+    current = [str(d) for d in (task.depends_on or [])]
+    if str(depends_on) in current:
+        current.remove(str(depends_on))
+        task.depends_on = current
+        await db.flush()
+    return _task_to_dict(task)
+
+
+async def decompose_goal(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    goal_id: uuid.UUID,
+    subtasks: list[dict[str, Any]],
+) -> dict[str, Any] | None:
+    """Decompose a goal into subtasks (milestones/todos)."""
+    goal = await _get_task(db, tenant_id, goal_id)
+    if goal is None:
+        return None
+    if goal.task_type != "goal":
+        goal.task_type = "goal"
+    created: list[dict[str, Any]] = []
+    for item in subtasks:
+        data = dict(item)
+        data["parent_task_id"] = str(goal_id)
+        data.setdefault("task_type", "milestone" if item.get("milestone") else "todo")
+        created.append(await create_task(db, tenant_id, user_id, data))
+    await _recompute_progress(db, tenant_id, goal)
+    await _propagate_parent_status(db, tenant_id, goal)
+    return {"goal": _task_to_dict(goal), "subtasks": created}
+
+
 async def get_due_tasks(
     db: AsyncSession,
     tenant_id: uuid.UUID,