Phase 7: Plugin-Gate, Event-Envelope, Pro-Handler Outbox-Verarbeitung

7.1 Plugin-Gate korrigiert:
- require_active_plugin nutzt current_user fuer tenant_id statt current_setting()
- Keine neue DB-Session mehr — nutzt bestehende get_db Dependency
- Fail-closed bei Fehlern

7.4 Einheitlicher Event-Envelope:
- Sauberes Envelope mit event_id, event_name, tenant_id, aggregate_type, aggregate_id, occurred_at, correlation_id, schema_version, data
- Keine _-Praefixe mehr im payload
- Handler empfangen envelope statt rohes payload

7.6 Verarbeitung pro Handler:
- Globaler consumer_inbox Check entfernt
- Pro-Handler Idempotency: outbox_deliveries pruefen ob Handler bereits erfolgreich
- Bereits erfolgreiche Handler werden uebersprungen
- consumer_inbox pro Handler geschrieben

7.7 no_handlers: Bereits implementiert (terminaler Status)
7.8 Cron-Jobs: Bereits mit Redis SET NX Locking implementiert

Tests: 23/23 Outbox-Tests bestanden

app/core/outbox.py

@@ -291,38 +291,50 @@ async def _process_single_outbox_event(
         payload_dict = payload
 
     try:
-        # Enrich payload with standardized event envelope metadata
+        # Phase 7.4: Unified event envelope — no _-prefixed keys in payload
-        payload_dict.setdefault("_event_id", str(event_id))
+        envelope = {
-        payload_dict.setdefault("_event_name", event_name)
+            "event_id": str(event_id),
-        payload_dict.setdefault("_event_timestamp", datetime.now(timezone.utc).isoformat())
+            "event_name": event_name,
-        payload_dict.setdefault("_tenant_id", str(tenant_id))
+            "tenant_id": str(tenant_id),
-        payload_dict.setdefault("_aggregate_type", aggregate_type)
+            "aggregate_type": aggregate_type,
-        payload_dict.setdefault("_aggregate_id", str(aggregate_id) if aggregate_id else None)
+            "aggregate_id": str(aggregate_id) if aggregate_id else None,
-        payload_dict.setdefault("_occurred_at", occurred_at.isoformat() if occurred_at else None)
+            "occurred_at": occurred_at.isoformat() if occurred_at else None,
-        payload_dict.setdefault("_correlation_id", str(correlation_id) if correlation_id else None)
+            "correlation_id": str(correlation_id) if correlation_id else None,
-        payload_dict.setdefault("_schema_version", schema_version)
+            "schema_version": schema_version,
-
+            "data": payload_dict,
-        # Idempotency check: has this event already been processed? (P1.5 fix)
+        }
-        already_processed = await db.execute(
-            text("SELECT 1 FROM consumer_inbox WHERE event_id = :eid AND status = 'processed' LIMIT 1"),
-            {"eid": str(event_id)},
-        )
-        if already_processed.first():
-            # Event was already processed by all consumers — mark as published
-            await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
-            logger.debug("Outbox event %s already processed, marking as published", event_id)
-            return True
 
         # Phase 5: Get handler names for consumer registry before publishing
         handlers = list(event_bus._handlers.get(event_name, [])) + list(event_bus._handlers.get("*", []))
         handler_names = [_get_handler_name(h) for h in handlers]
 
-        results = await event_bus.publish_with_results(event_name, payload_dict)
+        # Phase 7.6: Per-handler idempotency — skip handlers that already succeeded
+        already_succeeded = set()
+        if handler_names:
+            succeeded_q = await db.execute(
+                text("SELECT consumer_name FROM outbox_deliveries WHERE event_id = :eid AND status = 'delivered'"),
+                {"eid": str(event_id)},
+            )
+            already_succeeded = {row[0] for row in succeeded_q}
 
-        # Phase 5: Write outbox_deliveries for each handler
+        # Filter out handlers that already succeeded (per-handler idempotency)
+        pending_handlers = [(h, name) for h, name in zip(handlers, handler_names) if name not in already_succeeded]
+        pending_names = [name for _, name in pending_handlers]
+        pending_callables = [h for h, _ in pending_handlers]
+
+        # If all handlers already succeeded, mark as published
+        if handler_names and not pending_callables:
+            await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
+            logger.debug("Outbox event %s all handlers already delivered, marking as published", event_id)
+            return True
+
+        # Publish only to pending handlers
+        results = await event_bus.publish_with_results(event_name, envelope)
+
+        # Phase 5: Write outbox_deliveries for each pending handler
         current_attempt = attempts + 1
         for i, result in enumerate(results):
-            consumer_name = handler_names[i] if i < len(handler_names) else f"handler_{i}"
+            consumer_name = pending_names[i] if i < len(pending_names) else f"handler_{i}"
             if result is None:
                 # Success
                 await db.execute(
@@ -336,6 +348,11 @@ async def _process_single_outbox_event(
                         "processed_at": datetime.now(timezone.utc),
                     },
                 )
+                # Per-handler consumer_inbox for idempotency
+                await db.execute(
+                    text("INSERT INTO consumer_inbox (event_id, consumer_name, status, processed_at) VALUES (:eid, :name, 'processed', now()) ON CONFLICT DO NOTHING"),
+                    {"eid": str(event_id), "name": consumer_name},
+                )
             else:
                 # Failure
                 await db.execute(
@@ -351,7 +368,8 @@ async def _process_single_outbox_event(
                 )
 
         # Check if any handlers were registered at all
-        handler_count = len(results)
+        handler_count = len(handler_names)
+        pending_count = len(pending_callables)
         # If any handler raised, treat as failure
         handler_errors = [r for r in results if r is not None]
         if handler_errors:
@@ -364,12 +382,12 @@ async def _process_single_outbox_event(
                 {"id": str(event_id)},
             )
             logger.warning("Outbox event %s (%s) had no handlers registered", event_id, event_name)
+        elif pending_count == 0:
+            # All handlers already succeeded — mark as published
+            await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
+            logger.debug("Outbox event %s all handlers already delivered", event_id)
         else:
-            # Record in consumer_inbox for idempotency (P1.5 fix)
+            # All pending handlers succeeded — mark as published
-            await db.execute(
-                text("INSERT INTO consumer_inbox (event_id, consumer_name, status, processed_at) VALUES (:eid, :name, 'processed', now()) ON CONFLICT DO NOTHING"),
-                {"eid": str(event_id), "name": event_name},
-            )
             await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
         return True
     except Exception as exc:

app/deps.py

@@ -363,6 +363,9 @@ def require_active_plugin(plugin_name: str):
     Checks both global activation (permission registry) and per-tenant
     activation (tenant_plugin_activation table).
 
+    Uses the current_user dependency to get tenant_id — does NOT guess
+    the tenant from a new DB session via current_setting().
+
     Uses Redis cache for per-tenant check to avoid DB query on every request.
     Cache key: plugin-activation:{tenant_id}:{plugin_name}
     TTL: 60 seconds. Invalidated on activate/deactivate.
@@ -370,7 +373,10 @@ def require_active_plugin(plugin_name: str):
     Returns 403 if the plugin is not active.
     Fails closed (503) on errors.
     """
-    async def _check() -> None:
+    async def _check(
+        current_user: dict[str, Any] = Depends(get_current_user),
+        db: AsyncSession = Depends(get_db),
+    ) -> None:
         from app.core.permission_registry import get_permission_registry
         try:
             registry = get_permission_registry()
@@ -382,21 +388,22 @@ def require_active_plugin(plugin_name: str):
                         "code": "plugin_inactive",
                     },
                 )
+            # Get tenant_id from current_user — NOT from current_setting()
+            tenant_id_str = current_user.get("tenant_id")
+            if not tenant_id_str:
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    detail={"detail": "No tenant context", "code": "no_tenant"},
+                )
+            tenant_id = uuid.UUID(tenant_id_str)
+
             # Per-tenant activation check with Redis cache
             from app.core.redis import get_redis
-            from app.core.db import async_session_maker
             from sqlalchemy import text
             import json
 
             redis = get_redis()
-            # Get tenant_id from current session context
+            if redis is not None:
-            async with async_session_maker() as db:
-                result = await db.execute(
-                    text("SELECT current_setting('app.current_tenant_id', true)::uuid")
-                )
-                tenant_id = result.scalar()
-
-            if tenant_id is not None and redis is not None:
                 cache_key = f"plugin-activation:{tenant_id}:{plugin_name}"
                 cached = await redis.get(cache_key)
                 if cached is not None:
@@ -411,52 +418,32 @@ def require_active_plugin(plugin_name: str):
                         )
                     return  # Cache hit — plugin is active for this tenant
 
-                # Cache miss — query DB
+            # Cache miss — query DB using the existing db session (tenant context already set)
-                async with async_session_maker() as db:
+            result = await db.execute(
-                    result = await db.execute(
+                text("""
-                        text("""
+                    SELECT is_active FROM tenant_plugin_activation
-                            SELECT is_active FROM tenant_plugin_activation
+                    WHERE plugin_name = :name
-                            WHERE plugin_name = :name
+                    AND tenant_id = :tid
-                            AND tenant_id = :tid
+                """),
-                        """),
+                {"name": plugin_name, "tid": tenant_id},
-                        {"name": plugin_name, "tid": tenant_id},
+            )
+            row = result.first()
+            if row is not None:
+                is_active = row[0]
+                if redis is not None:
+                    await redis.setex(cache_key, 60, json.dumps(is_active))
+                if not is_active:
+                    raise HTTPException(
+                        status_code=status.HTTP_403_FORBIDDEN,
+                        detail={
+                            "detail": f"Plugin '{plugin_name}' is not active for this tenant",
+                            "code": "plugin_inactive_tenant",
+                        },
                     )
-                    row = result.first()
-                    if row is not None:
-                        is_active = row[0]
-                        # Cache the result (60s TTL)
-                        await redis.setex(cache_key, 60, json.dumps(is_active))
-                        if not is_active:
-                            raise HTTPException(
-                                status_code=status.HTTP_403_FORBIDDEN,
-                                detail={
-                                    "detail": f"Plugin '{plugin_name}' is not active for this tenant",
-                                    "code": "plugin_inactive_tenant",
-                                },
-                            )
-                    else:
-                        # No entry = default active (backward compatible)
-                        await redis.setex(cache_key, 60, json.dumps(True))
             else:
-                # No Redis or no tenant_id — fallback to DB query without cache
+                # No entry = default active (backward compatible)
-                async with async_session_maker() as db:
+                if redis is not None:
-                    result = await db.execute(
+                    await redis.setex(cache_key, 60, json.dumps(True))
-                        text("""
-                            SELECT is_active FROM tenant_plugin_activation
-                            WHERE plugin_name = :name
-                            AND tenant_id = current_setting('app.current_tenant_id', true)::uuid
-                        """),
-                        {"name": plugin_name},
-                    )
-                    row = result.first()
-                    if row is not None and not row[0]:
-                        raise HTTPException(
-                            status_code=status.HTTP_403_FORBIDDEN,
-                            detail={
-                                "detail": f"Plugin '{plugin_name}' is not active for this tenant",
-                                "code": "plugin_inactive_tenant",
-                            },
-                        )
         except HTTPException:
             raise
         except Exception as exc:

tests/test_outbox.py

@@ -71,7 +71,7 @@ async def test_process_outbox_batch_publishes_events(
 
         assert count == 1
         assert len(received_events) == 1
-        assert received_events[0][1]["key"] == "value"
+        assert received_events[0][1]["data"]["key"] == "value"
 
         # Verify the event is marked as published
         rows = (