feat: Unified Messaging System — kommunikation plugin, AI/Proactive/System participants, MessageSidebar, Rich Content Renderer

Phase 1: Backend plugin kommunikation (13 files, 10 tables, REST API, WebSocket, RBAC, DMS Bridge, Participant Registry, Mini-App Registry, Search Provider)
Phase 2: AI plugins as participants (ai_assistant + ai_proactive dock as participants, heartbeat job)
Phase 3: system_notif plugin (system events → chat messages, pinned System room)
Phase 4: Frontend MessageSidebar (replaces AISidebar, same design, comm API client, WebSocket hook, commStore)
Phase 5: Rich Content Block Renderer (11 components: Markdown, HTML, Image, Audio, Video, File, ActionCard, ContactCard, MiniApp, BlockRenderer)
BasePlugin: added services property + _container in on_activate

app/plugins/builtins/kommunikation/content_types.py

@@ -0,0 +1,82 @@
+"""Content block type definitions for rich content in messages."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+# Known block types and their expected schema
+BLOCK_TYPES: dict[str, dict[str, Any]] = {
+    "text": {
+        "description": "Plain text fallback",
+        "fields": {"text": "str"},
+    },
+    "markdown": {
+        "description": "Markdown formatted text",
+        "fields": {"markdown": "str"},
+    },
+    "html": {
+        "description": "Sanitized HTML content",
+        "fields": {"html": "str"},
+    },
+    "image": {
+        "description": "Image attachment",
+        "fields": {"url": "str", "alt": "str", "width": "int?"},
+    },
+    "audio": {
+        "description": "Audio file",
+        "fields": {"url": "str", "duration": "int?", "waveform": "list?"},
+    },
+    "video": {
+        "description": "Video file",
+        "fields": {"url": "str", "duration": "int?", "thumbnail": "str?"},
+    },
+    "file": {
+        "description": "Generic file attachment",
+        "fields": {"url": "str", "name": "str", "size": "int?"},
+    },
+    "action_card": {
+        "description": "Interactive card with buttons",
+        "fields": {
+            "title": "str",
+            "body": "str",
+            "actions": "list[dict]",  # [{label, action, type}]
+        },
+    },
+    "contact_card": {
+        "description": "Contact reference card",
+        "fields": {"contact_id": "str", "name": "str"},
+    },
+    "miniapp": {
+        "description": "Embedded mini-app",
+        "fields": {"app_id": "str", "config": "dict?"},
+    },
+}
+
+
+def validate_block(block_type: str, block_data: dict[str, Any]) -> bool:
+    """Validate that a block has the required fields for its type.
+
+    Returns True if valid, False otherwise.
+    Unknown block types are allowed (forward-compatible) but logged.
+    """
+    if block_type not in BLOCK_TYPES:
+        # Unknown types are allowed — forward compatible
+        return True
+
+    required_fields = BLOCK_TYPES[block_type].get("fields", {})
+    for field_name, field_type in required_fields.items():
+        if field_type.endswith("?"):
+            continue  # Optional field
+        if field_name not in block_data:
+            return False
+
+    return True
+
+
+def list_block_types() -> list[dict[str, Any]]:
+    """List all known block types for frontend reference."""
+    return [
+        {"block_type": bt, "description": info["description"], "fields": info["fields"]}
+        for bt, info in BLOCK_TYPES.items()
+    ]

app/plugins/builtins/kommunikation/dms_bridge.py

@@ -0,0 +1,187 @@
+"""DMS Bridge — integrates with the DMS plugin for file storage."""
+
+from __future__ import annotations
+
+import logging
+import os
+import uuid
+from typing import Any
+
+from fastapi import UploadFile
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.plugins.builtins.dms.models import File as DmsFile, Folder
+
+logger = logging.getLogger(__name__)
+
+DMS_STORAGE_BASE = os.environ.get("DMS_STORAGE_BASE", "/tmp/dms")
+COMM_FOLDER_NAME = "_kommunikation"
+MAX_DIRECT_UPLOAD = 100 * 1024 * 1024  # 100 MB — larger files must be DMS references
+
+
+class DmsBridge:
+    """Bridge to the DMS plugin for attachment storage."""
+
+    @staticmethod
+    async def ensure_comm_folder(
+        db: AsyncSession,
+        tenant_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> Folder:
+        """Ensure the _kommunikation root folder exists in DMS."""
+        result = await db.execute(
+            select(Folder).where(
+                Folder.name == COMM_FOLDER_NAME,
+                Folder.parent_id.is_(None),
+                Folder.tenant_id == tenant_id,
+                Folder.deleted_at.is_(None),
+            )
+        )
+        folder = result.scalar_one_or_none()
+        if folder is None:
+            folder = Folder(
+                tenant_id=tenant_id,
+                name=COMM_FOLDER_NAME,
+                parent_id=None,
+                created_by=user_id,
+            )
+            db.add(folder)
+            await db.flush()
+        return folder
+
+    @staticmethod
+    async def ensure_conversation_folder(
+        db: AsyncSession,
+        tenant_id: uuid.UUID,
+        user_id: uuid.UUID,
+        conversation_id: uuid.UUID,
+    ) -> Folder:
+        """Ensure a sub-folder for a specific conversation exists."""
+        root_folder = await DmsBridge.ensure_comm_folder(db, tenant_id, user_id)
+        conv_name = str(conversation_id)
+        result = await db.execute(
+            select(Folder).where(
+                Folder.name == conv_name,
+                Folder.parent_id == root_folder.id,
+                Folder.tenant_id == tenant_id,
+                Folder.deleted_at.is_(None),
+            )
+        )
+        folder = result.scalar_one_or_none()
+        if folder is None:
+            folder = Folder(
+                tenant_id=tenant_id,
+                name=conv_name,
+                parent_id=root_folder.id,
+                created_by=user_id,
+            )
+            db.add(folder)
+            await db.flush()
+        return folder
+
+    @staticmethod
+    async def store_attachment(
+        db: AsyncSession,
+        tenant_id: uuid.UUID,
+        conversation_id: uuid.UUID,
+        user_id: uuid.UUID,
+        file: UploadFile,
+    ) -> dict[str, Any]:
+        """Store an uploaded file in the DMS under _kommunikation/{conversation_id}/.
+
+        Returns dict with file_id, file_name, file_type, file_size.
+        """
+        # Check size limit
+        content = await file.read()
+        file_size = len(content)
+
+        if file_size > MAX_DIRECT_UPLOAD:
+            raise ValueError(
+                f"File size {file_size} exceeds direct upload limit ({MAX_DIRECT_UPLOAD} bytes). "
+                f"Use DMS reference instead."
+            )
+
+        # Ensure conversation folder
+        folder = await DmsBridge.ensure_conversation_folder(
+            db, tenant_id, user_id, conversation_id
+        )
+
+        # Save file to disk
+        file_id = uuid.uuid4()
+        file_ext = os.path.splitext(file.filename or "")[1] or ""
+        storage_path = os.path.join(
+            DMS_STORAGE_BASE,
+            str(tenant_id),
+            str(folder.id),
+            f"{file_id}{file_ext}",
+        )
+        os.makedirs(os.path.dirname(storage_path), exist_ok=True)
+        with open(storage_path, "wb") as f:
+            f.write(content)
+
+        # Create DMS file record
+        dms_file = DmsFile(
+            tenant_id=tenant_id,
+            name=file.filename or f"{file_id}",
+            folder_id=folder.id,
+            uploaded_by=user_id,
+            mime_type=file.content_type or "application/octet-stream",
+            size_bytes=file_size,
+            storage_path=storage_path,
+        )
+        db.add(dms_file)
+        await db.flush()
+
+        return {
+            "file_id": str(dms_file.id),
+            "file_name": dms_file.name,
+            "file_type": dms_file.mime_type,
+            "file_size": dms_file.size_bytes,
+            "file_source": "comm",
+        }
+
+    @staticmethod
+    async def reference_external_file(
+        db: AsyncSession,
+        tenant_id: uuid.UUID,
+        file_id: uuid.UUID,
+    ) -> dict[str, Any] | None:
+        """Reference an existing DMS file without copying it.
+
+        Returns dict with file metadata or None if file not found.
+        """
+        result = await db.execute(
+            select(DmsFile).where(
+                DmsFile.id == file_id,
+                DmsFile.tenant_id == tenant_id,
+                DmsFile.deleted_at.is_(None),
+            )
+        )
+        dms_file = result.scalar_one_or_none()
+        if dms_file is None:
+            return None
+
+        return {
+            "file_id": str(dms_file.id),
+            "file_name": dms_file.name,
+            "file_type": dms_file.mime_type,
+            "file_size": dms_file.size_bytes,
+            "file_source": "dms",
+        }
+
+    @staticmethod
+    async def get_file(
+        db: AsyncSession,
+        tenant_id: uuid.UUID,
+        file_id: uuid.UUID,
+    ) -> DmsFile | None:
+        """Get a DMS file by ID."""
+        result = await db.execute(
+            select(DmsFile).where(
+                DmsFile.id == file_id,
+                DmsFile.tenant_id == tenant_id,
+                DmsFile.deleted_at.is_(None),
+            )
+        )
+        return result.scalar_one_or_none()

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

@@ -0,0 +1,160 @@
+-- kommunikation plugin initial migration: creates all 10 tables
+
+CREATE TABLE IF NOT EXISTS comm_conversations (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    title VARCHAR(255),
+    title_set_by UUID,
+    is_pinned BOOLEAN NOT NULL DEFAULT FALSE,
+    is_locked BOOLEAN NOT NULL DEFAULT FALSE,
+    locked_by VARCHAR(100),
+    is_direct BOOLEAN NOT NULL DEFAULT FALSE,
+    is_archived BOOLEAN NOT NULL DEFAULT FALSE,
+    created_by UUID,
+    created_by_type VARCHAR(20) NOT NULL DEFAULT 'user',
+    last_msg_at TIMESTAMPTZ,
+    last_msg_preview TEXT,
+    last_msg_sender_type VARCHAR(50),
+    metadata JSONB NOT NULL DEFAULT '{}',
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ
+);
+CREATE INDEX IF NOT EXISTS ix_comm_conversations_tenant ON comm_conversations(tenant_id);
+CREATE INDEX IF NOT EXISTS ix_comm_conversations_tenant_pinned ON comm_conversations(tenant_id, is_pinned);
+CREATE INDEX IF NOT EXISTS ix_comm_conversations_last_msg ON comm_conversations(tenant_id, last_msg_at DESC);
+
+CREATE TABLE IF NOT EXISTS comm_participants (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    conversation_id UUID NOT NULL REFERENCES comm_conversations(id) ON DELETE CASCADE,
+    participant_id UUID,
+    participant_type VARCHAR(50) NOT NULL,
+    display_name VARCHAR(255),
+    role VARCHAR(20) NOT NULL DEFAULT 'member',
+    joined_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    left_at TIMESTAMPTZ,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ,
+    UNIQUE(conversation_id, participant_id, participant_type)
+);
+CREATE INDEX IF NOT EXISTS ix_comm_participants_tenant_conv ON comm_participants(tenant_id, conversation_id);
+CREATE INDEX IF NOT EXISTS ix_comm_participants_tenant_user ON comm_participants(tenant_id, participant_id);
+
+CREATE TABLE IF NOT EXISTS comm_messages (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    conversation_id UUID NOT NULL REFERENCES comm_conversations(id) ON DELETE CASCADE,
+    sender_id UUID,
+    sender_type VARCHAR(50) NOT NULL,
+    content TEXT NOT NULL DEFAULT '',
+    content_format VARCHAR(20) NOT NULL DEFAULT 'text',
+    metadata JSONB NOT NULL DEFAULT '{}',
+    reply_to_id UUID REFERENCES comm_messages(id) ON DELETE SET NULL,
+    is_pinned BOOLEAN NOT NULL DEFAULT FALSE,
+    read_at TIMESTAMPTZ,
+    edited_at TIMESTAMPTZ,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ
+);
+CREATE INDEX IF NOT EXISTS ix_comm_messages_tenant_conv ON comm_messages(tenant_id, conversation_id, created_at);
+CREATE INDEX IF NOT EXISTS ix_comm_messages_tenant_sender ON comm_messages(tenant_id, sender_id);
+
+CREATE TABLE IF NOT EXISTS comm_message_blocks (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    message_id UUID NOT NULL REFERENCES comm_messages(id) ON DELETE CASCADE,
+    block_type VARCHAR(50) NOT NULL,
+    block_data JSONB NOT NULL,
+    sort_order INTEGER NOT NULL DEFAULT 0,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ
+);
+CREATE INDEX IF NOT EXISTS ix_comm_blocks_tenant_msg ON comm_message_blocks(tenant_id, message_id);
+
+CREATE TABLE IF NOT EXISTS comm_message_attachments (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    message_id UUID NOT NULL REFERENCES comm_messages(id) ON DELETE CASCADE,
+    file_id UUID,
+    file_source VARCHAR(10) NOT NULL DEFAULT 'comm',
+    file_name VARCHAR(255) NOT NULL,
+    file_type VARCHAR(255) NOT NULL,
+    file_size INTEGER,
+    thumbnail_path VARCHAR(1024),
+    metadata JSONB NOT NULL DEFAULT '{}',
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ
+);
+CREATE INDEX IF NOT EXISTS ix_comm_attachments_tenant_msg ON comm_message_attachments(tenant_id, message_id);
+
+CREATE TABLE IF NOT EXISTS comm_message_reactions (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    message_id UUID NOT NULL REFERENCES comm_messages(id) ON DELETE CASCADE,
+    user_id UUID NOT NULL,
+    emoji VARCHAR(50) NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ,
+    UNIQUE(message_id, user_id, emoji)
+);
+CREATE INDEX IF NOT EXISTS ix_comm_reactions_tenant_msg ON comm_message_reactions(tenant_id, message_id);
+
+CREATE TABLE IF NOT EXISTS comm_message_reads (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    conversation_id UUID NOT NULL REFERENCES comm_conversations(id) ON DELETE CASCADE,
+    user_id UUID NOT NULL,
+    last_read_msg_id UUID REFERENCES comm_messages(id) ON DELETE SET NULL,
+    last_read_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ,
+    UNIQUE(conversation_id, user_id)
+);
+CREATE INDEX IF NOT EXISTS ix_comm_reads_tenant_user ON comm_message_reads(tenant_id, user_id);
+
+CREATE TABLE IF NOT EXISTS comm_conversation_pins (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    conversation_id UUID NOT NULL REFERENCES comm_conversations(id) ON DELETE CASCADE,
+    user_id UUID NOT NULL,
+    pinned_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ,
+    UNIQUE(conversation_id, user_id)
+);
+CREATE INDEX IF NOT EXISTS ix_comm_pins_tenant_user ON comm_conversation_pins(tenant_id, user_id);
+
+CREATE TABLE IF NOT EXISTS comm_conversation_mutes (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    conversation_id UUID NOT NULL REFERENCES comm_conversations(id) ON DELETE CASCADE,
+    user_id UUID NOT NULL,
+    muted_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ,
+    UNIQUE(conversation_id, user_id)
+);
+CREATE INDEX IF NOT EXISTS ix_comm_mutes_tenant_user ON comm_conversation_mutes(tenant_id, user_id);
+
+CREATE TABLE IF NOT EXISTS comm_message_edits (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    message_id UUID NOT NULL REFERENCES comm_messages(id) ON DELETE CASCADE,
+    old_content TEXT NOT NULL,
+    old_blocks JSONB NOT NULL DEFAULT '[]',
+    edited_by UUID NOT NULL,
+    edited_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    deleted_at TIMESTAMPTZ
+);
+CREATE INDEX IF NOT EXISTS ix_comm_edits_tenant_msg ON comm_message_edits(tenant_id, message_id);

app/plugins/builtins/kommunikation/miniapp_registry.py

@@ -0,0 +1,73 @@
+"""Mini-App registry for plugin-provided interactive chat components."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable, Awaitable
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+
+
+class MiniAppDef(BaseModel):
+    """Definition of a mini-app that plugins can register."""
+
+    app_id: str = Field(..., description="Unique app identifier")
+    name: str = Field(..., description="Display name")
+    icon: str = Field(default="app", description="Icon name")
+    description: str = Field(default="", description="App description")
+    plugin_name: str = Field(..., description="Plugin that registered this app")
+    render_schema: dict[str, Any] = Field(
+        default_factory=dict, description="JSON schema for frontend rendering"
+    )
+
+
+class MiniAppRegistry:
+    """Registry for mini-apps that plugins provide for chat embedding."""
+
+    def __init__(self) -> None:
+        self._apps: dict[str, MiniAppDef] = {}
+
+    def register(
+        self,
+        app_id: str,
+        name: str,
+        icon: str,
+        description: str,
+        plugin_name: str,
+        render_schema: dict[str, Any] | None = None,
+    ) -> None:
+        """Register a mini-app."""
+        app = MiniAppDef(
+            app_id=app_id,
+            name=name,
+            icon=icon,
+            description=description,
+            plugin_name=plugin_name,
+            render_schema=render_schema or {},
+        )
+        self._apps[app_id] = app
+        logger.info(f"Mini-app registered: {app_id} by {plugin_name}")
+
+    def unregister(self, app_id: str) -> None:
+        """Unregister a mini-app."""
+        app = self._apps.pop(app_id, None)
+        if app:
+            logger.info(f"Mini-app unregistered: {app_id}")
+
+    def unregister_plugin(self, plugin_name: str) -> None:
+        """Unregister all mini-apps from a specific plugin."""
+        to_remove = [app_id for app_id, app in self._apps.items() if app.plugin_name == plugin_name]
+        for app_id in to_remove:
+            self._apps.pop(app_id, None)
+        if to_remove:
+            logger.info(f"Unregistered {len(to_remove)} mini-apps from plugin {plugin_name}")
+
+    def list_apps(self) -> list[dict[str, Any]]:
+        """List all available mini-apps for frontend."""
+        return [app.model_dump() for app in self._apps.values()]
+
+    def get_app(self, app_id: str) -> MiniAppDef | None:
+        """Get a specific mini-app definition."""
+        return self._apps.get(app_id)

