.gitignore

@@ -160,3 +160,4 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 
+task.instructions.md

app/api/deps.py

@@ -10,10 +10,13 @@ from app.core.config import settings
 from app.core.database import get_session
 from app.core.security import jwt_service, password_hasher
 from app.models.user import User
+from app.repositories.activity_repo import ActivityRepository
 from app.repositories.deal_repo import DealRepository
 from app.repositories.org_repo import OrganizationRepository
+from app.repositories.task_repo import TaskRepository
 from app.repositories.user_repo import UserRepository
 from app.services.auth_service import AuthService
+from app.services.activity_service import ActivityService
 from app.services.deal_service import DealService
 from app.services.organization_service import (
     OrganizationAccessDeniedError,
@@ -21,6 +24,7 @@ from app.services.organization_service import (
     OrganizationContextMissingError,
     OrganizationService,
 )
+from app.services.task_service import TaskService
 from app.services.user_service import UserService
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.api_v1_prefix}/auth/token")
@@ -44,6 +48,14 @@ def get_deal_repository(session: AsyncSession = Depends(get_db_session)) -> Deal
     return DealRepository(session=session)
 
 
+def get_task_repository(session: AsyncSession = Depends(get_db_session)) -> TaskRepository:
+    return TaskRepository(session=session)
+
+
+def get_activity_repository(session: AsyncSession = Depends(get_db_session)) -> ActivityRepository:
+    return ActivityRepository(session=session)
+
+
 def get_deal_service(repo: DealRepository = Depends(get_deal_repository)) -> DealService:
     return DealService(repository=repo)
 
@@ -68,6 +80,19 @@ def get_organization_service(
     return OrganizationService(repository=repo)
 
 
+def get_activity_service(
+    repo: ActivityRepository = Depends(get_activity_repository),
+) -> ActivityService:
+    return ActivityService(repository=repo)
+
+
+def get_task_service(
+    task_repo: TaskRepository = Depends(get_task_repository),
+    activity_repo: ActivityRepository = Depends(get_activity_repository),
+) -> TaskService:
+    return TaskService(task_repository=task_repo, activity_repository=activity_repo)
+
+
 async def get_current_user(
     token: str = Depends(oauth2_scheme),
     repo: UserRepository = Depends(get_user_repository),

app/api/v1/activities/models.py

@@ -1,11 +1,18 @@
 """Pydantic schemas for activity endpoints."""
 from __future__ import annotations
 
-from typing import Any, Literal
+from typing import Literal
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
+
+
+class ActivityCommentBody(BaseModel):
+    text: str = Field(..., min_length=1, max_length=2000)
 
 
 class ActivityCommentPayload(BaseModel):
-    type: Literal["comment"]
+    type: Literal["comment"] = "comment"
-    payload: dict[str, Any]
+    payload: ActivityCommentBody
+
+    def extract_text(self) -> str:
+        return self.payload.text.strip()

app/api/v1/activities/views.py

@@ -1,9 +1,16 @@
-"""Activity timeline API stubs."""
+"""Activity timeline endpoints backed by ActivityService."""
 from __future__ import annotations
 
-from fastapi import APIRouter, Depends, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 
-from app.api.deps import get_organization_context
+from app.api.deps import get_activity_service, get_organization_context
+from app.models.activity import ActivityRead
+from app.services.activity_service import (
+    ActivityForbiddenError,
+    ActivityListFilters,
+    ActivityService,
+    ActivityValidationError,
+)
 from app.services.organization_service import OrganizationContext
 
 from .models import ActivityCommentPayload
@@ -11,26 +18,44 @@ from .models import ActivityCommentPayload
 router = APIRouter(prefix="/deals/{deal_id}/activities", tags=["activities"])
 
 
-def _stub(endpoint: str) -> dict[str, str]:
+@router.get("/", response_model=list[ActivityRead])
-    return {"detail": f"{endpoint} is not implemented yet"}
-
-
-@router.get("/", status_code=status.HTTP_501_NOT_IMPLEMENTED)
 async def list_activities(
     deal_id: int,
+    limit: int = Query(50, ge=1, le=200),
+    offset: int = Query(0, ge=0),
     context: OrganizationContext = Depends(get_organization_context),
-) -> dict[str, str]:
+    service: ActivityService = Depends(get_activity_service),
-    """Placeholder for listing deal activities."""
+) -> list[ActivityRead]:
-    _ = (deal_id, context)
+    """Fetch paginated activities for the deal within the current organization."""
-    return _stub("GET /deals/{deal_id}/activities")
+
+    filters = ActivityListFilters(deal_id=deal_id, limit=limit, offset=offset)
+    try:
+        activities = await service.list_activities(filters=filters, context=context)
+    except ActivityForbiddenError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+
+    return [ActivityRead.model_validate(activity) for activity in activities]
 
 
-@router.post("/", status_code=status.HTTP_501_NOT_IMPLEMENTED)
+@router.post("/", response_model=ActivityRead, status_code=status.HTTP_201_CREATED)
 async def create_activity_comment(
     deal_id: int,
     payload: ActivityCommentPayload,
     context: OrganizationContext = Depends(get_organization_context),
-) -> dict[str, str]:
+    service: ActivityService = Depends(get_activity_service),
-    """Placeholder for adding a comment activity to a deal."""
+) -> ActivityRead:
-    _ = (deal_id, payload, context)
+    """Add a comment to the deal timeline."""
-    return _stub("POST /deals/{deal_id}/activities")
+
+    try:
+        activity = await service.add_comment(
+            deal_id=deal_id,
+            author_id=context.user_id,
+            text=payload.extract_text(),
+            context=context,
+        )
+    except ActivityValidationError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+    except ActivityForbiddenError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+
+    return ActivityRead.model_validate(activity)

