fix: comprehensive production bug fix — admin auth, endpoints, seeding

- Create admin_auth.py with POST /login and GET /me endpoints
- Register admin_auth_router in main.py
- Implement ~20 missing admin endpoints in admin.py (users, workspaces,
webhooks, dispatch, automations, prompt-configs, zoho-health,
integrations, executions, email retry, impersonation)
- Fix wrap_error() calls with invalid code= kwarg (8 call sites across
auth.py, admin.py, integrations.py)
- Fix SQLAlchemy == None to .is_(None) in auth.py (2 places)
- Seed admin@leadpilot.io / LeadPilot@password123 in create_test_user.py
- Delete broken seed_admin.py and seed_rbac.py scripts

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

backend/app/api/v1/admin.py

@@ -5,9 +5,24 @@ from sqlmodel import select, func
 from pydantic import BaseModel
 import platform
 
+from datetime import timedelta
+from uuid import UUID
+
 from app.core.db import get_db
-from app.api.deps import get_current_user   # ← uses OAuth2PasswordBearer / Bearer header
-from app.models.models import EmailLog, SystemModuleConfig, User, AdminAuditLog
+from app.core import security
+from app.core.config import settings
+from app.api.deps import get_current_user
+from app.models.models import (
+    EmailLog, EmailOutbox, EmailOutboxStatus,
+    SystemModuleConfig, User, AdminAuditLog,
+    Workspace, WorkspaceMember,
+    WebhookEventLog,
+    Message, DeliveryStatus,
+    Flow, FlowStatus,
+    PromptConfig,
+    Integration, ZohoLeadMapping,
+    ExecutionInstance,
+)
 from app.schemas.envelope import ResponseEnvelope, wrap_data, wrap_error
 from app.core.modules import module_cache, ALL_MODULES, MODULE_ADMIN_PORTAL
 from app.core.audit import log_admin_action
@@ -82,7 +97,7 @@ async def get_email_log(
     """Get single email log by id."""
     log = await db.get(EmailLog, log_id)
     if not log:
-        return wrap_error("Email log not found", code="NOT_FOUND")
+        return wrap_error("Email log not found")
     return wrap_data(log.model_dump())
 
 
@@ -261,3 +276,577 @@ async def get_audit_log(
         "skip": skip,
         "limit": limit,
     })
