feat: implement collaboration module

- Backend (FastAPI):
  - Task comments with nested replies and soft delete
  - @mention parsing with 10-mention limit per comment
  - Notification system with read/unread tracking
  - Blocker management with project owner notification
  - WebSocket endpoint with JWT auth and keepalive
  - User search API for @mention autocomplete
  - Alembic migration for 4 new tables

- Frontend (React + Vite):
  - Comments component with @mention autocomplete
  - NotificationBell with real-time WebSocket updates
  - BlockerDialog for task blocking workflow
  - NotificationContext for state management

- OpenSpec:
  - 4 requirements with scenarios defined
  - add-collaboration change archived

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

backend/app/api/blockers/__init__.py

@@ -0,0 +1,3 @@
+from app.api.blockers.router import router
+
+__all__ = ["router"]

backend/app/api/blockers/router.py

@@ -0,0 +1,200 @@
+import uuid
+from datetime import datetime
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.models import User, Task, Blocker
+from app.schemas.blocker import (
+    BlockerCreate, BlockerResolve, BlockerResponse, BlockerListResponse, BlockerUserInfo
+)
+from app.middleware.auth import get_current_user, check_task_access, check_task_edit_access
+from app.services.notification_service import NotificationService
+
+router = APIRouter(tags=["blockers"])
+
+
+def blocker_to_response(blocker: Blocker) -> BlockerResponse:
+    """Convert Blocker model to BlockerResponse."""
+    return BlockerResponse(
+        id=blocker.id,
+        task_id=blocker.task_id,
+        reason=blocker.reason,
+        resolution_note=blocker.resolution_note,
+        created_at=blocker.created_at,
+        resolved_at=blocker.resolved_at,
+        reporter=BlockerUserInfo(
+            id=blocker.reporter.id,
+            name=blocker.reporter.name,
+            email=blocker.reporter.email,
+        ),
+        resolver=BlockerUserInfo(
+            id=blocker.resolver.id,
+            name=blocker.resolver.name,
+            email=blocker.resolver.email,
+        ) if blocker.resolver else None,
+    )
+
+
+@router.post("/api/tasks/{task_id}/blockers", response_model=BlockerResponse, status_code=status.HTTP_201_CREATED)
+async def create_blocker(
+    task_id: str,
+    blocker_data: BlockerCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Mark a task as blocked with a reason."""
+    task = db.query(Task).filter(Task.id == task_id).first()
+
+    if not task:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Task not found",
+        )
+
+    if not check_task_edit_access(current_user, task, task.project):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Permission denied",
+        )
+
+    # Check if task is already blocked with an unresolved blocker
+    existing_blocker = db.query(Blocker).filter(
+        Blocker.task_id == task_id,
+        Blocker.resolved_at == None,
+    ).first()
+
+    if existing_blocker:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Task already has an unresolved blocker",
+        )
+
+    # Create blocker record
+    blocker = Blocker(
+        id=str(uuid.uuid4()),
+        task_id=task_id,
+        reported_by=current_user.id,
+        reason=blocker_data.reason,
+    )
+    db.add(blocker)
+
+    # Update task blocker_flag
+    task.blocker_flag = True
+
+    # Notify project owner
+    NotificationService.notify_blocker(db, task, current_user, blocker_data.reason)
+
+    db.commit()
+    db.refresh(blocker)
+
+    return blocker_to_response(blocker)
+
+
+@router.put("/api/blockers/{blocker_id}/resolve", response_model=BlockerResponse)
+async def resolve_blocker(
+    blocker_id: str,
+    resolve_data: BlockerResolve,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Resolve a blocker with a resolution note."""
+    blocker = db.query(Blocker).filter(Blocker.id == blocker_id).first()
+
+    if not blocker:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Blocker not found",
+        )
+
+    if blocker.resolved_at is not None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Blocker is already resolved",
+        )
+
+    task = blocker.task
+    if not check_task_edit_access(current_user, task, task.project):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Permission denied",
+        )
+
+    # Update blocker
+    blocker.resolved_by = current_user.id
+    blocker.resolution_note = resolve_data.resolution_note
+    blocker.resolved_at = datetime.utcnow()
+
+    # Check if there are other unresolved blockers
+    other_blockers = db.query(Blocker).filter(
+        Blocker.task_id == task.id,
+        Blocker.id != blocker_id,
+        Blocker.resolved_at == None,
+    ).count()
+
+    if other_blockers == 0:
+        task.blocker_flag = False
+
+    # Notify reporter that blocker is resolved
+    NotificationService.notify_blocker_resolved(db, task, current_user, blocker.reported_by)
+
+    db.commit()
+    db.refresh(blocker)
+
+    return blocker_to_response(blocker)
+
+
+@router.get("/api/tasks/{task_id}/blockers", response_model=BlockerListResponse)
+async def list_blockers(
+    task_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """List all blockers (history) for a task."""
+    task = db.query(Task).filter(Task.id == task_id).first()
+
+    if not task:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Task not found",
+        )
+
+    if not check_task_access(current_user, task, task.project):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Access denied",
+        )
+
+    blockers = db.query(Blocker).filter(
+        Blocker.task_id == task_id,
+    ).order_by(Blocker.created_at.desc()).all()
+
+    return BlockerListResponse(
+        blockers=[blocker_to_response(b) for b in blockers],
+        total=len(blockers),
+    )
+
+
+@router.get("/api/blockers/{blocker_id}", response_model=BlockerResponse)
+async def get_blocker(
+    blocker_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Get a specific blocker by ID."""
+    blocker = db.query(Blocker).filter(Blocker.id == blocker_id).first()
+
+    if not blocker:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Blocker not found",
+        )
+
+    task = blocker.task
+    if not check_task_access(current_user, task, task.project):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Access denied",
+        )
+
+    return blocker_to_response(blocker)

backend/app/api/comments/__init__.py

@@ -0,0 +1,3 @@
+from app.api.comments.router import router
+
+__all__ = ["router"]

backend/app/api/comments/router.py

@@ -0,0 +1,257 @@
+import uuid
+from typing import List
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.models import User, Task, Comment
+from app.schemas.comment import (
+    CommentCreate, CommentUpdate, CommentResponse, CommentListResponse,
+    CommentAuthor, MentionedUser
+)
+from app.middleware.auth import get_current_user, check_task_access
+from app.services.notification_service import NotificationService
+
+router = APIRouter(tags=["comments"])
+
+
+def comment_to_response(comment: Comment) -> CommentResponse:
+    """Convert Comment model to CommentResponse."""
+    mentioned_users = [
+        MentionedUser(
+            id=m.mentioned_user.id,
+            name=m.mentioned_user.name,
+            email=m.mentioned_user.email,
+        )
+        for m in comment.mentions
+        if m.mentioned_user
+    ]
+
+    reply_count = len([r for r in comment.replies if not r.is_deleted]) if comment.replies else 0
+
+    return CommentResponse(
+        id=comment.id,
+        task_id=comment.task_id,
+        parent_comment_id=comment.parent_comment_id,
+        content=comment.content if not comment.is_deleted else "[This comment has been deleted]",
+        is_edited=comment.is_edited,
+        is_deleted=comment.is_deleted,
+        created_at=comment.created_at,
+        updated_at=comment.updated_at,
+        author=CommentAuthor(
+            id=comment.author.id,
+            name=comment.author.name,
+            email=comment.author.email,
+        ),
+        mentions=mentioned_users,
+        reply_count=reply_count,
+    )
+
+
+@router.post("/api/tasks/{task_id}/comments", response_model=CommentResponse, status_code=status.HTTP_201_CREATED)
+async def create_comment(
+    task_id: str,
+    comment_data: CommentCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Create a new comment on a task."""
+    task = db.query(Task).filter(Task.id == task_id).first()
+
+    if not task:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Task not found",
+        )
+
+    if not check_task_access(current_user, task, task.project):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Access denied",
+        )
+
+    # Validate parent comment if provided
+    parent_author_id = None
+    if comment_data.parent_comment_id:
+        parent_comment = db.query(Comment).filter(
+            Comment.id == comment_data.parent_comment_id,
+            Comment.task_id == task_id,
+        ).first()
+
+        if not parent_comment:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Parent comment not found",
+            )
+        parent_author_id = parent_comment.author_id
+
+    # Check @mention limit
+    mention_count = NotificationService.count_mentions(comment_data.content)
+    if mention_count > NotificationService.MAX_MENTIONS_PER_COMMENT:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Maximum {NotificationService.MAX_MENTIONS_PER_COMMENT} mentions allowed per comment",
+        )
+
+    # Create comment
+    comment = Comment(
+        id=str(uuid.uuid4()),
+        task_id=task_id,
+        parent_comment_id=comment_data.parent_comment_id,
+        author_id=current_user.id,
+        content=comment_data.content,
+    )
+    db.add(comment)
+    db.flush()
+
+    # Process mentions and create notifications
+    NotificationService.process_mentions(db, comment, task, current_user)
+
+    # Notify parent comment author if this is a reply
+    if parent_author_id:
+        NotificationService.notify_comment_reply(db, comment, task, current_user, parent_author_id)
+
+    db.commit()
+    db.refresh(comment)
+
+    return comment_to_response(comment)
+
+
+@router.get("/api/tasks/{task_id}/comments", response_model=CommentListResponse)
+async def list_comments(
+    task_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """List all comments on a task."""
+    task = db.query(Task).filter(Task.id == task_id).first()
+
+    if not task:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Task not found",
+        )
+
+    if not check_task_access(current_user, task, task.project):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Access denied",
+        )
+
+    # Get root comments (no parent) ordered by creation time
+    comments = db.query(Comment).filter(
+        Comment.task_id == task_id,
+        Comment.parent_comment_id == None,
+    ).order_by(Comment.created_at).all()
+
+    return CommentListResponse(
+        comments=[comment_to_response(c) for c in comments],
+        total=len(comments),
+    )
+
+
+@router.get("/api/comments/{comment_id}/replies", response_model=CommentListResponse)
+async def list_replies(
+    comment_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """List replies to a comment."""
+    comment = db.query(Comment).filter(Comment.id == comment_id).first()
+
+    if not comment:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Comment not found",
+        )
+
+    task = comment.task
+    if not check_task_access(current_user, task, task.project):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Access denied",
+        )
+
+    replies = db.query(Comment).filter(
+        Comment.parent_comment_id == comment_id,
+    ).order_by(Comment.created_at).all()
+
+    return CommentListResponse(
+        comments=[comment_to_response(r) for r in replies],
+        total=len(replies),
+    )
+
+
+@router.put("/api/comments/{comment_id}", response_model=CommentResponse)
+async def update_comment(
+    comment_id: str,
+    comment_data: CommentUpdate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Update a comment (only by author)."""
+    comment = db.query(Comment).filter(Comment.id == comment_id).first()
+
+    if not comment:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Comment not found",
+        )
+
+    if comment.is_deleted:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Cannot edit a deleted comment",
+        )
+
+    if comment.author_id != current_user.id and not current_user.is_system_admin:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only the author can edit this comment",
+        )
+
+    # Check @mention limit in updated content
+    mention_count = NotificationService.count_mentions(comment_data.content)
+    if mention_count > NotificationService.MAX_MENTIONS_PER_COMMENT:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Maximum {NotificationService.MAX_MENTIONS_PER_COMMENT} mentions allowed per comment",
+        )
+
+    comment.content = comment_data.content
+    comment.is_edited = True
+
+    db.commit()
+    db.refresh(comment)
+
+    return comment_to_response(comment)
+
+
+@router.delete("/api/comments/{comment_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_comment(
+    comment_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Delete a comment (soft delete, only by author or admin)."""
+    comment = db.query(Comment).filter(Comment.id == comment_id).first()
+
+    if not comment:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Comment not found",
+        )
+
+    if comment.author_id != current_user.id and not current_user.is_system_admin:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only the author can delete this comment",
+        )
+
+    # Soft delete - mark as deleted but keep for reply chain
+    comment.is_deleted = True
+    comment.content = ""
+
+    db.commit()
+
+    return None