app/api/v1/tasks/models.py

@@ -1,13 +1,36 @@
-"""Task API schemas."""
+"""Task API schemas and helpers."""
 from __future__ import annotations
 
-from datetime import date
+from datetime import date, datetime, time, timezone
 
 from pydantic import BaseModel
 
+from app.models.task import TaskCreate
+
 
 class TaskCreatePayload(BaseModel):
     deal_id: int
     title: str
     description: str | None = None
-    due_date: date
+    due_date: date | None = None
+
+    def to_domain(self) -> TaskCreate:
+        return TaskCreate(
+            deal_id=self.deal_id,
+            title=self.title,
+            description=self.description,
+            due_date=_date_to_datetime(self.due_date) if self.due_date else None,
+        )
+
+
+def to_range_boundary(value: date | None, *, end_of_day: bool) -> datetime | None:
+    """Convert a date query param to an inclusive datetime boundary."""
+
+    if value is None:
+        return None
+    boundary_time = time(23, 59, 59, 999999) if end_of_day else time(0, 0, 0)
+    return datetime.combine(value, boundary_time, tzinfo=timezone.utc)
+
+
+def _date_to_datetime(value: date) -> datetime:
+    return datetime.combine(value, time(0, 0, 0), tzinfo=timezone.utc)

app/api/v1/tasks/views.py

@@ -1,40 +1,62 @@
-"""Task API stubs supporting list/create operations."""
+"""Task API endpoints backed by TaskService."""
 from __future__ import annotations
 
 from datetime import date
 
-from fastapi import APIRouter, Depends, Query, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 
-from app.api.deps import get_organization_context
+from app.api.deps import get_organization_context, get_task_service
+from app.models.task import TaskRead
 from app.services.organization_service import OrganizationContext
+from app.services.task_service import (
+    TaskDueDateError,
+    TaskForbiddenError,
+    TaskListFilters,
+    TaskOrganizationError,
+    TaskService,
+)
 
-from .models import TaskCreatePayload
+from .models import TaskCreatePayload, to_range_boundary
 
 router = APIRouter(prefix="/tasks", tags=["tasks"])
 
 
-def _stub(endpoint: str) -> dict[str, str]:
+@router.get("/", response_model=list[TaskRead])
-    return {"detail": f"{endpoint} is not implemented yet"}
-
-
-@router.get("/", status_code=status.HTTP_501_NOT_IMPLEMENTED)
 async def list_tasks(
     deal_id: int | None = None,
     only_open: bool = False,
     due_before: date | None = Query(default=None),
     due_after: date | None = Query(default=None),
     context: OrganizationContext = Depends(get_organization_context),
-) -> dict[str, str]:
+    service: TaskService = Depends(get_task_service),
-    """Placeholder for task filtering endpoint."""
+) -> list[TaskRead]:
-    _ = context
+    """Filter tasks by deal, state, or due date range."""
-    return _stub("GET /tasks")
+
+    filters = TaskListFilters(
+        deal_id=deal_id,
+        only_open=only_open,
+        due_before=to_range_boundary(due_before, end_of_day=True),
+        due_after=to_range_boundary(due_after, end_of_day=False),
+    )
+    tasks = await service.list_tasks(filters=filters, context=context)
+    return [TaskRead.model_validate(task) for task in tasks]
 
 
-@router.post("/", status_code=status.HTTP_501_NOT_IMPLEMENTED)
+@router.post("/", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
 async def create_task(
     payload: TaskCreatePayload,
     context: OrganizationContext = Depends(get_organization_context),
-) -> dict[str, str]:
+    service: TaskService = Depends(get_task_service),
-    """Placeholder for creating a task linked to a deal."""
+) -> TaskRead:
-    _ = (payload, context)
+    """Create a task ensuring due-date and ownership constraints."""
-    return _stub("POST /tasks")
+
+    try:
+        task = await service.create_task(payload.to_domain(), context=context)
+    except TaskDueDateError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+    except TaskForbiddenError as exc:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
+    except TaskOrganizationError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+
+    return TaskRead.model_validate(task)

app/repositories/activity_repo.py

