T03: plugin system framework + lifecycle + migrations + event bus + DI

- Plugin registry with discover/install/activate/deactivate/uninstall lifecycle
- PluginManifest Pydantic v2 schema (name, version, dependencies, routes, events, migrations)
- BasePlugin abstract class with lifecycle hooks (on_install/activate/deactivate/uninstall)
- Migration runner with tenant_id validator (rejects tables without tenant_id)
- Event bus integration: register/unregister listeners on activate/deactivate
- Service container DI: plugins receive db, cache, event_bus, storage, notifications
- Idempotent operations (activate active=200, deactivate inactive=200)
- UI registry for frontend component registration
- 47 new tests (14 ACs + 33 unit tests), 103 total tests pass
- Migration 0003: plugins + plugin_migrations tables
- Coverage: 85.92% for plugin modules

alembic/versions/0003_plugin_system.py

@@ -0,0 +1,55 @@
+"""T03: plugins + plugin_migrations tables.
+
+Revision ID: 0003_plugin_system
+Revises: 0002_contacts_fts
+Create Date: 2026-06-29
+"""
+
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0003_plugin_system"
+down_revision: Union[str, None] = "0002_contacts_fts"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # --- plugins table (system-wide, NOT tenant-scoped) ---
+    op.create_table(
+        "plugins",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column("name", sa.String(80), nullable=False, unique=True),
+        sa.Column("display_name", sa.String(120), nullable=False),
+        sa.Column("version", sa.String(40), nullable=False),
+        sa.Column("status", sa.String(20), nullable=False, server_default="installed"),
+        sa.Column("installed", sa.Boolean, nullable=False, server_default=sa.text("true")),
+        sa.Column("active", sa.Boolean, nullable=False, server_default=sa.text("false")),
+        sa.Column("config", sa.Text, nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+    op.create_index("ix_plugins_name", "plugins", ["name"], unique=True)
+
+    # --- plugin_migrations table (tracks which migrations have been applied) ---
+    op.create_table(
+        "plugin_migrations",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column("plugin_name", sa.String(80), nullable=False),
+        sa.Column("migration_file", sa.String(255), nullable=False),
+        sa.Column("status", sa.String(20), nullable=False, server_default="applied"),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.UniqueConstraint("plugin_name", "migration_file", name="ix_plugin_migrations_unique"),
+    )
+    op.create_index("ix_plugin_migrations_plugin", "plugin_migrations", ["plugin_name"])
+
+
+def downgrade() -> None:
+    op.drop_table("plugin_migrations")
+    op.drop_table("plugins")

app/main.py

@@ -9,14 +9,26 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from app.config import get_settings
 from app.core.middleware import CSRFMiddleware
-from app.core.db import close_engine
-from app.routes import auth, users, roles, tenants, health, notifications, companies, contacts, import_export
+from app.core.db import close_engine, get_engine
+from app.core.service_container import get_container
+from app.plugins.registry import get_registry
+from app.routes import auth, users, roles, tenants, health, notifications, companies, contacts, import_export, plugins
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Application lifespan: startup and shutdown."""
+    # Initialize service container
+    container = get_container()
+    await container.initialize()
+
+    # Initialize plugin registry and discover built-in plugins
+    registry = get_registry()
+    registry.initialize(get_engine(), app)
+    registry.discover_builtins()
+
     yield
+
     await close_engine()
 
 
@@ -44,6 +56,7 @@ def create_app() -> FastAPI:
     app.include_router(companies.router)
     app.include_router(contacts.router)
     app.include_router(import_export.router)
+    app.include_router(plugins.router)
 
     return app
 

app/models/__init__.py

@@ -9,10 +9,12 @@ from app.models.notification import Notification
 from app.models.auth import PasswordResetToken, ApiToken
 from app.models.company import Company
 from app.models.contact import Contact, CompanyContact
+from app.models.plugin import Plugin, PluginMigration
 
 __all__ = [
     "Tenant", "User", "UserTenant", "Role", "Session",
     "AuditLog", "DeletionLog", "Notification",
     "PasswordResetToken", "ApiToken", "Company",
     "Contact", "CompanyContact",
+    "Plugin", "PluginMigration",
 ]

app/models/plugin.py

@@ -0,0 +1,68 @@
+"""Plugin and PluginMigration models — tracks installed plugins and their migrations."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import String, Boolean, Text, DateTime, func, Index
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, TimestampMixin
+
+
+class Plugin(Base, TimestampMixin):
+    """Plugin record — tracks installation and activation status.
+
+    This table is NOT tenant-scoped (plugins are system-wide, not per-tenant).
+    The tables *created by* plugin migrations MUST have tenant_id.
+    """
+
+    __tablename__ = "plugins"
+    __table_args__ = (
+        Index("ix_plugins_name", "name", unique=True),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    name: Mapped[str] = mapped_column(String(80), nullable=False, unique=True)
+    display_name: Mapped[str] = mapped_column(String(120), nullable=False)
+    version: Mapped[str] = mapped_column(String(40), nullable=False)
+    status: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="installed"
+    )
+    installed: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True
+    )
+    active: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
+    config: Mapped[dict[str, Any] | None] = mapped_column(
+        Text, nullable=True  # JSON string for plugin configuration
+    )
+
+    # Transient attribute for response (not persisted)
+    # Used by uninstall to report dropped tables
+    # This is set dynamically by registry.uninstall()
+
+
+class PluginMigration(Base, TimestampMixin):
+    """Tracks individual migration files applied for each plugin."""
+
+    __tablename__ = "plugin_migrations"
+    __table_args__ = (
+        Index("ix_plugin_migrations_plugin", "plugin_name"),
+        Index("ix_plugin_migrations_unique", "plugin_name", "migration_file", unique=True),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    plugin_name: Mapped[str] = mapped_column(String(80), nullable=False)
+    migration_file: Mapped[str] = mapped_column(String(255), nullable=False)
+    status: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="applied"
+    )

app/plugins/__init__.py

@@ -0,0 +1,8 @@
+"""Plugin system framework for LeoCRM."""
+
+from app.plugins.manifest import PluginManifest
+from app.plugins.base import BasePlugin
+from app.plugins.registry import get_registry
+from app.plugins.migration_runner import MigrationRunner
+
+__all__ = ["PluginManifest", "BasePlugin", "get_registry", "MigrationRunner"]

app/plugins/base.py

@@ -0,0 +1,125 @@
+"""Base plugin abstract class with lifecycle hooks."""
+
+from __future__ import annotations
+
+from abc import ABC
+from typing import TYPE_CHECKING, Any
+
+from fastapi import APIRouter
+
+from app.plugins.manifest import PluginManifest
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+    from app.core.event_bus import EventBus
+    from app.core.service_container import ServiceContainer
+
+
+class BasePlugin(ABC):
+    """Abstract base class for all LeoCRM plugins.
+
+    Subclasses must define a ``manifest`` class attribute (PluginManifest)
+    and implement the lifecycle hooks.
+    """
+
+    manifest: PluginManifest
+
+    def __init__(self) -> None:
+        if not hasattr(self, "manifest") or self.manifest is None:
+            raise ValueError(f"{self.__class__.__name__} must define a 'manifest' attribute")
+        self._event_handlers: dict[str, Any] = {}
+        self._routers: list[APIRouter] = []
+
+    # ─── Lifecycle Hooks ───
+
+    async def on_install(self, db: AsyncSession, service_container: ServiceContainer) -> None:
+        """Called when the plugin is installed (after migrations are run).
+
+        Override to perform seed data or initial setup.
+        """
+        return None
+
+    async def on_activate(
+        self, db: AsyncSession, service_container: ServiceContainer, event_bus: EventBus
+    ) -> None:
+        """Called when the plugin is activated.
+
+        Override to register event listeners and prepare runtime state.
+        Default implementation subscribes to events listed in the manifest.
+        """
+        for event_name in self.manifest.events:
+            handler = self._make_event_handler(event_name)
+            self._event_handlers[event_name] = handler
+            event_bus.subscribe(event_name, handler)
+
+    async def on_deactivate(
+        self, db: AsyncSession, service_container: ServiceContainer, event_bus: EventBus
+    ) -> None:
+        """Called when the plugin is deactivated.
+
+        Override to clean up runtime state. Default implementation unsubscribes
+        all event listeners that were registered during activation.
+        """
+        for event_name, handler in self._event_handlers.items():
+            event_bus.unsubscribe(event_name, handler)
+        self._event_handlers.clear()
+
+    async def on_uninstall(self, db: AsyncSession, service_container: ServiceContainer) -> None:
+        """Called when the plugin is uninstalled (before data tables are dropped).
+
+        Override to clean up any external resources.
+        """
+        return None
+
+    # ─── Route Registration ───
+
+    def get_routes(self) -> list[APIRouter]:
+        """Return list of APIRouter instances to mount on the FastAPI app.
+
+        Default implementation loads routers from manifest route definitions
+        by importing the module and getting the specified attribute.
+        """
+        if self._routers:
+            return self._routers
+
+        routers: list[APIRouter] = []
+        for route_def in self.manifest.routes:
+            import importlib
+            module = importlib.import_module(route_def.module)
+            router: APIRouter = getattr(module, route_def.router_attr)
+            routers.append(router)
+        self._routers = routers
+        return routers
+
+    # ─── Event Handler Factory ───
+
+    def _make_event_handler(self, event_name: str) -> Any:
+        """Create an event handler coroutine for the given event name.
+
+        Looks for a method named ``on_<event_name>`` with dots replaced by underscores.
+        Falls back to ``on_event`` if it exists, otherwise uses a no-op handler.
+        """
+        method_name = f"on_{event_name.replace('.', '_')}"
+        specific = getattr(self, method_name, None)
+        if specific is not None:
+            return specific
+        fallback = getattr(self, "on_event", None)
+        if fallback is not None:
+            return fallback
+        # Default no-op handler
+        async def _noop_handler(payload: dict[str, Any]) -> None:
+            pass
+        return _noop_handler
+
+    # ─── Utility ───
+
+    @property
+    def name(self) -> str:
+        return self.manifest.name
+
+    @property
+    def version(self) -> str:
+        return self.manifest.version
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} name={self.manifest.name} version={self.manifest.version}>"

app/plugins/builtins/__init__.py

