feat(5.21): Tasks Plugin — activities with status, priority, due dates, ARQ reminders

- New builtin plugin app/plugins/builtins/tasks/ with full CRUD
- Task model with TenantMixin (title, description, status, priority, due_date, assigned_to, contact_id)
- Routes: GET/POST /tasks, GET/PATCH/DELETE /tasks/{id}, POST /tasks/{id}/assign, POST /tasks/{id}/status
- All routes RBAC-protected (tasks:read, tasks:write, tasks:delete)
- ARQ cron job tasks_due_reminder (daily 8:00) sends notifications for due tasks
- Migration 0001_initial.sql creates tasks table with indexes
- Frontend: Tasks.tsx page with list, filter, create/edit modal, detail modal
- Frontend: api/tasks.ts with React Query hooks
- Route /tasks in routes/index.tsx, sidebar entry via plugin manifest
- i18n keys for nav.tasks and tasks.* in de.json and en.json
- Tests: test_tasks.py (11 tests) + Tasks.test.tsx (3 tests)
- Registered TasksPlugin in conftest.py and worker.py

app/core/worker.py

@@ -53,6 +53,7 @@ from app.plugins.builtins.automation.scheduler import scheduler_tick
 from app.plugins.builtins.automation.workflow_timeout import check_workflow_timeouts
 from app.plugins.builtins.automation.agent_runner import run_agent
 from app.plugins.builtins.automation.execution_engine import run_automation
+from app.plugins.builtins.tasks.jobs import tasks_due_reminder
 
 
 class WorkerSettings:
@@ -69,6 +70,7 @@ class WorkerSettings:
         check_workflow_timeouts,
         run_agent,
         run_automation,
+        tasks_due_reminder,
     ]
     redis_settings = _get_redis_settings()
     on_startup = on_startup
@@ -79,4 +81,5 @@ class WorkerSettings:
     cron_jobs = [
         cron(scheduler_tick, second={0, 30}),
         cron(check_workflow_timeouts, minute={0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55}),
+        cron(tasks_due_reminder, hour=8, minute=0),
     ]

app/plugins/builtins/tasks/__init__.py

@@ -0,0 +1,5 @@
+"""Tasks builtin plugin."""
+
+from app.plugins.builtins.tasks.plugin import TasksPlugin
+
+__all__ = ["TasksPlugin"]

app/plugins/builtins/tasks/jobs.py

@@ -0,0 +1,51 @@
+"""ARQ reminder job for due tasks — sends notifications for overdue/due tasks."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from app.core.db import get_session_factory
+
+logger = logging.getLogger(__name__)
+
+
+async def tasks_due_reminder(ctx: dict) -> None:
+    """Check for due tasks and send notifications to assigned users.
+
+    Runs daily at 8:00 via cron. Finds all non-done tasks with due_date <= now
+    and creates a notification for the assigned user.
+    """
+    from app.plugins.builtins.tasks.services import get_due_tasks
+    from app.models.tenant import Tenant
+    from app.core.notifications import create_notification
+    from sqlalchemy import select
+
+    factory = get_session_factory()
+    async with factory() as db:
+        result = await db.execute(select(Tenant))
+        tenants = result.scalars().all()
+
+        total_notified = 0
+        for tenant in tenants:
+            try:
+                due_tasks = await get_due_tasks(db, tenant.id)
+                for task in due_tasks:
+                    if not task.get("assigned_to"):
+                        continue
+                    await create_notification(
+                        db,
+                        user_id=task["assigned_to"],
+                        tenant_id=str(tenant.id),
+                        type="task_due",
+                        title=f"Task due: {task['title']}",
+                        body=f"Task '{task['title']}' is due. Priority: {task['priority']}",
+                        data={"task_id": task["id"], "due_date": task.get("due_date")},
+                    )
+                    total_notified += 1
+            except Exception:
+                logger.warning(f"Failed to process due tasks for tenant {tenant.id}", exc_info=True)
+
+        if total_notified > 0:
+            logger.info(f"Tasks reminder: sent {total_notified} notifications")
+        await db.commit()

app/plugins/builtins/tasks/migrations/0001_initial.sql

@@ -0,0 +1,21 @@
+-- Tasks plugin initial migration: creates tasks table
+CREATE TABLE IF NOT EXISTS tasks (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    title VARCHAR(255) NOT NULL,
+    description TEXT,
+    status VARCHAR(20) NOT NULL DEFAULT 'open',
+    priority VARCHAR(10) NOT NULL DEFAULT 'medium',
+    due_date TIMESTAMPTZ,
+    assigned_to UUID REFERENCES users(id) ON DELETE SET NULL,
+    contact_id UUID REFERENCES contacts(id) ON DELETE SET NULL,
+    created_by UUID REFERENCES users(id) ON DELETE SET NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ
+);
+CREATE INDEX IF NOT EXISTS ix_tasks_tenant_deleted ON tasks(tenant_id, deleted_at);
+CREATE INDEX IF NOT EXISTS ix_tasks_tenant_status ON tasks(tenant_id, status);
+CREATE INDEX IF NOT EXISTS ix_tasks_tenant_assigned ON tasks(tenant_id, assigned_to);
+CREATE INDEX IF NOT EXISTS ix_tasks_tenant_due ON tasks(tenant_id, due_date);
+CREATE INDEX IF NOT EXISTS ix_tasks_contact ON tasks(contact_id);

app/plugins/builtins/tasks/models.py

@@ -0,0 +1,49 @@
+"""Task model for the Tasks plugin."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Index, String, Text
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, TenantMixin
+
+
+class Task(Base, TenantMixin):
+    """Task entity — free activities (calls, notes, visits) linked to contacts."""
+
+    __tablename__ = "tasks"
+    __table_args__ = (
+        Index("ix_tasks_tenant_deleted", "tenant_id", "deleted_at"),
+        Index("ix_tasks_tenant_status", "tenant_id", "status"),
+        Index("ix_tasks_tenant_assigned", "tenant_id", "assigned_to"),
+        Index("ix_tasks_tenant_due", "tenant_id", "due_date"),
+        Index("ix_tasks_contact", "contact_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    title: Mapped[str] = mapped_column(String(255), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    status: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="open"
+    )  # open, in_progress, done
+    priority: Mapped[str] = mapped_column(
+        String(10), nullable=False, default="medium"
+    )  # low, medium, high, urgent
+    due_date: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    assigned_to: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
+    )
+    contact_id: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True), ForeignKey("contacts.id", ondelete="SET NULL"), nullable=True
+    )
+    created_by: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
+    )

