diff --git a/app/services/user_service.py b/app/services/user_service.py
new file mode 100644
index 0000000..132d550
--- /dev/null
+++ b/app/services/user_service.py
@@ -0,0 +1,120 @@
+"""User service: read, update, soft-delete, list."""
+
+from __future__ import annotations
+
+from datetime import datetime, UTC
+from typing import Optional
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.security import hash_password
+from app.models.user import User, UserRole
+from app.schemas.user import UserCreateRequest, UserUpdate
+
+
+class UserNotFound(Exception):
+    """Raised when a user lookup fails."""
+
+
+class EmailAlreadyTaken(Exception):
+    """Raised when attempting to create/update a user with an existing email."""
+
+
+async def get_user_by_id(db: AsyncSession, user_id: int) -> Optional[User]:
+    """Fetch a user by ID (active only, soft-deleted excluded)."""
+    result = await db.execute(
+        select(User).where(User.id == user_id, User.deleted_at.is_(None))
+    )
+    return result.scalar_one_or_none()
+
+
+async def get_user_by_email(
+    db: AsyncSession, email: str, org_id: Optional[int] = None
+) -> Optional[User]:
+    """Fetch a user by email, optionally scoped to an org."""
+    stmt = select(User).where(
+        User.email == email.lower(),
+        User.deleted_at.is_(None),
+    )
+    if org_id is not None:
+        stmt = stmt.where(User.org_id == org_id)
+    result = await db.execute(stmt)
+    return result.scalar_one_or_none()
+
+
+async def update_user_profile(
+    db: AsyncSession, user: User, payload: UserUpdate, *, is_admin: bool = False
+) -> User:
+    """Apply partial updates to a user.
+
+    Non-admin callers cannot change the role. All other fields are optional.
+    """
+    data = payload.model_dump(exclude_unset=True)
+
+    if "role" in data and not is_admin:
+        # Silently drop role change for non-admin callers
+        data.pop("role")
+
+    for field, value in data.items():
+        if value is not None:
+            setattr(user, field, value)
+
+    await db.commit()
+    await db.refresh(user)
+    return user
+
+
+async def soft_delete_user(db: AsyncSession, user: User) -> User:
+    """Soft-delete a user by setting deleted_at to now."""
+    user.deleted_at = datetime.now(UTC)
+    await db.commit()
+    await db.refresh(user)
+    return user
+
+
+async def create_user_as_admin(
+    db: AsyncSession, payload: UserCreateRequest, org_id: int
+) -> User:
+    """Create a new user in the given org (admin-only flow)."""
+    existing = await get_user_by_email(db, payload.email, org_id=org_id)
+    if existing is not None:
+        raise EmailAlreadyTaken(
+            f"A user with email {payload.email!r} already exists in this org."
+        )
+
+    user = User(
+        org_id=org_id,
+        email=payload.email.lower(),
+        password_hash=hash_password(payload.password),
+        name=payload.name,
+        role=payload.role if payload.role else UserRole.sales_rep,
+    )
+    db.add(user)
+    await db.commit()
+    await db.refresh(user)
+    return user
+
+
+async def list_users(
+    db: AsyncSession, org_id: int, skip: int = 0, limit: int = 50
+) -> list[User]:
+    """List active users in an org, paginated."""
+    result = await db.execute(
+        select(User)
+        .where(User.org_id == org_id, User.deleted_at.is_(None))
+        .order_by(User.id)
+        .offset(skip)
+        .limit(limit)
+    )
+    return list(result.scalars().all())
+
+
+async def count_users_in_org(db: AsyncSession, org_id: int) -> int:
+    """Count active users in an org."""
+    result = await db.execute(
+        select(func.count())
+        .select_from(User)
+        .where(User.org_id == org_id, User.deleted_at.is_(None))
+    )
+    return int(result.scalar_one())