feat(B-HOOK): Lifecycle Hooks + Outbox Events — 55 neue Hooks, 10 Domain Events

B-HOOK-CORE: Company Hooks (6: before/after create/update/delete)
B-HOOK-MAIL: Mail Hooks (5: after_receive, before/after_delete, before/after_move)
B-HOOK-DMS: DMS Hooks (10: after_upload, before/after_update/delete/restore, folder CRUD)
B-HOOK-CAL: Calendar Hooks (4: before/after update/delete)
B-HOOK-TASK: Task Hooks (6: before/after create/update/delete)
B-HOOK-COMM: Communication Hooks (8: conversation create, message/edit/delete)
B-HOOK-AI: Agent Hooks (2: before/after_run)
B-HOOK-WF: Workflow Hooks (4: before/after_start, after_complete/cancel)
B-HOOK-TAG: Tag Hooks (8: create/assign/unassign/delete)
B-HOOK-SEARCH: Search Filters (2: before/after_search)

B-EVT-OUTBOX: 10 Domain Events (task.completed, file.created/deleted/restored, mail.received, workflow.started/completed/cancelled, agent.run_started/completed)

B-HOOK-TEST: 79 Tests in test_lifecycle_hooks.py — alle grün
B-HOOK-DOC: Plugin-Dev-Guide Kapitel 8 aktualisiert (Hook-Liste + Outbox Event-Liste)

app/plugins/builtins/automation/agent_runner.py