@@ -0,0 +1,5 @@
+"""Built-in plugins directory.
+
+Each module in this package that exports a BasePlugin subclass will be
+discovered automatically by the plugin registry on application startup.
+"""

app/plugins/builtins/migrations/0001_bad_migration.sql

@@ -0,0 +1,7 @@
+-- Bad migration: creates table WITHOUT tenant_id (should be rejected by validator)
+CREATE TABLE IF NOT EXISTS plugin_bad_table (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    title VARCHAR(200) NOT NULL,
+    content TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);

app/plugins/builtins/migrations/0001_test_plugin.sql

@@ -0,0 +1,10 @@
+-- Test plugin migration: creates plugin_test_data table with tenant_id
+CREATE TABLE IF NOT EXISTS plugin_test_data (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    title VARCHAR(200) NOT NULL,
+    content TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS ix_plugin_test_data_tenant ON plugin_test_data(tenant_id);

app/plugins/builtins/test_sample.py

@@ -0,0 +1,52 @@
+"""Sample test plugin for LeoCRM plugin system testing."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from app.plugins.base import BasePlugin
+from app.plugins.manifest import PluginManifest
+
+
+class TestSamplePlugin(BasePlugin):
+    """A sample plugin for testing the plugin lifecycle."""
+
+    manifest = PluginManifest(
+        name="test_sample",
+        version="1.0.0",
+        display_name="Test Sample Plugin",
+        description="A sample plugin for testing install/activate/deactivate/uninstall lifecycle.",
+        dependencies=[],
+        routes=[],
+        events=["company.created", "contact.created"],
+        migrations=["0001_test_plugin.sql"],
+        permissions=[],
+    )
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.install_called = False
+        self.activate_called = False
+        self.deactivate_called = False
+        self.uninstall_called = False
+        self.event_log: list[dict[str, Any]] = []
+
+    async def on_install(self, db, service_container) -> None:
+        self.install_called = True
+
+    async def on_activate(self, db, service_container, event_bus) -> None:
+        self.activate_called = True
+        await super().on_activate(db, service_container, event_bus)
+
+    async def on_deactivate(self, db, service_container, event_bus) -> None:
+        self.deactivate_called = True
+        await super().on_deactivate(db, service_container, event_bus)
+
+    async def on_uninstall(self, db, service_container) -> None:
+        self.uninstall_called = True
+
+    async def on_company_created(self, payload: dict[str, Any]) -> None:
+        self.event_log.append({"event": "company.created", "payload": payload})
+
+    async def on_contact_created(self, payload: dict[str, Any]) -> None:
+        self.event_log.append({"event": "contact.created", "payload": payload})

app/plugins/manifest.py

@@ -0,0 +1,70 @@
+"""Plugin manifest schema (Pydantic v2)."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field, field_validator
+
+
+class PluginRouteDef(BaseModel):
+    """Route definition within a plugin manifest."""
+
+    path: str = Field(..., description="URL path prefix, e.g. /api/v1/plugin-mail")
+    module: str = Field(..., description="Dotted path to the module containing the APIRouter")
+    router_attr: str = Field(default="router", description="Attribute name of the APIRouter in the module")
+
+
+class PluginManifest(BaseModel):
+    """Manifest describing a plugin's metadata, dependencies, and capabilities."""
+
+    name: str = Field(..., min_length=1, max_length=80, description="Unique plugin identifier (snake_case)")
+    version: str = Field(..., min_length=1, max_length=40, description="Semantic version string")
+    display_name: str = Field(..., min_length=1, max_length=120)
+    description: str = Field(default="", max_length=500)
+    dependencies: list[str] = Field(default_factory=list, description="Other plugin names this plugin depends on")
+    routes: list[PluginRouteDef] = Field(default_factory=list, description="Route definitions to register on activation")
+    events: list[str] = Field(default_factory=list, description="Event names this plugin listens to")
+    migrations: list[str] = Field(default_factory=list, description="Migration file names (ordered, e.g. 0001_initial.sql)")
+    permissions: list[str] = Field(default_factory=list, description="Required permissions for this plugin")
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, v: str) -> str:
+        if not v.replace("_", "").isalnum():
+            raise ValueError("Plugin name must be alphanumeric with underscores only")
+        return v.lower()
+
+    model_config = {"extra": "forbid"}
+
+
+class ManifestSchemaResponse(BaseModel):
+    """Response model describing the manifest schema for API consumers."""
+
+    fields: dict[str, dict[str, str]]
+    example: PluginManifest
+
+
+# Pre-built schema documentation for GET /api/v1/plugins/manifest endpoint
+MANIFEST_SCHEMA_DOC = ManifestSchemaResponse(
+    fields={
+        "name": {"type": "str", "required": "true", "description": "Unique plugin identifier (snake_case, max 80 chars)"},
+        "version": {"type": "str", "required": "true", "description": "Semantic version string"},
+        "display_name": {"type": "str", "required": "true", "description": "Human-readable plugin name"},
+        "description": {"type": "str", "required": "false", "description": "Plugin description (max 500 chars)"},
+        "dependencies": {"type": "list[str]", "required": "false", "description": "Other plugin names required"},
+        "routes": {"type": "list[PluginRouteDef]", "required": "false", "description": "Route definitions to register"},
+        "events": {"type": "list[str]", "required": "false", "description": "Event names to listen to"},
+        "migrations": {"type": "list[str]", "required": "false", "description": "Migration file names (ordered)"},
+        "permissions": {"type": "list[str]", "required": "false", "description": "Required permissions"},
+    },
+    example=PluginManifest(
+        name="example_plugin",
+        version="1.0.0",
+        display_name="Example Plugin",
+        description="An example plugin demonstrating the manifest schema.",
+        dependencies=[],
+        routes=[],
+        events=["company.created", "contact.created"],
+        migrations=["0001_initial.sql"],
+        permissions=["companies.read"],
+    ),
+)

app/plugins/migration_runner.py

