diff --git a/app/ai/onboarding.py b/app/ai/onboarding.py new file mode 100644 index 0000000..8149328 --- /dev/null +++ b/app/ai/onboarding.py @@ -0,0 +1,154 @@ +"""Feature onboarding — setup wizard backend (I-ONB). + +Provides API endpoints for the setup wizard that guides users through +configuring agents, workflows, knowledge, workstreams/miniapps, +and proactive collaboration. +""" + +from __future__ import annotations + +import logging +import uuid +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +logger = logging.getLogger(__name__) + + +async def get_onboarding_status( + db: AsyncSession, + tenant_id: uuid.UUID, + user_id: uuid.UUID, +) -> dict[str, Any]: + """Get the onboarding progress for a user. + + Returns which steps are completed: agent_created, workflow_created, + knowledge_enabled, workstream_enabled, proactive_enabled. + """ + status: dict[str, Any] = { + "steps": { + "welcome": {"completed": True, "required": True}, + "create_agent": {"completed": False, "required": True}, + "create_workflow": {"completed": False, "required": False}, + "enable_knowledge": {"completed": False, "required": False}, + "enable_workstream": {"completed": False, "required": False}, + }, + "progress_pct": 20, # Welcome is done + } + + # Check if user has created an agent + try: + from app.models.workflow import AgentDefinition + from sqlalchemy import select, func + agent_count = await db.scalar( + select(func.count(AgentDefinition.id)).where( + AgentDefinition.tenant_id == tenant_id, + AgentDefinition.created_by == user_id, + ) + ) + if agent_count and agent_count > 0: + status["steps"]["create_agent"]["completed"] = True + status["progress_pct"] += 20 + except Exception as e: + logger.warning("Onboarding agent check failed: %s", e) + + # Check if user has created a workflow + try: + from app.models.workflow import WorkflowDefinition + from sqlalchemy import select, func + wf_count = await db.scalar( + select(func.count(WorkflowDefinition.id)).where( + WorkflowDefinition.tenant_id == tenant_id, + WorkflowDefinition.created_by == user_id, + ) + ) + if wf_count and wf_count > 0: + status["steps"]["create_workflow"]["completed"] = True + status["progress_pct"] += 20 + except Exception as e: + logger.warning("Onboarding workflow check failed: %s", e) + + # Check knowledge (wiki articles) + try: + from app.plugins.builtins.wiki.models import WikiArticle + from sqlalchemy import select, func + wiki_count = await db.scalar( + select(func.count(WikiArticle.id)).where( + WikiArticle.tenant_id == tenant_id, + WikiArticle.deleted_at.is_(None), + ) + ) + if wiki_count and wiki_count > 0: + status["steps"]["enable_knowledge"]["completed"] = True + status["progress_pct"] += 20 + except Exception as e: + logger.warning("Onboarding knowledge check failed: %s", e) + + # Check workstream (communication messages) + try: + from app.plugins.builtins.kommunikation.models import CommMessage + from sqlalchemy import select, func + msg_count = await db.scalar( + select(func.count(CommMessage.id)).where( + CommMessage.tenant_id == tenant_id, + ) + ) + if msg_count and msg_count > 0: + status["steps"]["enable_workstream"]["completed"] = True + status["progress_pct"] += 20 + except Exception as e: + logger.warning("Onboarding workstream check failed: %s", e) + + return status + + +def get_onboarding_guide() -> dict[str, Any]: + """Get the onboarding guide content for the setup wizard. + + Returns step-by-step instructions for each onboarding step. + """ + return { + "steps": [ + { + "id": "welcome", + "title": "Welcome to LeoCRM", + "description": "Get started with your AI-powered CRM platform.", + "icon": "sparkles", + }, + { + "id": "create_agent", + "title": "Create Your First Agent", + "description": "Set up an AI agent to help with email triage, contact enrichment, or follow-ups.", + "icon": "bot", + "action_url": "/agents/new", + }, + { + "id": "create_workflow", + "title": "Create Your First Workflow", + "description": "Automate repetitive tasks with workflows. Start with a template or build your own.", + "icon": "workflow", + "action_url": "/workflows/new", + }, + { + "id": "enable_knowledge", + "title": "Enable Knowledge Base", + "description": "Create wiki articles and let AI find answers from your company knowledge.", + "icon": "book-open", + "action_url": "/wiki", + }, + { + "id": "enable_workstream", + "title": "Enable Human-AI Workstream", + "description": "Connect humans, agents, and workflows in a unified communication stream.", + "icon": "message-square", + "action_url": "/workstream", + }, + ], + } + + +__all__ = [ + "get_onboarding_status", + "get_onboarding_guide", +] diff --git a/tests/test_phase_i_integration.py b/tests/test_phase_i_integration.py index 7a4e6e5..822da1a 100644 --- a/tests/test_phase_i_integration.py +++ b/tests/test_phase_i_integration.py @@ -514,3 +514,47 @@ class TestDSGVOExport: fields = _get_sensitive_fields() assert isinstance(fields, dict) assert len(fields) > 0 + + +# ─── I-ONB: Onboarding ────────────────────────────────────────────────────── + + +class TestOnboarding: + """Test the onboarding module (I-ONB).""" + + def test_onboarding_functions_importable(self): + """All onboarding functions are importable.""" + from app.ai.onboarding import get_onboarding_status, get_onboarding_guide + assert callable(get_onboarding_status) + assert callable(get_onboarding_guide) + + @pytest.mark.asyncio + async def test_get_onboarding_status_returns_dict(self): + """get_onboarding_status returns a dict with steps and progress.""" + from app.ai.onboarding import get_onboarding_status + + mock_db = AsyncMock() + mock_db.scalar = AsyncMock(return_value=0) + + result = await get_onboarding_status(mock_db, uuid.uuid4(), uuid.uuid4()) + assert isinstance(result, dict) + assert "steps" in result + assert "progress_pct" in result + assert "welcome" in result["steps"] + assert "create_agent" in result["steps"] + assert result["steps"]["welcome"]["completed"] is True + + def test_get_onboarding_guide_returns_steps(self): + """get_onboarding_guide returns a list of steps.""" + from app.ai.onboarding import get_onboarding_guide + + result = get_onboarding_guide() + assert isinstance(result, dict) + assert "steps" in result + assert len(result["steps"]) == 5 + step_ids = [s["id"] for s in result["steps"]] + assert "welcome" in step_ids + assert "create_agent" in step_ids + assert "create_workflow" in step_ids + assert "enable_knowledge" in step_ids + assert "enable_workstream" in step_ids