"""Business logic for the Tasks plugin.""" from __future__ import annotations import uuid from datetime import UTC, datetime from typing import Any from uuid import UUID def _to_uuid(val: str | UUID | None) -> UUID | None: """Convert a string or UUID to UUID, handling both types safely.""" if val is None: return None if isinstance(val, UUID): return val return uuid.UUID(str(val)) from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession 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 { "id": str(task.id), "title": task.title, "description": task.description, "status": task.status, "priority": task.priority, "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.entity_id) if task.entity_type == "contact" and task.entity_id else (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) # If parent has success_criteria and all are met, mark as done. if parent.task_type == "goal" and await _evaluate_success_criteria(parent): parent.status = "done" 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": ""}]}`` 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, *, page: int = 1, page_size: int = 25, status: str | None = None, priority: str | None = None, 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]: """List tasks with filtering and pagination.""" query = select(Task).where(Task.tenant_id == tenant_id, Task.deleted_at.is_(None)) if user_id and not is_system_admin: query = await apply_visibility_filter( db, query, "task", Task, user_id, tenant_id, is_system_admin ) if status: query = query.where(Task.status == status) if priority: query = query.where(Task.priority == priority) if assigned_to: query = query.where(Task.assigned_to == _to_uuid(assigned_to)) if contact_id: query = query.where(Task.contact_id == _to_uuid(contact_id)) if entity_type: query = query.where(Task.entity_type == entity_type) if entity_id: query = query.where(Task.entity_id == _to_uuid(entity_id)) if assignee_type: query = query.where(Task.assignee_type == assignee_type) if assignee_id: query = query.where(Task.assignee_id == _to_uuid(assignee_id)) if parent_task_id: query = query.where(Task.parent_task_id == _to_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}%")) # Count total count_query = select(func.count()).select_from(query.subquery()) total_result = await db.execute(count_query) total = total_result.scalar() or 0 # Paginate query = query.order_by(Task.due_date.asc().nulls_last(), Task.created_at.desc()) query = query.offset((page - 1) * page_size).limit(page_size) result = await db.execute(query) tasks = result.scalars().all() return { "items": [_task_to_dict(t) for t in tasks], "total": total, "page": page, "page_size": page_size, } async def get_task(db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID) -> dict[str, Any] | None: """Get a single task by ID.""" task = await _get_task(db, tenant_id, task_id) if task is None: return None await db.refresh(task) return _task_to_dict(task) async def create_task( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict[str, Any], ) -> dict[str, Any]: """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 # Don't store contact_id in FK column if it's just a polymorphic entity link # — contact_id FK requires a real Contact row. Use entity_id instead. contact_id_for_fk = _to_uuid(contact_id) if contact_id else None # 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"], description=data.get("description"), status=data.get("status", "open"), priority=data.get("priority", "medium"), due_date=data.get("due_date"), assigned_to=_to_uuid(assigned_to) if assigned_to else None, contact_id=None, # Don't store in FK column — derived from entity_id in _task_to_dict assignee_type=assignee_type, assignee_id=_to_uuid(assignee_id) if assignee_id else None, entity_type=entity_type, entity_id=_to_uuid(entity_id) if entity_id else None, creator_type=creator_type, creator_id=_to_uuid(creator_id) if creator_id else None, parent_task_id=_to_uuid(data.get("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) from app.services.entity_history_service import record_history await record_history(db, tenant_id, user_id, "task", task.id, "create", snapshot_after=snapshot) from app.core.hooks import do_action await do_action("task.after_create", snapshot, db=db, tenant_id=tenant_id, user_id=user_id) # Publish task.created event from app.core.event_bus import get_event_bus event_bus = get_event_bus() await event_bus.publish('task.created', { 'task_id': str(task.id), 'tenant_id': str(tenant_id), '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, }) await db.refresh(task) return _task_to_dict(task) async def update_task( db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID, data: dict[str, Any], ) -> dict[str, Any] | None: """Update a task.""" task = await _get_task(db, tenant_id, task_id) if task is None: return None from app.core.hooks import do_action await do_action("task.before_update", data, db=db, tenant_id=tenant_id, task_id=str(task_id)) # Capture snapshot before update (D-PLUG) snapshot_before = _task_to_dict(task) if "title" in data and data["title"] is not None: task.title = data["title"] if "description" in data: task.description = data["description"] if "status" in data and data["status"] is not None: task.status = data["status"] if "priority" in data and data["priority"] is not None: task.priority = data["priority"] if "due_date" in data: task.due_date = data["due_date"] if "assigned_to" in data: task.assigned_to = _to_uuid(data.get("assigned_to")) if data["assigned_to"] else None if "contact_id" in data: task.contact_id = _to_uuid(data.get("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 = _to_uuid(data.get("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 = _to_uuid(data.get("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 = _to_uuid(data.get("creator_id")) if data["creator_id"] else None if "parent_task_id" in data: task.parent_task_id = _to_uuid(data.get("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 = {} for key, new_val in snapshot_after.items(): old_val = snapshot_before.get(key) if old_val != new_val: changes[key] = {"old": old_val, "new": new_val} # Record history (D-PLUG) from app.services.entity_history_service import record_history as _rh await _rh(db, tenant_id, None, "task", task.id, "update", snapshot_before=snapshot_before, snapshot_after=snapshot_after, changes=changes or None) from app.core.hooks import do_action await do_action("task.after_update", snapshot_after, db=db, tenant_id=tenant_id, task_id=str(task_id)) return snapshot_after async def delete_task(db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID) -> bool: """Soft-delete a task.""" task = await _get_task(db, tenant_id, task_id) if task is None: return False from app.core.hooks import do_action await do_action("task.before_delete", db=db, tenant_id=tenant_id, task_id=str(task_id)) # Capture snapshot before delete (D-PLUG) snapshot_before = _task_to_dict(task) task.deleted_at = datetime.now(UTC) await db.flush() # Record history (D-PLUG) from app.services.entity_history_service import record_history as _rh await _rh(db, tenant_id, None, "task", task.id, "delete", snapshot_before=snapshot_before) from app.core.hooks import do_action await do_action("task.after_delete", db=db, tenant_id=tenant_id, task_id=str(task_id)) return True async def assign_task( db: AsyncSession, tenant_id: uuid.UUID, task_id: 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, agent, or group (polymorphic).""" task = await _get_task(db, tenant_id, task_id) if task is None: return None 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() await db.refresh(task) return _task_to_dict(task) async def update_task_status( db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID, new_status: str, ) -> dict[str, Any] | 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) await db.refresh(task) 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() await db.refresh(task) 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() await db.refresh(task) 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, *, before: datetime | None = None, ) -> list[dict[str, Any]]: """Get tasks that are due (for ARQ reminder job).""" now = before or datetime.now(UTC) result = await db.execute( select(Task).where( Task.tenant_id == tenant_id, Task.deleted_at.is_(None), Task.status != "done", Task.due_date.is_not(None), Task.due_date <= now, ) ) tasks = result.scalars().all() return [_task_to_dict(t) for t in tasks]