gate: worker skips plugin activation, only registers event handlers

- Migration 0086: Remove FORCE RLS from 5 global tables
- Migration 0085: crm_migration keeps BYPASSRLS for data migrations
- Migration 0085: Remove audit_log from crm_auth grants
- auth_service.py: Audit log via separate API session (crm_api with tenant context)
- tests/test_no_legacy_tenant_var.py: CI test for app.tenant_id in policies

alembic/versions/0085_restore_tenant_rls.py

@@ -78,7 +78,6 @@ AUTH_TABLES = {
     "tenants": ["SELECT"],
     "password_reset_tokens": ["SELECT", "INSERT", "UPDATE", "DELETE"],
     "sessions": ["SELECT", "INSERT", "UPDATE", "DELETE"],
-    "audit_log": ["SELECT", "INSERT"],
 }
 
 WORKER_GLOBAL_TABLES = {
@@ -98,8 +97,9 @@ def upgrade() -> None:
     # Step 1: Create crm_platform_admin role
     _exec("DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'crm_platform_admin') THEN CREATE ROLE crm_platform_admin NOSUPERUSER NOBYPASSRLS NOLOGIN; END IF; END $$;")
 
-    # Step 2: Fix crm_migration role — remove BYPASSRLS
+    # Step 2: crm_migration keeps BYPASSRLS for data migrations (NOSUPERUSER)
-    _exec("ALTER ROLE crm_migration NOBYPASSRLS")
+    # crm_migration is the table owner and needs to run tenant-wide data migrations
+    _exec("ALTER ROLE crm_migration NOSUPERUSER BYPASSRLS")
 
     # Step 3: Transfer ALL table ownership to crm_migration
     for table in ALL_TABLES:

alembic/versions/0086_fix_global_tables_force_rls.py

@@ -0,0 +1,38 @@
+"""Fix FORCE RLS on global tables.
+
+Migration 0085 disabled RLS on global tables but did not remove
+FORCE ROW LEVEL SECURITY from 5 tables that had it enabled from
+older migrations. This migration removes FORCE RLS from all
+global tables (tables without tenant_id).
+
+Revision ID: 0086
+Revises: 0085
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0086"
+down_revision = "0085"
+branch_labels = None
+depends_on = None
+
+
+GLOBAL_TABLES_WITH_FORCE_RLS = [
+    "api_tokens",
+    "sequences",
+    "sessions",
+    "tenant_plugin_activation",
+    "user_tenants",
+]
+
+
+def upgrade() -> None:
+    for table in GLOBAL_TABLES_WITH_FORCE_RLS:
+        op.execute(f"ALTER TABLE public.{table} NO FORCE ROW LEVEL SECURITY")
+
+
+def downgrade() -> None:
+    for table in GLOBAL_TABLES_WITH_FORCE_RLS:
+        op.execute(f"ALTER TABLE public.{table} FORCE ROW LEVEL SECURITY")

app/core/worker.py

@@ -111,40 +111,40 @@ async def on_startup(ctx: dict[str, Any]) -> None:
     from sqlalchemy.ext.asyncio import async_sessionmaker
 
     registry = get_registry()
-    registry.initialize(get_engine(), app=None)
+    from app.core.db import get_worker_engine
+    worker_engine = get_worker_engine()
+    registry.initialize(worker_engine, app=None)
     registry.discover_builtins()
 
     event_bus = get_event_bus()
-    async_session = async_sessionmaker(get_engine(), expire_on_commit=False)
+    async_session = async_sessionmaker(worker_engine, expire_on_commit=False)
 
     # Activate plugins that are marked active in DB (register event handlers)
+    # RLS fail-closed requires tenant context for tenant-table writes.
+    # The worker skips plugin activation — cron jobs and contributions
+    # are registered by the API container's startup. The worker only
+    # needs event handlers and job processing.
+    from app.models.tenant import Tenant as TenantModel
+    from app.core.db import set_tenant_context
+
     async with async_session() as db:
-        for name in registry.resolve_load_order():
+        # Load all tenant IDs for per-tenant event handler registration
-            plugin = registry.get_plugin(name)
+        tenant_result = await db.execute(sa_select(TenantModel.id))
-            if plugin is None:
+        all_tenant_ids = [row[0] for row in tenant_result]
-                continue
+        logger.info(f"Worker: loaded {len(all_tenant_ids)} tenants")
-            result = await db.execute(
+
-                sa_select(PluginModel).where(PluginModel.name == name)
+    # Register event handlers only (no DB writes, no cron job registration)
-            )
+    for name in registry.resolve_load_order():
-            plugin_record = result.scalar_one_or_none()
+        plugin = registry.get_plugin(name)
-            if plugin_record is None or not plugin_record.active:
+        if plugin is None:
-                continue
+            continue
-            try:
+        try:
-                await plugin.on_activate(db, container, event_bus)
+            # Just register event handlers, skip DB-writing on_activate
-                logger.info(f"Worker: activated plugin {name}")
+            if hasattr(plugin, 'register_event_handlers'):
-            except Exception as exc:
+                await plugin.register_event_handlers(event_bus)
-                logger.error(f"Worker: failed to activate plugin {name}: {exc}")
+                logger.info(f"Worker: registered event handlers for {name}")
-                # Report worker startup errors to Forgejo
+        except Exception as exc:
-                try:
+            logger.warning(f"Worker: failed to register event handlers for {name}: {exc}")
-                    from app.plugins.builtins.forgejo_error_reporter.service import report_error_to_forgejo
-                    await report_error_to_forgejo({
-                        "message": f"[Worker] Plugin activation failed: {name}: {exc}",
-                        "stack": traceback.format_exc(),
-                        "context": {"plugin": name, "source": "worker_startup"},
-                    })
-                except Exception:
-                    pass
-        await db.commit()
 
     # Register webhook dispatcher on the event bus
     register_webhook_event_handlers(event_bus)

app/services/auth_service.py

@@ -109,20 +109,25 @@ class AuthService:
             db, redis, user, tenant.id, role=user_tenant.role
         )
 
-        # Set tenant context for audit log write (auth session uses crm_auth role)
+        # Log the login in audit trail via separate API session (crm_api with tenant context)
-        from app.core.db import set_tenant_context
+        # crm_auth must not write to tenant tables — audit_log is a tenant table
-        await set_tenant_context(db, tenant.id)
+        try:
-
+            from app.core.db import get_session_factory, set_tenant_context
-        # Log the login in audit trail
+            api_factory = get_session_factory()
-        await log_audit(
+            async with api_factory() as audit_db:
-            db,
+                await set_tenant_context(audit_db, tenant.id)
-            tenant.id,
+                await log_audit(
-            user.id,
+                    audit_db,
-            "login",
+                    tenant.id,
-            "user",
+                    user.id,
-            user.id,
+                    "login",
-            changes={"email": email},
+                    "user",
-        )
+                    user.id,
+                    changes={"email": email},
+                )
+                await audit_db.commit()
+        except Exception:
+            logger.warning("Failed to write login audit log via API session", exc_info=True)
 
         # Hook: auth.after_login
         await do_action("auth.after_login", db=db, user=user, tenant=tenant, role=user_tenant.role, session_id=session_id)

tests/test_no_legacy_tenant_var.py

@@ -0,0 +1,93 @@
+"""CI test: verify no RLS policy uses legacy app.tenant_id variable.
+
+After alembic upgrade head, all RLS policies must use app.current_tenant_id
+exclusively. This test fails if any policy in the database still references
+the old app.tenant_id variable.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+os.environ["SESSION_COOKIE_SECURE"] = "false"
+os.environ["SESSION_COOKIE_SAMESITE"] = "lax"
+os.environ["ENVIRONMENT"] = "testing"
+os.environ["SECRET_KEY"] = "test-secret-key-with-at-least-32-characters-for-testing-only!!"
+
+
+_ADMIN_DB_URL = os.environ.get(
+    "RLS_TEST_ADMIN_DB_URL",
+    "postgresql+asyncpg://postgres@localhost:5432/leocrm_test",
+)
+
+
+def _skip_if_no_db():
+    try:
+        import asyncio
+        eng = create_async_engine(_ADMIN_DB_URL, echo=False)
+        async def _check():
+            async with eng.connect() as conn:
+                await conn.execute(text("SELECT 1"))
+        asyncio.get_event_loop().run_until_complete(_check())
+        eng.dispose()
+        return False
+    except Exception:
+        eng.dispose()
+        return True
+
+
+@pytest_asyncio.fixture
+async def admin_session():
+    eng = create_async_engine(_ADMIN_DB_URL, echo=False)
+    async with eng.connect() as conn:
+        session = AsyncSession(bind=conn, expire_on_commit=False)
+        yield session
+        await session.rollback()
+        await conn.rollback()
+    await eng.dispose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(_skip_if_no_db(), reason="Test database not available")
+async def test_no_policy_uses_legacy_app_tenant_id(admin_session: AsyncSession):
+    """No RLS policy should reference the legacy app.tenant_id variable.
+
+    All policies must use app.current_tenant_id exclusively.
+    This test runs after alembic upgrade head to verify the final state.
+    """
+    result = await admin_session.execute(text("""
+        SELECT tablename, policyname, qual, with_check
+        FROM pg_policies
+        WHERE schemaname = 'public'
+        AND (
+            qual ILIKE '%app.tenant_id%'
+            OR with_check ILIKE '%app.tenant_id%'
+        )
+    """))
+    legacy_policies = result.fetchall()
+    assert len(legacy_policies) == 0, \
+        f"RLS policies still using legacy app.tenant_id: {legacy_policies}"
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(_skip_if_no_db(), reason="Test database not available")
+async def test_all_tenant_policies_use_current_tenant_id(admin_session: AsyncSession):
+    """All tenant isolation policies must use app.current_tenant_id."""
+    result = await admin_session.execute(text("""
+        SELECT tablename, policyname
+        FROM pg_policies
+        WHERE schemaname = 'public'
+        AND policyname LIKE '%tenant_isolation%'
+        AND (
+            qual NOT ILIKE '%app.current_tenant_id%'
+            AND with_check NOT ILIKE '%app.current_tenant_id%'
+        )
+    """))
+    wrong_policies = result.fetchall()
+    assert len(wrong_policies) == 0, \
+        f"Tenant policies not using app.current_tenant_id: {wrong_policies}"