diff --git a/app/plugins/builtins/automation/agent_runner.py b/app/plugins/builtins/automation/agent_runner.py index 2182012..a6b1db9 100644 --- a/app/plugins/builtins/automation/agent_runner.py +++ b/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: diff --git a/app/plugins/builtins/calendar/routes.py b/app/plugins/builtins/calendar/routes.py index ff056b3..6b28d92 100644 --- a/app/plugins/builtins/calendar/routes.py +++ b/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) diff --git a/app/plugins/builtins/dms/routes.py b/app/plugins/builtins/dms/routes.py index cabf739..844077c 100644 --- a/app/plugins/builtins/dms/routes.py +++ b/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, diff --git a/app/plugins/builtins/kommunikation/services.py b/app/plugins/builtins/kommunikation/services.py index 4fbfb32..ff22028 100644 --- a/app/plugins/builtins/kommunikation/services.py +++ b/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 diff --git a/app/plugins/builtins/mail/services.py b/app/plugins/builtins/mail/services.py index e236df4..51d70b3 100644 --- a/app/plugins/builtins/mail/services.py +++ b/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 diff --git a/app/plugins/builtins/tags/routes.py b/app/plugins/builtins/tags/routes.py index 3303b1e..db98e30 100644 --- a/app/plugins/builtins/tags/routes.py +++ b/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) diff --git a/app/plugins/builtins/tasks/services.py b/app/plugins/builtins/tasks/services.py index 1aa3d45..c9f33ea 100644 --- a/app/plugins/builtins/tasks/services.py +++ b/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) diff --git a/app/plugins/builtins/unified_search/search_engine.py b/app/plugins/builtins/unified_search/search_engine.py index 1efa293..71e10ad 100644 --- a/app/plugins/builtins/unified_search/search_engine.py +++ b/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] diff --git a/app/routes/companies.py b/app/routes/companies.py index a0a7791..735d77f 100644 --- a/app/routes/companies.py +++ b/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) diff --git a/app/services/workflow_service.py b/app/services/workflow_service.py index 0befe05..c953c80 100644 --- a/app/services/workflow_service.py +++ b/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() diff --git a/docs/plugin-development-guide.md b/docs/plugin-development-guide.md index 837c2b1..d1fcaee 100644 --- a/docs/plugin-development-guide.md +++ b/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. diff --git a/tests/test_lifecycle_hooks.py b/tests/test_lifecycle_hooks.py new file mode 100644 index 0000000..4674f48 --- /dev/null +++ b/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"]