app/plugins/builtins/kommunikation/models.py

@@ -0,0 +1,283 @@
+"""SQLAlchemy models for the kommunikation plugin."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.dialects.postgresql import JSONB, UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, TenantMixin
+
+
+class CommConversation(Base, TenantMixin):
+    """Conversation / Room — tenant-scoped, supports pinning, locking, archiving."""
+
+    __tablename__ = "comm_conversations"
+    __table_args__ = (
+        Index("ix_comm_conversations_tenant", "tenant_id"),
+        Index("ix_comm_conversations_tenant_pinned", "tenant_id", "is_pinned"),
+        Index("ix_comm_conversations_last_msg", "tenant_id", "last_msg_at"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    title: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    title_set_by: Mapped[uuid.UUID | None] = mapped_column(PGUUID(as_uuid=True), nullable=True)
+    is_pinned: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    is_locked: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    locked_by: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    is_direct: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    is_archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    created_by: Mapped[uuid.UUID | None] = mapped_column(PGUUID(as_uuid=True), nullable=True)
+    created_by_type: Mapped[str] = mapped_column(String(20), nullable=False, default="user")
+    last_msg_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    last_msg_preview: Mapped[str | None] = mapped_column(Text, nullable=True)
+    last_msg_sender_type: Mapped[str | None] = mapped_column(String(50), nullable=True)
+    metadata_: Mapped[dict[str, Any]] = mapped_column("metadata", JSONB, default=dict, nullable=False)
+
+
+class CommParticipant(Base, TenantMixin):
+    """Participant in a conversation — user, ai, system, gateway, etc."""
+
+    __tablename__ = "comm_participants"
+    __table_args__ = (
+        UniqueConstraint(
+            "conversation_id", "participant_id", "participant_type",
+            name="uq_comm_participants_conv_part_type",
+        ),
+        Index("ix_comm_participants_tenant_conv", "tenant_id", "conversation_id"),
+        Index("ix_comm_participants_tenant_user", "tenant_id", "participant_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    conversation_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_conversations.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    participant_id: Mapped[uuid.UUID | None] = mapped_column(PGUUID(as_uuid=True), nullable=True)
+    participant_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    display_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    role: Mapped[str] = mapped_column(String(20), nullable=False, default="member")
+    joined_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
+    left_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+
+class CommMessage(Base, TenantMixin):
+    """Message in a conversation — text content plus rich content blocks."""
+
+    __tablename__ = "comm_messages"
+    __table_args__ = (
+        Index("ix_comm_messages_tenant_conv", "tenant_id", "conversation_id", "created_at"),
+        Index("ix_comm_messages_tenant_sender", "tenant_id", "sender_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    conversation_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_conversations.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    sender_id: Mapped[uuid.UUID | None] = mapped_column(PGUUID(as_uuid=True), nullable=True)
+    sender_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    content: Mapped[str] = mapped_column(Text, nullable=False, default="")
+    content_format: Mapped[str] = mapped_column(String(20), nullable=False, default="text")
+    metadata_: Mapped[dict[str, Any]] = mapped_column("metadata", JSONB, default=dict, nullable=False)
+    reply_to_id: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_messages.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    is_pinned: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    read_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    edited_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+
+class CommMessageBlock(Base, TenantMixin):
+    """Rich content block attached to a message."""
+
+    __tablename__ = "comm_message_blocks"
+    __table_args__ = (
+        Index("ix_comm_blocks_tenant_msg", "tenant_id", "message_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    message_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_messages.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    block_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    block_data: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
+    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+
+class CommMessageAttachment(Base, TenantMixin):
+    """File attachment on a message — DMS reference or comm-internal upload."""
+
+    __tablename__ = "comm_message_attachments"
+    __table_args__ = (
+        Index("ix_comm_attachments_tenant_msg", "tenant_id", "message_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    message_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_messages.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    file_id: Mapped[uuid.UUID | None] = mapped_column(PGUUID(as_uuid=True), nullable=True)
+    file_source: Mapped[str] = mapped_column(String(10), nullable=False, default="comm")
+    file_name: Mapped[str] = mapped_column(String(255), nullable=False)
+    file_type: Mapped[str] = mapped_column(String(255), nullable=False)
+    file_size: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    thumbnail_path: Mapped[str | None] = mapped_column(String(1024), nullable=True)
+    metadata_: Mapped[dict[str, Any]] = mapped_column("metadata", JSONB, default=dict, nullable=False)
+
+
+class CommMessageReaction(Base, TenantMixin):
+    """Emoji reaction on a message."""
+
+    __tablename__ = "comm_message_reactions"
+    __table_args__ = (
+        UniqueConstraint("message_id", "user_id", "emoji", name="uq_comm_reactions_msg_user_emoji"),
+        Index("ix_comm_reactions_tenant_msg", "tenant_id", "message_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    message_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_messages.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False)
+    emoji: Mapped[str] = mapped_column(String(50), nullable=False)
+
+
+class CommMessageRead(Base, TenantMixin):
+    """Read state per user per conversation."""
+
+    __tablename__ = "comm_message_reads"
+    __table_args__ = (
+        UniqueConstraint("conversation_id", "user_id", name="uq_comm_reads_conv_user"),
+        Index("ix_comm_reads_tenant_user", "tenant_id", "user_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    conversation_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_conversations.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False)
+    last_read_msg_id: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_messages.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    last_read_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+
+class CommConversationPin(Base, TenantMixin):
+    """User-specific conversation pinning."""
+
+    __tablename__ = "comm_conversation_pins"
+    __table_args__ = (
+        UniqueConstraint("conversation_id", "user_id", name="uq_comm_pins_conv_user"),
+        Index("ix_comm_pins_tenant_user", "tenant_id", "user_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    conversation_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_conversations.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False)
+    pinned_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+
+class CommConversationMute(Base, TenantMixin):
+    """User-specific conversation muting."""
+
+    __tablename__ = "comm_conversation_mutes"
+    __table_args__ = (
+        UniqueConstraint("conversation_id", "user_id", name="uq_comm_mutes_conv_user"),
+        Index("ix_comm_mutes_tenant_user", "tenant_id", "user_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    conversation_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_conversations.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False)
+    muted_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+
+class CommMessageEdit(Base, TenantMixin):
+    """Edit history for messages — stores old content before each edit."""
+
+    __tablename__ = "comm_message_edits"
+    __table_args__ = (
+        Index("ix_comm_edits_tenant_msg", "tenant_id", "message_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    message_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("comm_messages.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    old_content: Mapped[str] = mapped_column(Text, nullable=False)
+    old_blocks: Mapped[list[Any]] = mapped_column(JSONB, default=list, nullable=False)
+    edited_by: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False)
+    edited_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )

app/plugins/builtins/kommunikation/participant_registry.py

@@ -0,0 +1,89 @@
+"""Participant Registry — Andockpunkt für Plugins als Teilnehmer."""
+
+from __future__ import annotations
+
+import logging
+from abc import ABC, abstractmethod
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class ParticipantHandler(ABC):
+    """Interface that plugins implement to act as a conversation participant."""
+
+    @abstractmethod
+    async def on_message_received(
+        self,
+        conversation_id: Any,
+        message: dict[str, Any],
+        conversation: dict[str, Any],
+        mentions: list[str],
+        context: dict[str, Any],
+    ) -> list[dict[str, Any]] | None:
+        """Called when a new message arrives in a conversation this participant is part of.
+
+        Args:
+            conversation_id: UUID of the conversation.
+            message: The new message dict.
+            conversation: Full conversation dict with participants.
+            mentions: Parsed @mention participant types.
+            context: Tenant, user, and other context.
+
+        Returns:
+            Optional list of new message dicts (e.g. AI response).
+            For passive readers (system) → return None.
+            For reactive participants (ai) → return [message_dict, ...].
+        """
+        pass
+
+    @abstractmethod
+    def get_participant_info(self) -> dict[str, Any]:
+        """Return metadata: display_name, avatar_url, capabilities, description."""
+        pass
+
+
+class ParticipantRegistry:
+    """Global registry for plugin participants."""
+
+    def __init__(self) -> None:
+        self._handlers: dict[str, ParticipantHandler] = {}
+
+    def register(self, participant_type: str, handler: ParticipantHandler) -> None:
+        """Register a plugin as a participant type."""
+        self._handlers[participant_type] = handler
+        logger.info(f"Participant registered: {participant_type}")
+
+    def unregister(self, participant_type: str) -> None:
+        """Unregister a participant type."""
+        handler = self._handlers.pop(participant_type, None)
+        if handler:
+            logger.info(f"Participant unregistered: {participant_type}")
+
+    def get_handler(self, participant_type: str) -> ParticipantHandler | None:
+        """Get the handler for a participant type."""
+        return self._handlers.get(participant_type)
+
+    def list_types(self) -> list[str]:
+        """List all registered participant types."""
+        return list(self._handlers.keys())
+
+    def get_all_info(self) -> dict[str, dict[str, Any]]:
+        """Get info for all registered participants."""
+        return {ptype: handler.get_participant_info() for ptype, handler in self._handlers.items()}
+
+
+# Global instance
+_registry = ParticipantRegistry()
+
+
+def get_participant_registry() -> ParticipantRegistry:
+    """Get the global participant registry."""
+    return _registry
+
+
+def reset_participant_registry_for_testing() -> ParticipantRegistry:
+    """Create a fresh registry for testing."""
+    global _registry
+    _registry = ParticipantRegistry()
+    return _registry

app/plugins/builtins/kommunikation/plugin.py

@@ -0,0 +1,91 @@
+"""Kommunikation plugin — unified messaging: chat, AI, system, messenger."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.plugins.base import BasePlugin
+from app.plugins.manifest import PluginManifest, PluginRouteDef
+
+logger = logging.getLogger(__name__)
+
+
+class KommunikationPlugin(BasePlugin):
+    """Unified messaging plugin: conversations, messages, participants, WebSocket, rich content."""
+
+    manifest = PluginManifest(
+        name="kommunikation",
+        version="1.0.0",
+        display_name="Kommunikation",
+        description=(
+            "Unified Messaging: Chat, KI, System, Messenger — "
+            "alles ist ein Teilnehmer."
+        ),
+        dependencies=["permissions", "dms"],
+        routes=[
+            PluginRouteDef(
+                path="/api/v1/comm",
+                module="app.plugins.builtins.kommunikation.routes",
+                router_attr="router",
+            ),
+        ],
+        events=[
+            "message.received",
+            "message.sent",
+            "conversation.created",
+            "conversation.updated",
+            "participant.joined",
+            "participant.left",
+            "reaction.added",
+        ],
+        migrations=["0001_initial.sql"],
+        permissions=[
+            "comm:read",
+            "comm:write",
+            "comm:create",
+            "comm:manage",
+            "comm:admin",
+            "comm:delete",
+        ],
+        is_core=True,
+    )
+
+    async def on_activate(self, db, service_container, event_bus) -> None:
+        """Register participant registry and WebSocket manager."""
+        await super().on_activate(db, service_container, event_bus)
+
+        # Register WebSocket manager as a shared service
+        from app.plugins.builtins.kommunikation.websocket_manager import WebSocketManager
+        ws_manager = WebSocketManager()
+        service_container.register("comm_websocket", ws_manager)
+
+        # Register Mini-App registry as a shared service
+        from app.plugins.builtins.kommunikation.miniapp_registry import MiniAppRegistry
+        miniapp_registry = MiniAppRegistry()
+        service_container.register("comm_miniapps", miniapp_registry)
+
+        logger.info("Kommunikation plugin activated — WebSocket + MiniApp registries ready")
+
+    async def on_deactivate(self, db, service_container, event_bus) -> None:
+        """Clean up registries."""
+        await super().on_deactivate(db, service_container, event_bus)
+        logger.info("Kommunikation plugin deactivated")
+
+    def get_notification_types(self) -> list[dict[str, Any]]:
+        return [
+            {
+                "type_key": "comm_message",
+                "category": "communication",
+                "label": "Neue Nachricht",
+                "description": "Neue Nachricht in einer Konversation",
+                "is_enabled_by_default": True,
+            },
+            {
+                "type_key": "comm_mention",
+                "category": "communication",
+                "label": "Erwähnung",
+                "description": "Du wurdest in einer Nachricht erwähnt",
+                "is_enabled_by_default": True,
+            },
+        ]

