diff --git a/app/core/trigger_dispatcher.py b/app/core/trigger_dispatcher.py index d073cef..51f5e0b 100644 --- a/app/core/trigger_dispatcher.py +++ b/app/core/trigger_dispatcher.py @@ -100,6 +100,13 @@ class TriggerDispatcher: trigger_type=trigger_type, payload=payload, ) + # F-PROACTIVE: Also check for matching agent definitions on context/UI events + if is_ui_event or event_name.startswith("context."): + await self._dispatch_matching_agents( + event_name=event_name, + trigger_type=trigger_type, + payload=payload, + ) except Exception: logger.exception( "TriggerDispatcher: error dispatching event '%s'", event_name @@ -164,6 +171,72 @@ class TriggerDispatcher: trigger_data=payload, ) + async def _dispatch_matching_agents( + self, + event_name: str, + trigger_type: str, + payload: dict[str, Any], + ) -> None: + """F-PROACTIVE: Query DB for active agents matching *event_name* and dispatch. + + Checks AgentDefinition.trigger_config for matching context/UI events. + If match found, creates an AgentRun and dispatches via run_agent. + """ + from app.core.db import get_session_factory + from app.plugins.builtins.contracts import get_contract + + automation_contract = get_contract("automation") + if automation_contract is None: + return + + AgentDefinition = automation_contract.AgentDefinition # noqa: N806 + if AgentDefinition is None: + return + + factory = get_session_factory() + tenant_id = payload.get("tenant_id") + + async with factory() as db: + query = ( + select(AgentDefinition) + .where(AgentDefinition.is_active.is_(True)) + .where(AgentDefinition.mode == "proactive") + ) + if tenant_id is not None: + query = query.where(AgentDefinition.tenant_id == tenant_id) + + result = await db.execute(query) + agents = list(result.scalars().all()) + + if not agents: + return + + for agent in agents: + config = agent.trigger_config or {} + configured_event = config.get("event_name", "") + if configured_event != event_name: + continue + + logger.info( + "TriggerDispatcher: dispatching agent '%s' (%s) for event '%s'", + agent.name, + agent.id, + event_name, + ) + + try: + await automation_contract.run_agent( + ctx={}, + agent_id=str(agent.id), + trigger_type=trigger_type, + trigger_data=payload, + ) + except Exception: + logger.exception( + "TriggerDispatcher: run_agent failed for agent_id=%s", + agent.id, + ) + async def _enqueue_automation( self, automation_id: str, diff --git a/app/plugins/builtins/tasks/ai_tools.py b/app/plugins/builtins/tasks/ai_tools.py new file mode 100644 index 0000000..c4e2396 --- /dev/null +++ b/app/plugins/builtins/tasks/ai_tools.py @@ -0,0 +1,191 @@ +"""AI Tools for the Tool Registry — task tools for the AI Assistant. + +Each tool is an async function (arguments: dict, context: dict) -> str +that delegates to the tasks service layer and returns a JSON-string result. +All tools require the ``tasks:write`` permission. +""" + +from __future__ import annotations + +import json +import logging +import uuid +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.plugins.builtins.tasks import services + +logger = logging.getLogger(__name__) + + +def _get_db_and_tenant(context: dict[str, Any]) -> tuple[AsyncSession, uuid.UUID, uuid.UUID]: + """Extract tenant_id, user_id from context and return the DB session.""" + tenant_id = uuid.UUID(context.get("tenant_id", "")) + user_id = uuid.UUID(context.get("user_id", "")) + db = context.get("db") + if db is not None and isinstance(db, AsyncSession): + return db, tenant_id, user_id + raise RuntimeError("No db session in context") + + +async def create_task_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str: + """Create a new task (supports polymorphic assignee/entity/creator).""" + try: + db, tenant_id, user_id = _get_db_and_tenant(context) + data = dict(arguments) + result = await services.create_task(db, tenant_id, user_id, data) + return json.dumps(result, default=str) + except Exception as e: + logger.exception("create_task_handler failed") + return json.dumps({"error": str(e)}) + + +async def assign_task_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str: + """Assign a task to a user, agent, or group (polymorphic).""" + try: + db, tenant_id, _ = _get_db_and_tenant(context) + task_id = uuid.UUID(arguments["task_id"]) + assignee_type = arguments.get("assignee_type", "user") + assignee_id = uuid.UUID(arguments["assignee_id"]) if arguments.get("assignee_id") else None + result = await services.assign_task( + db, tenant_id, task_id, None, + assignee_type=assignee_type, assignee_id=assignee_id, + ) + if result is None: + return json.dumps({"error": "Task not found"}) + return json.dumps(result, default=str) + except Exception as e: + logger.exception("assign_task_handler failed") + return json.dumps({"error": str(e)}) + + +async def update_task_status_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str: + """Update a task's lifecycle status.""" + try: + db, tenant_id, _ = _get_db_and_tenant(context) + task_id = uuid.UUID(arguments["task_id"]) + status = arguments["status"] + result = await services.update_task_status(db, tenant_id, task_id, status) + if result is None: + return json.dumps({"error": "Task not found"}) + return json.dumps(result, default=str) + except Exception as e: + logger.exception("update_task_status_handler failed") + return json.dumps({"error": str(e)}) + + +async def decompose_goal_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str: + """Decompose a goal into subtasks (milestones/todos).""" + try: + db, tenant_id, user_id = _get_db_and_tenant(context) + goal_id = uuid.UUID(arguments["goal_id"]) + subtasks = arguments.get("subtasks") or [] + result = await services.decompose_goal(db, tenant_id, user_id, goal_id, subtasks) + if result is None: + return json.dumps({"error": "Goal not found"}) + return json.dumps(result, default=str) + except Exception as e: + logger.exception("decompose_goal_handler failed") + return json.dumps({"error": str(e)}) + + +# ─── Registration ─── + + +def register_task_tools(registry) -> None: + """Register task AI tools into the existing Tool Registry.""" + registry.register( + name="create_task", + description="Neue Aufgabe erstellen (Unified Task System). Unterstützt polymorphe Zuweisung (user/agent/group), Entity-Verknüpfung (contact/company/...), Subtasks, Dependencies, Goals/Milestones und Agent-Subtasks.", + parameters={ + "type": "object", + "properties": { + "title": {"type": "string", "description": "Aufgabentitel"}, + "description": {"type": "string", "description": "Beschreibung"}, + "status": {"type": "string", "enum": ["open", "in_progress", "review", "blocked", "done", "cancelled"], "default": "open"}, + "priority": {"type": "string", "enum": ["low", "medium", "high", "urgent"], "default": "medium"}, + "due_date": {"type": "string", "description": "ISO Datum"}, + "assignee_type": {"type": "string", "enum": ["user", "agent", "group"], "default": "user"}, + "assignee_id": {"type": "string", "description": "UUID des Assignees"}, + "entity_type": {"type": "string", "description": "z.B. contact, company"}, + "entity_id": {"type": "string", "description": "UUID der Entity"}, + "parent_task_id": {"type": "string", "description": "UUID des Parent-Tasks (Subtask)"}, + "depends_on": {"type": "array", "items": {"type": "string"}, "description": "Task-IDs von denen diese Aufgabe abhängt"}, + "task_type": {"type": "string", "enum": ["todo", "approval", "follow_up", "review", "goal", "milestone", "agent_subtask"], "default": "todo"}, + "success_criteria": {"type": "object", "description": "Strukturierte Erfolgskriterien für Goals"}, + "target_date": {"type": "string", "description": "Deadline für Goal/Milestone"}, + }, + "required": ["title"], + }, + handler=create_task_handler, + plugin_name="tasks", + required_permission="tasks:write", + category="tasks", + ) + + registry.register( + name="assign_task", + description="Aufgabe einem User, Agent oder einer Gruppe zuweisen (polymorph).", + parameters={ + "type": "object", + "properties": { + "task_id": {"type": "string", "description": "UUID der Aufgabe"}, + "assignee_type": {"type": "string", "enum": ["user", "agent", "group"], "default": "user"}, + "assignee_id": {"type": "string", "description": "UUID des Assignees"}, + }, + "required": ["task_id", "assignee_id"], + }, + handler=assign_task_handler, + plugin_name="tasks", + required_permission="tasks:write", + category="tasks", + ) + + registry.register( + name="update_task_status", + description="Status einer Aufgabe aktualisieren (open/in_progress/review/blocked/done/cancelled).", + parameters={ + "type": "object", + "properties": { + "task_id": {"type": "string", "description": "UUID der Aufgabe"}, + "status": {"type": "string", "enum": ["open", "in_progress", "review", "blocked", "done", "cancelled"]}, + }, + "required": ["task_id", "status"], + }, + handler=update_task_status_handler, + plugin_name="tasks", + required_permission="tasks:write", + category="tasks", + ) + + registry.register( + name="decompose_goal", + description="Ein Goal in Subtasks (Milestones/Todos) zerlegen.", + parameters={ + "type": "object", + "properties": { + "goal_id": {"type": "string", "description": "UUID des Goals"}, + "subtasks": { + "type": "array", + "items": { + "type": "object", + "properties": { + "title": {"type": "string"}, + "description": {"type": "string"}, + "milestone": {"type": "boolean", "default": False}, + "assignee_type": {"type": "string", "enum": ["user", "agent", "group"]}, + "assignee_id": {"type": "string"}, + }, + "required": ["title"], + }, + }, + }, + "required": ["goal_id", "subtasks"], + }, + handler=decompose_goal_handler, + plugin_name="tasks", + required_permission="tasks:write", + category="tasks", + ) + logger.info("Task AI tools registered") diff --git a/app/plugins/builtins/tasks/models.py b/app/plugins/builtins/tasks/models.py index 67af958..030f67a 100644 --- a/app/plugins/builtins/tasks/models.py +++ b/app/plugins/builtins/tasks/models.py @@ -4,8 +4,10 @@ from __future__ import annotations import uuid from datetime import datetime +from typing import Any -from sqlalchemy import DateTime, ForeignKey, Index, String, Text +from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, Text +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import UUID as PGUUID from sqlalchemy.orm import Mapped, mapped_column @@ -14,7 +16,13 @@ from app.models.owned_mixin import OwnedMixin class Task(Base, TenantMixin, OwnedMixin): - """Task entity — free activities (calls, notes, visits) linked to contacts.""" + """Task entity — free activities (calls, notes, visits) linked to entities. + + Supports polymorphic assignment (user/agent/group), polymorphic entity + links (contact/company/...), subtasks, dependencies, goals/milestones and + agent subtasks. Legacy ``contact_id`` and ``assigned_to`` columns are kept + for backward compatibility and mirrored into the polymorphic fields. + """ __tablename__ = "tasks" __table_args__ = ( @@ -23,6 +31,10 @@ class Task(Base, TenantMixin, OwnedMixin): Index("ix_tasks_tenant_assigned", "tenant_id", "assigned_to"), Index("ix_tasks_tenant_due", "tenant_id", "due_date"), Index("ix_tasks_contact", "contact_id"), + Index("ix_tasks_tenant_entity", "tenant_id", "entity_type", "entity_id"), + Index("ix_tasks_tenant_assignee", "tenant_id", "assignee_type", "assignee_id"), + Index("ix_tasks_tenant_parent", "tenant_id", "parent_task_id"), + Index("ix_tasks_tenant_type", "tenant_id", "task_type"), ) id: Mapped[uuid.UUID] = mapped_column( @@ -32,7 +44,7 @@ class Task(Base, TenantMixin, OwnedMixin): description: Mapped[str | None] = mapped_column(Text, nullable=True) status: Mapped[str] = mapped_column( String(20), nullable=False, default="open" - ) # open, in_progress, done + ) # open, in_progress, review, blocked, done, cancelled priority: Mapped[str] = mapped_column( String(10), nullable=False, default="medium" ) # low, medium, high, urgent @@ -49,3 +61,46 @@ class Task(Base, TenantMixin, OwnedMixin): PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True ) deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + # ── F.14 Unified Task System ──────────────────────────────────────────── + # Polymorphic assignee (user/agent/group) + assignee_type: Mapped[str] = mapped_column( + String(20), nullable=False, default="user" + ) # user, agent, group + assignee_id: Mapped[uuid.UUID | None] = mapped_column( + PGUUID(as_uuid=True), nullable=True + ) + + # Polymorphic entity link (contact/company/...) + entity_type: Mapped[str | None] = mapped_column(String(80), nullable=True) + entity_id: Mapped[uuid.UUID | None] = mapped_column(PGUUID(as_uuid=True), nullable=True) + + # Polymorphic creator (user/agent/workflow/system) + creator_type: Mapped[str] = mapped_column( + String(20), nullable=False, default="user" + ) # user, agent, workflow, system + creator_id: Mapped[uuid.UUID | None] = mapped_column( + PGUUID(as_uuid=True), nullable=True + ) + + # Subtasks (self-reference) + parent_task_id: Mapped[uuid.UUID | None] = mapped_column( + PGUUID(as_uuid=True), + ForeignKey("tasks.id", ondelete="CASCADE"), + nullable=True, + ) + + # Task dependencies (blocked-by task IDs) + depends_on: Mapped[list[Any]] = mapped_column(JSONB, nullable=False, default=list) + + # Task type + task_type: Mapped[str] = mapped_column( + String(30), nullable=False, default="todo" + ) # todo, approval, follow_up, review, goal, milestone, agent_subtask + + # Goal / Milestone support + success_criteria: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) + target_date: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + progress: Mapped[int] = mapped_column(Integer, nullable=False, default=0) # 0-100 diff --git a/app/plugins/builtins/tasks/plugin.py b/app/plugins/builtins/tasks/plugin.py index cbf0d76..6ff81ee 100644 --- a/app/plugins/builtins/tasks/plugin.py +++ b/app/plugins/builtins/tasks/plugin.py @@ -76,9 +76,18 @@ class TasksPlugin(BasePlugin): return {"task": Task} async def on_activate(self, db, service_container, event_bus) -> None: - """Activate plugin: register restore config + history hooks.""" + """Activate plugin: register restore config + history hooks + AI tools.""" await super().on_activate(db, service_container, event_bus) + # Register task AI tools (F-TASK-AGENT) + try: + from app.plugins.builtins.ai_assistant.contracts import get_tool_registry + from app.plugins.builtins.tasks.ai_tools import register_task_tools + register_task_tools(get_tool_registry()) + except Exception: + import logging + logging.getLogger(__name__).exception("Failed to register task AI tools") + # Register restore config for Task entities (P0-7 fix) from app.core.restore_registry import RestoreConfig, get_restore_registry from app.plugins.builtins.tasks.models import Task @@ -101,7 +110,15 @@ class TasksPlugin(BasePlugin): async def on_deactivate( self, db, service_container, event_bus ) -> None: - """Deactivate plugin: unregister contract, restore, history, events.""" + """Deactivate plugin: unregister contract, restore, history, events, AI tools.""" + # Unregister task AI tools (F-TASK-AGENT) + try: + from app.plugins.builtins.ai_assistant.contracts import get_tool_registry + get_tool_registry().unregister_plugin("tasks") + except Exception: + import logging + logging.getLogger(__name__).exception("Failed to unregister task AI tools") + # Contract abmelden from app.plugins.builtins.contracts import get_contract_registry get_contract_registry().unregister(self.manifest.name) diff --git a/app/plugins/builtins/tasks/routes.py b/app/plugins/builtins/tasks/routes.py index 4af251e..be01769 100644 --- a/app/plugins/builtins/tasks/routes.py +++ b/app/plugins/builtins/tasks/routes.py @@ -1,4 +1,4 @@ -"""Tasks plugin routes — CRUD, assign, status update.""" +"""Tasks plugin routes — CRUD, assign, status update, subtasks, dependencies.""" from __future__ import annotations @@ -13,6 +13,7 @@ from app.plugins.builtins.tasks import services from app.plugins.builtins.tasks.schemas import ( TaskAssignRequest, TaskCreate, + TaskDependencyRequest, TaskStatusRequest, TaskUpdate, ) @@ -20,6 +21,11 @@ from app.plugins.builtins.tasks.schemas import ( router = APIRouter(prefix="/api/v1/tasks", tags=["tasks"]) +TASK_STATUS_PATTERN = "^(open|in_progress|review|blocked|done|cancelled)$" +TASK_TYPE_PATTERN = "^(todo|approval|follow_up|review|goal|milestone|agent_subtask)$" +ASSIGNEE_TYPE_PATTERN = "^(user|agent|group)$" + + def _parse_uuid(val: str, field: str) -> uuid.UUID: try: return uuid.UUID(val) @@ -33,10 +39,16 @@ def _parse_uuid(val: str, field: str) -> uuid.UUID: async def list_tasks( page: int = Query(1, ge=1), page_size: int = Query(25, ge=1, le=100), - status: str | None = Query(None, pattern="^(open|in_progress|done)$"), + status: str | None = Query(None, pattern=TASK_STATUS_PATTERN), priority: str | None = Query(None, pattern="^(low|medium|high|urgent)$"), assigned_to: str | None = Query(None), contact_id: str | None = Query(None), + entity_type: str | None = Query(None), + entity_id: str | None = Query(None), + assignee_type: str | None = Query(None, pattern=ASSIGNEE_TYPE_PATTERN), + assignee_id: str | None = Query(None), + parent_task_id: str | None = Query(None), + task_type: str | None = Query(None, pattern=TASK_TYPE_PATTERN), search: str | None = Query(None), db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), @@ -50,6 +62,9 @@ async def list_tasks( page=page, page_size=page_size, status=status, priority=priority, assigned_to=assigned_to, contact_id=contact_id, + entity_type=entity_type, entity_id=entity_id, + assignee_type=assignee_type, assignee_id=assignee_id, + parent_task_id=parent_task_id, task_type=task_type, search=search, user_id=user_id, is_system_admin=is_system_admin, ) @@ -122,11 +137,15 @@ async def assign_task( db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): - """Assign a task to a user.""" + """Assign a task to a user, agent, or group (polymorphic).""" tenant_id = uuid.UUID(current_user["tenant_id"]) tid = _parse_uuid(task_id, "task_id") - assigned_to = _parse_uuid(body.assigned_to, "assigned_to") - result = await services.assign_task(db, tenant_id, tid, assigned_to) + assigned_to = _parse_uuid(body.assigned_to, "assigned_to") if body.assigned_to else None + assignee_id = _parse_uuid(body.assignee_id, "assignee_id") if body.assignee_id else None + result = await services.assign_task( + db, tenant_id, tid, assigned_to, + assignee_type=body.assignee_type, assignee_id=assignee_id, + ) if result is None: raise HTTPException(404, detail={"detail": "Task not found", "code": "not_found"}) return result @@ -146,3 +165,94 @@ async def update_task_status( if result is None: raise HTTPException(404, detail={"detail": "Task not found", "code": "not_found"}) return result + + +# ── F.14 Subtasks ──────────────────────────────────────────────────────────── + + +@router.post("/{task_id}/subtasks", status_code=status.HTTP_201_CREATED, dependencies=[Depends(require_permission("tasks:write"))]) +async def create_subtask( + task_id: str, + body: TaskCreate, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), +): + """Create a subtask under a parent task.""" + tenant_id = uuid.UUID(current_user["tenant_id"]) + user_id = uuid.UUID(current_user["user_id"]) + tid = _parse_uuid(task_id, "task_id") + result = await services.create_subtask(db, tenant_id, user_id, tid, body.model_dump()) + if result is None: + raise HTTPException(404, detail={"detail": "Parent task not found", "code": "not_found"}) + return result + + +@router.get("/{task_id}/subtasks", dependencies=[Depends(require_permission("tasks:read"))]) +async def list_subtasks( + task_id: str, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), +): + """List subtasks for a parent task.""" + tenant_id = uuid.UUID(current_user["tenant_id"]) + tid = _parse_uuid(task_id, "task_id") + return await services.list_subtasks(db, tenant_id, tid) + + +# ── F.14 Dependencies ──────────────────────────────────────────────────────── + + +@router.post("/{task_id}/dependencies", dependencies=[Depends(require_permission("tasks:write"))]) +async def add_dependency( + task_id: str, + body: TaskDependencyRequest, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), +): + """Add a dependency (this task depends on another task).""" + tenant_id = uuid.UUID(current_user["tenant_id"]) + tid = _parse_uuid(task_id, "task_id") + dep_id = _parse_uuid(body.depends_on, "depends_on") + result = await services.add_dependency(db, tenant_id, tid, dep_id) + if result is None: + raise HTTPException(404, detail={"detail": "Task or dependency not found", "code": "not_found"}) + return result + + +@router.delete("/{task_id}/dependencies/{depends_on}", dependencies=[Depends(require_permission("tasks:write"))]) +async def remove_dependency( + task_id: str, + depends_on: str, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), +): + """Remove a dependency from a task.""" + tenant_id = uuid.UUID(current_user["tenant_id"]) + tid = _parse_uuid(task_id, "task_id") + dep_id = _parse_uuid(depends_on, "depends_on") + result = await services.remove_dependency(db, tenant_id, tid, dep_id) + if result is None: + raise HTTPException(404, detail={"detail": "Task not found", "code": "not_found"}) + return result + + +# ── F.14 Goal decomposition ───────────────────────────────────────────────── + + +@router.post("/{task_id}/decompose", dependencies=[Depends(require_permission("tasks:write"))]) +async def decompose_goal( + task_id: str, + body: list[TaskCreate], + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), +): + """Decompose a goal into subtasks (milestones/todos).""" + tenant_id = uuid.UUID(current_user["tenant_id"]) + user_id = uuid.UUID(current_user["user_id"]) + tid = _parse_uuid(task_id, "task_id") + result = await services.decompose_goal( + db, tenant_id, user_id, tid, [item.model_dump() for item in body] + ) + if result is None: + raise HTTPException(404, detail={"detail": "Goal not found", "code": "not_found"}) + return result diff --git a/app/plugins/builtins/tasks/schemas.py b/app/plugins/builtins/tasks/schemas.py index 86f4ef3..388e87a 100644 --- a/app/plugins/builtins/tasks/schemas.py +++ b/app/plugins/builtins/tasks/schemas.py @@ -3,40 +3,79 @@ from __future__ import annotations from datetime import datetime +from typing import Any from pydantic import BaseModel, Field +TASK_STATUSES = "^(open|in_progress|review|blocked|done|cancelled)$" +TASK_TYPES = "^(todo|approval|follow_up|review|goal|milestone|agent_subtask)$" +ASSIGNEE_TYPES = "^(user|agent|group)$" +CREATOR_TYPES = "^(user|agent|workflow|system)$" + + class TaskCreate(BaseModel): """Schema for creating a task.""" title: str = Field(..., min_length=1, max_length=255) description: str | None = None - status: str = Field(default="open", pattern="^(open|in_progress|done)$") + status: str = Field(default="open", pattern=TASK_STATUSES) priority: str = Field(default="medium", pattern="^(low|medium|high|urgent)$") due_date: datetime | None = None assigned_to: str | None = None contact_id: str | None = None + # F.14 polymorphic fields + assignee_type: str = Field(default="user", pattern=ASSIGNEE_TYPES) + assignee_id: str | None = None + entity_type: str | None = None + entity_id: str | None = None + creator_type: str = Field(default="user", pattern=CREATOR_TYPES) + creator_id: str | None = None + parent_task_id: str | None = None + depends_on: list[str] = Field(default_factory=list) + task_type: str = Field(default="todo", pattern=TASK_TYPES) + success_criteria: dict[str, Any] | None = None + target_date: datetime | None = None + progress: int = Field(default=0, ge=0, le=100) class TaskUpdate(BaseModel): """Schema for updating a task.""" title: str | None = Field(default=None, min_length=1, max_length=255) description: str | None = None - status: str | None = Field(default=None, pattern="^(open|in_progress|done)$") + status: str | None = Field(default=None, pattern=TASK_STATUSES) priority: str | None = Field(default=None, pattern="^(low|medium|high|urgent)$") due_date: datetime | None = None assigned_to: str | None = None contact_id: str | None = None + assignee_type: str | None = Field(default=None, pattern=ASSIGNEE_TYPES) + assignee_id: str | None = None + entity_type: str | None = None + entity_id: str | None = None + creator_type: str | None = Field(default=None, pattern=CREATOR_TYPES) + creator_id: str | None = None + parent_task_id: str | None = None + depends_on: list[str] | None = None + task_type: str | None = Field(default=None, pattern=TASK_TYPES) + success_criteria: dict[str, Any] | None = None + target_date: datetime | None = None + progress: int | None = Field(default=None, ge=0, le=100) class TaskAssignRequest(BaseModel): - """Schema for assigning a task.""" - assigned_to: str = Field(..., description="User ID to assign the task to") + """Schema for assigning a task (polymorphic).""" + assigned_to: str | None = None + assignee_type: str = Field(default="user", pattern=ASSIGNEE_TYPES) + assignee_id: str | None = None class TaskStatusRequest(BaseModel): """Schema for updating task status.""" - status: str = Field(..., pattern="^(open|in_progress|done)$") + status: str = Field(..., pattern=TASK_STATUSES) + + +class TaskDependencyRequest(BaseModel): + """Schema for adding/removing a task dependency.""" + depends_on: str = Field(..., description="Task ID this task depends on") class TaskResponse(BaseModel): @@ -49,6 +88,18 @@ class TaskResponse(BaseModel): due_date: datetime | None = None assigned_to: str | None = None contact_id: str | None = None + assignee_type: str = "user" + assignee_id: str | None = None + entity_type: str | None = None + entity_id: str | None = None + creator_type: str = "user" + creator_id: str | None = None + parent_task_id: str | None = None + depends_on: list[str] = Field(default_factory=list) + task_type: str = "todo" + success_criteria: dict[str, Any] | None = None + target_date: datetime | None = None + progress: int = 0 created_by: str | None = None created_at: datetime updated_at: datetime diff --git a/app/plugins/builtins/tasks/services.py b/app/plugins/builtins/tasks/services.py index 32b94ef..ad7595d 100644 --- a/app/plugins/builtins/tasks/services.py +++ b/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": ""}]}`` + + 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,