fix(arch-a2): deactivation cleanup - container services, search provider, hook deregistration, notification sync, task state, activation order

app/core/service_container.py

@@ -29,6 +29,14 @@ class ServiceContainer:
         """Check if a service is registered."""
         return name in self._services
 
+    def remove(self, name: str) -> None:
+        """Remove a service registration (no-op if absent).
+
+        Used by plugin deactivation hooks to clean up services they
+        registered during activation.
+        """
+        self._services.pop(name, None)
+
     async def initialize(self) -> None:
         """Initialize core services."""
         if self._initialized:

app/plugins/builtins/ai_ui_control/plugin.py

@@ -62,7 +62,10 @@ class AIUIControlPlugin(BasePlugin):
         from app.plugins.builtins.contracts import get_contract_registry
         get_contract_registry().unregister(self.manifest.name)
 
-        await super().on_deactivate(db, service_container, event_bus)
+        # Remove the WebSocket manager BEFORE super() so that event handlers
+        # being unsubscribed can no longer reach it (ARCH-044).
         if service_container.has("ai_ui_control_ws"):
             service_container.remove("ai_ui_control_ws")
             logger.info("AI UI Control WebSocket manager removed")
+
+        await super().on_deactivate(db, service_container, event_bus)

app/plugins/builtins/graph_rag/plugin.py

@@ -40,6 +40,10 @@ class GraphRAGPlugin(BasePlugin):
 
     async def on_activate(self, db, service_container, event_bus) -> None:
         """Activate plugin: register GraphRAG search provider."""
+        # Base class first (event subscriptions etc.), then own registrations
+        # so a failure in our code cannot leave the base half-initialized (ARCH-037).
+        await super().on_activate(db, service_container, event_bus)
+
         from app.plugins.builtins.graph_rag.provider import GraphRAGSearchProvider
         from app.plugins.builtins.unified_search.contracts import get_search_registry
 
@@ -50,8 +54,6 @@ class GraphRAGPlugin(BasePlugin):
             import logging
             logging.getLogger(__name__).exception("Failed to register GraphRAGSearchProvider")
 
-        await super().on_activate(db, service_container, event_bus)
-
     async def on_deactivate(self, db, service_container, event_bus) -> None:
         """Deactivate plugin: unregister search provider and contract."""
         from app.plugins.builtins.unified_search.contracts import get_search_registry

app/plugins/builtins/knowledge/plugin.py

@@ -149,8 +149,8 @@ class KnowledgePlugin(BasePlugin):
 
     async def on_deactivate(self, db, service_container, event_bus) -> None:
         """Clean up on deactivation."""
-        from app.core.hooks import unregister_actions_by_owner
-        unregister_actions_by_owner("knowledge")
+        from app.core.hooks import get_hook_registry
+        get_hook_registry().unregister_all_for_plugin("knowledge")
         # Unregister knowledge agent tools from the core AI tool registry
         try:
             from app.ai.tool_registry import get_tool_registry

app/plugins/builtins/kommunikation/plugin.py

@@ -136,6 +136,13 @@ class KommunikationPlugin(BasePlugin):
         from app.plugins.builtins.contracts import get_contract_registry
         get_contract_registry().unregister(self.manifest.name)
 
+        # Remove services registered in on_activate BEFORE super() so that
+        # event handlers being unsubscribed can no longer reach them (ARCH-033).
+        for service_name in ("comm_websocket", "comm_miniapps"):
+            if service_container.has(service_name):
+                service_container.remove(service_name)
+                logger.info("Removed '%s' from service container", service_name)
+
         await super().on_deactivate(db, service_container, event_bus)
         logger.info("Kommunikation plugin deactivated")
 

app/plugins/builtins/mail/plugin.py

@@ -115,7 +115,11 @@ async def _auto_sync_loop() -> None:
 class MailPlugin(BasePlugin):
     """Mail plugin for email management: IMAP sync, SMTP send, threading, rules, PGP."""
 