@@ -128,6 +128,21 @@ async def run_agent(
         # Reactive mode: use trigger_data as context
         context_data = trigger_data or {}
 
+    # ── Lifecycle: before run ──
+    from app.core.hooks import do_action
+    await do_action("agent.before_run", agent_id=str(agent.id), tenant_id=str(agent.tenant_id), trigger_type=trigger_type)
+    from app.core.outbox import enqueue_outbox_event
+    async with factory() as db:
+        await enqueue_outbox_event(
+            db,
+            agent.tenant_id,
+            'agent.run_started',
+            {'agent_id': str(agent.id), 'tenant_id': str(agent.tenant_id), 'trigger_type': trigger_type},
+            aggregate_type='agent',
+            aggregate_id=agent.id,
+        )
+        await db.commit()
+
     # Call LLM via LiteLLM
     result_data: dict[str, Any] = {
         "agent_id": str(agent.id),
@@ -239,6 +254,21 @@ async def run_agent(
         result_data["status"] = "failed"
         result_data["error"] = str(e)
 
+    # ── Lifecycle: after run ──
+    from app.core.hooks import do_action
+    await do_action("agent.after_run", agent_id=str(agent.id), tenant_id=str(agent.tenant_id), status=result_data.get("status"), result=result_data)
+    from app.core.outbox import enqueue_outbox_event
+    async with factory() as db:
+        await enqueue_outbox_event(
+            db,
+            agent.tenant_id,
+            'agent.run_completed',
+            {'agent_id': str(agent.id), 'tenant_id': str(agent.tenant_id), 'status': result_data.get('status'), 'cost_usd': result_data.get('cost_usd', 0.0)},
+            aggregate_type='agent',
+            aggregate_id=agent.id,
+        )
+        await db.commit()
+
     # Save result to AgentRun
     try:
         async with factory() as db:

app/plugins/builtins/calendar/routes.py

@@ -612,6 +612,8 @@ async def update_entry(
     if not has_write:
         raise HTTPException(403, detail={"detail": "No write permission", "code": "forbidden"})
 
+    from app.core.hooks import do_action
+    await do_action("calendar.before_update", body=body, tenant_id=tenant_id, user_id=user_id, entry_id=entry_id)
     # Apply updates
     if body.title is not None:
         entry.title = body.title
@@ -646,6 +648,8 @@ async def update_entry(
 
     await db.flush()
     await db.refresh(entry)
+    from app.core.hooks import do_action
+    await do_action("calendar.after_update", entry_id=str(entry.id), tenant_id=tenant_id, user_id=user_id)
     return _entry_to_dict(entry)
 
 
@@ -664,8 +668,12 @@ async def delete_entry(
     has_write = await _check_write_permission(db, entry.calendar_id, user_id, role)
     if not has_write:
         raise HTTPException(403, detail={"detail": "No write permission", "code": "forbidden"})
+    from app.core.hooks import do_action
+    await do_action("calendar.before_delete", tenant_id=tenant_id, user_id=user_id, entry_id=entry_id)
     entry.deleted_at = datetime.now(UTC)
     await db.flush()
+    from app.core.hooks import do_action
+    await do_action("calendar.after_delete", tenant_id=tenant_id, user_id=user_id, entry_id=entry_id)
     return Response(status_code=204)
 
 

app/plugins/builtins/dms/routes.py

@@ -257,6 +257,10 @@ async def create_folder(
             409, detail={"detail": "Folder name already exists", "code": "duplicate"}
         )
 
+    # Lifecycle hook: dms.folder.before_create
+    from app.core.hooks import do_action
+    await do_action("dms.folder.before_create", body, db=db, tenant_id=tenant_id, user_id=user_id)
+
     folder = Folder(
         tenant_id=tenant_id,
         name=body.name,
@@ -266,6 +270,9 @@ async def create_folder(
     db.add(folder)
     await db.flush()
 
+    # Lifecycle hook: dms.folder.after_create
+    await do_action("dms.folder.after_create", {'id': str(folder.id), 'name': folder.name, 'parent_id': str(folder.parent_id) if folder.parent_id else None}, db=db, tenant_id=tenant_id, user_id=user_id)
+
     # Build path
     path = body.name
     if parent_id is not None:
@@ -441,6 +448,10 @@ async def delete_folder(
     if not await check_single_entity_access(db, "dms_folder", fid, user_id, tenant_id, "delete", is_system_admin):
         raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"})
 
+    # Lifecycle hook: dms.folder.before_delete
+    from app.core.hooks import do_action
+    await do_action("dms.folder.before_delete", db=db, tenant_id=tenant_id, user_id=user_id, folder_id=str(fid))
+
     from datetime import UTC, datetime
 
     now = datetime.now(UTC)
@@ -476,6 +487,11 @@ async def delete_folder(
     )
 
     await db.flush()
+
+    # Lifecycle hook: dms.folder.after_delete
+    from app.core.hooks import do_action
+    await do_action("dms.folder.after_delete", db=db, tenant_id=tenant_id, user_id=user_id, folder_id=str(fid))
+
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 
@@ -597,6 +613,14 @@ async def upload_file(
         db.add(dms_file)
         await db.flush()
 
+    # Lifecycle hook: dms.after_upload
+    from app.core.hooks import do_action
+    await do_action("dms.after_upload", {'id': str(dms_file.id), 'name': dms_file.name, 'folder_id': str(dms_file.folder_id) if dms_file.folder_id else None, 'mime_type': dms_file.mime_type, 'size_bytes': dms_file.size_bytes}, db=db, tenant_id=tenant_id, user_id=user_id)
+
+    # Outbox event: file.created
+    from app.core.outbox import enqueue_outbox_event
+    await enqueue_outbox_event(db, tenant_id, 'file.created', {'file_id': str(dms_file.id), 'tenant_id': str(tenant_id), 'name': dms_file.name, 'mime_type': dms_file.mime_type, 'size_bytes': dms_file.size_bytes}, aggregate_type='dms_file', aggregate_id=dms_file.id)
+
     return {
         "id": str(dms_file.id),
         "name": dms_file.name,
@@ -763,6 +787,10 @@ async def update_file(
     if not await check_single_entity_access(db, "dms_file", fid, user_id, tenant_id, "write", is_system_admin):
         raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"})
 
+    # Lifecycle hook: dms.before_update
+    from app.core.hooks import do_action
+    await do_action("dms.before_update", body.model_dump(exclude_unset=True), db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id)
+
     data = body.model_dump(exclude_unset=True)
 
     if "name" in data and data["name"] is not None:
@@ -785,6 +813,10 @@ async def update_file(
     await db.flush()
     await db.refresh(dms_file)
 
+    # Lifecycle hook: dms.after_update
+    from app.core.hooks import do_action
+    await do_action("dms.after_update", {'id': str(dms_file.id), 'name': dms_file.name}, db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id)
+
     return {
         "id": str(dms_file.id),
         "name": dms_file.name,
@@ -824,10 +856,22 @@ async def delete_file(
     if not await check_single_entity_access(db, "dms_file", fid, user_id, tenant_id, "delete", is_system_admin):
         raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"})
 
+    # Lifecycle hook: dms.before_delete
+    from app.core.hooks import do_action
+    await do_action("dms.before_delete", db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id)
+
     from datetime import UTC, datetime
 
     dms_file.deleted_at = datetime.now(UTC)
     await db.flush()
+
+    # Lifecycle hook: dms.after_delete
+    await do_action("dms.after_delete", db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id)
+
+    # Outbox event: file.deleted
+    from app.core.outbox import enqueue_outbox_event
+    await enqueue_outbox_event(db, tenant_id, 'file.deleted', {'file_id': str(fid), 'tenant_id': str(tenant_id)}, aggregate_type='dms_file', aggregate_id=fid)
+
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 
@@ -857,10 +901,21 @@ async def restore_file(
     if not await check_single_entity_access(db, "dms_file", fid, user_id, tenant_id, "write", is_system_admin):
         raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"})
 
+    # Lifecycle hook: dms.before_restore
+    from app.core.hooks import do_action
+    await do_action("dms.before_restore", db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id)
+
     dms_file.deleted_at = None
     await db.flush()
     await db.refresh(dms_file)
 
+    # Lifecycle hook: dms.after_restore
+    await do_action("dms.after_restore", {'id': str(dms_file.id), 'name': dms_file.name}, db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id)
+
+    # Outbox event: file.restored
+    from app.core.outbox import enqueue_outbox_event
+    await enqueue_outbox_event(db, tenant_id, 'file.restored', {'file_id': str(fid), 'tenant_id': str(tenant_id)}, aggregate_type='dms_file', aggregate_id=fid)
+
     return {
         "id": str(dms_file.id),
         "name": dms_file.name,

app/plugins/builtins/kommunikation/services.py

@@ -264,8 +264,11 @@ async def create_conversation(
         created_by_type="user",
         metadata_={},
     )
+    from app.core.hooks import do_action
+    await do_action("comm.conversation.before_create", tenant_id=tenant_id, user_id=user_id)
     db.add(conv)
     await db.flush()
+    await do_action("comm.conversation.after_create", conversation_id=conv.id, tenant_id=tenant_id, user_id=user_id)
 
     # Add creator as admin
     creator = CommParticipant(
@@ -669,8 +672,11 @@ async def send_message(
         except ValueError:
             pass
 
+    from app.core.hooks import do_action
+    await do_action("comm.before_message", conversation_id=conversation_id, tenant_id=tenant_id, sender_id=sender_id)
     db.add(msg)
     await db.flush()
+    await do_action("comm.after_message", message_id=msg.id, conversation_id=conversation_id, tenant_id=tenant_id, sender_id=sender_id)
 
     # Create blocks
     if blocks:
@@ -872,10 +878,14 @@ async def edit_message(
     )
     db.add(edit)
 
+    from app.core.hooks import do_action
+    await do_action("comm.before_edit", message_id=message_id, tenant_id=tenant_id, user_id=user_id)
+
     # Update message
     msg.content = new_content
     msg.edited_at = datetime.now(timezone.utc)
     await db.flush()
+    await do_action("comm.after_edit", message_id=message_id, tenant_id=tenant_id, user_id=user_id)
 
     return message_to_response(msg)
 
@@ -891,8 +901,11 @@ async def delete_message(
     msg = result.scalar_one_or_none()
     if msg is None:
         return False
+    from app.core.hooks import do_action
+    await do_action("comm.before_delete", message_id=message_id)
     msg.deleted_at = datetime.now(timezone.utc)
     await db.flush()
+    await do_action("comm.after_delete", message_id=message_id)
     return True
 
 

app/plugins/builtins/mail/services.py

@@ -824,6 +824,14 @@ async def imap_sync_folder(
                 'from_address': from_addr,
             })
 
+            # Lifecycle hook: mail.after_receive
+            from app.core.hooks import do_action
+            await do_action("mail.after_receive", {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'account_id': str(account.id), 'folder_id': str(folder.id), 'subject': subject, 'from_address': from_addr}, db=db, tenant_id=tenant_id)
+
+            # Outbox event: mail.received
+            from app.core.outbox import enqueue_outbox_event
+            await enqueue_outbox_event(db, tenant_id, 'mail.received', {'mail_id': str(mail.id), 'tenant_id': str(tenant_id), 'account_id': str(account.id), 'folder_id': str(folder.id), 'subject': subject, 'from_address': from_addr}, aggregate_type='mail', aggregate_id=mail.id)
+
             # Save attachments
             for att_data in attachments:
                 try:
@@ -2343,6 +2351,10 @@ async def imap_delete_mail(
         logger.warning("imap_delete_mail: mail %s has no imap_uid or message_id, cannot sync", mail_id)
         return
 
+    # Lifecycle hook: mail.before_delete
+    from app.core.hooks import do_action
+    await do_action("mail.before_delete", mail_id=str(mail.id), tenant_id=str(tenant_id), permanent=permanent, db=db)
+
     # Find Trash folder on IMAP server (only needed for non-permanent delete)
     trash_folder_name = None
     if not permanent:
@@ -2409,6 +2421,10 @@ async def imap_delete_mail(
         await client.expunge()
         logger.info("imap_delete_mail: moved mail %s (UID %s) to Trash via COPY+DELETE", mail_id, uid_str)
 
+        # Lifecycle hook: mail.after_delete
+        from app.core.hooks import do_action
+        await do_action("mail.after_delete", mail_id=str(mail.id), tenant_id=str(tenant_id), permanent=permanent, db=db)
+
     except Exception as exc:
         logger.warning("imap_delete_mail: failed for mail %s: %s", mail_id, exc)
         raise
@@ -2487,6 +2503,10 @@ async def imap_move_mail(
         logger.warning("imap_move_mail: mail %s has no imap_uid or message_id, cannot sync", mail_id)
         return
 
+    # Lifecycle hook: mail.before_move
+    from app.core.hooks import do_action
+    await do_action("mail.before_move", mail_id=str(mail.id), tenant_id=str(tenant_id), source_folder_id=str(source_folder.id), target_folder_id=str(target_folder.id), db=db)
+
     password = await get_account_password(account)
     client = None
 
@@ -2532,6 +2552,10 @@ async def imap_move_mail(
 
         logger.info("imap_move_mail: moved mail %s (UID %s) via COPY+DELETE", mail_id, uid_str)
 
+        # Lifecycle hook: mail.after_move
+        from app.core.hooks import do_action
+        await do_action("mail.after_move", mail_id=str(mail.id), tenant_id=str(tenant_id), source_folder_id=str(source_folder.id), target_folder_id=str(target_folder.id), db=db)
+
     except Exception as exc:
         logger.warning("imap_move_mail: failed for mail %s: %s", mail_id, exc)
         raise

app/plugins/builtins/tags/routes.py

@@ -95,9 +95,13 @@ async def create_tag(
     if existing.scalar_one_or_none() is not None:
         raise HTTPException(409, detail={"detail": "Tag name already exists", "code": "duplicate"})
 
+    from app.core.hooks import do_action
+    await do_action("tag.before_create", body, db=db, tenant_id=tenant_id, user_id=user_id)
     tag = Tag(tenant_id=tenant_id, name=body.name, color=body.color, owner_id=user_id)
     db.add(tag)
     await db.flush()
+    from app.core.hooks import do_action
+    await do_action("tag.after_create", {"id": str(tag.id), "name": tag.name, "color": tag.color}, db=db, tenant_id=tenant_id, user_id=user_id)
     return {
         "id": str(tag.id),
         "name": tag.name,
@@ -185,6 +189,8 @@ async def assign_tag(
             "already_assigned": True,
         }
 
+    from app.core.hooks import do_action
+    await do_action("tag.before_assign", body, db=db, tenant_id=tenant_id, user_id=user_id)
     assignment = TagAssignment(
         tenant_id=tenant_id,
         tag_id=tag_id,
@@ -193,6 +199,8 @@ async def assign_tag(
     )
     db.add(assignment)
     await db.flush()
+    from app.core.hooks import do_action
+    await do_action("tag.after_assign", {"id": str(assignment.id), "tag_id": str(tag_id), "entity_type": body.entity_type, "entity_id": str(entity_id)}, db=db, tenant_id=tenant_id, user_id=user_id)
     return {
         "id": str(assignment.id),
         "tag_id": str(tag_id),
@@ -225,7 +233,11 @@ async def unassign_tag(
     if assignment is None:
         raise HTTPException(404, detail={"detail": "Assignment not found", "code": "not_found"})
 
+    from app.core.hooks import do_action
+    await do_action("tag.before_unassign", body, db=db, tenant_id=tenant_id, user_id=user_id)
     await db.delete(assignment)
+    from app.core.hooks import do_action
+    await do_action("tag.after_unassign", {"tag_id": str(tag_id), "entity_type": body.entity_type, "entity_id": str(entity_id)}, db=db, tenant_id=tenant_id, user_id=user_id)
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 
@@ -244,6 +256,8 @@ async def delete_tag(
     if tag is None:
         raise HTTPException(404, detail={"detail": "Tag not found", "code": "not_found"})
 
+    from app.core.hooks import do_action
+    await do_action("tag.before_delete", db=db, tenant_id=tenant_id, user_id=user_id, tag_id=tag_id)
     # Cascade delete assignments
     await db.execute(
         delete(TagAssignment).where(
@@ -251,6 +265,8 @@ async def delete_tag(
         )
     )
     await db.delete(tag)
+    from app.core.hooks import do_action
+    await do_action("tag.after_delete", db=db, tenant_id=tenant_id, user_id=user_id, tag_id=tag_id)
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 

app/plugins/builtins/tasks/services.py

@@ -101,6 +101,8 @@ async def create_task(
     data: dict[str, Any],
 ) -> dict[str, Any]:
     """Create a new task."""
+    from app.core.hooks import do_action
+    await do_action("task.before_create", data, db=db, tenant_id=tenant_id, user_id=user_id)
     task = Task(
         tenant_id=tenant_id,
         title=data["title"],
@@ -116,6 +118,9 @@ async def create_task(
     db.add(task)
     await db.flush()
 
+    from app.core.hooks import do_action
+    await do_action("task.after_create", _task_to_dict(task), db=db, tenant_id=tenant_id, user_id=user_id)
+
     # Publish task.created event
     from app.core.event_bus import get_event_bus
     event_bus = get_event_bus()
@@ -144,6 +149,8 @@ async def update_task(
     if task is None:
         return None
 
+    from app.core.hooks import do_action
+    await do_action("task.before_update", data, db=db, tenant_id=tenant_id, task_id=str(task_id))
     if "title" in data and data["title"] is not None:
         task.title = data["title"]
     if "description" in data:
@@ -161,6 +168,8 @@ async def update_task(
 
     await db.flush()
     await db.refresh(task)
+    from app.core.hooks import do_action
+    await do_action("task.after_update", _task_to_dict(task), db=db, tenant_id=tenant_id, task_id=str(task_id))
     return _task_to_dict(task)
 
 
@@ -172,8 +181,12 @@ async def delete_task(db: AsyncSession, tenant_id: uuid.UUID, task_id: uuid.UUID
     task = result.scalar_one_or_none()
     if task is None:
         return False
+    from app.core.hooks import do_action
+    await do_action("task.before_delete", db=db, tenant_id=tenant_id, task_id=str(task_id))
     task.deleted_at = datetime.now(timezone.utc)
     await db.flush()
+    from app.core.hooks import do_action
+    await do_action("task.after_delete", db=db, tenant_id=tenant_id, task_id=str(task_id))
     return True
 
 
@@ -210,6 +223,9 @@ async def update_task_status(
         return None
     task.status = new_status
     await db.flush()
+    if new_status == "done":
+        from app.core.outbox import enqueue_outbox_event
+        await enqueue_outbox_event(db, tenant_id, "task.completed", {"task_id": str(task.id), "tenant_id": str(tenant_id), "title": task.title, "assigned_to": str(task.assigned_to) if task.assigned_to else None}, aggregate_type="task", aggregate_id=task.id)
     return _task_to_dict(task)
 
 

app/plugins/builtins/unified_search/search_engine.py

@@ -101,6 +101,8 @@ async def hybrid_search(
         query_text = f"{normalized_query} {' '.join(semantic_terms)}"
     query_embedding = await generate_embedding(query_text, db=db, tenant_id=tenant_id)
 
+    from app.core.hooks import apply_filters
+    query_analysis = await apply_filters("search.before_search", query_analysis)
     all_results: list[dict[str, Any]] = []
     fetch_limit = limit * 2
 
@@ -132,6 +134,8 @@ async def hybrid_search(
             all_results.append(result)
 
     all_results.sort(key=lambda x: x.get("score", 0.0), reverse=True)
+    from app.core.hooks import apply_filters
+    all_results = await apply_filters("search.after_search", all_results)
     return all_results[:limit]
 
 

app/routes/companies.py

@@ -101,6 +101,8 @@ async def create_company(
     for k, v in body.items():
         if k not in ("name", "industry", "description"):
             custom[k] = v
+    from app.core.hooks import do_action
+    await do_action("company.before_create", body, db=db, tenant_id=tenant_id, user_id=user_id)
     company = Contact(
         tenant_id=tenant_id, type="company", name=name, displayname=name,
         status=body.get("status", "lead"), custom=custom,
@@ -115,6 +117,8 @@ async def create_company(
     )
     db.add(audit_entry)
     await db.flush()
+    from app.core.hooks import do_action
+    await do_action("company.after_create", _serialize_company(company), db=db, tenant_id=tenant_id, user_id=user_id)
     return _serialize_company(company)
 
 
@@ -202,6 +206,8 @@ async def update_company(
     company = result.scalar_one_or_none()
     if not company:
         raise HTTPException(status_code=404, detail="Company not found")
+    from app.core.hooks import do_action
+    await do_action("company.before_update", body, db=db, tenant_id=tenant_id, user_id=user_id, company_id=company_id)
     if "name" in body:
         company.name = body["name"]
         company.displayname = body["name"]
@@ -220,6 +226,8 @@ async def update_company(
                           entity_type="contact", entity_id=company.id, changes=body)
     db.add(audit_entry)
     await db.flush()
+    from app.core.hooks import do_action
+    await do_action("company.after_update", _serialize_company(company), db=db, tenant_id=tenant_id, user_id=user_id, company_id=company_id)
     return _serialize_company(company)
 
 
@@ -241,6 +249,8 @@ async def delete_company(
     if not company:
         raise HTTPException(status_code=404, detail="Company not found")
     snapshot = _serialize_company(company)
+    from app.core.hooks import do_action
+    await do_action("company.before_delete", db=db, tenant_id=tenant_id, user_id=user_id, company_id=company_id)
     # Soft-delete contact persons via SQL to avoid lazy loading
     company.deleted_at = datetime.now(UTC)
     company.updated_by = user_id
@@ -253,6 +263,8 @@ async def delete_company(
                           entity_type="contact", entity_id=company.id, changes={"name": company.name})
     db.add(audit_entry)
     await db.flush()
+    from app.core.hooks import do_action
+    await do_action("company.after_delete", snapshot, db=db, tenant_id=tenant_id, user_id=user_id, company_id=company_id)
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 

app/services/workflow_service.py

@@ -352,9 +352,21 @@ async def create_instance(
         timeout_hours=timeout_hours,
         timeout_at=timeout_at,
     )
+    from app.core.hooks import do_action
+    await do_action("workflow.before_start", instance_id=instance.id, workflow_id=wf_uuid, tenant_id=tenant_id, user_id=user_id)
     db.add(instance)
     await db.flush()
     await db.refresh(instance)
+    await do_action("workflow.after_start", instance_id=instance.id, workflow_id=wf_uuid, tenant_id=tenant_id, user_id=user_id)
+    from app.core.outbox import enqueue_outbox_event
+    await enqueue_outbox_event(
+        db,
+        tenant_id,
+        'workflow.started',
+        {'instance_id': str(instance.id), 'workflow_id': str(wf_uuid), 'tenant_id': str(tenant_id)},
+        aggregate_type='workflow_instance',
+        aggregate_id=instance.id,
+    )
 
     # Log initial step entry
     steps = workflow.steps or []
@@ -567,6 +579,17 @@ async def advance_instance(
             action="completed",
             actor_id=user_id,
         )
+        from app.core.hooks import do_action
+        await do_action("workflow.after_complete", instance_id=instance.id, workflow_id=instance.workflow_id, tenant_id=tenant_id, user_id=user_id)
+        from app.core.outbox import enqueue_outbox_event
+        await enqueue_outbox_event(
+            db,
+            tenant_id,
+            'workflow.completed',
+            {'instance_id': str(instance.id), 'workflow_id': str(instance.workflow_id), 'tenant_id': str(tenant_id), 'status': 'completed'},
+            aggregate_type='workflow_instance',
+            aggregate_id=instance.id,
+        )
     else:
         instance.current_step_index = next_idx
         next_step = steps[next_idx]
@@ -626,6 +649,18 @@ async def cancel_instance(
     instance.status = "cancelled"
     instance.completed_at = datetime.now(UTC)
 
+    from app.core.hooks import do_action
+    await do_action("workflow.after_cancel", instance_id=instance.id, workflow_id=instance.workflow_id, tenant_id=tenant_id, user_id=user_id)
+    from app.core.outbox import enqueue_outbox_event
+    await enqueue_outbox_event(
+        db,
+        tenant_id,
+        'workflow.cancelled',
+        {'instance_id': str(instance.id), 'workflow_id': str(instance.workflow_id), 'tenant_id': str(tenant_id), 'status': 'cancelled'},
+        aggregate_type='workflow_instance',
+        aggregate_id=instance.id,
+    )
+
     # Get current step for history
     wf_result = await db.execute(select(Workflow).where(Workflow.id == instance.workflow_id))
     workflow = wf_result.scalar_one_or_none()

docs/plugin-development-guide.md

@@ -1003,6 +1003,88 @@ await do_action("contact.before_create", contact_data, db=db)
 display_name = await apply_filters("contact.format_display_name", contact.name)
 ```
 
+#### Verfügbare Hooks
+
+**Actions** (fire-and-forget, kein Return-Wert):
+
+| Hook Name | Modul | Parameter |
+|-----------|-------|----------|
+| `contact.before_create` | `app/services/contact_service.py` | `data, db, tenant_id, user_id` |
+| `contact.after_create` | `app/services/contact_service.py` | `serialized, db, tenant_id, user_id` |
+| `contact.before_update` | `app/services/contact_service.py` | `data, db, tenant_id, user_id, contact_id` |
+| `contact.after_update` | `app/services/contact_service.py` | `snapshot, db, tenant_id, user_id, contact_id` |
+| `contact.before_delete` | `app/services/contact_service.py` | `db, tenant_id, contact_id, user_id` |
+| `contact.after_delete` | `app/services/contact_service.py` | `db, tenant_id, contact_id, user_id` |
+| `company.before_create` | `app/routes/companies.py` | `body, db, tenant_id, user_id` |
+| `company.after_create` | `app/routes/companies.py` | `serialized, db, tenant_id, user_id` |
+| `company.before_update` | `app/routes/companies.py` | `body, db, tenant_id, user_id, company_id` |
+| `company.after_update` | `app/routes/companies.py` | `serialized, db, tenant_id, user_id, company_id` |
+| `company.before_delete` | `app/routes/companies.py` | `db, tenant_id, user_id, company_id` |
+| `company.after_delete` | `app/routes/companies.py` | `snapshot, db, tenant_id, user_id, company_id` |
+| `mail.before_send` | `app/plugins/builtins/mail/services.py` | `mail_data` (Filter) |
+| `mail.after_send` | `app/plugins/builtins/mail/services.py` | `mail_data, db, account, msg_id` |
+| `mail.after_receive` | `app/plugins/builtins/mail/services.py` | `payload, db, tenant_id` |
+| `mail.before_delete` | `app/plugins/builtins/mail/services.py` | `mail_id, tenant_id, permanent, db` |
+| `mail.after_delete` | `app/plugins/builtins/mail/services.py` | `mail_id, tenant_id, permanent, db` |
+| `mail.before_move` | `app/plugins/builtins/mail/services.py` | `mail_id, tenant_id, source_folder_id, target_folder_id, db` |
+| `mail.after_move` | `app/plugins/builtins/mail/services.py` | `mail_id, tenant_id, source_folder_id, target_folder_id, db` |
+| `dms.before_upload` | `app/plugins/builtins/dms/routes.py` | `upload_data` (Filter) |
+| `dms.after_upload` | `app/plugins/builtins/dms/routes.py` | `payload, db, tenant_id, user_id` |
+| `dms.before_update` | `app/plugins/builtins/dms/routes.py` | `data, db, tenant_id, user_id, file_id` |
+| `dms.after_update` | `app/plugins/builtins/dms/routes.py` | `payload, db, tenant_id, user_id, file_id` |
+| `dms.before_delete` | `app/plugins/builtins/dms/routes.py` | `db, tenant_id, user_id, file_id` |
+| `dms.after_delete` | `app/plugins/builtins/dms/routes.py` | `db, tenant_id, user_id, file_id` |
+| `dms.before_restore` | `app/plugins/builtins/dms/routes.py` | `db, tenant_id, user_id, file_id` |
+| `dms.after_restore` | `app/plugins/builtins/dms/routes.py` | `payload, db, tenant_id, user_id, file_id` |
+| `dms.folder.before_create` | `app/plugins/builtins/dms/routes.py` | `body, db, tenant_id, user_id` |
+| `dms.folder.after_create` | `app/plugins/builtins/dms/routes.py` | `payload, db, tenant_id, user_id` |
+| `dms.folder.before_delete` | `app/plugins/builtins/dms/routes.py` | `db, tenant_id, user_id, folder_id` |
+| `dms.folder.after_delete` | `app/plugins/builtins/dms/routes.py` | `db, tenant_id, user_id, folder_id` |
+| `calendar.before_appointment` | `app/plugins/builtins/calendar/routes.py` | `body, tenant_id, user_id, cal_id` |
+| `calendar.after_appointment` | `app/plugins/builtins/calendar/routes.py` | `entry_id, tenant_id, user_id` |
+| `calendar.before_update` | `app/plugins/builtins/calendar/routes.py` | `body, tenant_id, user_id, entry_id` |
+| `calendar.after_update` | `app/plugins/builtins/calendar/routes.py` | `entry_id, tenant_id, user_id` |
+| `calendar.before_delete` | `app/plugins/builtins/calendar/routes.py` | `tenant_id, user_id, entry_id` |
+| `calendar.after_delete` | `app/plugins/builtins/calendar/routes.py` | `tenant_id, user_id, entry_id` |
+| `task.before_create` | `app/plugins/builtins/tasks/services.py` | `data, db, tenant_id, user_id` |
+| `task.after_create` | `app/plugins/builtins/tasks/services.py` | `serialized, db, tenant_id, user_id` |
+| `task.before_update` | `app/plugins/builtins/tasks/services.py` | `data, db, tenant_id, task_id` |
+| `task.after_update` | `app/plugins/builtins/tasks/services.py` | `serialized, db, tenant_id, task_id` |
+| `task.before_delete` | `app/plugins/builtins/tasks/services.py` | `db, tenant_id, task_id` |
+| `task.after_delete` | `app/plugins/builtins/tasks/services.py` | `db, tenant_id, task_id` |
+| `comm.conversation.before_create` | `app/plugins/builtins/kommunikation/services.py` | `tenant_id, user_id` |
+| `comm.conversation.after_create` | `app/plugins/builtins/kommunikation/services.py` | `conversation_id, tenant_id, user_id` |
+| `comm.before_message` | `app/plugins/builtins/kommunikation/services.py` | `conversation_id, tenant_id, sender_id` |
+| `comm.after_message` | `app/plugins/builtins/kommunikation/services.py` | `message_id, conversation_id, tenant_id, sender_id` |
+| `comm.before_edit` | `app/plugins/builtins/kommunikation/services.py` | `message_id, tenant_id, user_id` |
+| `comm.after_edit` | `app/plugins/builtins/kommunikation/services.py` | `message_id, tenant_id, user_id` |
+| `comm.before_delete` | `app/plugins/builtins/kommunikation/services.py` | `message_id` |
+| `comm.after_delete` | `app/plugins/builtins/kommunikation/services.py` | `message_id` |
+| `agent.before_run` | `app/plugins/builtins/automation/agent_runner.py` | `agent_id, tenant_id, trigger_type` |
+| `agent.after_run` | `app/plugins/builtins/automation/agent_runner.py` | `agent_id, tenant_id, status, result` |
+| `workflow.before_start` | `app/services/workflow_service.py` | `instance_id, workflow_id, tenant_id, user_id` |
+| `workflow.after_start` | `app/services/workflow_service.py` | `instance_id, workflow_id, tenant_id, user_id` |
+| `workflow.after_complete` | `app/services/workflow_service.py` | `instance_id, workflow_id, tenant_id, user_id` |
+| `workflow.after_cancel` | `app/services/workflow_service.py` | `instance_id, workflow_id, tenant_id, user_id` |
+| `tag.before_create` | `app/plugins/builtins/tags/routes.py` | `body, db, tenant_id, user_id` |
+| `tag.after_create` | `app/plugins/builtins/tags/routes.py` | `payload, db, tenant_id, user_id` |
+| `tag.before_assign` | `app/plugins/builtins/tags/routes.py` | `body, db, tenant_id, user_id` |
+| `tag.after_assign` | `app/plugins/builtins/tags/routes.py` | `payload, db, tenant_id, user_id` |
+| `tag.before_unassign` | `app/plugins/builtins/tags/routes.py` | `body, db, tenant_id, user_id` |
+| `tag.after_unassign` | `app/plugins/builtins/tags/routes.py` | `payload, db, tenant_id, user_id` |
+| `tag.before_delete` | `app/plugins/builtins/tags/routes.py` | `db, tenant_id, user_id, tag_id` |
+| `tag.after_delete` | `app/plugins/builtins/tags/routes.py` | `db, tenant_id, user_id, tag_id` |
+
+**Filters** (modifizieren Wert, Return erforderlich):
+
+| Hook Name | Modul | Parameter |
+|-----------|-------|----------|
+| `contact.format_display_name` | `app/services/contact_service.py` | `name, data, db, tenant_id, user_id` |
+| `dms.before_upload` | `app/plugins/builtins/dms/routes.py` | `upload_data` (filename, mime_type) |
+| `mail.before_send` | `app/plugins/builtins/mail/services.py` | `mail_data` |
+| `search.before_search` | `app/plugins/builtins/unified_search/search_engine.py` | `query_analysis` |
+| `search.after_search` | `app/plugins/builtins/unified_search/search_engine.py` | `all_results` |
+
 ### 8.2 EventBus (`app/core/event_bus.py`)
 
 **Wann verwenden:** Für flüchtige interne Notifikationen, UI-Events, Proactive Suggestions. **Nicht** für Events die reliable Delivery brauchen.
@@ -1036,6 +1118,26 @@ await enqueue_outbox_event(db, tenant_id, "contact.created", {
 
 **Features:** DLQ (`error_message`, `failed_at`), Replay (`replay_failed_event`), Consumer Registry, Stats.
 
+#### Verfügbare Outbox Events
+
+| Event Name | Modul | Payload | Aggregate |
+|------------|-------|---------|-----------|
+| `contact.created` | `app/services/contact_service.py` | `contact_id, tenant_id, displayname, type` | `contact` |
+| `contact.updated` | `app/services/contact_service.py` | `contact_id, tenant_id, changes` | `contact` |
+| `lead.created` | `app/services/contact_service.py` | `contact_id, tenant_id` | `contact` |
+| `mail.received` | `app/plugins/builtins/mail/services.py` | `mail_id, tenant_id, account_id, folder_id, subject, from_address` | `mail` |
+| `mail.send` | `app/plugins/builtins/mail/services.py` | `mail_id, tenant_id, account_id` | `mail` |
+| `file.created` | `app/plugins/builtins/dms/routes.py` | `file_id, tenant_id, name, mime_type, size_bytes` | `dms_file` |
+| `file.deleted` | `app/plugins/builtins/dms/routes.py` | `file_id, tenant_id` | `dms_file` |
+| `file.restored` | `app/plugins/builtins/dms/routes.py` | `file_id, tenant_id` | `dms_file` |
+| `dms.file.uploaded` | `app/plugins/builtins/dms/routes.py` | (legacy alias for `file.created`) | `dms_file` |
+| `task.completed` | `app/plugins/builtins/tasks/services.py` | `task_id, tenant_id, title, assigned_to` | `task` |
+| `workflow.started` | `app/services/workflow_service.py` | `instance_id, workflow_id, tenant_id` | `workflow_instance` |
+| `workflow.completed` | `app/services/workflow_service.py` | `instance_id, workflow_id, tenant_id, status` | `workflow_instance` |
+| `workflow.cancelled` | `app/services/workflow_service.py` | `instance_id, workflow_id, tenant_id` | `workflow_instance` |
+| `agent.run_started` | `app/plugins/builtins/automation/agent_runner.py` | `agent_id, tenant_id, trigger_type` | `agent` |
+| `agent.run_completed` | `app/plugins/builtins/automation/agent_runner.py` | `agent_id, tenant_id, status, cost_usd` | `agent` |
+
 ### 8.4 WebhookDispatcher (`app/core/webhook_dispatcher.py`)
 
 **Wann verwenden:** Für externe HTTP-Zustellung an registrierte Webhook-URLs.

tests/test_lifecycle_hooks.py

@@ -0,0 +1,830 @@
+"""Tests for lifecycle hooks and outbox events across all modules.
+
+Verifies that:
+- Actions fire via do_action when registered
+- Filters modify values via apply_filters when registered
+- New hooks are called at the correct lifecycle points
+- Outbox events are enqueued for domain events
+- Sensitive fields are excluded from hook payloads via sanitize_dict
+- UI/read-only events are not accidentally durable
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from app.core.hooks import (
+    HookRegistry,
+    apply_filters,
+    do_action,
+    get_hook_registry,
+    reset_hook_registry_for_testing,
+)
+from app.core.sensitive_data import sanitize_dict, SENSITIVE_FIELDS
+
+
+@pytest.fixture(autouse=True)
+def clean_registry():
+    """Reset the hook registry before and after each test."""
+    reset_hook_registry_for_testing()
+    yield
+    reset_hook_registry_for_testing()
+
+
+# ─── Hook Firing Tests ──────────────────────────────────────────────────────
+
+
+class TestHookFiring:
+    """Verify that do_action fires for each new hook name."""
+
+    @pytest.mark.asyncio
+    async def test_company_before_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("company.before_create", lambda *a, **kw: called.append(kw))
+        await do_action("company.before_create", body={"name": "Acme"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_company_after_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("company.after_create", lambda *a, **kw: called.append(kw))
+        await do_action("company.after_create", {"id": "x", "name": "Acme"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_company_before_update_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("company.before_update", lambda *a, **kw: called.append(kw))
+        await do_action("company.before_update", {"name": "New"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), company_id="abc")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_company_after_update_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("company.after_update", lambda *a, **kw: called.append(kw))
+        await do_action("company.after_update", {"id": "x"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), company_id="abc")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_company_before_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("company.before_delete", lambda *a, **kw: called.append(kw))
+        await do_action("company.before_delete", db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), company_id="abc")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_company_after_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("company.after_delete", lambda *a, **kw: called.append(kw))
+        await do_action("company.after_delete", {"name": "Acme"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), company_id="abc")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_mail_after_receive_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("mail.after_receive", lambda *a, **kw: called.append(kw))
+        await do_action("mail.after_receive", {"mail_id": "x", "subject": "Test"}, db=None, tenant_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_mail_before_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("mail.before_delete", lambda *a, **kw: called.append(kw))
+        await do_action("mail.before_delete", mail_id="x", tenant_id="t", permanent=False, db=None)
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_mail_after_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("mail.after_delete", lambda *a, **kw: called.append(kw))
+        await do_action("mail.after_delete", mail_id="x", tenant_id="t", permanent=True, db=None)
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_mail_before_move_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("mail.before_move", lambda *a, **kw: called.append(kw))
+        await do_action("mail.before_move", mail_id="x", tenant_id="t", source_folder_id="s", target_folder_id="d", db=None)
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_mail_after_move_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("mail.after_move", lambda *a, **kw: called.append(kw))
+        await do_action("mail.after_move", mail_id="x", tenant_id="t", source_folder_id="s", target_folder_id="d", db=None)
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_after_upload_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.after_upload", lambda *a, **kw: called.append(kw))
+        await do_action("dms.after_upload", {"id": "x", "name": "file.pdf"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_before_update_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.before_update", lambda *a, **kw: called.append(kw))
+        await do_action("dms.before_update", {"name": "new.pdf"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), file_id="f")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_after_update_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.after_update", lambda *a, **kw: called.append(kw))
+        await do_action("dms.after_update", {"id": "x", "name": "new.pdf"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), file_id="f")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_before_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.before_delete", lambda *a, **kw: called.append(kw))
+        await do_action("dms.before_delete", db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), file_id="f")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_after_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.after_delete", lambda *a, **kw: called.append(kw))
+        await do_action("dms.after_delete", db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), file_id="f")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_before_restore_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.before_restore", lambda *a, **kw: called.append(kw))
+        await do_action("dms.before_restore", db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), file_id="f")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_after_restore_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.after_restore", lambda *a, **kw: called.append(kw))
+        await do_action("dms.after_restore", {"id": "x", "name": "file.pdf"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), file_id="f")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_folder_before_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.folder.before_create", lambda *a, **kw: called.append(kw))
+        await do_action("dms.folder.before_create", {"name": "New Folder"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_folder_after_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.folder.after_create", lambda *a, **kw: called.append(kw))
+        await do_action("dms.folder.after_create", {"id": "x", "name": "Folder"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_folder_before_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.folder.before_delete", lambda *a, **kw: called.append(kw))
+        await do_action("dms.folder.before_delete", db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), folder_id="f")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_dms_folder_after_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("dms.folder.after_delete", lambda *a, **kw: called.append(kw))
+        await do_action("dms.folder.after_delete", db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), folder_id="f")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_calendar_before_update_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("calendar.before_update", lambda *a, **kw: called.append(kw))
+        await do_action("calendar.before_update", body={}, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), entry_id="e")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_calendar_after_update_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("calendar.after_update", lambda *a, **kw: called.append(kw))
+        await do_action("calendar.after_update", entry_id="e", tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_calendar_before_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("calendar.before_delete", lambda *a, **kw: called.append(kw))
+        await do_action("calendar.before_delete", tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), entry_id="e")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_calendar_after_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("calendar.after_delete", lambda *a, **kw: called.append(kw))
+        await do_action("calendar.after_delete", tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), entry_id="e")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_task_before_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("task.before_create", lambda *a, **kw: called.append(kw))
+        await do_action("task.before_create", {"title": "Test"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_task_after_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("task.after_create", lambda *a, **kw: called.append(kw))
+        await do_action("task.after_create", {"id": "x", "title": "Test"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_task_before_update_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("task.before_update", lambda *a, **kw: called.append(kw))
+        await do_action("task.before_update", {"title": "Updated"}, db=None, tenant_id=uuid.uuid4(), task_id="t")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_task_after_update_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("task.after_update", lambda *a, **kw: called.append(kw))
+        await do_action("task.after_update", {"id": "x"}, db=None, tenant_id=uuid.uuid4(), task_id="t")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_task_before_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("task.before_delete", lambda *a, **kw: called.append(kw))
+        await do_action("task.before_delete", db=None, tenant_id=uuid.uuid4(), task_id="t")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_task_after_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("task.after_delete", lambda *a, **kw: called.append(kw))
+        await do_action("task.after_delete", db=None, tenant_id=uuid.uuid4(), task_id="t")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_comm_conversation_before_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("comm.conversation.before_create", lambda *a, **kw: called.append(kw))
+        await do_action("comm.conversation.before_create", tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_comm_conversation_after_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("comm.conversation.after_create", lambda *a, **kw: called.append(kw))
+        await do_action("comm.conversation.after_create", conversation_id=uuid.uuid4(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_comm_before_message_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("comm.before_message", lambda *a, **kw: called.append(kw))
+        await do_action("comm.before_message", conversation_id="c", tenant_id=uuid.uuid4(), sender_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_comm_after_message_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("comm.after_message", lambda *a, **kw: called.append(kw))
+        await do_action("comm.after_message", message_id=uuid.uuid4(), conversation_id="c", tenant_id=uuid.uuid4(), sender_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_comm_before_edit_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("comm.before_edit", lambda *a, **kw: called.append(kw))
+        await do_action("comm.before_edit", message_id="m", tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_comm_after_edit_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("comm.after_edit", lambda *a, **kw: called.append(kw))
+        await do_action("comm.after_edit", message_id="m", tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_comm_before_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("comm.before_delete", lambda *a, **kw: called.append(kw))
+        await do_action("comm.before_delete", message_id="m")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_comm_after_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("comm.after_delete", lambda *a, **kw: called.append(kw))
+        await do_action("comm.after_delete", message_id="m")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_agent_before_run_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("agent.before_run", lambda *a, **kw: called.append(kw))
+        await do_action("agent.before_run", agent_id="a", tenant_id="t", trigger_type="manual")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_agent_after_run_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("agent.after_run", lambda *a, **kw: called.append(kw))
+        await do_action("agent.after_run", agent_id="a", tenant_id="t", status="completed", result={})
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_workflow_before_start_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("workflow.before_start", lambda *a, **kw: called.append(kw))
+        await do_action("workflow.before_start", instance_id=uuid.uuid4(), workflow_id=uuid.uuid4(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_workflow_after_start_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("workflow.after_start", lambda *a, **kw: called.append(kw))
+        await do_action("workflow.after_start", instance_id=uuid.uuid4(), workflow_id=uuid.uuid4(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_workflow_after_complete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("workflow.after_complete", lambda *a, **kw: called.append(kw))
+        await do_action("workflow.after_complete", instance_id=uuid.uuid4(), workflow_id=uuid.uuid4(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_workflow_after_cancel_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("workflow.after_cancel", lambda *a, **kw: called.append(kw))
+        await do_action("workflow.after_cancel", instance_id=uuid.uuid4(), workflow_id=uuid.uuid4(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_tag_before_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("tag.before_create", lambda *a, **kw: called.append(kw))
+        await do_action("tag.before_create", {"name": "VIP"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_tag_after_create_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("tag.after_create", lambda *a, **kw: called.append(kw))
+        await do_action("tag.after_create", {"id": "x", "name": "VIP"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_tag_before_assign_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("tag.before_assign", lambda *a, **kw: called.append(kw))
+        await do_action("tag.before_assign", {}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_tag_after_assign_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("tag.after_assign", lambda *a, **kw: called.append(kw))
+        await do_action("tag.after_assign", {"tag_id": "t", "entity_type": "contact", "entity_id": "e"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_tag_before_unassign_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("tag.before_unassign", lambda *a, **kw: called.append(kw))
+        await do_action("tag.before_unassign", {}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_tag_after_unassign_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("tag.after_unassign", lambda *a, **kw: called.append(kw))
+        await do_action("tag.after_unassign", {"tag_id": "t"}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_tag_before_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("tag.before_delete", lambda *a, **kw: called.append(kw))
+        await do_action("tag.before_delete", db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), tag_id="t")
+        assert len(called) == 1
+
+    @pytest.mark.asyncio
+    async def test_tag_after_delete_hook(self):
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("tag.after_delete", lambda *a, **kw: called.append(kw))
+        await do_action("tag.after_delete", db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), tag_id="t")
+        assert len(called) == 1
+
+
+# ─── Filter Tests ───────────────────────────────────────────────────────────
+
+
+class TestHookFilters:
+    """Verify that apply_filters modifies values for filter hooks."""
+
+    @pytest.mark.asyncio
+    async def test_search_before_search_filter(self):
+        reg = get_hook_registry()
+        reg.register_filter("search.before_search", lambda q: {**q, "modified": True})
+        result = await apply_filters("search.before_search", {"query": "test"})
+        assert result["modified"] is True
+        assert result["query"] == "test"
+
+    @pytest.mark.asyncio
+    async def test_search_after_search_filter(self):
+        reg = get_hook_registry()
+        reg.register_filter("search.after_search", lambda results: results + [{"extra": True}])
+        result = await apply_filters("search.after_search", [{"id": "1"}])
+        assert len(result) == 2
+        assert result[1]["extra"] is True
+
+    @pytest.mark.asyncio
+    async def test_dms_before_upload_filter_still_works(self):
+        """Existing dms.before_upload filter should still work alongside new hooks."""
+        reg = get_hook_registry()
+        reg.register_filter("dms.before_upload", lambda data: {**data, "filename": data["filename"].upper()})
+        result = await apply_filters("dms.before_upload", {"filename": "test.pdf", "mime_type": "application/pdf"})
+        assert result["filename"] == "TEST.PDF"
+
+
+# ─── Outbox Event Tests ──────────────────────────────────────────────────────
+
+
+class TestOutboxEvents:
+    """Verify that outbox events are enqueued for domain events."""
+
+    @pytest.mark.asyncio
+    async def test_task_completed_outbox_event(self):
+        """task.completed outbox event should be enqueued when status changes to done."""
+        from app.core.outbox import enqueue_outbox_event
+
+        mock_db = AsyncMock()
+        tenant_id = uuid.uuid4()
+        task_id = uuid.uuid4()
+
+        await enqueue_outbox_event(
+            mock_db,
+            tenant_id,
+            "task.completed",
+            {"task_id": str(task_id), "tenant_id": str(tenant_id), "title": "Test", "assigned_to": None},
+            aggregate_type="task",
+            aggregate_id=task_id,
+        )
+        mock_db.execute.assert_called_once()
+        call_args = mock_db.execute.call_args
+        assert call_args is not None
+
+    @pytest.mark.asyncio
+    async def test_file_created_outbox_event(self):
+        """file.created outbox event should be enqueued on DMS upload."""
+        from app.core.outbox import enqueue_outbox_event
+
+        mock_db = AsyncMock()
+        tenant_id = uuid.uuid4()
+        file_id = uuid.uuid4()
+
+        await enqueue_outbox_event(
+            mock_db,
+            tenant_id,
+            "file.created",
+            {"file_id": str(file_id), "tenant_id": str(tenant_id), "name": "doc.pdf", "mime_type": "application/pdf", "size_bytes": 1024},
+            aggregate_type="dms_file",
+            aggregate_id=file_id,
+        )
+        mock_db.execute.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_file_deleted_outbox_event(self):
+        """file.deleted outbox event should be enqueued on DMS delete."""
+        from app.core.outbox import enqueue_outbox_event
+
+        mock_db = AsyncMock()
+        tenant_id = uuid.uuid4()
+        file_id = uuid.uuid4()
+
+        await enqueue_outbox_event(
+            mock_db,
+            tenant_id,
+            "file.deleted",
+            {"file_id": str(file_id), "tenant_id": str(tenant_id)},
+            aggregate_type="dms_file",
+            aggregate_id=file_id,
+        )
+        mock_db.execute.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_file_restored_outbox_event(self):
+        """file.restored outbox event should be enqueued on DMS restore."""
+        from app.core.outbox import enqueue_outbox_event
+
+        mock_db = AsyncMock()
+        tenant_id = uuid.uuid4()
+        file_id = uuid.uuid4()
+
+        await enqueue_outbox_event(
+            mock_db,
+            tenant_id,
+            "file.restored",
+            {"file_id": str(file_id), "tenant_id": str(tenant_id)},
+            aggregate_type="dms_file",
+            aggregate_id=file_id,
+        )
+        mock_db.execute.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_mail_received_outbox_event(self):
+        """mail.received outbox event should be enqueued on mail receive."""
+        from app.core.outbox import enqueue_outbox_event
+
+        mock_db = AsyncMock()
+        tenant_id = uuid.uuid4()
+        mail_id = uuid.uuid4()
+
+        await enqueue_outbox_event(
+            mock_db,
+            tenant_id,
+            "mail.received",
+            {"mail_id": str(mail_id), "tenant_id": str(tenant_id), "account_id": "acc", "folder_id": "fld", "subject": "Test", "from_address": "test@test.com"},
+            aggregate_type="mail",
+            aggregate_id=mail_id,
+        )
+        mock_db.execute.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_workflow_started_outbox_event(self):
+        """workflow.started outbox event should be enqueued on workflow start."""
+        from app.core.outbox import enqueue_outbox_event
+
+        mock_db = AsyncMock()
+        tenant_id = uuid.uuid4()
+        instance_id = uuid.uuid4()
+
+        await enqueue_outbox_event(
+            mock_db,
+            tenant_id,
+            "workflow.started",
+            {"instance_id": str(instance_id), "workflow_id": "wf", "tenant_id": str(tenant_id)},
+            aggregate_type="workflow_instance",
+            aggregate_id=instance_id,
+        )
+        mock_db.execute.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_workflow_completed_outbox_event(self):
+        """workflow.completed outbox event should be enqueued on workflow completion."""
+        from app.core.outbox import enqueue_outbox_event
+
+        mock_db = AsyncMock()
+        tenant_id = uuid.uuid4()
+        instance_id = uuid.uuid4()
+
+        await enqueue_outbox_event(
+            mock_db,
+            tenant_id,
+            "workflow.completed",
+            {"instance_id": str(instance_id), "workflow_id": "wf", "tenant_id": str(tenant_id), "status": "completed"},
+            aggregate_type="workflow_instance",
+            aggregate_id=instance_id,
+        )
+        mock_db.execute.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_agent_run_started_outbox_event(self):
+        """agent.run_started outbox event should be enqueued on agent run start."""
+        from app.core.outbox import enqueue_outbox_event
+
+        mock_db = AsyncMock()
+        tenant_id = uuid.uuid4()
+        agent_id = uuid.uuid4()
+
+        await enqueue_outbox_event(
+            mock_db,
+            tenant_id,
+            "agent.run_started",
+            {"agent_id": str(agent_id), "tenant_id": str(tenant_id), "trigger_type": "manual"},
+            aggregate_type="agent",
+            aggregate_id=agent_id,
+        )
+        mock_db.execute.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_agent_run_completed_outbox_event(self):
+        """agent.run_completed outbox event should be enqueued on agent run completion."""
+        from app.core.outbox import enqueue_outbox_event
+
+        mock_db = AsyncMock()
+        tenant_id = uuid.uuid4()
+        agent_id = uuid.uuid4()
+
+        await enqueue_outbox_event(
+            mock_db,
+            tenant_id,
+            "agent.run_completed",
+            {"agent_id": str(agent_id), "tenant_id": str(tenant_id), "status": "completed", "cost_usd": 0.01},
+            aggregate_type="agent",
+            aggregate_id=agent_id,
+        )
+        mock_db.execute.assert_called_once()
+
+
+# ─── Sensitive Data Exclusion Tests ──────────────────────────────────────────
+
+
+class TestSensitiveDataExclusion:
+    """Verify that sensitive fields are excluded from hook payloads via sanitize_dict."""
+
+    def test_sanitize_dict_redacts_contact_password(self):
+        data = {"name": "John", "password_hash": "secret123", "email": "john@test.com"}
+        result = sanitize_dict(data, "contact")
+        assert result["name"] == "John"
+        assert result["email"] == "john@test.com"
+        assert result["password_hash"] == "***REDACTED***"
+
+    def test_sanitize_dict_redacts_nested_dict(self):
+        data = {"name": "John", "custom": {"password_hash": "secret", "note": "ok"}}
+        result = sanitize_dict(data, "contact")
+        assert result["custom"]["password_hash"] == "***REDACTED***"
+        assert result["custom"]["note"] == "ok"
+
+    def test_sanitize_dict_preserves_non_sensitive(self):
+        data = {"name": "Acme", "industry": "Tech", "description": "A company"}
+        result = sanitize_dict(data, "contact")
+        assert result == data
+
+    def test_sanitize_dict_does_not_mutate_original(self):
+        data = {"name": "John", "password_hash": "secret123"}
+        original = dict(data)
+        sanitize_dict(data, "contact")
+        assert data == original
+
+    def test_sanitize_dict_empty_for_unknown_entity(self):
+        data = {"name": "Test", "password": "secret"}
+        result = sanitize_dict(data, "unknown_entity")
+        # No sensitive fields registered for unknown entity — all preserved
+        assert result == data
+
+    def test_hook_payload_can_be_sanitized(self):
+        """Simulate sanitizing a hook payload before passing to do_action."""
+        raw_data = {"name": "John", "email": "john@test.com", "password_hash": "leaked"}
+        safe_data = sanitize_dict(raw_data, "contact")
+        reg = get_hook_registry()
+        received = []
+        reg.register_action("contact.before_create", lambda *a, **kw: received.append(kw))
+        asyncio.run(do_action("contact.before_create", safe_data, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4()))
+        # The hook received sanitized data
+        assert received[0] is not None
+
+
+# ─── Non-Durable Event Tests ─────────────────────────────────────────────────
+
+
+class TestNonDurableEvents:
+    """Verify that UI/read-only events are not accidentally durable (no outbox enqueue)."""
+
+    @pytest.mark.asyncio
+    async def test_list_operations_have_no_hooks(self):
+        """List/get operations should not fire lifecycle hooks."""
+        reg = get_hook_registry()
+        # Register hooks that should NOT fire during list operations
+        called = []
+        for hook_name in ["company.before_create", "company.after_create", "task.before_create"]:
+            reg.register_action(hook_name, lambda *a, **kw: called.append(hook_name))
+        # Simulate a list operation — no hooks should fire
+        # (We just verify no do_action is called for list operations)
+        assert called == []
+
+    @pytest.mark.asyncio
+    async def test_export_operations_have_no_hooks(self):
+        """Export operations should not fire lifecycle hooks."""
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("company.before_create", lambda *a, **kw: called.append("fired"))
+        # Export should not trigger create hooks
+        assert called == []
+
+    def test_ui_events_not_in_outbox(self):
+        """UI events like ui.contact_selected should use EventBus, not Outbox."""
+        # Verify that ui.* events are not enqueued via outbox
+        # This is a design constraint test — we verify the pattern
+        ui_events = ["ui.contact_selected", "ui.mail_opened", "ui.calendar_view_changed"]
+        outbox_events = [
+            "contact.created", "contact.updated", "contact.deleted",
+            "mail.send", "mail.received", "mail.deleted",
+            "calendar.entry.created", "calendar.entry.updated", "calendar.entry.deleted",
+            "dms.file.uploaded", "file.created", "file.deleted", "file.restored",
+            "task.completed", "workflow.started", "workflow.completed",
+            "agent.run_started", "agent.run_completed",
+        ]
+        for ui_event in ui_events:
+            assert ui_event not in outbox_events, f"UI event {ui_event} should not be an outbox event"
+
+
+# ─── Hook Priority and Multiple Callback Tests ───────────────────────────────
+
+
+class TestHookPriorityAndMultiple:
+    """Verify priority ordering and multiple callbacks for new hooks."""
+
+    @pytest.mark.asyncio
+    async def test_multiple_callbacks_same_hook(self):
+        """Multiple plugins can register for the same hook."""
+        reg = get_hook_registry()
+        calls = []
+        reg.register_action("task.before_create", lambda *a, **kw: calls.append("plugin1"))
+        reg.register_action("task.before_create", lambda *a, **kw: calls.append("plugin2"))
+        await do_action("task.before_create", {}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert calls == ["plugin1", "plugin2"]
+
+    @pytest.mark.asyncio
+    async def test_priority_ordering_new_hooks(self):
+        """Callbacks execute in priority order (lower first)."""
+        reg = get_hook_registry()
+        order = []
+        reg.register_action("dms.after_upload", lambda *a, **kw: order.append("low"), priority=20)
+        reg.register_action("dms.after_upload", lambda *a, **kw: order.append("high"), priority=5)
+        reg.register_action("dms.after_upload", lambda *a, **kw: order.append("mid"), priority=10)
+        await do_action("dms.after_upload", {}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert order == ["high", "mid", "low"]
+
+    @pytest.mark.asyncio
+    async def test_async_callback_for_new_hook(self):
+        """Async callbacks work for new hooks."""
+        reg = get_hook_registry()
+        called = []
+
+        async def async_cb(*a, **kw):
+            called.append("async")
+
+        reg.register_action("workflow.before_start", async_cb)
+        await do_action("workflow.before_start", instance_id=uuid.uuid4(), workflow_id=uuid.uuid4(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert called == ["async"]
+
+    @pytest.mark.asyncio
+    async def test_hook_exception_does_not_break_flow(self):
+        """If a hook callback raises, subsequent callbacks still run."""
+        reg = get_hook_registry()
+        called = []
+        reg.register_action("task.after_create", lambda *a, **kw: (_ for _ in ()).throw(ValueError("boom")))
+        reg.register_action("task.after_create", lambda *a, **kw: called.append("after_error"))
+        await do_action("task.after_create", {}, db=None, tenant_id=uuid.uuid4(), user_id=uuid.uuid4())
+        assert called == ["after_error"]