backend/app/api/notifications/__init__.py

@@ -0,0 +1,3 @@
+from app.api.notifications.router import router
+
+__all__ = ["router"]

backend/app/api/notifications/router.py

@@ -0,0 +1,119 @@
+from typing import Optional
+from datetime import datetime
+from fastapi import APIRouter, Depends, HTTPException, status, Query
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.models import User, Notification
+from app.schemas.notification import (
+    NotificationResponse, NotificationListResponse, UnreadCountResponse
+)
+from app.middleware.auth import get_current_user
+
+router = APIRouter(tags=["notifications"])
+
+
+def notification_to_response(notification: Notification) -> NotificationResponse:
+    """Convert Notification model to NotificationResponse."""
+    return NotificationResponse(
+        id=notification.id,
+        type=notification.type,
+        reference_type=notification.reference_type,
+        reference_id=notification.reference_id,
+        title=notification.title,
+        message=notification.message,
+        is_read=notification.is_read,
+        created_at=notification.created_at,
+        read_at=notification.read_at,
+    )
+
+
+@router.get("/api/notifications", response_model=NotificationListResponse)
+async def list_notifications(
+    is_read: Optional[bool] = Query(None, description="Filter by read status"),
+    limit: int = Query(50, ge=1, le=100, description="Number of notifications to return"),
+    offset: int = Query(0, ge=0, description="Offset for pagination"),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """List notifications for the current user."""
+    query = db.query(Notification).filter(Notification.user_id == current_user.id)
+
+    if is_read is not None:
+        query = query.filter(Notification.is_read == is_read)
+
+    total = query.count()
+    unread_count = db.query(Notification).filter(
+        Notification.user_id == current_user.id,
+        Notification.is_read == False,
+    ).count()
+
+    notifications = query.order_by(Notification.created_at.desc()).offset(offset).limit(limit).all()
+
+    return NotificationListResponse(
+        notifications=[notification_to_response(n) for n in notifications],
+        total=total,
+        unread_count=unread_count,
+    )
+
+
+@router.get("/api/notifications/unread-count", response_model=UnreadCountResponse)
+async def get_unread_count(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Get count of unread notifications."""
+    count = db.query(Notification).filter(
+        Notification.user_id == current_user.id,
+        Notification.is_read == False,
+    ).count()
+
+    return UnreadCountResponse(unread_count=count)
+
+
+@router.put("/api/notifications/{notification_id}/read", response_model=NotificationResponse)
+async def mark_as_read(
+    notification_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Mark a notification as read."""
+    notification = db.query(Notification).filter(
+        Notification.id == notification_id,
+        Notification.user_id == current_user.id,
+    ).first()
+
+    if not notification:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Notification not found",
+        )
+
+    if not notification.is_read:
+        notification.is_read = True
+        notification.read_at = datetime.utcnow()
+        db.commit()
+        db.refresh(notification)
+
+    return notification_to_response(notification)
+
+
+@router.put("/api/notifications/read-all", response_model=dict)
+async def mark_all_as_read(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Mark all notifications as read."""
+    now = datetime.utcnow()
+
+    updated_count = db.query(Notification).filter(
+        Notification.user_id == current_user.id,
+        Notification.is_read == False,
+    ).update({
+        Notification.is_read: True,
+        Notification.read_at: now,
+    })
+
+    db.commit()
+
+    return {"updated_count": updated_count}

backend/app/api/users/router.py

@@ -1,5 +1,6 @@
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, status, Query
 from sqlalchemy.orm import Session
+from sqlalchemy import or_
 from typing import List
 
 from app.core.database import get_db
@@ -16,6 +17,33 @@ from app.middleware.auth import (
 router = APIRouter()
 
 
+@router.get("/search", response_model=List[UserResponse])
+async def search_users(
+    q: str = Query(..., min_length=1, max_length=100, description="Search query"),
+    limit: int = Query(10, ge=1, le=50, description="Max results"),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """
+    Search users by name or email. Used for @mention autocomplete.
+    Returns users matching the query, limited to same department unless system admin.
+    """
+    query = db.query(User).filter(
+        User.is_active == True,
+        or_(
+            User.name.ilike(f"%{q}%"),
+            User.email.ilike(f"%{q}%"),
+        )
+    )
+
+    # Filter by department if not system admin
+    if not current_user.is_system_admin and current_user.department_id:
+        query = query.filter(User.department_id == current_user.department_id)
+
+    users = query.limit(limit).all()
+    return users
+
+
 @router.get("", response_model=List[UserResponse])
 async def list_users(
     skip: int = 0,

backend/app/api/websocket/__init__.py

@@ -0,0 +1,3 @@
+from app.api.websocket.router import router
+
+__all__ = ["router"]

backend/app/api/websocket/router.py

@@ -0,0 +1,98 @@
+import asyncio
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Query
+from sqlalchemy.orm import Session
+
+from app.core.database import SessionLocal
+from app.core.security import decode_access_token
+from app.core.redis import get_redis_sync
+from app.models import User
+from app.services.websocket_manager import manager
+
+router = APIRouter(tags=["websocket"])
+
+
+async def get_user_from_token(token: str) -> tuple[str | None, User | None]:
+    """Validate token and return user_id and user object."""
+    payload = decode_access_token(token)
+    if payload is None:
+        return None, None
+
+    user_id = payload.get("sub")
+    if user_id is None:
+        return None, None
+
+    # Verify session in Redis
+    redis_client = get_redis_sync()
+    stored_token = redis_client.get(f"session:{user_id}")
+    if stored_token is None or stored_token != token:
+        return None, None
+
+    # Get user from database
+    db = SessionLocal()
+    try:
+        user = db.query(User).filter(User.id == user_id).first()
+        if user is None or not user.is_active:
+            return None, None
+        return user_id, user
+    finally:
+        db.close()
+
+
+@router.websocket("/ws/notifications")
+async def websocket_notifications(
+    websocket: WebSocket,
+    token: str = Query(..., description="JWT token for authentication"),
+):
+    """
+    WebSocket endpoint for real-time notifications.
+
+    Connect with: ws://host/ws/notifications?token=<jwt_token>
+
+    Messages sent by server:
+    - {"type": "notification", "data": {...}}  - New notification
+    - {"type": "unread_count", "data": {"unread_count": N}}  - Unread count update
+    - {"type": "pong"}  - Response to ping
+
+    Messages accepted from client:
+    - {"type": "ping"}  - Keepalive ping
+    """
+    user_id, user = await get_user_from_token(token)
+
+    if user_id is None:
+        await websocket.close(code=4001, reason="Invalid or expired token")
+        return
+
+    await manager.connect(websocket, user_id)
+
+    try:
+        # Send initial connection success message
+        await websocket.send_json({
+            "type": "connected",
+            "data": {"user_id": user_id, "message": "Connected to notification service"},
+        })
+
+        while True:
+            try:
+                # Wait for messages from client (ping/pong for keepalive)
+                data = await asyncio.wait_for(
+                    websocket.receive_json(),
+                    timeout=60.0  # 60 second timeout
+                )
+
+                # Handle ping message
+                if data.get("type") == "ping":
+                    await websocket.send_json({"type": "pong"})
+
+            except asyncio.TimeoutError:
+                # Send keepalive ping if no message received
+                try:
+                    await websocket.send_json({"type": "ping"})
+                except Exception:
+                    break
+
+    except WebSocketDisconnect:
+        pass
+    except Exception:
+        pass
+    finally:
+        await manager.disconnect(websocket, user_id)