@@ -0,0 +1,68 @@
+"""Repository helpers for deal activities."""
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from sqlalchemy import Select, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.activity import Activity, ActivityCreate
+from app.models.deal import Deal
+
+
+class ActivityOrganizationMismatchError(Exception):
+    """Raised when a deal/activity pair targets another organization."""
+
+
+@dataclass(slots=True)
+class ActivityQueryParams:
+    """Filtering options for fetching activities."""
+
+    organization_id: int
+    deal_id: int
+    limit: int | None = None
+    offset: int = 0
+
+
+class ActivityRepository:
+    """Provides CRUD helpers for Activity model."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
+    @property
+    def session(self) -> AsyncSession:
+        return self._session
+
+    async def list(self, *, params: ActivityQueryParams) -> Sequence[Activity]:
+        stmt = (
+            select(Activity)
+            .join(Deal, Deal.id == Activity.deal_id)
+            .where(Activity.deal_id == params.deal_id, Deal.organization_id == params.organization_id)
+            .order_by(Activity.created_at)
+        )
+        stmt = self._apply_window(stmt, params)
+        result = await self._session.scalars(stmt)
+        return result.all()
+
+    async def create(self, data: ActivityCreate, *, organization_id: int) -> Activity:
+        deal = await self._session.get(Deal, data.deal_id)
+        if deal is None or deal.organization_id != organization_id:
+            raise ActivityOrganizationMismatchError("Deal belongs to another organization")
+
+        activity = Activity(**data.model_dump())
+        self._session.add(activity)
+        await self._session.flush()
+        return activity
+
+    def _apply_window(
+        self,
+        stmt: Select[tuple[Activity]],
+        params: ActivityQueryParams,
+    ) -> Select[tuple[Activity]]:
+        if params.offset:
+            stmt = stmt.offset(params.offset)
+        if params.limit is not None:
+            stmt = stmt.limit(params.limit)
+        return stmt

app/repositories/task_repo.py

@@ -0,0 +1,123 @@
+"""Task repository providing role-aware CRUD helpers."""
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import Select, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.models.deal import Deal
+from app.models.organization_member import OrganizationRole
+from app.models.task import Task, TaskCreate
+
+
+class TaskAccessError(Exception):
+    """Raised when a user attempts to modify a forbidden task."""
+
+
+class TaskOrganizationMismatchError(Exception):
+    """Raised when a task or deal belongs to another organization."""
+
+
+@dataclass(slots=True)
+class TaskQueryParams:
+    """Filtering options supported by list queries."""
+
+    organization_id: int
+    deal_id: int | None = None
+    only_open: bool = False
+    due_before: datetime | None = None
+    due_after: datetime | None = None
+
+
+class TaskRepository:
+    """Encapsulates database access for Task entities."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
+    @property
+    def session(self) -> AsyncSession:
+        return self._session
+
+    async def list(self, *, params: TaskQueryParams) -> Sequence[Task]:
+        stmt = (
+            select(Task)
+            .join(Deal, Deal.id == Task.deal_id)
+            .where(Deal.organization_id == params.organization_id)
+            .options(selectinload(Task.deal))
+            .order_by(Task.due_date.is_(None), Task.due_date, Task.id)
+        )
+        stmt = self._apply_filters(stmt, params)
+        result = await self._session.scalars(stmt)
+        return result.all()
+
+    async def get(self, task_id: int, *, organization_id: int) -> Task | None:
+        stmt = (
+            select(Task)
+            .join(Deal, Deal.id == Task.deal_id)
+            .where(Task.id == task_id, Deal.organization_id == organization_id)
+            .options(selectinload(Task.deal))
+        )
+        result = await self._session.scalars(stmt)
+        return result.first()
+
+    async def create(
+        self,
+        data: TaskCreate,
+        *,
+        organization_id: int,
+        role: OrganizationRole,
+        user_id: int,
+    ) -> Task:
+        deal = await self._session.get(Deal, data.deal_id)
+        if deal is None or deal.organization_id != organization_id:
+            raise TaskOrganizationMismatchError("Deal belongs to another organization")
+        if role == OrganizationRole.MEMBER and deal.owner_id != user_id:
+            raise TaskAccessError("Members can only create tasks for their own deals")
+
+        task = Task(**data.model_dump())
+        self._session.add(task)
+        await self._session.flush()
+        return task
+
+    async def update(
+        self,
+        task: Task,
+        updates: Mapping[str, Any],
+        *,
+        role: OrganizationRole,
+        user_id: int,
+    ) -> Task:
+        owner_id = await self._resolve_task_owner(task)
+        if owner_id is None:
+            raise TaskOrganizationMismatchError("Task is missing an owner context")
+        if role == OrganizationRole.MEMBER and owner_id != user_id:
+            raise TaskAccessError("Members can only modify their own tasks")
+
+        for field, value in updates.items():
+            if hasattr(task, field):
+                setattr(task, field, value)
+        await self._session.flush()
+        return task
+
+    def _apply_filters(self, stmt: Select[tuple[Task]], params: TaskQueryParams) -> Select[tuple[Task]]:
+        if params.deal_id is not None:
+            stmt = stmt.where(Task.deal_id == params.deal_id)
+        if params.only_open:
+            stmt = stmt.where(Task.is_done.is_(False))
+        if params.due_before is not None:
+            stmt = stmt.where(Task.due_date <= params.due_before)
+        if params.due_after is not None:
+            stmt = stmt.where(Task.due_date >= params.due_after)
+        return stmt
+
+    async def _resolve_task_owner(self, task: Task) -> int | None:
+        if task.deal is not None:
+            return task.deal.owner_id
+        stmt = select(Deal.owner_id).where(Deal.id == task.deal_id)
+        return await self._session.scalar(stmt)

app/services/__init__.py

@@ -1,9 +1,26 @@
 """Business logic services."""
+from .activity_service import (  # noqa: F401
+    ActivityForbiddenError,
+    ActivityListFilters,
+    ActivityService,
+    ActivityServiceError,
+    ActivityValidationError,
+)
+from .auth_service import AuthService  # noqa: F401
 from .organization_service import (  # noqa: F401
     OrganizationAccessDeniedError,
     OrganizationContext,
     OrganizationContextMissingError,
     OrganizationService,
 )
-from .user_service import UserService  # noqa: F401
+from .task_service import (  # noqa: F401
-from .auth_service import AuthService  # noqa: F401
+    TaskDueDateError,
+    TaskForbiddenError,
+    TaskListFilters,
+    TaskNotFoundError,
+    TaskOrganizationError,
+    TaskService,
+    TaskServiceError,
+    TaskUpdateData,
+)
+from .user_service import UserService  # noqa: F401

app/services/activity_service.py