@@ -0,0 +1,282 @@
+"""Plugin DB migration runner with tenant_id column validation."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import text, inspect
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
+
+from app.models.plugin import PluginMigration
+
+
+class MigrationValidationError(Exception):
+    """Raised when a plugin migration creates a table without tenant_id column."""
+
+
+class MigrationRunner:
+    """Runs plugin SQL migrations and validates that all created tables have tenant_id."""
+
+    def __init__(self, engine: AsyncEngine, migrations_dir: str | None = None) -> None:
+        self._engine = engine
+        self._migrations_dir = Path(migrations_dir or os.path.join(os.path.dirname(__file__), "migrations"))
+
+    def _resolve_migration_path(self, filename: str) -> Path:
+        """Resolve a migration filename to an absolute path."""
+        # Try migrations_dir first, then app/plugins/migrations/
+        candidate = self._migrations_dir / filename
+        if candidate.exists():
+            return candidate
+        # Fallback: builtins directory
+        builtins_dir = Path(__file__).parent / "builtins" / "migrations"
+        candidate2 = builtins_dir / filename
+        if candidate2.exists():
+            return candidate2
+        raise FileNotFoundError(f"Migration file not found: {filename}")
+
+    async def run_migration(
+        self,
+        db: AsyncSession,
+        plugin_name: str,
+        migration_filename: str,
+        tenant_id: Any | None = None,
+    ) -> PluginMigration:
+        """Run a single SQL migration file for a plugin.
+
+        Reads the SQL file, executes it, validates created tables have tenant_id,
+        and records the migration in plugin_migrations table.
+
+        Args:
+            db: Async database session.
+            plugin_name: Name of the plugin being migrated.
+            migration_filename: Filename of the migration SQL file.
+            tenant_id: Optional tenant UUID for tenant-scoped migrations.
+
+        Returns:
+            PluginMigration record.
+
+        Raises:
+            FileNotFoundError: If migration file doesn't exist.
+            MigrationValidationError: If a created table lacks tenant_id column.
+        """
+        migration_path = self._resolve_migration_path(migration_filename)
+        sql_content = migration_path.read_text(encoding="utf-8")
+
+        # Capture existing table names before migration (static analysis of SQL)
+        existing_tables = await self._get_table_names_via_session(db)
+
+        # Execute the migration SQL
+        # Split on semicolons but handle potential function definitions
+        statements = self._split_sql(sql_content)
+        for stmt in statements:
+            stmt = stmt.strip()
+            if stmt:
+                await db.execute(text(stmt))
+        await db.flush()
+
+        # Capture table names after migration (using same session connection)
+        new_tables = await self._get_table_names_via_session(db)
+        created_tables = new_tables - existing_tables
+
+        # Validate: all newly created tables must have tenant_id column
+        tables_without_tenant = []
+        for table_name in created_tables:
+            if not await self._table_has_column_via_session(db, table_name, "tenant_id"):
+                tables_without_tenant.append(table_name)
+
+        if tables_without_tenant:
+            # Rollback the migration
+            await db.rollback()
+            raise MigrationValidationError(
+                f"Plugin '{plugin_name}' migration '{migration_filename}' created tables without tenant_id column: "
+                f"{', '.join(tables_without_tenant)}. All plugin tables must have tenant_id for multi-tenant isolation."
+            )
+
+        # Record the migration in plugin_migrations table
+        migration_record = PluginMigration(
+            plugin_name=plugin_name,
+            migration_file=migration_filename,
+            status="applied",
+        )
+        db.add(migration_record)
+        await db.flush()
+
+        return migration_record
+
+    async def run_all_migrations(
+        self,
+        db: AsyncSession,
+        plugin_name: str,
+        migration_files: list[str],
+        tenant_id: Any | None = None,
+    ) -> list[PluginMigration]:
+        """Run all migration files for a plugin in order.
+
+        Skips already-applied migrations (idempotent).
+        """
+        # Check which migrations are already applied
+        from sqlalchemy import select
+
+        result = await db.execute(
+            select(PluginMigration).where(
+                PluginMigration.plugin_name == plugin_name,
+                PluginMigration.status == "applied",
+            )
+        )
+        applied_files = {row.migration_file for row in result.scalars().all()}
+
+        records: list[PluginMigration] = []
+        for filename in migration_files:
+            if filename in applied_files:
+                continue  # Already applied — idempotent skip
+            record = await self.run_migration(db, plugin_name, filename, tenant_id)
+            records.append(record)
+
+        return records
+
+    async def drop_plugin_tables(self, db: AsyncSession, plugin_name: str) -> list[str]:
+        """Drop all tables that were created by a plugin's migrations.
+
+        Args:
+            db: Async database session.
+            plugin_name: Name of the plugin to remove tables for.
+
+        Returns:
+            List of dropped table names.
+        """
+        from sqlalchemy import select
+
+        # Get all migration records for this plugin
+        result = await db.execute(
+            select(PluginMigration).where(PluginMigration.plugin_name == plugin_name)
+        )
+        migration_records = result.scalars().all()
+
+        dropped_tables: list[str] = []
+        for record in migration_records:
+            migration_path = self._resolve_migration_path(record.migration_file)
+            sql_content = migration_path.read_text(encoding="utf-8")
+
+            # Parse table names from CREATE TABLE statements
+            table_names = self._extract_table_names(sql_content)
+            for table_name in table_names:
+                try:
+                    await db.execute(text(f'DROP TABLE IF EXISTS "{table_name}" CASCADE'))
+                    dropped_tables.append(table_name)
+                except Exception:
+                    pass  # Table may already be gone
+            await db.flush()
+
+        # Remove migration records
+        for record in migration_records:
+            await db.delete(record)
+        await db.flush()
+
+        return dropped_tables
+
+    async def _get_table_names_via_session(self, db: AsyncSession) -> set[str]:
+        """Get current table names using the session's own connection.
+
+        This ensures we see uncommitted DDL changes within the same transaction.
+        """
+        result = await db.execute(
+            text("SELECT tablename FROM pg_tables WHERE schemaname = 'public'")
+        )
+        return {row[0] for row in result.fetchall()}
+
+    async def _table_has_column_via_session(
+        self, db: AsyncSession, table_name: str, column_name: str
+    ) -> bool:
+        """Check if a table has a specific column using the session's connection."""
+        result = await db.execute(
+            text(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_schema = 'public' AND table_name = :table_name AND column_name = :column_name"
+            ),
+            {"table_name": table_name, "column_name": column_name},
+        )
+        return result.fetchone() is not None
+
+    async def _get_table_names(self) -> set[str]:
+        """Get current table names from the database (separate connection)."""
+        def _get_names(sync_conn):
+            insp = inspect(sync_conn)
+            return set(insp.get_table_names())
+
+        async with self._engine.connect() as conn:
+            return await conn.run_sync(_get_names)
+
+    async def _table_has_column(self, table_name: str, column_name: str) -> bool:
+        """Check if a table has a specific column (separate connection)."""
+        def _has_col(sync_conn):
+            insp = inspect(sync_conn)
+            if table_name not in insp.get_table_names():
+                return False
+            columns = [col["name"] for col in insp.get_columns(table_name)]
+            return column_name in columns
+
+        async with self._engine.connect() as conn:
+            return await conn.run_sync(_has_col)
+
+    @staticmethod
+    def _split_sql(sql: str) -> list[str]:
+        """Split SQL into individual statements, respecting basic semicolon boundaries."""
+        statements: list[str] = []
+        current: list[str] = []
+        in_dollar_quote = False
+        dollar_tag = ""
+
+        for line in sql.splitlines():
+            stripped = line.strip()
+
+            # Handle dollar-quoted blocks (PostgreSQL function bodies)
+            if "$$" in stripped and not in_dollar_quote:
+                parts = stripped.split("$$", 1)
+                if len(parts) > 1:
+                    dollar_tag = parts[0].strip()
+                    if dollar_tag:
+                        in_dollar_quote = True
+                    elif stripped.count("$$") >= 2:
+                        # Single-line $$ block
+                        current.append(line)
+                        if stripped.rstrip().endswith(";"):
+                            statements.append("\n".join(current))
+                            current = []
+                        continue
+                    else:
+                        in_dollar_quote = True
+                current.append(line)
+                continue
+
+            if in_dollar_quote:
+                current.append(line)
+                if "$$" in stripped:
+                    in_dollar_quote = False
+                    # After closing $$, check if the line ends with ; to flush statement
+                    after_dollar = stripped.split("$$", 1)[-1] if "$$" in stripped else ""
+                    if after_dollar.rstrip().endswith(";"):
+                        statements.append("\n".join(current))
+                        current = []
+                continue
+
+            current.append(line)
+            if stripped.endswith(";"):
+                statements.append("\n".join(current))
+                current = []
+
+        if current:
+            remaining = "\n".join(current).strip()
+            if remaining:
+                statements.append(remaining)
+
+        return statements
+
+    @staticmethod
+    def _extract_table_names(sql: str) -> list[str]:
+        """Extract table names from CREATE TABLE statements in SQL."""
+        import re
+        # Match CREATE TABLE [IF NOT EXISTS] "table_name" or CREATE TABLE table_name
+        pattern = r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?["\']?(\w+)["\']?'
+        return re.findall(pattern, sql, re.IGNORECASE)

app/plugins/registry.py

@@ -0,0 +1,385 @@
+"""Plugin registry — manages discovered, installed, and active plugins."""
+
+from __future__ import annotations
+
+import importlib
+import logging
+from typing import Any
+
+from fastapi import FastAPI
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, AsyncEngine
+
+from app.core.event_bus import EventBus, get_event_bus
+from app.core.service_container import ServiceContainer, get_container
+from app.models.plugin import Plugin as PluginModel
+from app.plugins.base import BasePlugin
+from app.plugins.migration_runner import MigrationRunner
+
+logger = logging.getLogger(__name__)
+
+
+class PluginRegistry:
+    """Registry for managing plugin lifecycle.
+
+    Maintains in-memory state for discovered plugins and their runtime instances,
+    backed by database records for persistent status tracking.
+    """
+
+    def __init__(self) -> None:
+        # name -> BasePlugin instance (discovered/loaded)
+        self._plugins: dict[str, BasePlugin] = {}
+        # name -> PluginModel DB record (installed status)
+        self._db_status: dict[str, PluginModel] = {}
+        # name -> list of mounted APIRouter references (for unregistration)
+        self._mounted_routers: dict[str, list[Any]] = {}
+        self._engine: AsyncEngine | None = None
+        self._event_bus: EventBus = get_event_bus()
+        self._container: ServiceContainer = get_container()
+        self._migration_runner: MigrationRunner | None = None
+        self._app: FastAPI | None = None
+        self._initialized = False
+
+    def initialize(self, engine: AsyncEngine, app: FastAPI | None = None) -> None:
+        """Initialize the registry with the DB engine and optional FastAPI app."""
+        self._engine = engine
+        self._app = app
+        self._migration_runner = MigrationRunner(engine)
+        self._initialized = True
+
+    @property
+    def migration_runner(self) -> MigrationRunner:
+        if self._migration_runner is None:
+            raise RuntimeError("Registry not initialized — call initialize() first")
+        return self._migration_runner
+
+    @property
+    def engine(self) -> AsyncEngine:
+        if self._engine is None:
+            raise RuntimeError("Registry not initialized — call initialize() first")
+        return self._engine
+
+    # ─── Discovery ───
+
+    def discover_builtins(self) -> list[str]:
+        """Discover and load built-in plugins from app.plugins.builtins.
+
+        Scans the builtins package for modules that export a BasePlugin subclass.
+        Returns list of discovered plugin names.
+        """
+        discovered: list[str] = []
+        try:
+            builtins_pkg = importlib.import_module("app.plugins.builtins")
+        except ImportError:
+            return discovered
+
+        pkg_path = getattr(builtins_pkg, "__path__", None)
+        if pkg_path is None:
+            return discovered
+
+        import pkgutil
+        for importer, modname, ispkg in pkgutil.iter_modules(pkg_path):
+            if modname.startswith("_"):
+                continue
+            try:
+                full_name = f"app.plugins.builtins.{modname}"
+                module = importlib.import_module(full_name)
+                # Look for BasePlugin subclass in module
+                for attr_name in dir(module):
+                    attr = getattr(module, attr_name)
+                    if (
+                        isinstance(attr, type)
+                        and issubclass(attr, BasePlugin)
+                        and attr is not BasePlugin
+                    ):
+                        instance = attr()
+                        if instance.name not in self._plugins:
+                            self._plugins[instance.name] = instance
+                            discovered.append(instance.name)
+                            logger.info(f"Discovered plugin: {instance.name} v{instance.version}")
+            except Exception as exc:
+                logger.warning(f"Failed to load builtin plugin module {modname}: {exc}")
+
+        return discovered
+
+    def register_plugin(self, plugin: BasePlugin) -> None:
+        """Manually register a plugin instance."""
+        self._plugins[plugin.name] = plugin
+        logger.info(f"Registered plugin: {plugin.name} v{plugin.version}")
+
+    def get_plugin(self, name: str) -> BasePlugin | None:
+        """Get a registered plugin instance by name."""
+        return self._plugins.get(name)
+
+    def list_discovered(self) -> list[str]:
+        """List all discovered plugin names."""
+        return list(self._plugins.keys())
+
+    # ─── DB Status Sync ───
+
+    async def sync_db_status(self, db: AsyncSession) -> None:
+        """Load plugin status records from the database."""
+        result = await db.execute(select(PluginModel))
+        self._db_status = {row.name: row for row in result.scalars().all()}
+
+    def get_db_status(self, name: str) -> PluginModel | None:
+        """Get the DB status record for a plugin."""
+        return self._db_status.get(name)
+
+    # ─── Install / Activate / Deactivate / Uninstall ───
+
+    async def install(self, db: AsyncSession, name: str) -> PluginModel:
+        """Install a plugin: run migrations and create DB record.
+
+        Idempotent: if already installed, returns existing record.
+        """
+        plugin = self.get_plugin(name)
+        if plugin is None:
+            raise ValueError(f"Plugin '{name}' not found in registry")
+
+        # Check if already installed in DB
+        existing = await self._get_plugin_record(db, name)
+        if existing is not None:
+            return existing
+
+        # Run migrations
+        if plugin.manifest.migrations:
+            await self.migration_runner.run_all_migrations(
+                db, name, plugin.manifest.migrations
+            )
+
+        # Call on_install hook
+        await plugin.on_install(db, self._container)
+
+        # Create DB record
+        record = PluginModel(
+            name=name,
+            display_name=plugin.manifest.display_name,
+            version=plugin.manifest.version,
+            status="installed",
+            installed=True,
+            active=False,
+        )
+        db.add(record)
+        await db.flush()
+        self._db_status[name] = record
+
+        return record
+
+    async def activate(self, db: AsyncSession, name: str) -> PluginModel:
+        """Activate a plugin: register routes, event listeners, set status=active.
+
+        Idempotent: if already active, returns existing record without error.
+        """
+        plugin = self.get_plugin(name)
+        if plugin is None:
+            raise ValueError(f"Plugin '{name}' not found in registry")
+
+        record = await self._get_plugin_record(db, name)
+        if record is None:
+            raise ValueError(f"Plugin '{name}' is not installed — install first")
+
+        # Idempotent: already active
+        if record.active and record.status == "active":
+            return record
+
+        # Call on_activate hook (registers event listeners)
+        await plugin.on_activate(db, self._container, self._event_bus)
+
+        # Register routes on FastAPI app if available
+        if self._app is not None:
+            routers = plugin.get_routes()
+            mounted = []
+            for router in routers:
+                self._app.include_router(router)
+                mounted.append(router)
+            self._mounted_routers[name] = mounted
+
+        # Update DB record
+        record.status = "active"
+        record.active = True
+        await db.flush()
+        self._db_status[name] = record
+
+        return record
+
+    async def deactivate(self, db: AsyncSession, name: str) -> PluginModel:
+        """Deactivate a plugin: unregister event listeners, set status=inactive.
+
+        Idempotent: if already inactive, returns existing record without error.
+        """
+        plugin = self.get_plugin(name)
+        if plugin is None:
+            raise ValueError(f"Plugin '{name}' not found in registry")
+
+        record = await self._get_plugin_record(db, name)
+        if record is None:
+            raise ValueError(f"Plugin '{name}' is not installed")
+
+        # Idempotent: already inactive
+        if not record.active and record.status == "inactive":
+            return record
+
+        # Call on_deactivate hook (unregisters event listeners)
+        await plugin.on_deactivate(db, self._container, self._event_bus)
+
+        # Unregister routes from FastAPI app if available
+        if self._app is not None and name in self._mounted_routers:
+            mounted = self._mounted_routers.pop(name, [])
+            # Collect all route paths from mounted routers for removal
+            paths_to_remove: set[str] = set()
+            for router in mounted:
+                for route in router.routes:
+                    if hasattr(route, "path"):
+                        paths_to_remove.add(route.path)
+            # Remove matching routes from app
+            self._app.router.routes = [
+                r for r in self._app.router.routes
+                if not (hasattr(r, "path") and r.path in paths_to_remove)
+            ]
+
+        # Update DB record
+        record.status = "inactive"
+        record.active = False
+        await db.flush()
+        self._db_status[name] = record
+
+        return record
+
+    async def uninstall(
+        self, db: AsyncSession, name: str, remove_data: bool = False
+    ) -> PluginModel:
+        """Uninstall a plugin: deactivate, optionally drop tables, remove DB record.
+
+        Args:
+            db: Async database session.
+            name: Plugin name to uninstall.
+            remove_data: If True, drop all plugin-created tables.
+
+        Returns:
+            The plugin record before deletion (for response).
+
+        Raises:
+            ValueError if plugin not installed.
+        """
+        plugin = self.get_plugin(name)
+        if plugin is None:
+            raise ValueError(f"Plugin '{name}' not found in registry")
+
+        record = await self._get_plugin_record(db, name)
+        if record is None:
+            raise ValueError(f"Plugin '{name}' is not installed")
+
+        # Deactivate first if active
+        if record.active:
+            await self.deactivate(db, name)
+            # Refetch record after deactivate
+            record = await self._get_plugin_record(db, name)
+            if record is None:
+                raise ValueError(f"Plugin '{name}' disappeared during uninstall")
+
+        # Call on_uninstall hook
+        await plugin.on_uninstall(db, self._container)
+
+        # Optionally drop plugin tables
+        dropped_tables: list[str] = []
+        if remove_data:
+            dropped_tables = await self.migration_runner.drop_plugin_tables(db, name)
+
+        # Remove DB record
+        await db.delete(record)
+        await db.flush()
+        self._db_status.pop(name, None)
+
+        # Return a detached copy for response
+        record_dropped_tables = dropped_tables
+        record.status = "uninstalled"
+        record.dropped_tables = record_dropped_tables
+        return record
+
+    async def list_plugins(self, db: AsyncSession) -> list[dict[str, Any]]:
+        """List all plugins with their current status.
+
+        Merges discovered (in-memory) plugins with installed (DB) records.
+        """
+        result = await db.execute(select(PluginModel))
+        db_records = {row.name: row for row in result.scalars().all()}
+
+        plugins_list: list[dict[str, Any]] = []
+        for name, plugin in self._plugins.items():
+            record = db_records.get(name)
+            if record is not None:
+                plugins_list.append({
+                    "name": name,
+                    "display_name": record.display_name,
+                    "version": record.version,
+                    "status": record.status,
+                    "installed": record.installed,
+                    "active": record.active,
+                    "description": plugin.manifest.description,
+                    "dependencies": plugin.manifest.dependencies,
+                    "events": plugin.manifest.events,
+                    "migrations": plugin.manifest.migrations,
+                    "permissions": plugin.manifest.permissions,
+                })
+            else:
+                plugins_list.append({
+                    "name": name,
+                    "display_name": plugin.manifest.display_name,
+                    "version": plugin.version,
+                    "status": "discovered",
+                    "installed": False,
+                    "active": False,
+                    "description": plugin.manifest.description,
+                    "dependencies": plugin.manifest.dependencies,
+                    "events": plugin.manifest.events,
+                    "migrations": plugin.manifest.migrations,
+                    "permissions": plugin.manifest.permissions,
+                })
+
+        # Also include DB-only records (plugins that were installed but no longer discovered)
+        for name, record in db_records.items():
+            if name not in self._plugins:
+                plugins_list.append({
+                    "name": name,
+                    "display_name": record.display_name,
+                    "version": record.version,
+                    "status": record.status,
+                    "installed": record.installed,
+                    "active": record.active,
+                    "description": "",
+                    "dependencies": [],
+                    "events": [],
+                    "migrations": [],
+                    "permissions": [],
+                })
+
+        return plugins_list
+
+    # ─── Internal Helpers ───
+
+    async def _get_plugin_record(self, db: AsyncSession, name: str) -> PluginModel | None:
+        """Fetch a plugin record from DB by name."""
+        result = await db.execute(
+            select(PluginModel).where(PluginModel.name == name)
+        )
+        return result.scalar_one_or_none()
+
+
+# Global registry instance
+_registry: PluginRegistry | None = None
+
+
+def get_registry() -> PluginRegistry:
+    """Get the global plugin registry."""
+    global _registry
+    if _registry is None:
+        _registry = PluginRegistry()
+    return _registry
+
+
+def reset_registry_for_testing() -> PluginRegistry:
+    """Create a fresh registry for testing."""
+    global _registry
+    _registry = PluginRegistry()
+    return _registry

app/routes/__init__.py

@@ -1 +1,3 @@
 """Routes package."""
+
+from app.routes import auth, users, roles, tenants, health, notifications, companies, contacts, import_export, plugins

app/routes/plugins.py

@@ -0,0 +1,139 @@
+"""Plugin routes — list, install, activate, deactivate, uninstall, manifest schema."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_db
+from app.deps import require_admin
+from app.plugins.migration_runner import MigrationValidationError
+from app.services.plugin_service import get_plugin_service
+
+router = APIRouter(prefix="/api/v1/plugins", tags=["plugins"])
+
+
+@router.get("")
+async def list_plugins(
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_admin),
+):
+    """List all plugins with their current status (discovered, installed, active, inactive)."""
+    service = get_plugin_service()
+    plugins = await service.list_plugins(db)
+    return {"plugins": plugins, "total": len(plugins)}
+
+
+@router.get("/manifest")
+async def get_manifest_schema(
+    current_user: dict = Depends(require_admin),
+):
+    """Get the plugin manifest schema documentation."""
+    service = get_plugin_service()
+    return service.get_manifest_schema()
+
+
+@router.post("/{name}/install")
+async def install_plugin(
+    name: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_admin),
+):
+    """Install a plugin by name. Runs migrations and creates DB record.
+
+    Idempotent: returns 200 if already installed.
+    """
+    import uuid as uuid_mod
+    service = get_plugin_service()
+    try:
+        result = await service.install_plugin(
+            db, name,
+            tenant_id=uuid_mod.UUID(current_user["tenant_id"]),
+            user_id=uuid_mod.UUID(current_user["user_id"]),
+        )
+        return result
+    except ValueError as exc:
+        if "not found" in str(exc).lower():
+            raise HTTPException(404, detail={"detail": str(exc), "code": "plugin_not_found"})
+        raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_error"})
+    except MigrationValidationError as exc:
+        raise HTTPException(422, detail={"detail": str(exc), "code": "migration_validation_error"})
+
+
+@router.post("/{name}/activate")
+async def activate_plugin(
+    name: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_admin),
+):
+    """Activate a plugin by name. Registers event listeners and routes.
+
+    Idempotent: returns 200 if already active.
+    """
+    import uuid as uuid_mod
+    service = get_plugin_service()
+    try:
+        result = await service.activate_plugin(
+            db, name,
+            tenant_id=uuid_mod.UUID(current_user["tenant_id"]),
+            user_id=uuid_mod.UUID(current_user["user_id"]),
+        )
+        return result
+    except ValueError as exc:
+        if "not installed" in str(exc).lower():
+            raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_not_installed"})
+        if "not found" in str(exc).lower():
+            raise HTTPException(404, detail={"detail": str(exc), "code": "plugin_not_found"})
+        raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_error"})
+
+
+@router.post("/{name}/deactivate")
+async def deactivate_plugin(
+    name: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_admin),
+):
+    """Deactivate a plugin by name. Unregisters event listeners and routes.
+
+    Idempotent: returns 200 if already inactive.
+    """
+    import uuid as uuid_mod
+    service = get_plugin_service()
+    try:
+        result = await service.deactivate_plugin(
+            db, name,
+            tenant_id=uuid_mod.UUID(current_user["tenant_id"]),
+            user_id=uuid_mod.UUID(current_user["user_id"]),
+        )
+        return result
+    except ValueError as exc:
+        if "not found" in str(exc).lower():
+            raise HTTPException(404, detail={"detail": str(exc), "code": "plugin_not_found"})
+        raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_error"})
+
+
+@router.delete("/{name}")
+async def uninstall_plugin(
+    name: str,
+    remove_data: bool = Query(False),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_admin),
+):
+    """Uninstall a plugin. Optionally drop plugin-created tables with remove_data=true.
+
+    Returns 200 with uninstalled status. Idempotent for already-uninstalled plugins.
+    """
+    import uuid as uuid_mod
+    service = get_plugin_service()
+    try:
+        result = await service.uninstall_plugin(
+            db, name,
+            remove_data=remove_data,
+            tenant_id=uuid_mod.UUID(current_user["tenant_id"]),
+            user_id=uuid_mod.UUID(current_user["user_id"]),
+        )
+        return result
+    except ValueError as exc:
+        if "not found" in str(exc).lower():
+            raise HTTPException(404, detail={"detail": str(exc), "code": "plugin_not_found"})
+        raise HTTPException(400, detail={"detail": str(exc), "code": "plugin_error"})

app/schemas/__init__.py

@@ -1 +1,3 @@
 """Pydantic schemas package."""
+
+from app.schemas.plugin import PluginInfo, PluginListResponse, PluginActionResponse, PluginUninstallResponse

app/schemas/plugin.py

@@ -0,0 +1,61 @@
+"""Pydantic schemas for plugin API responses."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class PluginInfo(BaseModel):
+    """Plugin status information returned in list endpoint."""
+
+    name: str
+    display_name: str
+    version: str
+    status: str = Field(description="discovered | installed | active | inactive")
+    installed: bool = False
+    active: bool = False
+    description: str = ""
+    dependencies: list[str] = Field(default_factory=list)
+    events: list[str] = Field(default_factory=list)
+    migrations: list[str] = Field(default_factory=list)
+    permissions: list[str] = Field(default_factory=list)
+
+
+class PluginListResponse(BaseModel):
+    """Response for GET /api/v1/plugins."""
+
+    plugins: list[PluginInfo]
+    total: int
+
+
+class PluginActionResponse(BaseModel):
+    """Response for install/activate/deactivate/uninstall actions."""
+
+    name: str
+    display_name: str
+    version: str
+    status: str
+    installed: bool = False
+    active: bool = False
+    dropped_tables: list[str] = Field(default_factory=list, description="Tables dropped (uninstall with remove_data=true)")
+    message: str = ""
+
+
+class PluginUninstallResponse(PluginActionResponse):
+    """Response for DELETE /api/v1/plugins/{name}."""
+
+    pass
+
+
+class ManifestFieldDoc(BaseModel):
+    """Documentation for a single manifest field."""
+    type: str
+    required: str
+    description: str
+
+
+class ManifestSchemaResponse(BaseModel):
+    """Response for GET /api/v1/plugins/manifest."""
+
+    fields: dict[str, ManifestFieldDoc]
+    example: dict

app/services/__init__.py

@@ -1 +1,3 @@
 """Service layer package."""
+
+from app.services.plugin_service import PluginService, get_plugin_service

app/services/plugin_service.py

@@ -0,0 +1,196 @@
+"""Plugin service layer — business logic for plugin lifecycle operations."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.audit import log_audit
+from app.plugins.registry import PluginRegistry, get_registry
+from app.plugins.migration_runner import MigrationValidationError
+
+logger = logging.getLogger(__name__)
+
+
+class PluginService:
+    """Service layer for plugin lifecycle management.
+
+    Wraps the PluginRegistry with audit logging and error handling.
+    """
+
+    def __init__(self, registry: PluginRegistry | None = None) -> None:
+        self._registry = registry or get_registry()
+
+    @property
+    def registry(self) -> PluginRegistry:
+        return self._registry
+
+    async def list_plugins(self, db: AsyncSession) -> list[dict[str, Any]]:
+        """List all discovered and installed plugins with their status."""
+        return await self._registry.list_plugins(db)
+
+    async def install_plugin(
+        self,
+        db: AsyncSession,
+        name: str,
+        tenant_id: uuid.UUID | None = None,
+        user_id: uuid.UUID | None = None,
+    ) -> dict[str, Any]:
+        """Install a plugin by name.
+
+        Runs migrations, calls on_install hook, creates DB record.
+        Idempotent: returns existing record if already installed.
+        """
+        try:
+            record = await self._registry.install(db, name)
+            if tenant_id and user_id:
+                await log_audit(
+                    db, tenant_id, user_id,
+                    action="plugin.install",
+                    entity_type="plugin",
+                    entity_id=getattr(record, "id", None),
+                    changes={"name": name, "version": record.version},
+                )
+            return {
+                "name": record.name,
+                "display_name": record.display_name,
+                "version": record.version,
+                "status": record.status,
+                "installed": record.installed,
+                "active": record.active,
+                "message": "Plugin installed successfully",
+            }
+        except ValueError as exc:
+            raise ValueError(str(exc))
+        except MigrationValidationError as exc:
+            raise MigrationValidationError(str(exc))
+
+    async def activate_plugin(
+        self,
+        db: AsyncSession,
+        name: str,
+        tenant_id: uuid.UUID | None = None,
+        user_id: uuid.UUID | None = None,
+    ) -> dict[str, Any]:
+        """Activate a plugin by name.
+
+        Registers event listeners and routes. Idempotent.
+        """
+        try:
+            record = await self._registry.activate(db, name)
+            was_already_active = record.active and record.status == "active"
+            if tenant_id and user_id:
+                await log_audit(
+                    db, tenant_id, user_id,
+                    action="plugin.activate",
+                    entity_type="plugin",
+                    entity_id=getattr(record, "id", None),
+                    changes={"name": name, "was_already_active": was_already_active},
+                )
+            return {
+                "name": record.name,
+                "display_name": record.display_name,
+                "version": record.version,
+                "status": record.status,
+                "installed": record.installed,
+                "active": record.active,
+                "message": "Plugin is already active" if was_already_active else "Plugin activated successfully",
+            }
+        except ValueError as exc:
+            raise ValueError(str(exc))
+
+    async def deactivate_plugin(
+        self,
+        db: AsyncSession,
+        name: str,
+        tenant_id: uuid.UUID | None = None,
+        user_id: uuid.UUID | None = None,
+    ) -> dict[str, Any]:
+        """Deactivate a plugin by name.
+
+        Unregisters event listeners and routes. Idempotent.
+        """
+        try:
+            record = await self._registry.deactivate(db, name)
+            was_already_inactive = not record.active and record.status == "inactive"
+            if tenant_id and user_id:
+                await log_audit(
+                    db, tenant_id, user_id,
+                    action="plugin.deactivate",
+                    entity_type="plugin",
+                    entity_id=getattr(record, "id", None),
+                    changes={"name": name, "was_already_inactive": was_already_inactive},
+                )
+            return {
+                "name": record.name,
+                "display_name": record.display_name,
+                "version": record.version,
+                "status": record.status,
+                "installed": record.installed,
+                "active": record.active,
+                "message": "Plugin is already inactive" if was_already_inactive else "Plugin deactivated successfully",
+            }
+        except ValueError as exc:
+            raise ValueError(str(exc))
+
+    async def uninstall_plugin(
+        self,
+        db: AsyncSession,
+        name: str,
+        remove_data: bool = False,
+        tenant_id: uuid.UUID | None = None,
+        user_id: uuid.UUID | None = None,
+    ) -> dict[str, Any]:
+        """Uninstall a plugin by name.
+
+        Deactivates, calls on_uninstall hook, optionally drops tables, removes DB record.
+        """
+        try:
+            record = await self._registry.uninstall(db, name, remove_data=remove_data)
+            dropped_tables = getattr(record, "dropped_tables", [])
+            if tenant_id and user_id:
+                await log_audit(
+                    db, tenant_id, user_id,
+                    action="plugin.uninstall",
+                    entity_type="plugin",
+                    changes={"name": name, "remove_data": remove_data, "dropped_tables": dropped_tables},
+                )
+            return {
+                "name": name,
+                "display_name": record.display_name,
+                "version": record.version,
+                "status": "uninstalled",
+                "installed": False,
+                "active": False,
+                "dropped_tables": dropped_tables,
+                "message": f"Plugin uninstalled{' and data tables dropped' if remove_data else ''}",
+            }
+        except ValueError as exc:
+            raise ValueError(str(exc))
+
+    def get_manifest_schema(self) -> dict[str, Any]:
+        """Return the manifest schema documentation."""
+        from app.plugins.manifest import MANIFEST_SCHEMA_DOC
+        return MANIFEST_SCHEMA_DOC.model_dump()
+
+
+# Global service instance
+_service: PluginService | None = None
+
+
+def get_plugin_service() -> PluginService:
+    """Get the global plugin service instance."""
+    global _service
+    if _service is None:
+        _service = PluginService()
+    return _service
+
+
+def reset_plugin_service_for_testing(registry: PluginRegistry | None = None) -> PluginService:
+    """Create a fresh plugin service for testing."""
+    global _service
+    _service = PluginService(registry=registry)
+    return _service

tests/conftest.py

@@ -32,6 +32,7 @@ from app.models.user import User, UserTenant
 from app.models.role import Role
 from app.models.company import Company
 from app.models.contact import Contact, CompanyContact
+from app.models.plugin import Plugin, PluginMigration
 from app.main import create_app
 
 
@@ -87,7 +88,7 @@ def clean_tables(db_setup):
     sync_eng = _get_sync_engine()
     with sync_eng.connect() as conn:
         # TRUNCATE all tables with CASCADE — fast and reliable isolation
-        conn.execute(text("TRUNCATE TABLE company_contacts, contacts, api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, companies, user_tenants, users, tenants CASCADE;"))
+        conn.execute(text("TRUNCATE TABLE plugin_migrations, plugins, company_contacts, contacts, api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, companies, user_tenants, users, tenants CASCADE;"))
         conn.commit()
     sync_eng.dispose()
     yield

tests/test_plugins.py

@@ -0,0 +1,926 @@
+"""Tests for plugin system framework — all 14 acceptance criteria."""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from typing import Any
+
+import pytest
+import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import text, select, inspect
+from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, AsyncSession
+
+from app.core.db import Base, reset_engine_for_testing, close_engine, get_engine
+from app.core.event_bus import get_event_bus
+from app.core.service_container import get_container
+from app.main import create_app
+from app.models.plugin import Plugin as PluginModel, PluginMigration
+from app.plugins.base import BasePlugin
+from app.plugins.manifest import PluginManifest
+from app.plugins.registry import get_registry, reset_registry_for_testing
+from app.plugins.migration_runner import MigrationRunner, MigrationValidationError
+from app.plugins.builtins.test_sample import TestSamplePlugin
+from app.services.plugin_service import reset_plugin_service_for_testing
+from tests.conftest import TEST_DB_URL, seed_tenant_and_users, login_client, ORIGIN_HEADER
+
+
+# ─── Bad Plugin for AC11 (migration without tenant_id) ───
+
+class BadMigrationPlugin(BasePlugin):
+    """Plugin with a migration that creates a table WITHOUT tenant_id."""
+
+    manifest = PluginManifest(
+        name="bad_migration_plugin",
+        version="1.0.0",
+        display_name="Bad Migration Plugin",
+        description="Plugin with invalid migration (no tenant_id) for validator testing.",
+        dependencies=[],
+        routes=[],
+        events=[],
+        migrations=["0001_bad_migration.sql"],
+        permissions=[],
+    )
+
+
+# ─── Fixtures ───
+
+@pytest_asyncio.fixture
+async def plugin_app(engine: AsyncEngine, redis_client):
+    """FastAPI app with plugin registry initialized for testing."""
+    reset_engine_for_testing(engine)
+    app = create_app()
+
+    # Reset and initialize registry (lifespan doesn't run with ASGITransport)
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, app)
+
+    # Reset service container and event bus
+    container = get_container()
+    await container.initialize()
+
+    # Register test plugins manually (don't auto-discover to avoid side effects)
+    registry.register_plugin(TestSamplePlugin())
+    registry.register_plugin(BadMigrationPlugin())
+
+    # Reset plugin service to use the new registry
+    reset_plugin_service_for_testing(registry)
+
+    yield app
+    await close_engine()
+
+
+@pytest_asyncio.fixture
+async def plugin_client(plugin_app) -> AsyncClient:
+    """HTTP test client with plugin registry active."""
+    transport = ASGITransport(app=plugin_app)
+    async with AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+@pytest_asyncio.fixture
+async def authed_plugin_client(plugin_client: AsyncClient, db_session: AsyncSession) -> AsyncClient:
+    """Authenticated admin client for plugin tests."""
+    seed = await seed_tenant_and_users(db_session)
+    await login_client(plugin_client, "admin@tenanta.com")
+    return plugin_client
+
+
+@pytest_asyncio.fixture
+async def db_session_for_plugins(engine: AsyncEngine) -> AsyncSession:
+    """DB session for plugin test verification."""
+    factory = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
+    async with factory() as session:
+        yield session
+        await session.rollback()
+
+
+# ─── AC1: GET /api/v1/plugins → 200 + list of plugins with status ───
+
+@pytest.mark.asyncio
+async def test_ac01_list_plugins(authed_plugin_client: AsyncClient):
+    """AC1: GET /api/v1/plugins returns 200 with plugin list and status."""
+    resp = await authed_plugin_client.get("/api/v1/plugins", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert "plugins" in data
+    assert "total" in data
+    assert data["total"] >= 2
+    plugin_names = [p["name"] for p in data["plugins"]]
+    assert "test_sample" in plugin_names
+    assert "bad_migration_plugin" in plugin_names
+    # Check status field exists
+    for p in data["plugins"]:
+        assert "status" in p
+        assert "installed" in p
+        assert "active" in p
+
+
+# ─── AC2: POST /api/v1/plugins/{name}/install → 200, status=installed, migrations run ───
+
+@pytest.mark.asyncio
+async def test_ac02_install_plugin(authed_plugin_client: AsyncClient, db_session_for_plugins: AsyncSession):
+    """AC2: Install plugin runs migrations and sets status=installed."""
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["name"] == "test_sample"
+    assert data["status"] == "installed"
+    assert data["installed"] is True
+    assert data["active"] is False
+
+    # Verify migration table was created (tenant_id column exists)
+    result = await db_session_for_plugins.execute(
+        text("SELECT column_name FROM information_schema.columns WHERE table_name = 'plugin_test_data' AND column_name = 'tenant_id'")
+    )
+    assert result.fetchone() is not None, "plugin_test_data table should have tenant_id column"
+
+
+# ─── AC3: POST /api/v1/plugins/{name}/activate → 200, status=active, routes registered ───
+
+@pytest.mark.asyncio
+async def test_ac03_activate_plugin(authed_plugin_client: AsyncClient):
+    """AC3: Activate plugin sets status=active and registers event listeners."""
+    # Install first
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+
+    # Activate
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["name"] == "test_sample"
+    assert data["status"] == "active"
+    assert data["active"] is True
+
+
+# ─── AC4: POST /api/v1/plugins/{name}/deactivate → 200, status=inactive, routes unregistered ───
+
+@pytest.mark.asyncio
+async def test_ac04_deactivate_plugin(authed_plugin_client: AsyncClient):
+    """AC4: Deactivate plugin sets status=inactive and unregisters event listeners."""
+    # Install and activate first
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER)
+
+    # Deactivate
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/deactivate", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["name"] == "test_sample"
+    assert data["status"] == "inactive"
+    assert data["active"] is False
+
+
+# ─── AC5: DELETE /api/v1/plugins/{name} → 200, plugin removed ───
+
+@pytest.mark.asyncio
+async def test_ac05_uninstall_plugin(authed_plugin_client: AsyncClient, db_session_for_plugins: AsyncSession):
+    """AC5: Uninstall plugin removes DB record."""
+    # Install first
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+
+    # Uninstall
+    resp = await authed_plugin_client.delete("/api/v1/plugins/test_sample", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["status"] == "uninstalled"
+    assert data["installed"] is False
+
+    # Verify DB record is gone
+    result = await db_session_for_plugins.execute(
+        select(PluginModel).where(PluginModel.name == "test_sample")
+    )
+    assert result.scalar_one_or_none() is None
+
+
+# ─── AC6: DELETE /api/v1/plugins/{name}?remove_data=true → 200, plugin tables dropped ───
+
+@pytest.mark.asyncio
+async def test_ac06_uninstall_remove_data(authed_plugin_client: AsyncClient, engine: AsyncEngine):
+    """AC6: Uninstall with remove_data=true drops plugin tables."""
+    # Install first (creates plugin_test_data table)
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+
+    # Verify table exists
+    def _check_table(sync_conn):
+        insp = inspect(sync_conn)
+        return "plugin_test_data" in insp.get_table_names()
+    async with engine.connect() as conn:
+        table_exists_before = await conn.run_sync(_check_table)
+    assert table_exists_before, "plugin_test_data table should exist after install"
+
+    # Uninstall with remove_data=true
+    resp = await authed_plugin_client.delete(
+        "/api/v1/plugins/test_sample?remove_data=true", headers=ORIGIN_HEADER
+    )
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["status"] == "uninstalled"
+    assert "plugin_test_data" in data.get("dropped_tables", [])
+
+    # Verify table is gone
+    async with engine.connect() as conn:
+        table_exists_after = await conn.run_sync(_check_table)
+    assert not table_exists_after, "plugin_test_data table should be dropped after uninstall with remove_data=true"
+
+
+# ─── AC7: GET /api/v1/plugins/manifest → 200 + manifest schema documentation ───
+
+@pytest.mark.asyncio
+async def test_ac07_manifest_schema(authed_plugin_client: AsyncClient):
+    """AC7: GET /api/v1/plugins/manifest returns schema documentation."""
+    resp = await authed_plugin_client.get("/api/v1/plugins/manifest", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert "fields" in data
+    assert "example" in data
+    # Check required fields are documented
+    assert "name" in data["fields"]
+    assert "version" in data["fields"]
+    assert "display_name" in data["fields"]
+    assert "dependencies" in data["fields"]
+    assert "routes" in data["fields"]
+    assert "events" in data["fields"]
+    assert "migrations" in data["fields"]
+    # Check example has expected values
+    assert data["example"]["name"] == "example_plugin"
+
+
+# ─── AC8: Plugin activation registers event listeners on event bus ───
+
+@pytest.mark.asyncio
+async def test_ac08_activation_registers_event_listeners(
+    authed_plugin_client: AsyncClient, db_session_for_plugins: AsyncSession
+):
+    """AC8: Activating a plugin registers its event listeners on the event bus."""
+    registry = get_registry()
+    plugin = registry.get_plugin("test_sample")
+    assert plugin is not None
+
+    # Install and activate
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER)
+
+    event_bus = get_event_bus()
+
+    # Check that event bus has handlers for the plugin's events
+    assert len(event_bus._handlers.get("company.created", [])) > 0
+    assert len(event_bus._handlers.get("contact.created", [])) > 0
+
+    # Publish an event and verify the handler is called
+    await event_bus.publish("company.created", {"company_id": "test-123", "name": "Test Corp"})
+    # Give async tasks a moment to complete
+    await asyncio.sleep(0.1)
+
+    # Check that the plugin's event handler was called
+    assert len(plugin.event_log) > 0
+    assert plugin.event_log[0]["event"] == "company.created"
+
+
+# ─── AC9: Plugin deactivation unregisters event listeners ───
+
+@pytest.mark.asyncio
+async def test_ac09_deactivation_unregisters_event_listeners(authed_plugin_client: AsyncClient):
+    """AC9: Deactivating a plugin unregisters its event listeners from the event bus."""
+    registry = get_registry()
+    plugin = registry.get_plugin("test_sample")
+    assert plugin is not None
+
+    # Install, activate, then deactivate
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER)
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/deactivate", headers=ORIGIN_HEADER)
+
+    event_bus = get_event_bus()
+
+    # Publish an event — handler should NOT be called after deactivation
+    initial_log_count = len(plugin.event_log)
+    await event_bus.publish("company.created", {"company_id": "test-456", "name": "After Deactivate"})
+    await asyncio.sleep(0.1)
+
+    # Event log should not have grown
+    assert len(plugin.event_log) == initial_log_count, "Event handler should not be called after deactivation"
+
+    # Check event bus no longer has the plugin's handlers
+    handlers = event_bus._handlers.get("company.created", [])
+    # The specific handler reference should be removed
+    for handler in handlers:
+        assert handler not in plugin._event_handlers.values()
+
+
+# ─── AC10: Plugin migration creates tables with tenant_id column ───
+
+@pytest.mark.asyncio
+async def test_ac10_migration_creates_tenant_id(authed_plugin_client: AsyncClient, engine: AsyncEngine):
+    """AC10: Plugin migration creates tables that have tenant_id column."""
+    # Install the test_sample plugin which has a migration creating plugin_test_data
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+
+    # Verify the table has tenant_id column
+    def _get_columns(sync_conn):
+        insp = inspect(sync_conn)
+        if "plugin_test_data" not in insp.get_table_names():
+            return None
+        return [col["name"] for col in insp.get_columns("plugin_test_data")]
+
+    async with engine.connect() as conn:
+        columns = await conn.run_sync(_get_columns)
+
+    assert columns is not None, "plugin_test_data table should exist"
+    assert "tenant_id" in columns, "plugin_test_data table must have tenant_id column"
+
+
+# ─── AC11: Plugin migration validator rejects tables without tenant_id ───
+
+@pytest.mark.asyncio
+async def test_ac11_validator_rejects_no_tenant_id(authed_plugin_client: AsyncClient):
+    """AC11: Migration validator rejects tables created without tenant_id column."""
+    # Try to install the bad_migration_plugin — should fail with 422
+    resp = await authed_plugin_client.post(
+        "/api/v1/plugins/bad_migration_plugin/install", headers=ORIGIN_HEADER
+    )
+    assert resp.status_code == 422
+    data = resp.json()
+    # detail is a dict with {detail: str, code: str}
+    detail = data.get("detail", {})
+    if isinstance(detail, dict):
+        detail_text = detail.get("detail", "").lower()
+        detail_code = detail.get("code", "")
+    else:
+        detail_text = str(detail).lower()
+        detail_code = data.get("code", "")
+    assert "tenant_id" in detail_text or "migration_validation" in detail_code
+
+
+# ─── AC12: Plugin DB migrations tracked in plugin_migrations table ───
+
+@pytest.mark.asyncio
+async def test_ac12_migrations_tracked(authed_plugin_client: AsyncClient, db_session_for_plugins: AsyncSession):
+    """AC12: Plugin migrations are tracked in plugin_migrations table."""
+    # Install the test_sample plugin
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+
+    # Check plugin_migrations table has a record
+    result = await db_session_for_plugins.execute(
+        select(PluginMigration).where(PluginMigration.plugin_name == "test_sample")
+    )
+    migrations = result.scalars().all()
+    assert len(migrations) > 0, "plugin_migrations table should have a record for test_sample"
+    assert migrations[0].migration_file == "0001_test_plugin.sql"
+    assert migrations[0].status == "applied"
+
+
+# ─── AC13: Activating already-active plugin → idempotent (200, no error) ───
+
+@pytest.mark.asyncio
+async def test_ac13_activate_already_active(authed_plugin_client: AsyncClient):
+    """AC13: Activating an already-active plugin is idempotent (returns 200, no error)."""
+    # Install and activate
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "active"
+
+    # Activate again — should be idempotent
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["status"] == "active"
+    assert data["active"] is True
+    assert "already active" in data["message"].lower()
+
+
+# ─── AC14: Deactivating inactive plugin → idempotent (200) ───
+
+@pytest.mark.asyncio
+async def test_ac14_deactivate_already_inactive(authed_plugin_client: AsyncClient):
+    """AC14: Deactivating an already-inactive plugin is idempotent (returns 200, no error)."""
+    # Install and activate
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/install", headers=ORIGIN_HEADER)
+    await authed_plugin_client.post("/api/v1/plugins/test_sample/activate", headers=ORIGIN_HEADER)
+
+    # Deactivate
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/deactivate", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "inactive"
+
+    # Deactivate again — should be idempotent
+    resp = await authed_plugin_client.post("/api/v1/plugins/test_sample/deactivate", headers=ORIGIN_HEADER)
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["status"] == "inactive"
+    assert data["active"] is False
+    assert "already inactive" in data["message"].lower()
+
+
+# ─── Additional Tests: Direct MigrationRunner Unit Tests ───
+
+@pytest.mark.asyncio
+async def test_registry_discover_builtins(engine: AsyncEngine):
+    """Test that registry can discover built-in plugins."""
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    discovered = registry.discover_builtins()
+    # test_sample should be discovered from builtins package
+    assert "test_sample" in discovered
+    assert registry.get_plugin("test_sample") is not None
+
+
+@pytest.mark.asyncio
+async def test_registry_list_plugins_mixed_states(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test listing plugins in various states (discovered, installed, active, inactive)."""
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    registry.register_plugin(TestSamplePlugin())
+
+    # List before install — all discovered
+    plugins = await registry.list_plugins(db_session_for_plugins)
+    assert len(plugins) == 1
+    assert plugins[0]["status"] == "discovered"
+    assert plugins[0]["installed"] is False
+
+    # Install
+    await registry.install(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+
+    # List after install — installed
+    plugins = await registry.list_plugins(db_session_for_plugins)
+    assert len(plugins) == 1
+    assert plugins[0]["status"] == "installed"
+    assert plugins[0]["installed"] is True
+
+    # Activate
+    await registry.activate(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+
+    # List after activate — active
+    plugins = await registry.list_plugins(db_session_for_plugins)
+    assert plugins[0]["status"] == "active"
+    assert plugins[0]["active"] is True
+
+
+@pytest.mark.asyncio
+async def test_registry_install_idempotent(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test that installing an already-installed plugin is idempotent."""
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    registry.register_plugin(TestSamplePlugin())
+
+    record1 = await registry.install(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+
+    record2 = await registry.install(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+
+    # Same record returned
+    assert record1.id == record2.id
+
+
+@pytest.mark.asyncio
+async def test_registry_not_found_errors(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test registry raises ValueError for unknown plugins."""
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+
+    with pytest.raises(ValueError, match="not found"):
+        await registry.install(db_session_for_plugins, "nonexistent")
+
+    with pytest.raises(ValueError, match="not found"):
+        await registry.activate(db_session_for_plugins, "nonexistent")
+
+    with pytest.raises(ValueError, match="not found"):
+        await registry.deactivate(db_session_for_plugins, "nonexistent")
+
+    with pytest.raises(ValueError, match="not found"):
+        await registry.uninstall(db_session_for_plugins, "nonexistent")
+
+
+@pytest.mark.asyncio
+async def test_registry_activate_without_install(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test that activating without install raises error."""
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    registry.register_plugin(TestSamplePlugin())
+
+    with pytest.raises(ValueError, match="not installed"):
+        await registry.activate(db_session_for_plugins, "test_sample")
+
+
+@pytest.mark.asyncio
+async def test_migration_runner_drop_tables(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test MigrationRunner.drop_plugin_tables removes tables and migration records."""
+    runner = MigrationRunner(engine)
+
+    # Run migration first
+    await runner.run_migration(db_session_for_plugins, "drop_test", "0001_test_plugin.sql")
+    await db_session_for_plugins.commit()
+
+    # Verify table exists
+    result = await db_session_for_plugins.execute(
+        text("SELECT tablename FROM pg_tables WHERE tablename = 'plugin_test_data'")
+    )
+    assert result.fetchone() is not None
+
+    # Drop tables
+    dropped = await runner.drop_plugin_tables(db_session_for_plugins, "drop_test")
+    await db_session_for_plugins.commit()
+
+    assert "plugin_test_data" in dropped
+
+    # Verify table is gone
+    result = await db_session_for_plugins.execute(
+        text("SELECT tablename FROM pg_tables WHERE tablename = 'plugin_test_data'")
+    )
+    assert result.fetchone() is None
+
+    # Verify migration records are removed
+    result = await db_session_for_plugins.execute(
+        select(PluginMigration).where(PluginMigration.plugin_name == "drop_test")
+    )
+    assert result.scalars().all() == []
+
+
+@pytest.mark.asyncio
+async def test_migration_runner_file_not_found(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test MigrationRunner raises FileNotFoundError for missing migration file."""
+    runner = MigrationRunner(engine)
+    with pytest.raises(FileNotFoundError):
+        await runner.run_migration(db_session_for_plugins, "test", "nonexistent.sql")
+
+
+def test_migration_runner_split_sql():
+    """Test SQL splitting logic."""
+    sql = "CREATE TABLE a();\nCREATE INDEX idx ON a();\n"
+    statements = MigrationRunner._split_sql(sql)
+    assert len(statements) == 2
+
+    # Test with comments
+    sql2 = "-- comment\nCREATE TABLE b();\n"
+    statements2 = MigrationRunner._split_sql(sql2)
+    assert len(statements2) == 1
+
+    # Test empty SQL
+    assert MigrationRunner._split_sql("") == []
+
+
+def test_migration_runner_extract_table_names():
+    """Test table name extraction from SQL."""
+    sql = "CREATE TABLE foo (); CREATE TABLE IF NOT EXISTS bar ();"
+    names = MigrationRunner._extract_table_names(sql)
+    assert "foo" in names
+    assert "bar" in names
+
+
+def test_plugin_manifest_validation():
+    """Test PluginManifest field validation."""
+    # Valid manifest
+    m = PluginManifest(
+        name="my_plugin",
+        version="1.0.0",
+        display_name="My Plugin",
+    )
+    assert m.name == "my_plugin"
+    assert m.dependencies == []
+    assert m.routes == []
+
+    # Name is lowercased
+    m2 = PluginManifest(name="MyPlugin", version="1.0.0", display_name="Test")
+    assert m2.name == "myplugin"
+
+    # Invalid name with special chars
+    with pytest.raises(Exception):
+        PluginManifest(name="my-plugin!", version="1.0.0", display_name="Test")
+
+
+def test_base_plugin_repr():
+    """Test BasePlugin __repr__ method."""
+    plugin = TestSamplePlugin()
+    repr_str = repr(plugin)
+    assert "test_sample" in repr_str
+    assert "1.0.0" in repr_str
+
+
+def test_base_plugin_no_manifest_error():
+    """Test that BasePlugin without manifest raises ValueError."""
+    class NoManifestPlugin(BasePlugin):
+        pass
+
+    with pytest.raises(ValueError, match="manifest"):
+        NoManifestPlugin()
+
+
+def test_base_plugin_name_version_properties():
+    """Test BasePlugin name and version properties."""
+    plugin = TestSamplePlugin()
+    assert plugin.name == "test_sample"
+    assert plugin.version == "1.0.0"
+
+
+@pytest.mark.asyncio
+async def test_base_plugin_on_install_default():
+    """Test default on_install returns None (no-op)."""
+    class MinimalPlugin(BasePlugin):
+        manifest = PluginManifest(
+            name="minimal", version="1.0.0", display_name="Minimal",
+        )
+    plugin = MinimalPlugin()
+    result = await plugin.on_install(None, None)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_base_plugin_on_uninstall_default():
+    """Test default on_uninstall returns None (no-op)."""
+    class MinimalPlugin(BasePlugin):
+        manifest = PluginManifest(
+            name="minimal2", version="1.0.0", display_name="Minimal",
+        )
+    plugin = MinimalPlugin()
+    result = await plugin.on_uninstall(None, None)
+    assert result is None
+
+
+def test_base_plugin_get_routes_empty():
+    """Test get_routes returns empty list when manifest has no routes."""
+    plugin = TestSamplePlugin()
+    routes = plugin.get_routes()
+    assert routes == []
+
+
+def test_base_plugin_make_event_handler_noop():
+    """Test that _make_event_handler creates noop for unknown events."""
+    class MinimalPlugin(BasePlugin):
+        manifest = PluginManifest(
+            name="minimal3", version="1.0.0", display_name="Minimal",
+            events=["unknown.event"],
+        )
+    plugin = MinimalPlugin()
+    handler = plugin._make_event_handler("unknown.event")
+    assert handler is not None
+    # It should be callable
+    assert callable(handler)
+
+
+@pytest.mark.asyncio
+async def test_service_manifest_schema():
+    """Test plugin service get_manifest_schema returns dict."""
+    from app.services.plugin_service import PluginService
+    service = PluginService()
+    schema = service.get_manifest_schema()
+    assert isinstance(schema, dict)
+    assert "fields" in schema
+    assert "example" in schema
+
+
+@pytest.mark.asyncio
+async def test_uninstall_inactive_plugin(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test uninstalling a plugin that is installed but not active."""
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    registry.register_plugin(TestSamplePlugin())
+
+    # Install only (not activated)
+    await registry.install(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+
+    # Uninstall should work without deactivation
+    record = await registry.uninstall(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+    assert record.status == "uninstalled"
+
+
+@pytest.mark.asyncio
+async def test_registry_activate_with_app_routes(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test that activating a plugin with a FastAPI app registers routes."""
+    from fastapi import FastAPI, APIRouter
+    from app.plugins.manifest import PluginManifest, PluginRouteDef
+
+    # Create a plugin with a route
+    class RoutePlugin(BasePlugin):
+        manifest = PluginManifest(
+            name="route_plugin",
+            version="1.0.0",
+            display_name="Route Plugin",
+            events=["test.event"],
+        )
+        def get_routes(self) -> list:
+            test_router = APIRouter()
+            @test_router.get("/api/v1/route-plugin/test")
+            async def test_endpoint():
+                return {"status": "ok"}
+            return [test_router]
+
+    app = FastAPI()
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, app)
+    registry.register_plugin(RoutePlugin())
+
+    # Install and activate
+    await registry.install(db_session_for_plugins, "route_plugin")
+    await db_session_for_plugins.commit()
+    await registry.activate(db_session_for_plugins, "route_plugin")
+    await db_session_for_plugins.commit()
+
+    # Verify route was mounted
+    route_paths = [r.path for r in app.router.routes]
+    assert "/api/v1/route-plugin/test" in route_paths
+
+    # Deactivate — route should be unmounted
+    await registry.deactivate(db_session_for_plugins, "route_plugin")
+    await db_session_for_plugins.commit()
+    route_paths_after = [r.path for r in app.router.routes]
+    assert "/api/v1/route-plugin/test" not in route_paths_after
+
+
+@pytest.mark.asyncio
+async def test_registry_uninstall_with_remove_data(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test registry uninstall with remove_data drops tables."""
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    registry.register_plugin(TestSamplePlugin())
+
+    await registry.install(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+
+    record = await registry.uninstall(db_session_for_plugins, "test_sample", remove_data=True)
+    await db_session_for_plugins.commit()
+    assert record.status == "uninstalled"
+    assert "plugin_test_data" in getattr(record, "dropped_tables", [])
+
+
+@pytest.mark.asyncio
+async def test_registry_not_initialized_errors():
+    """Test that uninitialized registry raises RuntimeError."""
+    registry = reset_registry_for_testing()
+    with pytest.raises(RuntimeError, match="not initialized"):
+        _ = registry.migration_runner
+    with pytest.raises(RuntimeError, match="not initialized"):
+        _ = registry.engine
+
+
+@pytest.mark.asyncio
+async def test_registry_deactivate_already_inactive_direct(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test registry.deactivate is idempotent when already inactive."""
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    registry.register_plugin(TestSamplePlugin())
+
+    await registry.install(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+
+    # Deactivate without activating first — status is 'installed', not 'inactive'
+    # But calling deactivate should still work (sets to inactive)
+    record = await registry.deactivate(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+    assert record.status == "inactive"
+
+    # Deactivate again — idempotent
+    record2 = await registry.deactivate(db_session_for_plugins, "test_sample")
+    await db_session_for_plugins.commit()
+    assert record2.status == "inactive"
+
+
+@pytest.mark.asyncio
+async def test_plugin_service_install_error_handling(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test plugin service raises ValueError for unknown plugin."""
+    from app.services.plugin_service import PluginService
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    service = PluginService(registry=registry)
+
+    with pytest.raises(ValueError, match="not found"):
+        await service.install_plugin(db_session_for_plugins, "nonexistent")
+
+
+@pytest.mark.asyncio
+async def test_plugin_service_activate_error_handling(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test plugin service raises ValueError when activating uninstalled plugin."""
+    from app.services.plugin_service import PluginService
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    registry.register_plugin(TestSamplePlugin())
+    service = PluginService(registry=registry)
+
+    with pytest.raises(ValueError, match="not installed"):
+        await service.activate_plugin(db_session_for_plugins, "test_sample")
+
+
+@pytest.mark.asyncio
+async def test_plugin_service_deactivate_error_handling(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test plugin service raises ValueError for deactivating unknown plugin."""
+    from app.services.plugin_service import PluginService
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    service = PluginService(registry=registry)
+
+    with pytest.raises(ValueError, match="not found"):
+        await service.deactivate_plugin(db_session_for_plugins, "nonexistent")
+
+
+@pytest.mark.asyncio
+async def test_plugin_service_uninstall_error_handling(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test plugin service raises ValueError for uninstalling unknown plugin."""
+    from app.services.plugin_service import PluginService
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+    service = PluginService(registry=registry)
+
+    with pytest.raises(ValueError, match="not found"):
+        await service.uninstall_plugin(db_session_for_plugins, "nonexistent")
+
+
+@pytest.mark.asyncio
+async def test_migration_runner_split_sql_dollar_quotes():
+    """Test SQL splitting with dollar-quoted blocks."""
+    sql = "CREATE FUNCTION foo() RETURNS void AS $$\nBEGIN\nEND;\n$$ LANGUAGE plpgsql;\nCREATE TABLE bar();\n"
+    statements = MigrationRunner._split_sql(sql)
+    assert len(statements) >= 2
+
+
+@pytest.mark.asyncio
+async def test_registry_list_plugins_db_only_record(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test listing plugins includes DB-only records (installed but not discovered)."""
+    registry = reset_registry_for_testing()
+    registry.initialize(engine, None)
+
+    # Manually create a DB record without registering the plugin
+    record = PluginModel(
+        name="orphan_plugin",
+        display_name="Orphan Plugin",
+        version="0.1.0",
+        status="installed",
+        installed=True,
+        active=False,
+    )
+    db_session_for_plugins.add(record)
+    await db_session_for_plugins.commit()
+
+    plugins = await registry.list_plugins(db_session_for_plugins)
+    names = [p["name"] for p in plugins]
+    assert "orphan_plugin" in names
+    orphan = [p for p in plugins if p["name"] == "orphan_plugin"][0]
+    assert orphan["installed"] is True
+
+
+@pytest.mark.asyncio
+async def test_base_plugin_on_event_fallback():
+    """Test that on_event fallback handler works."""
+    class FallbackPlugin(BasePlugin):
+        manifest = PluginManifest(
+            name="fallback", version="1.0.0", display_name="Fallback",
+            events=["custom.event"],
+        )
+        async def on_event(self, payload: dict[str, Any]) -> None:
+            self.event_received = payload
+
+    plugin = FallbackPlugin()
+    handler = plugin._make_event_handler("custom.event")
+    await handler({"test": True})
+    assert plugin.event_received == {"test": True}
+
+
+@pytest.mark.asyncio
+async def test_migration_runner_valid_migration(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test MigrationRunner directly with a valid migration (has tenant_id)."""
+    runner = MigrationRunner(engine)
+    record = await runner.run_migration(
+        db_session_for_plugins,
+        "test_plugin",
+        "0001_test_plugin.sql",
+    )
+    await db_session_for_plugins.commit()
+    assert record.plugin_name == "test_plugin"
+    assert record.migration_file == "0001_test_plugin.sql"
+    assert record.status == "applied"
+
+
+@pytest.mark.asyncio
+async def test_migration_runner_invalid_migration(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test MigrationRunner directly rejects migration without tenant_id."""
+    runner = MigrationRunner(engine)
+    with pytest.raises(MigrationValidationError) as exc_info:
+        await runner.run_migration(
+            db_session_for_plugins,
+            "bad_plugin",
+            "0001_bad_migration.sql",
+        )
+    assert "tenant_id" in str(exc_info.value).lower()
+
+
+@pytest.mark.asyncio
+async def test_migration_runner_idempotent(engine: AsyncEngine, db_session_for_plugins: AsyncSession):
+    """Test MigrationRunner skips already-applied migrations."""
+    runner = MigrationRunner(engine)
+    # Run migration
+    await runner.run_migration(db_session_for_plugins, "test_idempotent", "0001_test_plugin.sql")
+    await db_session_for_plugins.commit()
+
+    # Run all migrations again — should skip
+    records = await runner.run_all_migrations(
+        db_session_for_plugins, "test_idempotent", ["0001_test_plugin.sql"]
+    )
+    assert len(records) == 0, "Already-applied migration should be skipped"