-    _auto_sync_task: asyncio.Task | None = None
+    def __init__(self) -> None:
+        super().__init__()
+        # Instance attribute: multiple plugin instances must not share the
+        # background task state (ARCH-036).
+        self._auto_sync_task: asyncio.Task | None = None
 
     manifest = PluginManifest(
         name="mail",

app/plugins/builtins/self_improvement/services.py

@@ -579,29 +579,30 @@ async def request_approval(
     proposal.approval_request_id = req.id
     await db.flush()
 
-    # Post to Communication via KommunikationContract
+    # Post to Communication via KommunikationContract (ARCH-013: contract
+    # only — no direct plugin imports; skip cleanly when contract is absent)
     try:
         from app.plugins.builtins.contracts import get_contract
-        # Find or create a system conversation for improvement proposals
         _komm = get_contract("kommunikation")
         if not _komm or not hasattr(_komm, "create_plugin_room"):
-            from app.plugins.builtins.kommunikation.services import create_plugin_room
-        room = await create_plugin_room(
-            db=db, tenant_id=tenant_id, user_id=requested_by,
-            plugin_name="self_improvement", title="Improvement Proposals",
-            participant_type="system",
-        )
-        conversation_id = room.get("conversation_id") if isinstance(room, dict) else None
-        if not conversation_id and hasattr(room, "id"):
-            conversation_id = room.id
-        if conversation_id:
-            await KommunikationContract.send_message(
-                db=db,
-                tenant_id=tenant_id,
-                conversation_id=conversation_id,
-                sender_id=requested_by,
-                sender_type="system",
-                content=f"Improvement Proposal: {proposal.title}",
+            logger.warning("kommunikation contract unavailable - skipping proposal notification")
+        else:
+            room = await _komm.create_plugin_room(
+                db=db, tenant_id=tenant_id, user_id=requested_by,
+                plugin_name="self_improvement", title="Improvement Proposals",
+                participant_type="system",
+            )
+            conversation_id = room.get("conversation_id") if isinstance(room, dict) else None
+            if not conversation_id and hasattr(room, "id"):
+                conversation_id = room.id
+            if conversation_id:
+                await _komm.send_message(
+                    db=db,
+                    tenant_id=tenant_id,
+                    conversation_id=conversation_id,
+                    sender_id=requested_by,
+                    sender_type="system",
+                    content=f"Improvement Proposal: {proposal.title}",
                 blocks=[
                     {
                         "type": "action_card",

app/plugins/builtins/wiki/plugin.py

@@ -27,9 +27,9 @@ class WikiPlugin(BasePlugin):
         try:
             from app.plugins.builtins.contracts import get_contract
             search_contract = get_contract("unified_search")
-            if search_contract and hasattr(search_contract, "register_provider"):
+            if search_contract and hasattr(search_contract, "get_search_registry"):
                 from app.plugins.builtins.unified_search.providers.wiki_provider import WikiSearchProvider
-                search_contract.register_provider(WikiSearchProvider())
+                search_contract.get_search_registry().register(WikiSearchProvider())
                 logger.info("Registered WikiSearchProvider via contract")
             else:
                 logger.warning("unified_search contract not available, skipping WikiSearchProvider registration")
@@ -37,6 +37,17 @@ class WikiPlugin(BasePlugin):
             logger.exception("Failed to register WikiSearchProvider")
 
     async def on_deactivate(self, db, service_container, event_bus) -> None:
-        from app.core.hooks import unregister_actions_by_owner
-        unregister_actions_by_owner("wiki")
+        from app.core.hooks import get_hook_registry
+        get_hook_registry().unregister_all_for_plugin("wiki")
+
+        # Unregister the search provider registered in on_activate (ARCH-012).
+        try:
+            from app.plugins.builtins.contracts import get_contract
+            search_contract = get_contract("unified_search")
+            if search_contract and hasattr(search_contract, "get_search_registry"):
+                search_contract.get_search_registry().unregister("wiki_article")
+                logger.info("Unregistered WikiSearchProvider via contract")
+        except Exception:
+            logger.exception("Failed to unregister WikiSearchProvider")
+
         await super().on_deactivate(db, service_container, event_bus)

app/plugins/registry.py

@@ -709,6 +709,13 @@ class PluginRegistry:
         await db.flush()
         self._db_status[name] = record
 
+        # Re-sync notification types so entries of the deactivated plugin are
+        # removed from the DB (ARCH-015) — must run AFTER the status update.
+        try:
+            await self.sync_notification_types(db)
+        except Exception:
+            logger.exception("Failed to sync notification types after deactivating '%s'", name)
+
         # Invalidate Redis cache for this plugin across all tenants
         try:
             from app.core.redis import get_redis

tests/test_arch_block_a.py

@@ -197,3 +197,85 @@ class TestManifestPermissionValidator:
                 display_name='X',
                 permissions=['core:contacts:read'],
             )
+
+
+# --- A2 deactivation cleanup: lifecycle symmetry ---
+
+
+class TestServiceContainerRemove:
+    def test_remove_registered_service(self):
+        from app.core.service_container import ServiceContainer
+
+        container = ServiceContainer()
+        container.register('svc', object())
+        assert container.has('svc') is True
+        container.remove('svc')
+        assert container.has('svc') is False
+
+    def test_remove_absent_service_is_noop(self):
+        from app.core.service_container import ServiceContainer
+
+        container = ServiceContainer()
+        container.remove('never_registered')  # must not raise
+        assert container.has('never_registered') is False
+
+
+class TestKommunikationLifecycle:
+    @pytest.mark.asyncio
+    async def test_deactivate_removes_container_services(self):
+        from app.core.event_bus import EventBus
+        from app.core.service_container import ServiceContainer
+        from app.plugins.builtins.kommunikation.plugin import KommunikationPlugin
+
+        plugin = KommunikationPlugin()
+        container = ServiceContainer()
+        event_bus = EventBus()
+
+        await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
+        assert container.has('comm_websocket') is True
+        assert container.has('comm_miniapps') is True
+
+        await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
+        assert container.has('comm_websocket') is False
+        assert container.has('comm_miniapps') is False
+
+    @pytest.mark.asyncio
+    async def test_reactivate_no_duplicates(self):
+        from app.core.event_bus import EventBus
+        from app.core.service_container import ServiceContainer
+        from app.plugins.builtins.kommunikation.plugin import KommunikationPlugin
+
+        plugin = KommunikationPlugin()
+        container = ServiceContainer()
+        event_bus = EventBus()
+
+        await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
+        await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
+        await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
+
+        ws = container.get('comm_websocket')
+        miniapps = container.get('comm_miniapps')
+        assert ws is not None and miniapps is not None
+        assert len(miniapps._apps) == len(set(miniapps._apps.keys()))
+
+
+class TestWikiProviderLifecycle:
+    @pytest.mark.asyncio
+    async def test_deactivate_unregisters_search_provider(self):
+        from app.core.event_bus import EventBus
+        from app.core.service_container import ServiceContainer
+        from app.plugins.builtins.contracts import get_contract
+        from app.plugins.builtins.wiki.plugin import WikiPlugin
+
+        plugin = WikiPlugin()
+        container = ServiceContainer()
+        event_bus = EventBus()
+
+        search_contract = get_contract('unified_search')
+        registry = search_contract.get_search_registry()
+
+        await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
+        assert registry.get('wiki_article') is not None
+
+        await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
+        assert registry.get('wiki_article') is None