@@ -0,0 +1,104 @@
+"""Business logic for timeline activities."""
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Any
+
+from app.models.activity import Activity, ActivityCreate, ActivityType
+from app.models.deal import Deal
+from app.repositories.activity_repo import (
+    ActivityOrganizationMismatchError,
+    ActivityQueryParams,
+    ActivityRepository,
+)
+from app.services.organization_service import OrganizationContext
+
+
+class ActivityServiceError(Exception):
+    """Base class for activity service errors."""
+
+
+class ActivityValidationError(ActivityServiceError):
+    """Raised when payload does not satisfy business constraints."""
+
+
+class ActivityForbiddenError(ActivityServiceError):
+    """Raised when a user accesses activities from another organization."""
+
+
+@dataclass(slots=True)
+class ActivityListFilters:
+    """Filtering helpers for listing activities."""
+
+    deal_id: int
+    limit: int | None = None
+    offset: int = 0
+
+
+class ActivityService:
+    """Encapsulates timeline-specific workflows."""
+
+    def __init__(self, repository: ActivityRepository) -> None:
+        self._repository = repository
+
+    async def list_activities(
+        self,
+        *,
+        filters: ActivityListFilters,
+        context: OrganizationContext,
+    ) -> Sequence[Activity]:
+        await self._ensure_deal_in_context(filters.deal_id, context)
+        params = ActivityQueryParams(
+            organization_id=context.organization_id,
+            deal_id=filters.deal_id,
+            limit=filters.limit,
+            offset=max(filters.offset, 0),
+        )
+        return await self._repository.list(params=params)
+
+    async def add_comment(
+        self,
+        *,
+        deal_id: int,
+        author_id: int,
+        text: str,
+        context: OrganizationContext,
+    ) -> Activity:
+        normalized = text.strip()
+        if not normalized:
+            raise ActivityValidationError("Comment text cannot be empty")
+        return await self.record_activity(
+            deal_id=deal_id,
+            activity_type=ActivityType.COMMENT,
+            payload={"text": normalized},
+            author_id=author_id,
+            context=context,
+        )
+
+    async def record_activity(
+        self,
+        *,
+        deal_id: int,
+        activity_type: ActivityType,
+        context: OrganizationContext,
+        payload: dict[str, Any] | None = None,
+        author_id: int | None = None,
+    ) -> Activity:
+        await self._ensure_deal_in_context(deal_id, context)
+        data = ActivityCreate(
+            deal_id=deal_id,
+            author_id=author_id,
+            type=activity_type,
+            payload=payload or {},
+        )
+        try:
+            return await self._repository.create(data, organization_id=context.organization_id)
+        except ActivityOrganizationMismatchError as exc:  # pragma: no cover - defensive
+            raise ActivityForbiddenError("Deal belongs to another organization") from exc
+
+    async def _ensure_deal_in_context(self, deal_id: int, context: OrganizationContext) -> Deal:
+        deal = await self._repository.session.get(Deal, deal_id)
+        if deal is None or deal.organization_id != context.organization_id:
+            raise ActivityForbiddenError("Deal not found in current organization")
+        return deal

app/services/task_service.py