app/plugins/builtins/tasks/plugin.py

@@ -0,0 +1,65 @@
+"""Tasks plugin — manage free tasks/activities linked to contacts."""
+
+from __future__ import annotations
+
+from app.plugins.base import BasePlugin
+from app.plugins.manifest import (
+    PluginManifest,
+    PluginRouteDef,
+    FrontendMenuItem,
+    FrontendPageRoute,
+    CronJobContribution,
+)
+
+
+class TasksPlugin(BasePlugin):
+    """Tasks plugin for managing activities (calls, notes, visits) linked to contacts."""
+
+    manifest = PluginManifest(
+        name="tasks",
+        version="1.0.0",
+        display_name="Tasks",
+        description="Manage free tasks/activities with status, priority, due dates, and contact links.",
+        dependencies=["permissions"],
+        routes=[
+            PluginRouteDef(
+                path="/api/v1/tasks",
+                module="app.plugins.builtins.tasks.routes",
+                router_attr="router",
+            ),
+        ],
+        events=[],
+        migrations=["0001_initial.sql"],
+        permissions=[
+            "tasks:read",
+            "tasks:write",
+            "tasks:delete",
+        ],
+        is_core=True,
+        menu_items=[
+            FrontendMenuItem(
+                label_key="nav.tasks",
+                label="Tasks",
+                path="/tasks",
+                icon="CheckSquare",
+                order=30,
+            ),
+        ],
+        page_routes=[
+            FrontendPageRoute(
+                path="/tasks",
+                component="@/pages/Tasks",
+                protected=True,
+                order=30,
+            ),
+        ],
+        cron_jobs=[
+            CronJobContribution(
+                name="tasks_due_reminder",
+                cron_expression="0 8 * * *",
+                job_type="custom",
+                target_name="tasks_due_reminder",
+                plugin_name="tasks",
+            ),
+        ],
+    )

app/plugins/builtins/tasks/routes.py