app/plugins/builtins/kommunikation/rbac.py

@@ -0,0 +1,136 @@
+"""Chat-internal RBAC logic — two-level permission checks."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.permissions import check_permission
+from app.plugins.builtins.kommunikation.models import CommParticipant
+
+logger = logging.getLogger(__name__)
+
+
+class CommRBAC:
+    """Chat-internal RBAC, integrated with the existing permission system."""
+
+    @staticmethod
+    async def get_user_role(
+        db: AsyncSession,
+        conversation_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> str | None:
+        """Get the user's role in a conversation, or None if not a participant."""
+        result = await db.execute(
+            select(CommParticipant).where(
+                CommParticipant.conversation_id == conversation_id,
+                CommParticipant.participant_id == user_id,
+                CommParticipant.participant_type == "user",
+                CommParticipant.left_at.is_(None),
+            )
+        )
+        participant = result.scalar_one_or_none()
+        return participant.role if participant else None
+
+    @staticmethod
+    async def is_participant(
+        db: AsyncSession,
+        conversation_id: uuid.UUID,
+        user_id: uuid.UUID,
+    ) -> bool:
+        """Check if a user is an active participant in a conversation."""
+        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
+        return role is not None
+
+    @staticmethod
+    async def can_user_write(
+        db: AsyncSession,
+        conversation_id: uuid.UUID,
+        current_user: dict[str, Any],
+    ) -> bool:
+        """Check if user can write messages.
+
+        1. System permission 'comm:write' via check_permission
+        2. is_system_admin → always allowed
+        3. User must be participant with role 'admin' or 'member'
+        """
+        if current_user.get("is_system_admin"):
+            return True
+        if not check_permission(current_user, "comm:write"):
+            return False
+        user_id = uuid.UUID(current_user["user_id"])
+        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
+        return role in ("admin", "member")
+
+    @staticmethod
+    async def can_user_manage(
+        db: AsyncSession,
+        conversation_id: uuid.UUID,
+        current_user: dict[str, Any],
+    ) -> bool:
+        """Check if user can manage participants.
+
+        1. System permission 'comm:manage' via check_permission
+        2. is_system_admin → always allowed
+        3. User must be conversation admin
+        """
+        if current_user.get("is_system_admin"):
+            return True
+        if not check_permission(current_user, "comm:manage"):
+            return False
+        user_id = uuid.UUID(current_user["user_id"])
+        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
+        return role == "admin"
+
+    @staticmethod
+    async def can_user_delete(
+        db: AsyncSession,
+        conversation_id: uuid.UUID,
+        current_user: dict[str, Any],
+        is_own_message: bool = False,
+    ) -> bool:
+        """Check if user can delete messages or conversation.
+
+        1. is_system_admin → always allowed
+        2. For own messages: comm:write suffices
+        3. For others' messages: comm:delete + conversation admin
+        4. For conversation: comm:delete + admin role
+        """
+        if current_user.get("is_system_admin"):
+            return True
+        if is_own_message:
+            return check_permission(current_user, "comm:write")
+        if not check_permission(current_user, "comm:delete"):
+            return False
+        user_id = uuid.UUID(current_user["user_id"])
+        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
+        return role == "admin"
+
+    @staticmethod
+    async def can_user_create(current_user: dict[str, Any]) -> bool:
+        """Check if user can create conversations."""
+        if current_user.get("is_system_admin"):
+            return True
+        return check_permission(current_user, "comm:create")
+
+    @staticmethod
+    async def can_user_change_role(
+        db: AsyncSession,
+        conversation_id: uuid.UUID,
+        current_user: dict[str, Any],
+    ) -> bool:
+        """Check if user can change participant roles.
+
+        Requires comm:admin permission + conversation admin role.
+        """
+        if current_user.get("is_system_admin"):
+            return True
+        if not check_permission(current_user, "comm:admin"):
+            return False
+        user_id = uuid.UUID(current_user["user_id"])
+        role = await CommRBAC.get_user_role(db, conversation_id, user_id)
+        return role == "admin"