@@ -0,0 +1,186 @@
+"""Business logic for tasks linked to deals."""
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any
+
+from app.models.activity import ActivityCreate, ActivityType
+from app.models.organization_member import OrganizationRole
+from app.models.task import Task, TaskCreate
+from app.repositories.activity_repo import ActivityRepository, ActivityOrganizationMismatchError
+from app.repositories.task_repo import (
+    TaskAccessError as RepoTaskAccessError,
+    TaskOrganizationMismatchError as RepoTaskOrganizationMismatchError,
+    TaskQueryParams,
+    TaskRepository,
+)
+from app.services.organization_service import OrganizationContext
+
+
+class TaskServiceError(Exception):
+    """Base class for task service errors."""
+
+
+class TaskDueDateError(TaskServiceError):
+    """Raised when due_date violates temporal constraints."""
+
+
+class TaskForbiddenError(TaskServiceError):
+    """Raised when the user lacks permissions for an operation."""
+
+
+class TaskOrganizationError(TaskServiceError):
+    """Raised when a task/deal belongs to another organization."""
+
+
+class TaskNotFoundError(TaskServiceError):
+    """Raised when task cannot be located in the current organization."""
+
+
+@dataclass(slots=True)
+class TaskListFilters:
+    """Filters accepted by the task listing endpoint."""
+
+    deal_id: int | None = None
+    only_open: bool = False
+    due_before: datetime | None = None
+    due_after: datetime | None = None
+
+
+@dataclass(slots=True)
+class TaskUpdateData:
+    """Subset of fields allowed for partial updates."""
+
+    title: str | None = None
+    description: str | None = None
+    due_date: datetime | None = None
+    is_done: bool | None = None
+
+
+class TaskService:
+    """Encapsulates task workflows and policy validations."""
+
+    def __init__(
+        self,
+        task_repository: TaskRepository,
+        activity_repository: ActivityRepository | None = None,
+    ) -> None:
+        self._task_repository = task_repository
+        self._activity_repository = activity_repository
+
+    async def list_tasks(
+        self,
+        *,
+        filters: TaskListFilters,
+        context: OrganizationContext,
+    ) -> Sequence[Task]:
+        params = TaskQueryParams(
+            organization_id=context.organization_id,
+            deal_id=filters.deal_id,
+            only_open=filters.only_open,
+            due_before=filters.due_before,
+            due_after=filters.due_after,
+        )
+        return await self._task_repository.list(params=params)
+
+    async def get_task(self, task_id: int, *, context: OrganizationContext) -> Task:
+        task = await self._task_repository.get(task_id, organization_id=context.organization_id)
+        if task is None:
+            raise TaskNotFoundError("Task not found")
+        return task
+
+    async def create_task(
+        self,
+        data: TaskCreate,
+        *,
+        context: OrganizationContext,
+    ) -> Task:
+        self._validate_due_date(data.due_date)
+        try:
+            task = await self._task_repository.create(
+                data,
+                organization_id=context.organization_id,
+                role=context.role,
+                user_id=context.user_id,
+            )
+        except RepoTaskOrganizationMismatchError as exc:
+            raise TaskOrganizationError("Deal belongs to another organization") from exc
+        except RepoTaskAccessError as exc:
+            raise TaskForbiddenError(str(exc)) from exc
+
+        await self._log_task_created(task, context=context)
+        return task
+
+    async def update_task(
+        self,
+        task_id: int,
+        updates: TaskUpdateData,
+        *,
+        context: OrganizationContext,
+    ) -> Task:
+        task = await self.get_task(task_id, context=context)
+        if updates.due_date is not None:
+            self._validate_due_date(updates.due_date)
+
+        payload = self._build_update_mapping(updates)
+        if not payload:
+            return task
+
+        try:
+            return await self._task_repository.update(
+                task,
+                payload,
+                role=context.role,
+                user_id=context.user_id,
+            )
+        except RepoTaskAccessError as exc:
+            raise TaskForbiddenError(str(exc)) from exc
+
+    async def delete_task(self, task_id: int, *, context: OrganizationContext) -> None:
+        task = await self.get_task(task_id, context=context)
+        self._ensure_member_owns_task(task, context)
+        await self._task_repository.session.delete(task)
+        await self._task_repository.session.flush()
+
+    def _ensure_member_owns_task(self, task: Task, context: OrganizationContext) -> None:
+        if context.role != OrganizationRole.MEMBER:
+            return
+        owner_id = task.deal.owner_id if task.deal is not None else None
+        if owner_id is None or owner_id != context.user_id:
+            raise TaskForbiddenError("Members can only modify their own tasks")
+
+    def _validate_due_date(self, due_date: datetime | None) -> None:
+        if due_date is None:
+            return
+        today = datetime.now(timezone.utc).date()
+        value_date = (due_date.astimezone(timezone.utc) if due_date.tzinfo else due_date).date()
+        if value_date < today:
+            raise TaskDueDateError("Task due date cannot be in the past")
+
+    def _build_update_mapping(self, updates: TaskUpdateData) -> Mapping[str, Any]:
+        payload: dict[str, Any] = {}
+        if updates.title is not None:
+            payload["title"] = updates.title
+        if updates.description is not None:
+            payload["description"] = updates.description
+        if updates.due_date is not None:
+            payload["due_date"] = updates.due_date
+        if updates.is_done is not None:
+            payload["is_done"] = updates.is_done
+        return payload
+
+    async def _log_task_created(self, task: Task, *, context: OrganizationContext) -> None:
+        if self._activity_repository is None:
+            return
+        data = ActivityCreate(
+            deal_id=task.deal_id,
+            author_id=context.user_id,
+            type=ActivityType.TASK_CREATED,
+            payload={"task_id": task.id, "title": task.title},
+        )
+        try:
+            await self._activity_repository.create(data, organization_id=context.organization_id)
+        except ActivityOrganizationMismatchError:  # pragma: no cover - defensive
+            raise TaskOrganizationError("Activity target does not belong to organization")

tests/api/v1/conftest.py

@@ -0,0 +1,38 @@
+"""Pytest fixtures shared across API v1 tests."""
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+
+import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from app.api.deps import get_db_session
+from app.main import create_app
+from app.models import Base
+
+
+@pytest_asyncio.fixture()
+async def session_factory() -> AsyncGenerator[async_sessionmaker[AsyncSession], None]:
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", future=True)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    factory = async_sessionmaker(engine, expire_on_commit=False)
+    yield factory
+    await engine.dispose()
+
+
+@pytest_asyncio.fixture()
+async def client(
+    session_factory: async_sessionmaker[AsyncSession],
+) -> AsyncGenerator[AsyncClient, None]:
+    app = create_app()
+
+    async def _get_session_override() -> AsyncGenerator[AsyncSession, None]:
+        async with session_factory() as session:
+            yield session
+
+    app.dependency_overrides[get_db_session] = _get_session_override
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://testserver") as test_client:
+        yield test_client

tests/api/v1/task_activity_shared.py

