T09: KI-Copilot API + Hybrid Workflow Engine + LLM client + event-triggered workflows
- KI-Copilot: NL query → proposed actions, execute with RBAC, history, audit logging - LLM client: mock mode (no API key) + OpenAI-compatible mode (AI_MODEL/AI_API_KEY) - Action mapper: NL intent → API calls (create/update/delete/search company/contact) - Workflow engine: step types (action/approval/notification/condition), JSONB steps - Workflow lifecycle: pending → in_progress → completed/rejected/cancelled - Event-triggered workflows: event bus → auto-start instances - Code-engine workflows: onboarding on user.created event - Approval timeout: auto-reject after configured hours - 5 new tenant-scoped tables with RLS: ai_conversations, ai_messages, workflows, workflow_instances, workflow_step_history - Migration 0004: all tables + RLS policies + tenant_id + indexes - 238 tests pass (30 AC + 105 coverage + 103 existing), 84.12% T09 module coverage - MissingGreenlet fix: safe accessor helpers for async ORM attribute access
This commit is contained in:
leocrm-bot
@@ -0,0 +1 @@
|
||||
"""Workflow engine package — execution engine and code-engine workflows."""
|
||||
@@ -0,0 +1 @@
|
||||
"""Code-engine workflows — hardcoded workflow definitions."""
|
||||
@@ -0,0 +1,112 @@
|
||||
"""Onboarding workflow — runs on user creation.
|
||||
|
||||
A code-engine workflow that creates a welcome notification and an
|
||||
approval step for admin confirmation of new users.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.services.workflow_service import create_instance
|
||||
|
||||
# Onboarding workflow definition (code-engine — hardcoded steps)
|
||||
ONBOARDING_WORKFLOW_STEPS = [
|
||||
{
|
||||
"name": "Send welcome notification",
|
||||
"type": "notification",
|
||||
"config": {
|
||||
"notification_type": "welcome",
|
||||
"title": "Welcome to LeoCRM!",
|
||||
"body": "Your account has been created. An admin will approve your access shortly.",
|
||||
},
|
||||
"description": "Send a welcome notification to the new user",
|
||||
},
|
||||
{
|
||||
"name": "Admin approval",
|
||||
"type": "approval",
|
||||
"config": {
|
||||
"required_role": "admin",
|
||||
"description": "Admin must approve the new user account",
|
||||
},
|
||||
"description": "Admin reviews and approves the new user",
|
||||
},
|
||||
{
|
||||
"name": "Send confirmation",
|
||||
"type": "notification",
|
||||
"config": {
|
||||
"notification_type": "onboarding_complete",
|
||||
"title": "Account approved",
|
||||
"body": "Your account has been approved. You can now use LeoCRM.",
|
||||
},
|
||||
"description": "Notify user that their account is approved",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def get_onboarding_workflow_definition() -> dict[str, Any]:
|
||||
"""Return the onboarding workflow definition for DB seeding."""
|
||||
return {
|
||||
"name": "User Onboarding",
|
||||
"description": "Automated onboarding workflow triggered on user creation",
|
||||
"trigger_event": "user.created",
|
||||
"steps": ONBOARDING_WORKFLOW_STEPS,
|
||||
"is_active": True,
|
||||
}
|
||||
|
||||
|
||||
async def ensure_onboarding_workflow_exists(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
) -> uuid.UUID | None:
|
||||
"""Ensure the onboarding workflow exists in the DB for this tenant.
|
||||
|
||||
If it doesn't exist yet, create it. Returns the workflow ID.
|
||||
"""
|
||||
from app.models.workflow import Workflow
|
||||
from sqlalchemy import select
|
||||
|
||||
result = await db.execute(
|
||||
select(Workflow).where(
|
||||
Workflow.tenant_id == tenant_id,
|
||||
Workflow.trigger_event == "user.created",
|
||||
Workflow.name == "User Onboarding",
|
||||
)
|
||||
)
|
||||
existing = result.scalar_one_or_none()
|
||||
if existing:
|
||||
return existing.id
|
||||
|
||||
from app.services.workflow_service import create_workflow
|
||||
wf_dict = await create_workflow(
|
||||
db, tenant_id, user_id,
|
||||
get_onboarding_workflow_definition(),
|
||||
)
|
||||
return uuid.UUID(wf_dict["id"]) if wf_dict else None
|
||||
|
||||
|
||||
async def trigger_onboarding(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
admin_user_id: uuid.UUID,
|
||||
new_user_id: uuid.UUID,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Trigger the onboarding workflow for a newly created user.
|
||||
|
||||
1. Ensure the onboarding workflow exists in DB
|
||||
2. Create a workflow instance with new_user_id in context
|
||||
"""
|
||||
wf_id = await ensure_onboarding_workflow_exists(db, tenant_id, admin_user_id)
|
||||
if wf_id is None:
|
||||
return None
|
||||
|
||||
instance = await create_instance(
|
||||
db, tenant_id, admin_user_id,
|
||||
str(wf_id),
|
||||
context={"new_user_id": str(new_user_id), "user_id": str(new_user_id)},
|
||||
)
|
||||
return instance
|
||||
@@ -0,0 +1,288 @@
|
||||
"""Workflow execution engine — step processing, conditions, approvals.
|
||||
|
||||
Processes workflow instances by evaluating steps sequentially.
|
||||
Supports step types: action, approval, notification, condition.
|
||||
Integrates with the event bus for event-triggered workflows.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.event_bus import get_event_bus
|
||||
from app.models.workflow import Workflow, WorkflowInstance, WorkflowStepHistory
|
||||
from app.models.notification import Notification
|
||||
from app.services.workflow_service import (
|
||||
_log_step_history,
|
||||
_instance_to_dict,
|
||||
create_instance,
|
||||
find_workflows_for_event,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class WorkflowEngine:
|
||||
"""Processes workflow instances through their defined steps.
|
||||
|
||||
Step types:
|
||||
- action: Executes a configured action (e.g. create entity, send notification)
|
||||
- approval: Pauses and waits for user approve/reject via API
|
||||
- notification: Sends a notification to specified users
|
||||
- condition: Evaluates a condition and branches accordingly
|
||||
"""
|
||||
|
||||
def __init__(self, db: AsyncSession, tenant_id: uuid.UUID):
|
||||
self.db = db
|
||||
self.tenant_id = tenant_id
|
||||
|
||||
async def process_step(self, instance: WorkflowInstance) -> dict[str, Any]:
|
||||
"""Process the current step of a workflow instance.
|
||||
|
||||
For action/notification/condition steps: executes and advances.
|
||||
For approval steps: sets status to in_progress and waits.
|
||||
Returns the updated instance dict.
|
||||
"""
|
||||
wf_result = await self.db.execute(
|
||||
select(Workflow).where(Workflow.id == instance.workflow_id)
|
||||
)
|
||||
workflow = wf_result.scalar_one_or_none()
|
||||
if workflow is None:
|
||||
return {"error": "Workflow not found", "status_code": 404}
|
||||
|
||||
steps = workflow.steps or []
|
||||
if instance.current_step_index >= len(steps):
|
||||
instance.status = "completed"
|
||||
instance.completed_at = datetime.now(timezone.utc)
|
||||
await self.db.flush()
|
||||
return _instance_to_dict(instance)
|
||||
|
||||
step = steps[instance.current_step_index]
|
||||
step_type = step.get("type", "action")
|
||||
|
||||
# Log step entry
|
||||
await _log_step_history(
|
||||
self.db, self.tenant_id, instance.id,
|
||||
step_index=instance.current_step_index,
|
||||
step_type=step_type,
|
||||
action="processing",
|
||||
details={"step_name": step.get("name")},
|
||||
)
|
||||
|
||||
if step_type == "approval":
|
||||
# Approval steps pause and wait for user input
|
||||
if instance.status == "pending":
|
||||
instance.status = "in_progress"
|
||||
await self.db.flush()
|
||||
return _instance_to_dict(instance)
|
||||
|
||||
elif step_type == "notification":
|
||||
return await self._process_notification(instance, step)
|
||||
|
||||
elif step_type == "condition":
|
||||
return await self._process_condition(instance, step, steps)
|
||||
|
||||
elif step_type == "action":
|
||||
return await self._process_action(instance, step, steps)
|
||||
|
||||
else:
|
||||
return {"error": f"Unknown step type: {step_type}", "status_code": 400}
|
||||
|
||||
async def _process_action(
|
||||
self, instance: WorkflowInstance, step: dict, steps: list
|
||||
) -> dict[str, Any]:
|
||||
"""Process an action step — executes the configured action and advances."""
|
||||
config = step.get("config", {})
|
||||
action_type = config.get("action_type", "noop")
|
||||
|
||||
# Execute action based on type
|
||||
if action_type == "create_notification":
|
||||
user_id = config.get("user_id") or (str(instance.initiated_by) if instance.initiated_by else None)
|
||||
if user_id:
|
||||
notification = Notification(
|
||||
tenant_id=self.tenant_id,
|
||||
user_id=uuid.UUID(user_id),
|
||||
type=config.get("notification_type", "workflow_action"),
|
||||
title=config.get("title", "Workflow notification"),
|
||||
body=config.get("body", ""),
|
||||
)
|
||||
self.db.add(notification)
|
||||
await self.db.flush()
|
||||
|
||||
elif action_type == "noop":
|
||||
pass # No operation — just advance
|
||||
|
||||
# Log action executed
|
||||
await _log_step_history(
|
||||
self.db, self.tenant_id, instance.id,
|
||||
step_index=instance.current_step_index,
|
||||
step_type="action",
|
||||
action="executed",
|
||||
details={"action_type": action_type},
|
||||
)
|
||||
|
||||
# Advance to next step
|
||||
next_idx = instance.current_step_index + 1
|
||||
if next_idx >= len(steps):
|
||||
instance.status = "completed"
|
||||
instance.completed_at = datetime.now(timezone.utc)
|
||||
else:
|
||||
instance.current_step_index = next_idx
|
||||
instance.status = "in_progress"
|
||||
|
||||
await self.db.flush()
|
||||
return _instance_to_dict(instance)
|
||||
|
||||
async def _process_notification(
|
||||
self, instance: WorkflowInstance, step: dict
|
||||
) -> dict[str, Any]:
|
||||
"""Process a notification step — sends notification and advances."""
|
||||
config = step.get("config", {})
|
||||
user_id = config.get("user_id") or (str(instance.initiated_by) if instance.initiated_by else None)
|
||||
|
||||
if user_id:
|
||||
notification = Notification(
|
||||
tenant_id=self.tenant_id,
|
||||
user_id=uuid.UUID(user_id),
|
||||
type=config.get("notification_type", "workflow_notification"),
|
||||
title=config.get("title", "Workflow notification"),
|
||||
body=config.get("body", ""),
|
||||
)
|
||||
self.db.add(notification)
|
||||
await self.db.flush()
|
||||
|
||||
await _log_step_history(
|
||||
self.db, self.tenant_id, instance.id,
|
||||
step_index=instance.current_step_index,
|
||||
step_type="notification",
|
||||
action="sent",
|
||||
details={"user_id": user_id},
|
||||
)
|
||||
|
||||
# Advance — notification steps auto-advance
|
||||
wf_result = await self.db.execute(
|
||||
select(Workflow).where(Workflow.id == instance.workflow_id)
|
||||
)
|
||||
workflow = wf_result.scalar_one_or_none()
|
||||
steps = workflow.steps if workflow else []
|
||||
next_idx = instance.current_step_index + 1
|
||||
if next_idx >= len(steps):
|
||||
instance.status = "completed"
|
||||
instance.completed_at = datetime.now(timezone.utc)
|
||||
else:
|
||||
instance.current_step_index = next_idx
|
||||
|
||||
await self.db.flush()
|
||||
return _instance_to_dict(instance)
|
||||
|
||||
async def _process_condition(
|
||||
self, instance: WorkflowInstance, step: dict, steps: list
|
||||
) -> dict[str, Any]:
|
||||
"""Process a condition step — evaluates condition and branches.
|
||||
|
||||
Config format:
|
||||
{
|
||||
"field": "context_key",
|
||||
"operator": "eq|ne|gt|lt|contains",
|
||||
"value": "expected_value",
|
||||
"on_true_step": optional_index,
|
||||
"on_false_step": optional_index
|
||||
}
|
||||
"""
|
||||
config = step.get("config", {})
|
||||
field = config.get("field", "")
|
||||
operator = config.get("operator", "eq")
|
||||
expected = config.get("value")
|
||||
actual = instance.context.get(field)
|
||||
|
||||
condition_met = False
|
||||
if operator == "eq":
|
||||
condition_met = actual == expected
|
||||
elif operator == "ne":
|
||||
condition_met = actual != expected
|
||||
elif operator == "gt":
|
||||
condition_met = actual is not None and expected is not None and actual > expected
|
||||
elif operator == "lt":
|
||||
condition_met = actual is not None and expected is not None and actual < expected
|
||||
elif operator == "contains":
|
||||
condition_met = actual is not None and expected in actual if isinstance(actual, (str, list)) else False
|
||||
|
||||
await _log_step_history(
|
||||
self.db, self.tenant_id, instance.id,
|
||||
step_index=instance.current_step_index,
|
||||
step_type="condition",
|
||||
action="evaluated",
|
||||
details={"field": field, "operator": operator, "condition_met": condition_met},
|
||||
)
|
||||
|
||||
# Branch or advance
|
||||
if condition_met and "on_true_step" in config:
|
||||
instance.current_step_index = config["on_true_step"]
|
||||
elif not condition_met and "on_false_step" in config:
|
||||
instance.current_step_index = config["on_false_step"]
|
||||
else:
|
||||
next_idx = instance.current_step_index + 1
|
||||
if next_idx >= len(steps):
|
||||
instance.status = "completed"
|
||||
instance.completed_at = datetime.now(timezone.utc)
|
||||
else:
|
||||
instance.current_step_index = next_idx
|
||||
|
||||
await self.db.flush()
|
||||
return _instance_to_dict(instance)
|
||||
|
||||
|
||||
async def handle_event(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
event_name: str,
|
||||
payload: dict[str, Any],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Handle an event by starting matching workflow instances.
|
||||
|
||||
Called by the event bus integration. Finds all active workflows
|
||||
with trigger_event matching event_name and creates instances.
|
||||
"""
|
||||
workflows = await find_workflows_for_event(db, tenant_id, event_name)
|
||||
instances: list[dict[str, Any]] = []
|
||||
for wf in workflows:
|
||||
inst = await create_instance(
|
||||
db, tenant_id,
|
||||
uuid.UUID(payload.get("user_id", str(uuid.uuid4()))) if payload.get("user_id") else None or uuid.uuid4(),
|
||||
str(wf.id),
|
||||
context=payload,
|
||||
)
|
||||
if inst:
|
||||
instances.append(inst)
|
||||
return instances
|
||||
|
||||
|
||||
def register_workflow_event_handlers() -> None:
|
||||
"""Register event bus handlers for workflow triggers.
|
||||
|
||||
Subscribes to the event bus to auto-start workflows when events fire.
|
||||
Should be called during application startup.
|
||||
"""
|
||||
event_bus = get_event_bus()
|
||||
|
||||
async def _workflow_event_handler(payload: dict[str, Any]) -> None:
|
||||
"""Handle events that may trigger workflows."""
|
||||
from app.core.db import create_db_session
|
||||
tenant_id_str = payload.get("tenant_id")
|
||||
event_name = payload.get("event", "")
|
||||
if not tenant_id_str or not event_name:
|
||||
return
|
||||
|
||||
tenant_id = uuid.UUID(tenant_id_str)
|
||||
async with create_db_session(tenant_id) as db:
|
||||
await handle_event(db, tenant_id, event_name, payload)
|
||||
|
||||
# Subscribe to common events
|
||||
for event_name in ("user.created", "company.created", "contact.created"):
|
||||
event_bus.subscribe(event_name, _workflow_event_handler)
|
||||
Reference in New Issue
Block a user