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,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