@@ -0,0 +1,101 @@
+"""Shared helpers for task and activity API tests."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import timedelta
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.core.security import jwt_service
+from app.models.contact import Contact
+from app.models.deal import Deal
+from app.models.organization import Organization
+from app.models.organization_member import OrganizationMember, OrganizationRole
+from app.models.user import User
+
+
+@dataclass(slots=True)
+class Scenario:
+    """Captures seeded entities for API tests."""
+
+    user_id: int
+    user_email: str
+    organization_id: int
+    contact_id: int
+    deal_id: int
+
+
+async def prepare_scenario(session_factory: async_sessionmaker[AsyncSession]) -> Scenario:
+    async with session_factory() as session:
+        user = User(email="owner@example.com", hashed_password="hashed", name="Owner", is_active=True)
+        org = Organization(name="Acme LLC")
+        session.add_all([user, org])
+        await session.flush()
+
+        membership = OrganizationMember(
+            organization_id=org.id,
+            user_id=user.id,
+            role=OrganizationRole.OWNER,
+        )
+        session.add(membership)
+
+        contact = Contact(
+            organization_id=org.id,
+            owner_id=user.id,
+            name="John Doe",
+            email="john@example.com",
+        )
+        session.add(contact)
+        await session.flush()
+
+        deal = Deal(
+            organization_id=org.id,
+            contact_id=contact.id,
+            owner_id=user.id,
+            title="Website redesign",
+            amount=None,
+        )
+        session.add(deal)
+        await session.commit()
+
+    return Scenario(
+        user_id=user.id,
+        user_email=user.email,
+        organization_id=org.id,
+        contact_id=contact.id,
+        deal_id=deal.id,
+    )
+
+
+async def create_deal(
+    session_factory: async_sessionmaker[AsyncSession],
+    *,
+    scenario: Scenario,
+    title: str,
+) -> int:
+    async with session_factory() as session:
+        deal = Deal(
+            organization_id=scenario.organization_id,
+            contact_id=scenario.contact_id,
+            owner_id=scenario.user_id,
+            title=title,
+            amount=None,
+        )
+        session.add(deal)
+        await session.commit()
+        return deal.id
+
+
+def auth_headers(token: str, scenario: Scenario) -> dict[str, str]:
+    return {
+        "Authorization": f"Bearer {token}",
+        "X-Organization-Id": str(scenario.organization_id),
+    }
+
+
+def make_token(user_id: int, email: str) -> str:
+    return jwt_service.create_access_token(
+        subject=str(user_id),
+        expires_delta=timedelta(minutes=30),
+        claims={"email": email},
+    )

tests/api/v1/test_activities.py

@@ -0,0 +1,63 @@
+"""API tests for activity endpoints."""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.models.activity import Activity, ActivityType
+
+from tests.api.v1.task_activity_shared import auth_headers, make_token, prepare_scenario
+
+
+@pytest.mark.asyncio
+async def test_create_activity_comment_endpoint(
+    session_factory: async_sessionmaker[AsyncSession], client: AsyncClient
+) -> None:
+    scenario = await prepare_scenario(session_factory)
+    token = make_token(scenario.user_id, scenario.user_email)
+
+    response = await client.post(
+        f"/api/v1/deals/{scenario.deal_id}/activities/",
+        json={"type": "comment", "payload": {"text": "  hello world  "}},
+        headers=auth_headers(token, scenario),
+    )
+
+    assert response.status_code == 201
+    payload = response.json()
+    assert payload["payload"]["text"] == "hello world"
+    assert payload["type"] == ActivityType.COMMENT.value
+
+
+@pytest.mark.asyncio
+async def test_list_activities_endpoint_supports_pagination(
+    session_factory: async_sessionmaker[AsyncSession], client: AsyncClient
+) -> None:
+    scenario = await prepare_scenario(session_factory)
+    token = make_token(scenario.user_id, scenario.user_email)
+
+    base_time = datetime.now(timezone.utc)
+    async with session_factory() as session:
+        for index in range(3):
+            activity = Activity(
+                deal_id=scenario.deal_id,
+                author_id=scenario.user_id,
+                type=ActivityType.COMMENT,
+                payload={"text": f"Entry {index}"},
+                created_at=base_time + timedelta(seconds=index),
+            )
+            session.add(activity)
+        await session.commit()
+
+    response = await client.get(
+        f"/api/v1/deals/{scenario.deal_id}/activities/?limit=2&offset=1",
+        headers=auth_headers(token, scenario),
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data) == 2
+    assert data[0]["payload"]["text"] == "Entry 1"
+    assert data[1]["payload"]["text"] == "Entry 2"

tests/api/v1/test_tasks.py

@@ -0,0 +1,78 @@
+"""API tests for task endpoints."""
+from __future__ import annotations
+
+from datetime import date, datetime, timedelta, timezone
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.models.task import Task
+
+from tests.api.v1.task_activity_shared import auth_headers, create_deal, make_token, prepare_scenario
+
+
+@pytest.mark.asyncio
+async def test_create_task_endpoint_creates_task_and_activity(
+    session_factory: async_sessionmaker[AsyncSession], client: AsyncClient
+) -> None:
+    scenario = await prepare_scenario(session_factory)
+    token = make_token(scenario.user_id, scenario.user_email)
+    due_date = (date.today() + timedelta(days=5)).isoformat()
+
+    response = await client.post(
+        "/api/v1/tasks/",
+        json={
+            "deal_id": scenario.deal_id,
+            "title": "Prepare proposal",
+            "description": "Send draft",
+            "due_date": due_date,
+        },
+        headers=auth_headers(token, scenario),
+    )
+
+    assert response.status_code == 201
+    payload = response.json()
+    assert payload["deal_id"] == scenario.deal_id
+    assert payload["title"] == "Prepare proposal"
+    assert payload["is_done"] is False
+
+
+@pytest.mark.asyncio
+async def test_list_tasks_endpoint_filters_by_deal(
+    session_factory: async_sessionmaker[AsyncSession], client: AsyncClient
+) -> None:
+    scenario = await prepare_scenario(session_factory)
+    token = make_token(scenario.user_id, scenario.user_email)
+    other_deal_id = await create_deal(session_factory, scenario=scenario, title="Renewal")
+
+    async with session_factory() as session:
+        session.add_all(
+            [
+                Task(
+                    deal_id=scenario.deal_id,
+                    title="Task A",
+                    description=None,
+                    due_date=datetime.now(timezone.utc) + timedelta(days=2),
+                    is_done=False,
+                ),
+                Task(
+                    deal_id=other_deal_id,
+                    title="Task B",
+                    description=None,
+                    due_date=datetime.now(timezone.utc) + timedelta(days=3),
+                    is_done=False,
+                ),
+            ]
+        )
+        await session.commit()
+
+    response = await client.get(
+        f"/api/v1/tasks/?deal_id={scenario.deal_id}",
+        headers=auth_headers(token, scenario),
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data) == 1
+    assert data[0]["title"] == "Task A"

tests/services/test_activity_service.py

@@ -0,0 +1,164 @@
+"""Unit tests for ActivityService."""
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+import uuid
+
+import pytest
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from app.models.activity import Activity, ActivityType
+from app.models.base import Base
+from app.models.contact import Contact
+from app.models.deal import Deal
+from app.models.organization import Organization
+from app.models.organization_member import OrganizationMember, OrganizationRole
+from app.models.user import User
+from app.repositories.activity_repo import ActivityRepository
+from app.services.activity_service import (
+    ActivityForbiddenError,
+    ActivityListFilters,
+    ActivityService,
+    ActivityValidationError,
+)
+from app.services.organization_service import OrganizationContext
+
+
+@pytest_asyncio.fixture()
+async def session() -> AsyncGenerator[AsyncSession, None]:
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        future=True,
+        poolclass=StaticPool,
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    session_factory = async_sessionmaker(engine, expire_on_commit=False)
+    async with session_factory() as session:
+        yield session
+    await engine.dispose()
+
+
+def _make_user(suffix: str) -> User:
+    return User(
+        email=f"user-{suffix}@example.com",
+        hashed_password="hashed",
+        name="Test",
+        is_active=True,
+    )
+
+
+async def _prepare_deal(
+    session: AsyncSession,
+    *,
+    role: OrganizationRole = OrganizationRole.MANAGER,
+) -> tuple[OrganizationContext, ActivityRepository, int, Organization]:
+    org = Organization(name=f"Org-{uuid.uuid4()}"[:8])
+    user = _make_user("owner")
+    session.add_all([org, user])
+    await session.flush()
+
+    contact = Contact(
+        organization_id=org.id,
+        owner_id=user.id,
+        name="Alice",
+        email="alice@example.com",
+    )
+    session.add(contact)
+    await session.flush()
+
+    deal = Deal(
+        organization_id=org.id,
+        contact_id=contact.id,
+        owner_id=user.id,
+        title="Activity",
+        amount=None,
+    )
+    session.add(deal)
+    await session.flush()
+
+    membership = OrganizationMember(organization_id=org.id, user_id=user.id, role=role)
+    context = OrganizationContext(organization=org, membership=membership)
+    return context, ActivityRepository(session=session), deal.id, org
+
+
+@pytest.mark.asyncio
+async def test_list_activities_returns_only_current_deal(session: AsyncSession) -> None:
+    context, repo, deal_id, _ = await _prepare_deal(session)
+    service = ActivityService(repository=repo)
+
+    session.add_all(
+        [
+            Activity(deal_id=deal_id, author_id=context.user_id, type=ActivityType.COMMENT, payload={"text": "hi"}),
+            Activity(deal_id=deal_id + 1, author_id=context.user_id, type=ActivityType.SYSTEM, payload={}),
+        ]
+    )
+    await session.flush()
+
+    activities = await service.list_activities(
+        filters=ActivityListFilters(deal_id=deal_id, limit=10, offset=0),
+        context=context,
+    )
+
+    assert len(activities) == 1
+    assert activities[0].deal_id == deal_id
+
+
+@pytest.mark.asyncio
+async def test_add_comment_rejects_empty_text(session: AsyncSession) -> None:
+    context, repo, deal_id, _ = await _prepare_deal(session)
+    service = ActivityService(repository=repo)
+
+    with pytest.raises(ActivityValidationError):
+        await service.add_comment(deal_id=deal_id, author_id=context.user_id, text="   ", context=context)
+
+
+@pytest.mark.asyncio
+async def test_record_activity_blocks_foreign_deal(session: AsyncSession) -> None:
+    context, repo, _deal_id, _ = await _prepare_deal(session)
+    service = ActivityService(repository=repo)
+    # Create a second deal in another organization
+    other_org = Organization(name="External")
+    other_user = _make_user("external")
+    session.add_all([other_org, other_user])
+    await session.flush()
+    other_contact = Contact(
+        organization_id=other_org.id,
+        owner_id=other_user.id,
+        name="Bob",
+        email="bob@example.com",
+    )
+    session.add(other_contact)
+    await session.flush()
+    other_deal = Deal(
+        organization_id=other_org.id,
+        contact_id=other_contact.id,
+        owner_id=other_user.id,
+        title="Foreign",
+        amount=None,
+    )
+    session.add(other_deal)
+    await session.flush()
+
+    with pytest.raises(ActivityForbiddenError):
+        await service.list_activities(
+            filters=ActivityListFilters(deal_id=other_deal.id),
+            context=context,
+        )
+
+
+@pytest.mark.asyncio
+async def test_add_comment_trims_payload_text(session: AsyncSession) -> None:
+    context, repo, deal_id, _ = await _prepare_deal(session)
+    service = ActivityService(repository=repo)
+
+    activity = await service.add_comment(
+        deal_id=deal_id,
+        author_id=context.user_id,
+        text="  trimmed text  ",
+        context=context,
+    )
+
+    assert activity.payload["text"] == "trimmed text"

tests/services/test_task_service.py

@@ -0,0 +1,199 @@
+"""Unit tests for TaskService."""
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from datetime import datetime, timedelta, timezone
+import uuid
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from app.models.activity import Activity, ActivityType
+from app.models.base import Base
+from app.models.contact import Contact
+from app.models.deal import Deal
+from app.models.organization import Organization
+from app.models.organization_member import OrganizationMember, OrganizationRole
+from app.models.task import TaskCreate
+from app.models.user import User
+from app.repositories.activity_repo import ActivityRepository
+from app.repositories.task_repo import TaskRepository
+from app.services.organization_service import OrganizationContext
+from app.services.task_service import (
+    TaskDueDateError,
+    TaskForbiddenError,
+    TaskService,
+    TaskUpdateData,
+)
+
+
+@pytest_asyncio.fixture()
+async def session() -> AsyncGenerator[AsyncSession, None]:
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        future=True,
+        poolclass=StaticPool,
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    session_factory = async_sessionmaker(engine, expire_on_commit=False)
+    async with session_factory() as session:
+        yield session
+    await engine.dispose()
+
+
+def _make_user(suffix: str) -> User:
+    return User(
+        email=f"user-{suffix}@example.com",
+        hashed_password="hashed",
+        name="Test User",
+        is_active=True,
+    )
+
+
+async def _setup_environment(
+    session: AsyncSession,
+    *,
+    role: OrganizationRole = OrganizationRole.MANAGER,
+    context_user: User | None = None,
+    owner_user: User | None = None,
+) -> tuple[OrganizationContext, User, User, int, TaskRepository, ActivityRepository]:
+    org = Organization(name=f"Org-{uuid.uuid4()}"[:8])
+    owner = owner_user or _make_user("owner")
+    ctx_user = context_user or owner
+    session.add_all([org, owner])
+    if ctx_user is not owner:
+        session.add(ctx_user)
+    await session.flush()
+
+    contact = Contact(
+        organization_id=org.id,
+        owner_id=owner.id,
+        name="John Doe",
+        email="john@example.com",
+    )
+    session.add(contact)
+    await session.flush()
+
+    deal = Deal(
+        organization_id=org.id,
+        contact_id=contact.id,
+        owner_id=owner.id,
+        title="Implementation",
+        amount=None,
+    )
+    session.add(deal)
+    await session.flush()
+
+    membership = OrganizationMember(organization_id=org.id, user_id=ctx_user.id, role=role)
+    context = OrganizationContext(organization=org, membership=membership)
+    task_repo = TaskRepository(session=session)
+    activity_repo = ActivityRepository(session=session)
+    return context, owner, ctx_user, deal.id, task_repo, activity_repo
+
+
+@pytest.mark.asyncio
+async def test_create_task_logs_activity(session: AsyncSession) -> None:
+    context, owner, _, deal_id, task_repo, activity_repo = await _setup_environment(session)
+    service = TaskService(task_repository=task_repo, activity_repository=activity_repo)
+
+    due_date = datetime.now(timezone.utc) + timedelta(days=2)
+    task = await service.create_task(
+        TaskCreate(
+            deal_id=deal_id,
+            title="Follow up",
+            description="Call client",
+            due_date=due_date,
+        ),
+        context=context,
+    )
+
+    result = await session.scalars(select(Activity).where(Activity.deal_id == deal_id))
+    activities = result.all()
+    assert len(activities) == 1
+    assert activities[0].type == ActivityType.TASK_CREATED
+    assert activities[0].payload["task_id"] == task.id
+    assert activities[0].payload["title"] == task.title
+
+
+@pytest.mark.asyncio
+async def test_member_cannot_create_task_for_foreign_deal(session: AsyncSession) -> None:
+    owner = _make_user("owner")
+    member = _make_user("member")
+    context, _, _, deal_id, task_repo, activity_repo = await _setup_environment(
+        session,
+        role=OrganizationRole.MEMBER,
+        context_user=member,
+        owner_user=owner,
+    )
+    service = TaskService(task_repository=task_repo, activity_repository=activity_repo)
+
+    with pytest.raises(TaskForbiddenError):
+        await service.create_task(
+            TaskCreate(
+                deal_id=deal_id,
+                title="Follow up",
+                description=None,
+                due_date=datetime.now(timezone.utc) + timedelta(days=1),
+            ),
+            context=context,
+        )
+
+
+@pytest.mark.asyncio
+async def test_due_date_cannot_be_in_past(session: AsyncSession) -> None:
+    context, _, _, deal_id, task_repo, activity_repo = await _setup_environment(session)
+    service = TaskService(task_repository=task_repo, activity_repository=activity_repo)
+
+    with pytest.raises(TaskDueDateError):
+        await service.create_task(
+            TaskCreate(
+                deal_id=deal_id,
+                title="Late",
+                description=None,
+                due_date=datetime.now(timezone.utc) - timedelta(days=1),
+            ),
+            context=context,
+        )
+
+
+@pytest.mark.asyncio
+async def test_member_cannot_update_foreign_task(session: AsyncSession) -> None:
+    # First create a task as the owner
+    owner = _make_user("owner")
+    context_owner, _, _, deal_id, task_repo, activity_repo = await _setup_environment(
+        session,
+        context_user=owner,
+        owner_user=owner,
+    )
+    service = TaskService(task_repository=task_repo, activity_repository=activity_repo)
+    task = await service.create_task(
+        TaskCreate(
+            deal_id=deal_id,
+            title="Prepare deck",
+            description=None,
+            due_date=datetime.now(timezone.utc) + timedelta(days=5),
+        ),
+        context=context_owner,
+    )
+
+    # Attempt to update it as another member
+    member = _make_user("member")
+    session.add(member)
+    await session.flush()
+    membership = OrganizationMember(
+        organization_id=context_owner.organization_id,
+        user_id=member.id,
+        role=OrganizationRole.MEMBER,
+    )
+    member_context = OrganizationContext(organization=context_owner.organization, membership=membership)
+
+    with pytest.raises(TaskForbiddenError):
+        await service.update_task(
+            task.id,
+            TaskUpdateData(is_done=True),
+            context=member_context,
+        )