@@ -0,0 +1,145 @@
+"""Tasks plugin routes — CRUD, assign, status update."""
+
+from __future__ import annotations
+
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_db
+from app.deps import get_current_user, require_permission
+from app.plugins.builtins.tasks import services
+from app.plugins.builtins.tasks.schemas import (
+    TaskAssignRequest,
+    TaskCreate,
+    TaskStatusRequest,
+    TaskUpdate,
+)
+
+router = APIRouter(prefix="/api/v1/tasks", tags=["tasks"])
+
+
+def _parse_uuid(val: str, field: str) -> uuid.UUID:
+    try:
+        return uuid.UUID(val)
+    except (ValueError, TypeError):
+        raise HTTPException(
+            400, detail={"detail": f"Invalid {field}", "code": "invalid_id"}
+        ) from None
+
+
+@router.get("", dependencies=[Depends(require_permission("tasks:read"))])
+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)$"),
+    priority: str | None = Query(None, pattern="^(low|medium|high|urgent)$"),
+    assigned_to: str | None = Query(None),
+    contact_id: str | None = Query(None),
+    search: str | None = Query(None),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """List tasks with filtering and pagination."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    return await services.list_tasks(
+        db, tenant_id,
+        page=page, page_size=page_size,
+        status=status, priority=priority,
+        assigned_to=assigned_to, contact_id=contact_id,
+        search=search,
+    )
+
+
+@router.post("", status_code=status.HTTP_201_CREATED, dependencies=[Depends(require_permission("tasks:write"))])
+async def create_task(
+    body: TaskCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Create a new task."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    data = body.model_dump()
+    return await services.create_task(db, tenant_id, user_id, data)
+
+
+@router.get("/{task_id}", dependencies=[Depends(require_permission("tasks:read"))])
+async def get_task(
+    task_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Get a single task by ID."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    tid = _parse_uuid(task_id, "task_id")
+    result = await services.get_task(db, tenant_id, tid)
+    if result is None:
+        raise HTTPException(404, detail={"detail": "Task not found", "code": "not_found"})
+    return result
+
+
+@router.patch("/{task_id}", dependencies=[Depends(require_permission("tasks:write"))])
+async def update_task(
+    task_id: str,
+    body: TaskUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Update a task."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    tid = _parse_uuid(task_id, "task_id")
+    data = body.model_dump(exclude_unset=True)
+    result = await services.update_task(db, tenant_id, tid, data)
+    if result is None:
+        raise HTTPException(404, detail={"detail": "Task not found", "code": "not_found"})
+    return result
+
+
+@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(require_permission("tasks:delete"))])
+async def delete_task(
+    task_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Delete a task (soft-delete)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    tid = _parse_uuid(task_id, "task_id")
+    deleted = await services.delete_task(db, tenant_id, tid)
+    if not deleted:
+        raise HTTPException(404, detail={"detail": "Task not found", "code": "not_found"})
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+@router.post("/{task_id}/assign", dependencies=[Depends(require_permission("tasks:write"))])
+async def assign_task(
+    task_id: str,
+    body: TaskAssignRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Assign a task to a user."""
+    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)
+    if result is None:
+        raise HTTPException(404, detail={"detail": "Task not found", "code": "not_found"})
+    return result
+
+
+@router.post("/{task_id}/status", dependencies=[Depends(require_permission("tasks:write"))])
+async def update_task_status(
+    task_id: str,
+    body: TaskStatusRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Update task status."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    tid = _parse_uuid(task_id, "task_id")
+    result = await services.update_task_status(db, tenant_id, tid, body.status)
+    if result is None:
+        raise HTTPException(404, detail={"detail": "Task not found", "code": "not_found"})
+    return result

app/plugins/builtins/tasks/schemas.py

@@ -0,0 +1,63 @@
+"""Pydantic schemas for the Tasks plugin."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+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)$")
+    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
+
+
+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)$")
+    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
+
+
+class TaskAssignRequest(BaseModel):
+    """Schema for assigning a task."""
+    assigned_to: str = Field(..., description="User ID to assign the task to")
+
+
+class TaskStatusRequest(BaseModel):
+    """Schema for updating task status."""
+    status: str = Field(..., pattern="^(open|in_progress|done)$")
+
+
+class TaskResponse(BaseModel):
+    """Schema for task response."""
+    id: str
+    title: str
+    description: str | None = None
+    status: str
+    priority: str
+    due_date: datetime | None = None
+    assigned_to: str | None = None
+    contact_id: str | None = None
+    created_by: str | None = None
+    created_at: datetime
+    updated_at: datetime
+
+
+class TaskListResponse(BaseModel):
+    """Paginated task list response."""
+    items: list[TaskResponse]
+    total: int
+    page: int
+    page_size: int

app/plugins/builtins/tasks/services.py

@@ -0,0 +1,212 @@
+"""Business logic for the Tasks plugin."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.plugins.builtins.tasks.models import Task
+
+
+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.contact_id) if task.contact_id else None,
+        "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 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,
+) -> dict[str, Any]:
+    """List tasks with filtering and pagination."""
+    query = select(Task).where(Task.tenant_id == tenant_id, Task.deleted_at.is_(None))
+
+    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 == uuid.UUID(assigned_to))
+    if contact_id:
+        query = query.where(Task.contact_id == uuid.UUID(contact_id))
+    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."""
+    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()
+    if task is None:
+        return None
+    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."""
+    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=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,
+        created_by=user_id,
+    )
+    db.add(task)
+    await db.flush()
+    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."""
+    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()
+    if task is None:
+        return None
+
+    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 = 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
+
+    await db.flush()
+    return _task_to_dict(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()
+    if task is None:
+        return False
+    task.deleted_at = datetime.now(timezone.utc)
+    await db.flush()
+    return True
+
+
+async def assign_task(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    task_id: uuid.UUID,
+    assigned_to: uuid.UUID,
+) -> 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()
+    if task is None:
+        return None
+    task.assigned_to = assigned_to
+    await db.flush()
+    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."""
+    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()
+    if task is None:
+        return None
+    task.status = new_status
+    await db.flush()
+    return _task_to_dict(task)
+
+
+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(timezone.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]