app/plugins/builtins/kommunikation/routes.py

@@ -0,0 +1,532 @@
+"""REST API routes for the kommunikation plugin."""
+
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, WebSocket, WebSocketDisconnect, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_db
+from app.deps import get_current_user
+from app.plugins.builtins.kommunikation.rbac import CommRBAC
+from app.plugins.builtins.kommunikation.schemas import (
+    ConversationCreate,
+    ConversationUpdate,
+    MessageCreate,
+    MessageUpdate,
+    ParticipantAdd,
+    ParticipantRoleUpdate,
+    ReactionCreate,
+    ReadStateUpdate,
+    MiniAppStartRequest,
+)
+from app.plugins.builtins.kommunikation.services import (
+    add_participant,
+    add_reaction,
+    change_role,
+    create_conversation,
+    delete_message,
+    edit_message,
+    get_conversation,
+    get_messages,
+    list_conversations,
+    mark_read,
+    mute_conversation,
+    pin_conversation,
+    remove_participant,
+    remove_reaction,
+    send_message,
+    unmute_conversation,
+    unpin_conversation,
+    update_conversation,
+)
+from app.plugins.builtins.kommunikation.content_types import list_block_types
+from app.plugins.builtins.kommunikation.dms_bridge import DmsBridge
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/v1/comm", tags=["kommunikation"])
+
+
+def _parse_uuid(val: str, field: str = "id") -> uuid.UUID:
+    try:
+        return uuid.UUID(val)
+    except (ValueError, TypeError):
+        raise HTTPException(400, detail={"detail": f"Invalid {field}", "code": "invalid_id"})
+
+
+# ─── Conversations ───
+
+@router.get("/conversations")
+async def list_user_conversations(
+    archived: bool = Query(False, description="Include archived conversations"),
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """List all conversations for the current user."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    convs = await list_conversations(db, tenant_id, user_id, include_archived=archived)
+    return {"items": convs, "total": len(convs)}
+
+
+@router.post("/conversations")
+async def create_new_conversation(
+    body: ConversationCreate,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Create a new conversation."""
+    if not await CommRBAC.can_user_create(current_user):
+        raise HTTPException(403, detail={"detail": "Permission denied", "code": "forbidden"})
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    return await create_conversation(
+        db, tenant_id, user_id,
+        title=body.title,
+        participant_ids=body.participant_ids,
+        is_direct=body.is_direct,
+        initial_message=body.initial_message,
+    )
+
+
+@router.get("/conversations/{conversation_id}")
+async def get_single_conversation(
+    conversation_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Get a single conversation with participants."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    conv = await get_conversation(db, tenant_id, conv_id, user_id)
+    if conv is None:
+        raise HTTPException(404, detail={"detail": "Conversation not found", "code": "not_found"})
+    return conv
+
+
+@router.patch("/conversations/{conversation_id}")
+async def update_single_conversation(
+    conversation_id: str,
+    body: ConversationUpdate,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Update a conversation (title, archive)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    conv = await update_conversation(db, tenant_id, conv_id, user_id, title=body.title, is_archived=body.is_archived)
+    if conv is None:
+        raise HTTPException(404, detail={"detail": "Conversation not found", "code": "not_found"})
+    return conv
+
+
+@router.delete("/conversations/{conversation_id}")
+async def leave_or_delete_conversation(
+    conversation_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Leave (member) or delete (admin) a conversation."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    # For now: just leave (set left_at)
+    success = await remove_participant(db, conv_id, user_id)
+    if not success:
+        raise HTTPException(404, detail={"detail": "Not a participant", "code": "not_found"})
+    return {"success": True}
+
+
+@router.post("/conversations/{conversation_id}/pin")
+async def pin_conv(
+    conversation_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Pin a conversation for the current user."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    await pin_conversation(db, tenant_id, conv_id, user_id)
+    return {"success": True}
+
+
+@router.delete("/conversations/{conversation_id}/pin")
+async def unpin_conv(
+    conversation_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Unpin a conversation."""
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    await unpin_conversation(db, conv_id, user_id)
+    return {"success": True}
+
+
+@router.post("/conversations/{conversation_id}/mute")
+async def mute_conv(
+    conversation_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Mute a conversation."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    await mute_conversation(db, tenant_id, conv_id, user_id)
+    return {"success": True}
+
+
+@router.delete("/conversations/{conversation_id}/mute")
+async def unmute_conv(
+    conversation_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Unmute a conversation."""
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    await unmute_conversation(db, conv_id, user_id)
+    return {"success": True}
+
+
+# ─── Participants ───
+
+@router.post("/conversations/{conversation_id}/participants")
+async def add_participant_endpoint(
+    conversation_id: str,
+    body: ParticipantAdd,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Add a participant to a conversation."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    if not await CommRBAC.can_user_manage(db, conv_id, current_user):
+        raise HTTPException(403, detail={"detail": "Cannot manage participants", "code": "forbidden"})
+    result = await add_participant(db, tenant_id, conv_id, body.participant_id, body.participant_type, body.role)
+    if result is None:
+        raise HTTPException(400, detail={"detail": "Already a participant or invalid ID", "code": "bad_request"})
+    return result
+
+
+@router.delete("/conversations/{conversation_id}/participants/{participant_id}")
+async def remove_participant_endpoint(
+    conversation_id: str,
+    participant_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Remove a participant from a conversation."""
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    pid = _parse_uuid(participant_id, "participant_id")
+    # Self-leave is always allowed
+    if pid != uuid.UUID(current_user["user_id"]):
+        if not await CommRBAC.can_user_manage(db, conv_id, current_user):
+            raise HTTPException(403, detail={"detail": "Cannot manage participants", "code": "forbidden"})
+    success = await remove_participant(db, conv_id, pid)
+    if not success:
+        raise HTTPException(404, detail={"detail": "Participant not found", "code": "not_found"})
+    return {"success": True}
+
+
+@router.patch("/conversations/{conversation_id}/participants/{participant_id}")
+async def change_participant_role(
+    conversation_id: str,
+    participant_id: str,
+    body: ParticipantRoleUpdate,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Change a participant's role."""
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    pid = _parse_uuid(participant_id, "participant_id")
+    if not await CommRBAC.can_user_change_role(db, conv_id, current_user):
+        raise HTTPException(403, detail={"detail": "Cannot change roles", "code": "forbidden"})
+    result = await change_role(db, conv_id, pid, body.role)
+    if result is None:
+        raise HTTPException(404, detail={"detail": "Participant not found", "code": "not_found"})
+    return result
+
+
+# ─── Messages ───
+
+@router.get("/conversations/{conversation_id}/messages")
+async def get_conv_messages(
+    conversation_id: str,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=100),
+    before: str | None = Query(None),
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Get paginated messages for a conversation."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    if not await CommRBAC.is_participant(db, conv_id, user_id):
+        raise HTTPException(403, detail={"detail": "Not a participant", "code": "forbidden"})
+    before_id = _parse_uuid(before, "before") if before else None
+    return await get_messages(db, tenant_id, conv_id, page, page_size, before_id)
+
+
+@router.post("/conversations/{conversation_id}/messages")
+async def send_conv_message(
+    conversation_id: str,
+    body: MessageCreate,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Send a message to a conversation."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    if not await CommRBAC.can_user_write(db, conv_id, current_user):
+        raise HTTPException(403, detail={"detail": "Cannot write to this conversation", "code": "forbidden"})
+    blocks_data = [b.model_dump() for b in body.blocks] if body.blocks else None
+    attachments_data = [a.model_dump() for a in body.attachments] if body.attachments else None
+    return await send_message(
+        db, tenant_id, conv_id, user_id, "user",
+        content=body.content,
+        content_format=body.content_format,
+        blocks=blocks_data,
+        reply_to_id=body.reply_to_id,
+        attachments=attachments_data,
+    )
+
+
+@router.patch("/messages/{message_id}")
+async def update_msg(
+    message_id: str,
+    body: MessageUpdate,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Edit a message or mark as read."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    msg_id = _parse_uuid(message_id, "message_id")
+    if body.content is not None:
+        result = await edit_message(db, tenant_id, msg_id, user_id, body.content)
+        if result is None:
+            raise HTTPException(404, detail={"detail": "Message not found", "code": "not_found"})
+        return result
+    return {"success": True}
+
+
+@router.delete("/messages/{message_id}")
+async def delete_msg(
+    message_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Delete a message."""
+    msg_id = _parse_uuid(message_id, "message_id")
+    success = await delete_message(db, msg_id)
+    if not success:
+        raise HTTPException(404, detail={"detail": "Message not found", "code": "not_found"})
+    return {"success": True}
+
+
+# ─── Attachments ───
+
+@router.post("/messages/{message_id}/attachments")
+async def upload_attachment(
+    message_id: str,
+    file: UploadFile = File(...),
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Upload a file attachment to a message."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    msg_id = _parse_uuid(message_id, "message_id")
+    # Get conversation_id from message
+    from sqlalchemy import select
+    from app.plugins.builtins.kommunikation.models import CommMessage
+    result = await db.execute(select(CommMessage).where(CommMessage.id == msg_id))
+    msg = result.scalar_one_or_none()
+    if msg is None:
+        raise HTTPException(404, detail={"detail": "Message not found", "code": "not_found"})
+    try:
+        return await DmsBridge.store_attachment(db, tenant_id, msg.conversation_id, user_id, file)
+    except ValueError as e:
+        raise HTTPException(413, detail={"detail": str(e), "code": "file_too_large"})
+
+
+# ─── Reactions ───
+
+@router.post("/messages/{message_id}/reactions")
+async def add_msg_reaction(
+    message_id: str,
+    body: ReactionCreate,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Add an emoji reaction to a message."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    msg_id = _parse_uuid(message_id, "message_id")
+    result = await add_reaction(db, tenant_id, msg_id, user_id, body.emoji)
+    if result is None:
+        raise HTTPException(409, detail={"detail": "Already reacted", "code": "conflict"})
+    return result
+
+
+@router.delete("/messages/{message_id}/reactions/{emoji}")
+async def remove_msg_reaction(
+    message_id: str,
+    emoji: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Remove an emoji reaction."""
+    user_id = uuid.UUID(current_user["user_id"])
+    msg_id = _parse_uuid(message_id, "message_id")
+    success = await remove_reaction(db, msg_id, user_id, emoji)
+    if not success:
+        raise HTTPException(404, detail={"detail": "Reaction not found", "code": "not_found"})
+    return {"success": True}
+
+
+# ─── Read State ───
+
+@router.post("/conversations/{conversation_id}/read")
+async def mark_conv_read(
+    conversation_id: str,
+    body: ReadStateUpdate,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Mark a conversation as read."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    await mark_read(db, tenant_id, conv_id, user_id, body.last_read_msg_id)
+    return {"success": True}
+
+
+# ─── Mini-Apps ───
+
+@router.get("/miniapps")
+async def list_miniapps(
+    current_user: dict = Depends(get_current_user),
+):
+    """List available mini-apps."""
+    from app.core.service_container import get_container
+    container = get_container()
+    if not container.has("comm_miniapps"):
+        return {"items": []}
+    registry = container.get("comm_miniapps")
+    return {"items": registry.list_apps()}
+
+
+@router.post("/conversations/{conversation_id}/miniapps")
+async def start_miniapp(
+    conversation_id: str,
+    body: MiniAppStartRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Start a mini-app in a conversation."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    conv_id = _parse_uuid(conversation_id, "conversation_id")
+    if not await CommRBAC.can_user_write(db, conv_id, current_user):
+        raise HTTPException(403, detail={"detail": "Cannot write", "code": "forbidden"})
+    # Create a miniapp block as a message
+    return await send_message(
+        db, tenant_id, conv_id, user_id, "user",
+        content=f"[Mini-App: {body.app_id}]",
+        blocks=[{
+            "block_type": "miniapp",
+            "block_data": {"app_id": body.app_id, "config": body.config},
+        }],
+    )
+
+
+# ─── Content Types ───
+
+@router.get("/block-types")
+async def get_block_types():
+    """List all known content block types."""
+    return {"items": list_block_types()}
+
+
+# ─── WebSocket ───
+
+@router.websocket("/ws")
+async def websocket_endpoint(
+    websocket: WebSocket,
+):
+    """WebSocket endpoint for real-time messaging.
+
+    Authenticates via session cookie. On connect, subscribes user to all their conversations.
+    """
+    # Authenticate via session cookie
+    from app.config import get_settings
+    from app.core.auth import get_session_data, get_redis
+
+    settings = get_settings()
+    session_id = websocket.cookies.get(settings.session_cookie_name)
+    if not session_id:
+        await websocket.close(code=4001, reason="Not authenticated")
+        return
+
+    redis = get_redis()
+    session_data = await get_session_data(redis, session_id)
+    if session_data is None:
+        await websocket.close(code=4001, reason="Session expired")
+        return
+
+    user_id = session_data["user_id"]
+    tenant_id = session_data["tenant_id"]
+
+    # Get WebSocket manager from service container
+    from app.core.service_container import get_container
+    container = get_container()
+    if not container.has("comm_websocket"):
+        await websocket.close(code=4003, reason="Messaging not available")
+        return
+
+    ws_manager = container.get("comm_websocket")
+    await ws_manager.connect(websocket, user_id)
+
+    try:
+        while True:
+            data = await websocket.receive_text()
+            msg = json.loads(data)
+            msg_type = msg.get("type")
+
+            if msg_type == "ping":
+                await ws_manager.send_to_user(user_id, {"type": "pong"})
+            elif msg_type == "subscribe":
+                conv_id = msg.get("conversation_id")
+                if conv_id:
+                    ws_manager.subscribe(conv_id, user_id)
+            elif msg_type == "unsubscribe":
+                conv_id = msg.get("conversation_id")
+                if conv_id:
+                    ws_manager.unsubscribe(conv_id, user_id)
+            elif msg_type == "typing":
+                conv_id = msg.get("conversation_id")
+                is_typing = msg.get("is_typing", False)
+                if conv_id:
+                    await ws_manager.send_to_conversation(
+                        conv_id,
+                        {"type": "typing", "conversation_id": conv_id, "user_id": user_id, "is_typing": is_typing},
+                        exclude_user=user_id,
+                    )
+    except WebSocketDisconnect:
+        await ws_manager.disconnect(websocket, user_id)
+    except Exception:
+        logger.exception("WebSocket error")
+        await ws_manager.disconnect(websocket, user_id)

app/plugins/builtins/kommunikation/schemas.py

@@ -0,0 +1,177 @@
+"""Pydantic schemas for the kommunikation plugin API."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+# ─── Conversation Schemas ───
+
+class ParticipantResponse(BaseModel):
+    id: str
+    conversation_id: str
+    participant_id: str | None = None
+    participant_type: str
+    display_name: str | None = None
+    role: str
+    joined_at: str | None = None
+
+
+class ConversationCreate(BaseModel):
+    title: str | None = None
+    participant_ids: list[str] = Field(default_factory=list)
+    participant_types: list[str] = Field(default_factory=lambda: ["user"])
+    initial_message: str | None = None
+    is_direct: bool = False
+
+
+class ConversationUpdate(BaseModel):
+    title: str | None = None
+    is_archived: bool | None = None
+
+
+class ConversationResponse(BaseModel):
+    id: str
+    title: str | None = None
+    is_locked: bool = False
+    locked_by: str | None = None
+    is_direct: bool = False
+    is_archived: bool = False
+    is_pinned: bool = False
+    created_by: str | None = None
+    created_by_type: str = "user"
+    last_msg_at: str | None = None
+    last_msg_preview: str | None = None
+    last_msg_sender_type: str | None = None
+    participants: list[ParticipantResponse] = Field(default_factory=list)
+    unread_count: int = 0
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class ConversationListResponse(BaseModel):
+    items: list[ConversationResponse]
+    total: int
+
+
+# ─── Participant Management ───
+
+class ParticipantAdd(BaseModel):
+    participant_id: str
+    participant_type: str = "user"
+    role: str = "member"
+
+
+class ParticipantRoleUpdate(BaseModel):
+    role: str = Field(..., pattern="^(admin|member|reader)$")
+
+
+# ─── Message Schemas ───
+
+class MessageBlockCreate(BaseModel):
+    block_type: str
+    block_data: dict[str, Any]
+
+
+class MessageBlockResponse(BaseModel):
+    id: str
+    block_type: str
+    block_data: dict[str, Any]
+    sort_order: int = 0
+
+
+class AttachmentCreate(BaseModel):
+    file_id: str | None = None
+    file_source: str = "comm"
+
+
+class AttachmentResponse(BaseModel):
+    id: str
+    file_id: str | None = None
+    file_source: str = "comm"
+    file_name: str
+    file_type: str
+    file_size: int | None = None
+    thumbnail_path: str | None = None
+
+
+class MessageCreate(BaseModel):
+    content: str = ""
+    content_format: str = "text"
+    blocks: list[MessageBlockCreate] = Field(default_factory=list)
+    reply_to_id: str | None = None
+    attachments: list[AttachmentCreate] = Field(default_factory=list)
+
+
+class MessageUpdate(BaseModel):
+    content: str | None = None
+    read: bool | None = None
+
+
+class MessageResponse(BaseModel):
+    id: str
+    conversation_id: str
+    sender_id: str | None = None
+    sender_type: str
+    content: str = ""
+    content_format: str = "text"
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    reply_to_id: str | None = None
+    is_pinned: bool = False
+    created_at: str | None = None
+    edited_at: str | None = None
+    blocks: list[MessageBlockResponse] = Field(default_factory=list)
+    attachments: list[AttachmentResponse] = Field(default_factory=list)
+    reactions: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class MessageListResponse(BaseModel):
+    items: list[MessageResponse]
+    total: int
+    page: int = 1
+    has_more: bool = False
+
+
+# ─── Reaction Schemas ───
+
+class ReactionCreate(BaseModel):
+    emoji: str
+
+
+class ReactionResponse(BaseModel):
+    id: str
+    message_id: str
+    user_id: str
+    emoji: str
+
+
+# ─── Read State ───
+
+class ReadStateUpdate(BaseModel):
+    last_read_msg_id: str | None = None
+
+
+# ─── Mini-App Schemas ───
+
+class MiniAppResponse(BaseModel):
+    app_id: str
+    name: str
+    icon: str
+    description: str
+    plugin_name: str
+    render_schema: dict[str, Any] = Field(default_factory=dict)
+
+
+class MiniAppStartRequest(BaseModel):
+    app_id: str
+    config: dict[str, Any] = Field(default_factory=dict)
+
+
+# ─── WebSocket Message Schemas ───
+
+class WSMessage(BaseModel):
+    type: str
+    data: dict[str, Any] = Field(default_factory=dict)

app/plugins/builtins/kommunikation/search_provider.py

@@ -0,0 +1,112 @@
+"""Search provider for unified_search integration."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy import select, func, or_
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.plugins.builtins.kommunikation.models import (
+    CommConversation,
+    CommMessage,
+    CommParticipant,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class CommSearchProvider:
+    """Provider for unified_search — searches conversations and messages."""
+
+    async def search(
+        self,
+        db: AsyncSession,
+        tenant_id: uuid.UUID,
+        user_id: uuid.UUID,
+        query: str,
+        limit: int = 20,
+    ) -> list[dict[str, Any]]:
+        """Search in messages and conversations the user has access to."""
+        results: list[dict[str, Any]] = []
+
+        # Get user's conversation IDs
+        conv_result = await db.execute(
+            select(CommParticipant.conversation_id).where(
+                CommParticipant.participant_id == user_id,
+                CommParticipant.participant_type == "user",
+                CommParticipant.left_at.is_(None),
+                CommParticipant.tenant_id == tenant_id,
+            )
+        )
+        conv_ids = [row[0] for row in conv_result.fetchall()]
+
+        if not conv_ids:
+            return results
+
+        # Search in messages
+        msg_result = await db.execute(
+            select(CommMessage, CommConversation.title)
+            .join(CommConversation, CommConversation.id == CommMessage.conversation_id)
+            .where(
+                CommMessage.conversation_id.in_(conv_ids),
+                CommMessage.tenant_id == tenant_id,
+                CommMessage.deleted_at.is_(None),
+                CommMessage.content.ilike(f"%{query}%"),
+            )
+            .order_by(CommMessage.created_at.desc())
+            .limit(limit)
+        )
+
+        for msg, conv_title in msg_result.fetchall():
+            # Build snippet around match
+            content = msg.content or ""
+            idx = content.lower().find(query.lower())
+            if idx >= 0:
+                start = max(0, idx - 30)
+                end = min(len(content), idx + len(query) + 30)
+                snippet = ("..." if start > 0 else "") + content[start:end] + ("..." if end < len(content) else "")
+            else:
+                snippet = content[:100]
+
+            results.append({
+                "type": "message",
+                "id": str(msg.id),
+                "conversation_id": str(msg.conversation_id),
+                "conversation_title": conv_title,
+                "content": msg.content,
+                "sender_type": msg.sender_type,
+                "created_at": msg.created_at.isoformat() if msg.created_at else None,
+                "snippet": snippet,
+            })
+
+        # Search in conversation titles
+        conv_title_result = await db.execute(
+            select(CommConversation).where(
+                CommConversation.id.in_(conv_ids),
+                CommConversation.tenant_id == tenant_id,
+                CommConversation.deleted_at.is_(None),
+                CommConversation.title.ilike(f"%{query}%"),
+            )
+            .limit(limit)
+        )
+
+        for conv in conv_title_result.scalars().all():
+            results.append({
+                "type": "conversation",
+                "id": str(conv.id),
+                "title": conv.title,
+                "last_msg_at": conv.last_msg_at.isoformat() if conv.last_msg_at else None,
+            })
+
+        return results
+
+    async def index_message(self, message: dict[str, Any]) -> None:
+        """Index a message for search (placeholder for future full-text indexing)."""
+        pass
+
+    async def reindex_all(self, db: AsyncSession, tenant_id: uuid.UUID) -> None:
+        """Full reindex (placeholder for future full-text indexing)."""
+        pass

app/plugins/builtins/kommunikation/websocket_manager.py

@@ -0,0 +1,89 @@
+"""WebSocket connection manager for the kommunikation plugin."""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from fastapi import WebSocket
+
+logger = logging.getLogger(__name__)
+
+
+class WebSocketManager:
+    """Manages WebSocket connections per user for real-time messaging."""
+
+    def __init__(self) -> None:
+        # user_id (str) → list of WebSocket connections
+        self._connections: dict[str, list[WebSocket]] = {}
+        # conversation_id (str) → set of user_ids subscribed
+        self._subscriptions: dict[str, set[str]] = {}
+
+    async def connect(self, websocket: WebSocket, user_id: str) -> None:
+        """Accept and register a new WebSocket connection."""
+        await websocket.accept()
+        if user_id not in self._connections:
+            self._connections[user_id] = []
+        self._connections[user_id].append(websocket)
+        logger.debug(f"WebSocket connected: user={user_id}, total={len(self._connections[user_id])}")
+
+    async def disconnect(self, websocket: WebSocket, user_id: str) -> None:
+        """Remove a WebSocket connection."""
+        conns = self._connections.get(user_id, [])
+        if websocket in conns:
+            conns.remove(websocket)
+        if not conns:
+            self._connections.pop(user_id, None)
+            # Remove from all subscriptions
+            for conv_id, users in self._subscriptions.items():
+                users.discard(user_id)
+        logger.debug(f"WebSocket disconnected: user={user_id}, remaining={len(conns)}")
+
+    def subscribe(self, conversation_id: str, user_id: str) -> None:
+        """Subscribe a user to a conversation's updates."""
+        if conversation_id not in self._subscriptions:
+            self._subscriptions[conversation_id] = set()
+        self._subscriptions[conversation_id].add(user_id)
+
+    def unsubscribe(self, conversation_id: str, user_id: str) -> None:
+        """Unsubscribe a user from a conversation."""
+        if conversation_id in self._subscriptions:
+            self._subscriptions[conversation_id].discard(user_id)
+
+    async def send_to_user(self, user_id: str, message: dict[str, Any]) -> None:
+        """Send a message to all connections of a specific user."""
+        conns = self._connections.get(user_id, [])
+        text = json.dumps(message, default=str)
+        for ws in conns:
+            try:
+                await ws.send_text(text)
+            except Exception:
+                logger.warning(f"Failed to send to user {user_id}, removing connection")
+                await self.disconnect(ws, user_id)
+
+    async def send_to_conversation(
+        self,
+        conversation_id: str,
+        message: dict[str, Any],
+        exclude_user: str | None = None,
+    ) -> None:
+        """Send a message to all users subscribed to a conversation."""
+        user_ids = self._subscriptions.get(conversation_id, set())
+        for user_id in list(user_ids):
+            if exclude_user and user_id == exclude_user:
+                continue
+            await self.send_to_user(user_id, message)
+
+    async def broadcast(self, message: dict[str, Any]) -> None:
+        """Broadcast a message to all connected users."""
+        for user_id in list(self._connections.keys()):
+            await self.send_to_user(user_id, message)
+
+    def get_online_users(self) -> list[str]:
+        """Get list of currently connected user IDs."""
+        return list(self._connections.keys())
+
+    def is_user_online(self, user_id: str) -> bool:
+        """Check if a user has any active connections."""
+        return user_id in self._connections and len(self._connections[user_id]) > 0