+
+
+# ─── Users Endpoints ─────────────────────────────────────────────────────────
+
+@router.get("/users", response_model=ResponseEnvelope[dict])
+async def list_users(
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=200),
+    query: Optional[str] = None,
+) -> Any:
+    """List all users with optional search."""
+    stmt = select(User)
+    if query:
+        stmt = stmt.where(
+            User.email.ilike(f"%{query}%") | User.full_name.ilike(f"%{query}%")
+        )
+
+    count_stmt = select(func.count(User.id))
+    if query:
+        count_stmt = count_stmt.where(
+            User.email.ilike(f"%{query}%") | User.full_name.ilike(f"%{query}%")
+        )
+    total_res = await db.execute(count_stmt)
+    total = total_res.scalar_one() or 0
+
+    stmt = stmt.order_by(User.created_at.desc()).offset(skip).limit(limit)
+    result = await db.execute(stmt)
+    users = result.scalars().all()
+
+    return wrap_data({
+        "items": [
+            {
+                "id": str(u.id),
+                "email": u.email,
+                "full_name": u.full_name,
+                "is_active": u.is_active,
+                "is_superuser": u.is_superuser,
+                "auth_provider": u.auth_provider,
+                "email_verified_at": u.email_verified_at.isoformat() if u.email_verified_at else None,
+                "created_at": u.created_at.isoformat(),
+            }
+            for u in users
+        ],
+        "total": total,
+    })
+
+
+class UserToggleRequest(BaseModel):
+    is_active: bool
+
+
+@router.post("/users/{user_id}/toggle", response_model=ResponseEnvelope[dict])
+async def toggle_user_status(
+    user_id: str,
+    payload: UserToggleRequest,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Enable or disable a user."""
+    user = await db.get(User, UUID(user_id))
+    if not user:
+        return wrap_error("User not found")
+
+    user.is_active = payload.is_active
+    await db.commit()
+
+    await log_admin_action(
+        db=db,
+        actor_user_id=admin_user.id,
+        action="user_toggle",
+        entity_type="user",
+        entity_id=user_id,
+        metadata={"is_active": payload.is_active},
+    )
+
+    return wrap_data({"id": str(user.id), "is_active": user.is_active})
+
+
+@router.post("/users/{user_id}/impersonate", response_model=ResponseEnvelope[dict])
+async def impersonate_user(
+    user_id: str,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Generate a short-lived impersonation token for a user."""
+    user = await db.get(User, UUID(user_id))
+    if not user:
+        return wrap_error("User not found")
+
+    # Get user's first workspace
+    result = await db.execute(
+        select(WorkspaceMember).where(WorkspaceMember.user_id == user.id).limit(1)
+    )
+    membership = result.scalars().first()
+    workspace_id = membership.workspace_id if membership else None
+
+    token = security.create_access_token(
+        user.id,
+        workspace_id=str(workspace_id) if workspace_id else None,
+        expires_delta=timedelta(minutes=30),
+    )
+
+    await log_admin_action(
+        db=db,
+        actor_user_id=admin_user.id,
+        action="impersonate",
+        entity_type="user",
+        entity_id=user_id,
+    )
+
+    return wrap_data({"access_token": token})
+
+
+# ─── Workspaces Endpoints ────────────────────────────────────────────────────
+
+@router.get("/workspaces", response_model=ResponseEnvelope[dict])
+async def list_workspaces(
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=200),
+    query: Optional[str] = None,
+) -> Any:
+    """List all workspaces."""
+    stmt = select(Workspace)
+    if query:
+        stmt = stmt.where(Workspace.name.ilike(f"%{query}%"))
+
+    count_stmt = select(func.count(Workspace.id))
+    if query:
+        count_stmt = count_stmt.where(Workspace.name.ilike(f"%{query}%"))
+    total_res = await db.execute(count_stmt)
+    total = total_res.scalar_one() or 0
+
+    stmt = stmt.order_by(Workspace.created_at.desc()).offset(skip).limit(limit)
+    result = await db.execute(stmt)
+    workspaces = result.scalars().all()
+
+    return wrap_data({
+        "items": [
+            {
+                "id": str(w.id),
+                "name": w.name,
+                "subscription_tier": w.subscription_tier,
+                "created_at": w.created_at.isoformat(),
+            }
+            for w in workspaces
+        ],
+        "total": total,
+    })
+
+
+@router.get("/workspaces/{workspace_id}", response_model=ResponseEnvelope[dict])
+async def get_workspace_detail(
+    workspace_id: str,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Get workspace detail including member count."""
+    ws = await db.get(Workspace, UUID(workspace_id))
+    if not ws:
+        return wrap_error("Workspace not found")
+
+    member_count_res = await db.execute(
+        select(func.count(WorkspaceMember.user_id)).where(
+            WorkspaceMember.workspace_id == ws.id
+        )
+    )
+    member_count = member_count_res.scalar_one() or 0
+
+    return wrap_data({
+        "id": str(ws.id),
+        "name": ws.name,
+        "subscription_tier": ws.subscription_tier,
+        "created_at": ws.created_at.isoformat(),
+        "member_count": member_count,
+    })
+
+
+@router.get("/workspaces/{workspace_id}/modules", response_model=ResponseEnvelope[list])
+async def get_workspace_modules(
+    workspace_id: str,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Get per-workspace module override status."""
+    # Get global module states
+    global_res = await db.execute(select(SystemModuleConfig))
+    global_modules = {m.module_name: m.is_enabled for m in global_res.scalars().all()}
+
+    output = []
+    for module_name in ALL_MODULES:
+        global_enabled = global_modules.get(module_name, True)
+        output.append({
+            "module_name": module_name,
+            "is_enabled": global_enabled,
+            "overridden": False,
+        })
+
+    return wrap_data(output)
+
+
+class WorkspaceModuleToggle(BaseModel):
+    is_enabled: bool
+
+
+@router.patch("/workspaces/{workspace_id}/modules/{module_name}", response_model=ResponseEnvelope[dict])
+async def set_workspace_module(
+    workspace_id: str,
+    module_name: str,
+    payload: WorkspaceModuleToggle,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Set a per-workspace module override (uses global toggle for now)."""
+    result = await db.execute(
+        select(SystemModuleConfig).where(SystemModuleConfig.module_name == module_name)
+    )
+    mod = result.scalars().first()
+    if not mod:
+        return wrap_error(f"Module '{module_name}' not found")
+
+    mod.is_enabled = payload.is_enabled
+    mod.updated_by_user_id = admin_user.id
+    await db.commit()
+    module_cache.invalidate(module_name)
+
+    return wrap_data({
+        "module_name": module_name,
+        "is_enabled": payload.is_enabled,
+        "workspace_id": workspace_id,
+    })
+
+
+# ─── Email Retry Endpoint ────────────────────────────────────────────────────
+
+@router.post("/email-logs/{outbox_id}/retry", response_model=ResponseEnvelope[dict])
+async def retry_email(
+    outbox_id: str,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Re-queue a failed email for retry."""
+    outbox = await db.get(EmailOutbox, UUID(outbox_id))
+    if not outbox:
+        return wrap_error("Email outbox entry not found")
+
+    outbox.status = EmailOutboxStatus.PENDING
+    outbox.attempt_count = 0
+    outbox.last_error = None
+    await db.commit()
+
+    try:
+        from app.workers.email_tasks import send_email_task_v2
+        send_email_task_v2.delay(str(outbox.id))
+    except Exception:
+        pass
+
+    return wrap_data({"message": "Email re-queued for retry", "outbox_id": str(outbox.id)})
+
+
+# ─── Webhooks Endpoints ──────────────────────────────────────────────────────
+
+@router.get("/webhooks", response_model=ResponseEnvelope[dict])
+async def list_webhooks(
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+    provider: Optional[str] = None,
+    status: Optional[str] = None,
+) -> Any:
+    """List webhook event logs."""
+    stmt = select(WebhookEventLog)
+    if provider:
+        stmt = stmt.where(WebhookEventLog.provider == provider)
+    if status:
+        stmt = stmt.where(WebhookEventLog.status == status)
+
+    stmt = stmt.order_by(WebhookEventLog.created_at.desc()).offset(skip).limit(limit)
+    result = await db.execute(stmt)
+    events = result.scalars().all()
+
+    return wrap_data({
+        "items": [
+            {
+                "id": str(e.id),
+                "provider": e.provider,
+                "provider_event_id": e.provider_event_id,
+                "status": e.status,
+                "attempts": e.attempts,
+                "last_error": e.last_error,
+                "created_at": e.created_at.isoformat(),
+                "processed_at": e.processed_at.isoformat() if e.processed_at else None,
+            }
+            for e in events
+        ],
+    })
+
+
+@router.post("/webhooks/{event_id}/replay", response_model=ResponseEnvelope[dict])
+async def replay_webhook(
+    event_id: str,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Reset a webhook event to RECEIVED so it gets reprocessed."""
+    event = await db.get(WebhookEventLog, UUID(event_id))
+    if not event:
+        return wrap_error("Webhook event not found")
+
+    event.status = "received"
+    event.attempts = 0
+    event.last_error = None
+    event.processed_at = None
+    await db.commit()
+
+    return wrap_data({"message": "Webhook event reset for replay", "id": str(event.id)})
+
+
+# ─── Dispatch Endpoints ──────────────────────────────────────────────────────
+
+@router.get("/dispatch", response_model=ResponseEnvelope[dict])
+async def list_dispatch_queue(
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+) -> Any:
+    """List messages in the dispatch queue."""
+    count_res = await db.execute(select(func.count(Message.id)))
+    total = count_res.scalar_one() or 0
+
+    stmt = (
+        select(Message)
+        .order_by(Message.created_at.desc())
+        .offset(skip)
+        .limit(limit)
+    )
+    result = await db.execute(stmt)
+    messages = result.scalars().all()
+
+    return wrap_data({
+        "items": [
+            {
+                "id": str(m.id),
+                "conversation_id": str(m.conversation_id),
+                "direction": m.direction,
+                "platform": m.platform,
+                "delivery_status": m.delivery_status,
+                "attempt_count": m.attempt_count,
+                "last_error": m.last_error,
+                "created_at": m.created_at.isoformat(),
+            }
+            for m in messages
+        ],
+        "total": total,
+    })
+
+
+@router.patch("/dispatch/{message_id}/retry", response_model=ResponseEnvelope[dict])
+async def retry_dispatch(
+    message_id: str,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Reset a failed message for retry."""
+    msg = await db.get(Message, UUID(message_id))
+    if not msg:
+        return wrap_error("Message not found")
+
+    msg.delivery_status = DeliveryStatus.PENDING
+    msg.attempt_count = 0
+    msg.last_error = None
+    await db.commit()
+
+    return wrap_data({"message": "Message reset for retry", "id": str(msg.id)})
+
+
+@router.patch("/dispatch/{message_id}/dead-letter", response_model=ResponseEnvelope[dict])
+async def dead_letter_dispatch(
+    message_id: str,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Move a message to dead-letter (mark as failed permanently)."""
+    msg = await db.get(Message, UUID(message_id))
+    if not msg:
+        return wrap_error("Message not found")
+
+    msg.delivery_status = DeliveryStatus.FAILED
+    msg.last_error = "Moved to dead-letter by admin"
+    await db.commit()
+
+    return wrap_data({"message": "Message moved to dead-letter", "id": str(msg.id)})
+
+
+# ─── Automations Endpoints ───────────────────────────────────────────────────
+
+@router.get("/automations", response_model=ResponseEnvelope[dict])
+async def list_automations(
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+) -> Any:
+    """List all automation flows across all workspaces."""
+    count_res = await db.execute(select(func.count(Flow.id)))
+    total = count_res.scalar_one() or 0
+
+    stmt = select(Flow).order_by(Flow.created_at.desc()).offset(skip).limit(limit)
+    result = await db.execute(stmt)
+    flows = result.scalars().all()
+
+    return wrap_data({
+        "items": [
+            {
+                "id": str(f.id),
+                "name": f.name,
+                "workspace_id": str(f.workspace_id),
+                "status": f.status,
+                "description": f.description,
+                "created_at": f.created_at.isoformat(),
+            }
+            for f in flows
+        ],
+        "total": total,
+    })
+
+
+@router.patch("/automations/{flow_id}/disable", response_model=ResponseEnvelope[dict])
+async def disable_flow(
+    flow_id: str,
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Disable (set to draft) an automation flow."""
+    flow = await db.get(Flow, UUID(flow_id))
+    if not flow:
+        return wrap_error("Flow not found")
+
+    flow.status = FlowStatus.DRAFT
+    await db.commit()
+
+    await log_admin_action(
+        db=db,
+        actor_user_id=admin_user.id,
+        action="flow_disable",
+        entity_type="flow",
+        entity_id=flow_id,
+        metadata={"workspace_id": str(flow.workspace_id)},
+    )
+
+    return wrap_data({"message": "Flow disabled", "id": str(flow.id)})
+
+
+# ─── Prompt Configs Endpoint ─────────────────────────────────────────────────
+
+@router.get("/prompt-configs", response_model=ResponseEnvelope[dict])
+async def list_prompt_configs(
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+) -> Any:
+    """List all prompt configs across all workspaces."""
+    count_res = await db.execute(select(func.count(PromptConfig.id)))
+    total = count_res.scalar_one() or 0
+
+    stmt = select(PromptConfig).order_by(PromptConfig.created_at.desc()).offset(skip).limit(limit)
+    result = await db.execute(stmt)
+    configs = result.scalars().all()
+
+    return wrap_data({
+        "items": [
+            {
+                "id": str(c.id),
+                "name": c.name,
+                "workspace_id": str(c.workspace_id),
+                "current_version_id": str(c.current_version_id) if c.current_version_id else None,
+                "created_at": c.created_at.isoformat(),
+            }
+            for c in configs
+        ],
+        "total": total,
+    })
+
+
+# ─── Zoho Health Endpoint ────────────────────────────────────────────────────
+
+@router.get("/zoho-health", response_model=ResponseEnvelope[dict])
+async def get_zoho_health(
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """Return Zoho integration health across all workspaces."""
+    result = await db.execute(
+        select(Integration).where(Integration.provider == "zoho")
+    )
+    integrations = result.scalars().all()
+
+    return wrap_data({
+        "items": [
+            {
+                "id": str(i.id),
+                "workspace_id": str(i.workspace_id),
+                "status": i.status,
+                "provider_workspace_id": i.provider_workspace_id,
+                "connected_at": i.connected_at.isoformat() if i.connected_at else None,
+                "last_checked_at": i.last_checked_at.isoformat() if i.last_checked_at else None,
+                "last_error": i.last_error,
+            }
+            for i in integrations
+        ],
+    })
+
+
+# ─── Monitoring Endpoints ────────────────────────────────────────────────────
+
+@router.get("/integrations", response_model=ResponseEnvelope[dict])
+async def list_all_integrations(
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """List all integrations across all workspaces (monitoring)."""
+    result = await db.execute(
+        select(Integration).order_by(Integration.created_at.desc())
+    )
+    integrations = result.scalars().all()
+
+    return wrap_data({
+        "items": [
+            {
+                "id": str(i.id),
+                "workspace_id": str(i.workspace_id),
+                "provider": i.provider,
+                "status": i.status,
+                "provider_workspace_id": i.provider_workspace_id,
+                "connected_at": i.connected_at.isoformat() if i.connected_at else None,
+                "last_error": i.last_error,
+            }
+            for i in integrations
+        ],
+    })
+
+
+@router.get("/executions", response_model=ResponseEnvelope[dict])
+async def list_executions(
+    db: AsyncSession = Depends(get_db),
+    admin_user: User = Depends(require_superadmin),
+) -> Any:
+    """List recent execution instances across all workspaces (monitoring)."""
+    stmt = (
+        select(ExecutionInstance)
+        .order_by(ExecutionInstance.created_at.desc())
+        .limit(100)
+    )
+    result = await db.execute(stmt)
+    executions = result.scalars().all()
+
+    return wrap_data({
+        "items": [
+            {
+                "id": str(e.id),
+                "workspace_id": str(e.workspace_id),
+                "flow_version_id": str(e.flow_version_id),
+                "status": e.status,
+                "created_at": e.created_at.isoformat(),
+            }
+            for e in executions
+        ],
+    })

backend/app/api/v1/admin_auth.py

@@ -0,0 +1,64 @@
+from datetime import timedelta
+from typing import Any
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel import select
+
+from app.core import security
+from app.core.config import settings
+from app.core.db import get_db
+from app.models.models import User
+from app.schemas.envelope import ResponseEnvelope, wrap_data, wrap_error
+from app.api.deps import get_current_user
+
+router = APIRouter()
+
+
+class AdminLoginRequest(BaseModel):
+    email: str
+    password: str
+
+
+@router.post("/login", response_model=ResponseEnvelope[dict])
+async def admin_login(
+    body: AdminLoginRequest,
+    db: AsyncSession = Depends(get_db),
+) -> Any:
+    """Admin login — JSON body {email, password}. Returns JWT if user is superadmin."""
+    result = await db.execute(select(User).where(User.email == body.email))
+    user = result.scalars().first()
+
+    if not user or not user.hashed_password:
+        return wrap_error("Invalid email or password")
+
+    if not security.verify_password(body.password, user.hashed_password):
+        return wrap_error("Invalid email or password")
+
+    if not user.is_active:
+        return wrap_error("Account is disabled")
+
+    if not user.is_superuser:
+        return wrap_error("Admin privileges required")
+
+    access_token = security.create_access_token(
+        user.id, expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    )
+
+    return wrap_data({"access_token": access_token, "token_type": "bearer"})
+
+
+@router.get("/me", response_model=ResponseEnvelope[dict])
+async def admin_me(
+    current_user: User = Depends(get_current_user),
+) -> Any:
+    """Return the authenticated admin user info."""
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Admin privileges required")
+
+    return wrap_data({
+        "id": str(current_user.id),
+        "email": current_user.email,
+        "full_name": current_user.full_name,
+        "is_superuser": current_user.is_superuser,
+    })

backend/app/api/v1/auth.py

@@ -179,7 +179,7 @@ async def verify_email(
     result = await db.execute(
         select(EmailVerificationToken).where(
             EmailVerificationToken.token_hash == token_hash,
-            EmailVerificationToken.used_at == None,
+            EmailVerificationToken.used_at.is_(None),
         )
     )
     db_token = result.scalars().first()
@@ -310,7 +310,7 @@ async def forgot_password(
             await redis_client.expire(rl_key, 3600)
             
         if attempts > 5:
-            return wrap_error("Rate limit exceeded for password resets", code="EMAIL_429")
+            return wrap_error("Rate limit exceeded for password resets")
     except Exception:
         # If redis fails, do we bypass rate limiting? Yes, degrades gracefully.
         pass
@@ -383,13 +383,13 @@ async def reset_password(
         .where(
             PasswordResetToken.token_hash == token_hash,
             PasswordResetToken.expires_at > datetime.now(timezone.utc),
-            PasswordResetToken.used_at == None
+            PasswordResetToken.used_at.is_(None)
         )
     )
     reset_token = result.scalars().first()
     
     if not reset_token:
-        return wrap_error("Invalid or expired reset token", code="AUTH_INVALID_TOKEN")
+        return wrap_error("Invalid or expired reset token")
     
     # Find user
     user = await db.get(User, reset_token.user_id)

backend/app/api/v1/integrations.py

@@ -45,20 +45,20 @@ async def connect_integration(
 
     # 1. Validation for provider-specific fields
     if "access_token" not in connect_in.config:
-        return wrap_error("Missing access_token in config", code="INTEGRATION_VALIDATION")
-    
+        return wrap_error("Missing access_token in config")
+
     # WhatsApp standard: provider_workspace_id = phone_number_id
     if provider == "whatsapp" and not connect_in.provider_workspace_id:
-        return wrap_error("phone_number_id is required for WhatsApp", code="INTEGRATION_VALIDATION")
-    
+        return wrap_error("phone_number_id is required for WhatsApp")
+
     # Meta standard: provider_workspace_id = page_id
     if provider == "meta" and not connect_in.provider_workspace_id:
-        return wrap_error("page_id is required for Meta", code="INTEGRATION_VALIDATION")
-    
+        return wrap_error("page_id is required for Meta")
+
     # Zoho standard: provider_workspace_id = org_id (optional but recommended)
     # We enforce NOT NULL for any CONNECTED status below
     if not connect_in.provider_workspace_id:
-        return wrap_error(f"Identifier (provider_workspace_id) is required for {provider} connection", code="INTEGRATION_VALIDATION")
+        return wrap_error(f"Identifier (provider_workspace_id) is required for {provider} connection")
 
     # 2. Encrypt the config
     encrypted_config = encrypt_data(connect_in.config)
@@ -92,8 +92,7 @@ async def connect_integration(
     except IntegrityError:
         await db.rollback()
         return wrap_error(
-            f"This connection ({connect_in.provider_workspace_id}) is already linked to another workspace.",
-            code="INTEGRATION_409"
+            f"This connection ({connect_in.provider_workspace_id}) is already linked to another workspace."
         )
 
     return wrap_data(integration)

backend/main.py

@@ -10,6 +10,7 @@ from app.api.v1.dispatch import router as dispatch_router
 from app.api.v1.inbox import router as inbox_router
 from app.api.v1.zoho import router as zoho_router
 from app.api.v1.admin import router as admin_router
+from app.api.v1.admin_auth import router as admin_auth_router
 from app.api.v1.diagnostics import router as diagnostics_router
 from fastapi import HTTPException
 import uuid
@@ -59,6 +60,7 @@ async def add_process_time_header(request: Request, call_next):
 app.include_router(health.router, prefix=f"{settings.API_V1_STR}", tags=["health"])
 app.include_router(auth.router, prefix=f"{settings.API_V1_STR}/auth", tags=["auth"])
 app.include_router(admin_router, prefix=f"{settings.API_V1_STR}/admin", tags=["admin"])
+app.include_router(admin_auth_router, prefix=f"{settings.API_V1_STR}/admin_auth", tags=["admin-auth"])
 
 logger = logging.getLogger("api")
 

backend/scripts/create_test_user.py

@@ -28,49 +28,45 @@ async def create_test_user():
         await conn.run_sync(SQLModel.metadata.create_all)
         
     async with async_session() as session:
-        # Check if user exists
+        # Check if test user exists
         email = "test@example.com"
         result = await session.execute(select(User).where(User.email == email))
         user = result.scalars().first()
-        
+
         if user:
             logger.info(f"User {email} already exists.")
-            return
-
-        # Create user
-        logger.info(f"Creating test user: {email}")
-        db_user = User(
-            email=email,
-            hashed_password=security.get_password_hash("Password123!"),
-            full_name="Test User",
-            is_active=True,
-            is_superuser=True
-        )
-        session.add(db_user)
-        await session.flush()
-
-        # Create workspace
-        logger.info("Creating workspace for test user...")
-        db_workspace = Workspace(name="Test Workspace")
-        session.add(db_workspace)
-        await session.flush()
-
-        # Create membership
-        logger.info("Creating workspace membership...")
-        db_membership = WorkspaceMember(
-            user_id=db_user.id,
-            workspace_id=db_workspace.id,
-            role=WorkspaceRole.OWNER
-        )
-        session.add(db_membership)
-        
-        # --- Mission 8: Zoho Integration Setup ---
-        from app.models.models import Integration, ZohoLeadMapping
-        
-        logger.info("Setting up Zoho Integration for test user...")
-        # Check if exists
-        integ = await session.execute(select(Integration).where(Integration.workspace_id == db_workspace.id, Integration.provider == "zoho"))
-        if not integ.scalars().first():
+        else:
+            # Create user
+            logger.info(f"Creating test user: {email}")
+            db_user = User(
+                email=email,
+                hashed_password=security.get_password_hash("Password123!"),
+                full_name="Test User",
+                is_active=True,
+                is_superuser=True
+            )
+            session.add(db_user)
+            await session.flush()
+
+            # Create workspace
+            logger.info("Creating workspace for test user...")
+            db_workspace = Workspace(name="Test Workspace")
+            session.add(db_workspace)
+            await session.flush()
+
+            # Create membership
+            logger.info("Creating workspace membership...")
+            db_membership = WorkspaceMember(
+                user_id=db_user.id,
+                workspace_id=db_workspace.id,
+                role=WorkspaceRole.OWNER
+            )
+            session.add(db_membership)
+
+            # --- Mission 8: Zoho Integration Setup ---
+            from app.models.models import Integration, ZohoLeadMapping
+
+            logger.info("Setting up Zoho Integration for test user...")
             db_integration = Integration(
                 workspace_id=db_workspace.id,
                 provider="zoho",
@@ -78,10 +74,7 @@ async def create_test_user():
                 encrypted_config='{"refresh_token": "mock_refresh", "client_id": "mock_client", "client_secret": "mock_secret", "access_token": "mock_access", "expires_at": "2099-01-01T00:00:00"}'
             )
             session.add(db_integration)
-            
-        # Check mapping
-        mapping = await session.execute(select(ZohoLeadMapping).where(ZohoLeadMapping.workspace_id == db_workspace.id))
-        if not mapping.scalars().first():
+
             db_mapping = ZohoLeadMapping(
                 workspace_id=db_workspace.id,
                 dedupe_strategy="EMAIL_OR_PHONE",
@@ -96,8 +89,41 @@ async def create_test_user():
             )
             session.add(db_mapping)
 
-        await session.commit()
-        logger.info("Test user and Zoho configuration created successfully.")
+            await session.commit()
+            logger.info("Test user and Zoho configuration created successfully.")
+
+    # --- Seed documented admin user (admin@leadpilot.io) ---
+    async with async_session() as session:
+        admin_email = "admin@leadpilot.io"
+        result = await session.execute(select(User).where(User.email == admin_email))
+        admin_user = result.scalars().first()
+
+        if admin_user:
+            logger.info(f"Admin user {admin_email} already exists.")
+        else:
+            logger.info(f"Creating admin user: {admin_email}")
+            db_admin = User(
+                email=admin_email,
+                hashed_password=security.get_password_hash("LeadPilot@password123"),
+                full_name="LeadPilot Admin",
+                is_active=True,
+                is_superuser=True,
+            )
+            session.add(db_admin)
+            await session.flush()
+
+            db_admin_ws = Workspace(name="Admin Workspace")
+            session.add(db_admin_ws)
+            await session.flush()
+
+            db_admin_member = WorkspaceMember(
+                user_id=db_admin.id,
+                workspace_id=db_admin_ws.id,
+                role=WorkspaceRole.OWNER,
+            )
+            session.add(db_admin_member)
+            await session.commit()
+            logger.info(f"Admin user {admin_email} created successfully.")
 
 if __name__ == "__main__":
     asyncio.run(create_test_user())

backend/scripts/seed_admin.py

@@ -1,37 +0,0 @@
-import asyncio
-import sys
-from sqlmodel import select
-
-from app.core.db import SessionLocal
-from app.models.admin_models import AdminUser
-from app.core.security import get_password_hash
-
-async def seed_initial_admin(email: str, password: str, name: str):
-    async with SessionLocal() as db:
-        # Check if any admin exists
-        result = await db.execute(select(AdminUser))
-        if result.scalars().first():
-            print("Admin users already exist. Skipping seed.")
-            return
-
-        admin = AdminUser(
-            email=email,
-            full_name=name,
-            hashed_password=get_password_hash(password),
-            is_superuser=True,
-            is_active=True
-        )
-        db.add(admin)
-        await db.commit()
-        print(f"Initial SuperAdmin created: {email}")
-
-if __name__ == "__main__":
-    if len(sys.argv) < 4:
-        print("Usage: python seed_admin.py <email> <password> <name>")
-        sys.exit(1)
-    
-    email = sys.argv[1]
-    password = sys.argv[2]
-    name = sys.argv[3]
-    
-    asyncio.run(seed_initial_admin(email, password, name))

backend/scripts/seed_rbac.py

@@ -1,75 +0,0 @@
-import asyncio
-from sqlmodel import select
-
-from app.core.db import SessionLocal
-from app.models.admin_models import AdminRole, AdminPermission, AdminRolePermission
-
-DEFAULT_PERMISSIONS = [
-    {"key": "users.read", "description": "View product users"},
-    {"key": "users.write", "description": "Manage product users (toggle active)"},
-    {"key": "workspaces.read", "description": "View workspaces"},
-    {"key": "workspaces.write", "description": "Manage workspaces"},
-    {"key": "modules.read", "description": "View module status"},
-    {"key": "modules.write", "description": "Toggle system modules"},
-    {"key": "audit.read", "description": "View admin audit logs"},
-    {"key": "monitoring.read", "description": "View integrations, webhooks, and execution logs"},
-]
-
-DEFAULT_ROLES = [
-    {
-        "name": "SuperAdmin",
-        "description": "Full system access (bypasses permission checks)",
-        "permissions": [] # is_superuser handles this
-    },
-    {
-        "name": "Support",
-        "description": "Standard support access",
-        "permissions": ["users.read", "workspaces.read", "monitoring.read"]
-    },
-    {
-        "name": "Admin",
-        "description": "Administrative access with management capabilities",
-        "permissions": ["users.read", "users.write", "workspaces.read", "workspaces.write", "modules.read", "monitoring.read", "audit.read"]
-    }
-]
-
-async def seed_rbac():
-    async with SessionLocal() as db:
-        # 1. Seed Permissions
-        perm_map = {}
-        for p_data in DEFAULT_PERMISSIONS:
-            result = await db.execute(select(AdminPermission).where(AdminPermission.key == p_data["key"]))
-            perm = result.scalars().first()
-            if not perm:
-                perm = AdminPermission(**p_data)
-                db.add(perm)
-                await db.flush()
-            perm_map[p_data["key"]] = perm.id
-        
-        # 2. Seed Roles
-        for r_data in DEFAULT_ROLES:
-            result = await db.execute(select(AdminRole).where(AdminRole.name == r_data["name"]))
-            role = result.scalars().first()
-            if not role:
-                role = AdminRole(name=r_data["name"], description=r_data["description"])
-                db.add(role)
-                await db.flush()
-            
-            # 3. Link Permissions
-            for p_key in r_data["permissions"]:
-                p_id = perm_map[p_key]
-                link_result = await db.execute(
-                    select(AdminRolePermission).where(
-                        AdminRolePermission.role_id == role.id,
-                        AdminRolePermission.permission_id == p_id
-                    )
-                )
-                if not link_result.scalars().first():
-                    link = AdminRolePermission(role_id=role.id, permission_id=p_id)
-                    db.add(link)
-        
-        await db.commit()
-        print("Admin RBAC seeded successfully.")
-
-if __name__ == "__main__":
-    asyncio.run(seed_rbac())