ساختار مالکیت پلتفرم: یک مالک، مدیرانِ ساختهٔ مالک، سازمان‌های مدیریت‌شده

- users.is_platform_owner → users.platform_role (owner|manager|user) با مهاجرت
  d4e5f6a7b8c9 و backfill. ایندکس یکتای جزئی uq_single_platform_owner تضمین
  می‌کند دقیقاً یک مالک وجود داشته باشد — در سطح دیتابیس، نه فقط در کد.
- ثبت‌نام عمومی حذف شد (POST /auth/signup). حساب‌ها فقط با POST /admin/users
  (فقط مالک، با رمز موقت یک‌بارمصرف) یا با پذیرش لینک دعوت ساخته می‌شوند.
  در مسیر دعوت، اگر حساب وجود نداشته باشد خودِ دعوت‌شده رمز تعیین می‌کند؛ اگر
  وجود داشته باشد لینک هرگز جای reset رمز را نمی‌گیرد.
- ساخت سازمان فقط برای مالک و مدیران؛ کاربر عادی با دعوت عضو می‌شود.
- تنزل یا حذف مدیر، سازمان‌هایش را به مالک سامانه منتقل می‌کند (مالکیت + عضویت،
  با audit). حساب مالک از API تغییرناپذیر است؛ جابه‌جایی صندلی فقط با
  POST /admin/transfer-ownership و در یک تراکنش انجام می‌شود.
- افزودن POST /auth/change-password برای تبدیل رمز موقت به رمز خصوصی.
- فرانت: صفحهٔ /admin/users، صفحهٔ /account، بازنویسی /accept-invite، حذف /signup،
  گارد /onboarding برای کاربر عادی، و نشان نقش پلتفرم در پوستهٔ اپ.
- PRAGMA foreign_keys=ON برای SQLite تا ON DELETE CASCADE در dev و prod یکسان
  رفتار کند (پیش‌تر حذف کاربر، عضویت‌هایش را یتیم می‌گذاشت).
- تست: ۳۰۰ تست بک‌اند (شامل tests/test_platform_hierarchy.py با ۲۲ تست) و ۶۴ تست
  فرانت‌اند؛ ruff، mypy، tsc، eslint و build تمیز.

apps/api/tests/conftest.py

@@ -12,7 +12,6 @@ from __future__ import annotations
 import pytest
 import pytest_asyncio
 from httpx import ASGITransport, AsyncClient
-from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.pool import StaticPool
 
@@ -21,9 +20,12 @@ from sqlalchemy.pool import StaticPool
 # and binding the package to the same name would shadow it.
 import app.infrastructure.db.models as _orm_models  # noqa: F401
 from app.core.config import settings
-from app.core.database import Base, get_session
+from app.core.database import Base, enable_sqlite_foreign_keys, get_session
+from app.core.security import create_access_token, hash_password
 from app.domain.base import new_id
-from app.infrastructure.db.models import UserRow
+from app.domain.identity.models import PlatformRole, User
+from app.infrastructure.repositories.identity import UserRepository
+
 # The internal "full" app holds every HTTP handler; the public app (app.main.app)
 # now serves only /health + /api/ws, so REST tests must target the full app.
 from app.interfaces.realtime.rpc_app import get_rpc_app
@@ -64,6 +66,9 @@ async def engine():
         connect_args={"check_same_thread": False},
         poolclass=StaticPool,
     )
+    # Same pragma the app engine gets, so ON DELETE CASCADE behaves here the
+    # way it does on Postgres.
+    enable_sqlite_foreign_keys(test_engine)
     async with test_engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
     yield test_engine
@@ -109,12 +114,20 @@ async def client(engine, session_factory):
                 await session.rollback()
                 raise
 
+    # Publish the factory for the module-level ``signup`` helper below. The
+    # in-memory engine each test builds is private to its fixtures; a helper
+    # that needs the same database has to reach it through this door.
+    _session_factory_holder["factory"] = session_factory
     app.dependency_overrides[get_session] = _override_session
     async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as test_client:
         yield test_client
     app.dependency_overrides.clear()
 
 
+#: Set by the ``client`` fixture; read by :func:`signup`.
+_session_factory_holder: dict[str, async_sessionmaker] = {}
+
+
 # ---------------------------------------------------------------------------
 # Fake model provider — the only sanctioned stand-in for agy in tests
 # ---------------------------------------------------------------------------
@@ -186,13 +199,45 @@ def _fake_model_provider(monkeypatch):
 
 
 async def signup(
-    client: AsyncClient, email: str, password: str = "password123", name: str = ""
+    client: AsyncClient,
+    email: str,
+    password: str = "password123",
+    name: str = "",
+    platform_role: PlatformRole = PlatformRole.MANAGER,
 ) -> dict:
-    response = await client.post(
-        "/api/auth/signup", json={"email": email, "password": password, "name": name}
-    )
-    assert response.status_code == 201, response.text
-    return response.json()
+    """Create an account directly in the database and mint a token for it.
+
+    There is no public signup endpoint anymore — accounts are created by the
+    platform owner or by redeeming an invite — so this helper writes the row
+    those flows would and returns the same shape ``POST /auth/signup`` used to.
+    The ``client`` argument is kept for the historical call signature.
+
+    The default platform role is ``manager`` because most tests exercise the
+    tenant surface, and creating an organization requires that tier. Tests
+    that model a plain invited user pass ``platform_role=PlatformRole.USER``.
+    """
+    factory = _session_factory_holder["factory"]
+    async with factory() as session:
+        user = await UserRepository(session).create(
+            User(
+                email=email,
+                password_hash=hash_password(password),
+                name=name,
+                platform_role=platform_role,
+            )
+        )
+        await session.commit()
+    return {
+        "access_token": create_access_token(user.id),
+        "user": {
+            "id": user.id,
+            "email": user.email,
+            "name": user.display_name,
+            "platform_role": user.platform_role.value,
+            "is_manager": user.is_manager,
+            "is_platform_owner": user.is_platform_owner,
+        },
+    }
 
 
 def auth_header(token: str) -> dict[str, str]:
@@ -201,22 +246,30 @@ def auth_header(token: str) -> dict[str, str]:
 
 @pytest_asyncio.fixture
 async def platform_owner_token(client: AsyncClient, session_factory) -> str:
-    """A signed-up user promoted to platform owner.
+    """A token for the one and only platform owner.
 
-    The promotion is done directly in the database because there is deliberately
-    no API that can grant platform-owner status.
+    Owner seats are not grantable through the API by design (a manager must
+    not be able to mint themselves an owner), so the fixture creates the
+    account with the owner role directly — the same seat the seed installs.
     """
-    # ``example.com`` is a documentation domain and passes email validation.
-    # Do NOT use ``.local``/``.test``: those are special-use TLDs and newer
-    # ``email-validator`` releases reject them, which fails the fixture.
     email = "platform-owner@example.com"
-    body = await signup(client, email, name="مالک سامانه")
-    async with session_factory() as session:
-        row = (
-            await session.execute(select(UserRow).where(UserRow.email == email))
-        ).scalar_one()
-        row.is_platform_owner = True
-        await session.commit()
+    body = await signup(client, email, name="مالک سامانه", platform_role=PlatformRole.OWNER)
+    return body["access_token"]
+
+
+@pytest_asyncio.fixture
+async def manager_token(client: AsyncClient) -> str:
+    """A token for a plain manager account (the org-creating tier)."""
+    body = await signup(client, "manager@example.com", name="مدیر")
+    return body["access_token"]
+
+
+@pytest_asyncio.fixture
+async def plain_user_token(client: AsyncClient) -> str:
+    """A token for a plain user — the tier that may not create organizations."""
+    body = await signup(
+        client, "plain-user@example.com", name="کاربر عادی", platform_role=PlatformRole.USER
+    )
     return body["access_token"]
 
 
@@ -328,6 +381,10 @@ async def seeded(session_factory):
                 id=provider_id, name="agy", type="agy", base_url="", is_active=True
             )
         )
+        # Flush the provider before its models: with foreign keys enforced the
+        # batched model INSERT must not run before the row it references
+        # exists.
+        await session.flush()
         for internal, display, capabilities, effort in SYSTEM_MODELS:
             session.add(
                 AiModelRow(

apps/api/tests/test_auth.py

@@ -13,37 +13,36 @@ async def test_health_is_public(client: AsyncClient):
     assert response.json()["status"] == "ok"
 
 
-async def test_signup_returns_token_and_user(client: AsyncClient):
+async def test_signup_helper_returns_token_and_user(client: AsyncClient):
+    """There is no public signup endpoint — the test helper is the owner-side
+    creation path — and what it returns must carry the platform role."""
     body = await signup(client, "alice@example.com", name="آلیس")
     assert body["access_token"]
     assert body["user"]["email"] == "alice@example.com"
     assert body["user"]["name"] == "آلیس"
+    assert body["user"]["platform_role"] == "manager"
+    assert body["user"]["is_manager"] is True
     assert body["user"]["is_platform_owner"] is False
     assert "password" not in str(body).lower() or "password_hash" not in str(body)
 
 
-async def test_signup_never_returns_password_hash(client: AsyncClient):
+async def test_signup_helper_never_returns_password_hash(client: AsyncClient):
     body = await signup(client, "hashcheck@example.com")
     assert "password_hash" not in body["user"]
     assert "password" not in body["user"]
 
 
-async def test_signup_rejects_duplicate_email(client: AsyncClient):
-    await signup(client, "dup@example.com")
+async def test_public_signup_endpoint_is_gone(client: AsyncClient):
+    """Signup is closed: accounts come from the owner or from an invite link.
+
+    A stray old client must not silently succeed — the route simply does not
+    exist anymore.
+    """
     response = await client.post(
         "/api/auth/signup",
-        json={"email": "dup@example.com", "password": "password123", "name": ""},
+        json={"email": "nope@example.com", "password": "password123", "name": ""},
     )
-    assert response.status_code == 409
-    assert response.json()["error"]["code"] == "conflict"
-
-
-async def test_signup_rejects_short_password(client: AsyncClient):
-    response = await client.post(
-        "/api/auth/signup",
-        json={"email": "short@example.com", "password": "123", "name": ""},
-    )
-    assert response.status_code == 422
+    assert response.status_code == 404
 
 
 async def test_login_succeeds_with_correct_password(client: AsyncClient):
@@ -53,6 +52,9 @@ async def test_login_succeeds_with_correct_password(client: AsyncClient):
     )
     assert response.status_code == 200
     assert response.json()["access_token"]
+    payload = response.json()["user"]
+    assert payload["platform_role"] == "manager"
+    assert payload["is_platform_owner"] is False
 
 
 async def test_login_fails_with_wrong_password(client: AsyncClient):
@@ -96,3 +98,49 @@ async def test_logout_is_recorded_and_returns_204(client: AsyncClient):
     body = await signup(client, "frank@example.com")
     response = await client.post("/api/auth/logout", headers=auth_header(body["access_token"]))
     assert response.status_code == 204
+
+
+async def test_change_password_replaces_the_old_one(client: AsyncClient):
+    """A temp password handed out by the owner must become a private one.
+
+    Without this, every account created by someone else would live forever on
+    a secret two people know.
+    """
+    body = await signup(client, "grace@example.com", password="password123")
+    headers = auth_header(body["access_token"])
+
+    response = await client.post(
+        "/api/auth/change-password",
+        json={"current_password": "password123", "new_password": "newpassword456"},
+        headers=headers,
+    )
+    assert response.status_code == 204, response.text
+
+    stale = await client.post(
+        "/api/auth/login", json={"email": "grace@example.com", "password": "password123"}
+    )
+    assert stale.status_code == 401
+    fresh = await client.post(
+        "/api/auth/login", json={"email": "grace@example.com", "password": "newpassword456"}
+    )
+    assert fresh.status_code == 200
+
+
+async def test_change_password_rejects_a_wrong_current_password(client: AsyncClient):
+    body = await signup(client, "heidi@example.com", password="password123")
+    response = await client.post(
+        "/api/auth/change-password",
+        json={"current_password": "not-my-password", "new_password": "newpassword456"},
+        headers=auth_header(body["access_token"]),
+    )
+    assert response.status_code == 401
+
+
+async def test_change_password_rejects_a_short_new_password(client: AsyncClient):
+    body = await signup(client, "ivan@example.com", password="password123")
+    response = await client.post(
+        "/api/auth/change-password",
+        json={"current_password": "password123", "new_password": "123"},
+        headers=auth_header(body["access_token"]),
+    )
+    assert response.status_code == 422

apps/api/tests/test_error_log.py

@@ -22,12 +22,13 @@ from app.core.errors import ProviderUnavailableError
 from app.infrastructure.db.models import ErrorEventRow
 from app.interfaces.http.deps import OrgContext, get_org_context
 from app.interfaces.http.request_context import REQUEST_ID_HEADER
+
 # The internal "full" app holds every HTTP handler; the public app serves only
 # /health + /api/ws, so this module's REST exercises must target the full app.
 from app.interfaces.realtime.rpc_app import get_rpc_app
+from tests.conftest import auth_header, signup
 
 app = get_rpc_app()
-from tests.conftest import auth_header, signup
 
 #: The password in this DSN must never reach the database.
 LEAKY_DSN = "postgresql://agy:sup3rs3cret@db.internal:5432/agy"
@@ -55,6 +56,11 @@ async def client(engine, session_factory):
                 await session.rollback()
                 raise
 
+    # The shared ``signup`` helper reads the factory from the conftest holder;
+    # this module shadows ``client``, so publish it here too.
+    from tests import conftest as _conftest
+
+    _conftest._session_factory_holder["factory"] = session_factory
     app.dependency_overrides[get_session] = _override_session
     async with AsyncClient(
         transport=ASGITransport(app=app, raise_app_exceptions=False),

apps/api/tests/test_invites.py

@@ -1,8 +1,13 @@
 """Invite redemption: the link an admin hands out must actually work.
 
 Covers the full loop — create the invite, preview it as the invitee, redeem it,
-and the three ways redemption can legitimately fail. Also pins the absolute
-invite URL, since a bare path is not clickable for the person receiving it.
+and the ways redemption can legitimately fail. Also pins the absolute invite
+URL, since a bare path is not clickable for the person receiving it.
+
+Because signup is closed, the invite link doubles as the account-creation
+path: a brand-new invitee redeems it with a password and gets an account bound
+to the invited email. The token can never reset an existing account's password
+— that is what the "account already exists" rejection guards.
 """
 
 from __future__ import annotations
@@ -13,7 +18,7 @@ from tests.conftest import auth_header, signup
 
 
 async def _owner_with_org(client: AsyncClient, name: str = "Acme") -> tuple[str, str]:
-    """``(token, org_id)`` for a freshly signed-up owner."""
+    """``(token, org_id)`` for a freshly created manager and their organization."""
     body = await signup(client, "owner@example.com", name="Owner")
     headers = auth_header(body["access_token"])
     org = await client.post("/api/organizations", json={"name": name}, headers=headers)
@@ -104,19 +109,121 @@ async def test_invite_cannot_be_redeemed_by_a_different_account(client: AsyncCli
 
 
 async def test_unknown_invite_token_is_not_found(client: AsyncClient):
-    user = await signup(client, "nobody@example.com")
-    response = await client.get(
-        "/api/invites/definitely-not-a-real-token",
-        headers=auth_header(user["access_token"]),
-    )
+    response = await client.get("/api/invites/definitely-not-a-real-token")
     assert response.status_code == 404
 
 
-async def test_accepting_an_invite_requires_authentication(client: AsyncClient):
+async def test_invite_preview_works_without_a_session(client: AsyncClient):
+    """A brand-new invitee must see what they were invited to *before* they can
+    log in — otherwise the login button on the accept screen is a dead end."""
+    owner_token, org_id = await _owner_with_org(client)
+    raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
+        "invite_token"
+    ]
+
+    preview = await client.get(f"/api/invites/{raw}")
+    assert preview.status_code == 200, preview.text
+    body = preview.json()
+    assert body["organization_name"] == "Acme"
+    assert body["email"] == "newbie@example.com"
+    assert body["role"] == "member"
+    # No account exists for the invited email yet, so the UI offers to create
+    # one instead of demanding a login.
+    assert body["has_account"] is False
+    assert body["already_member"] is False
+
+
+async def test_invite_preview_reports_an_existing_account(client: AsyncClient):
+    owner_token, org_id = await _owner_with_org(client)
+    raw = (await _invite(client, owner_token, org_id, "existing@example.com", "member"))[
+        "invite_token"
+    ]
+    await signup(client, "existing@example.com", name="Existing")
+
+    preview = await client.get(f"/api/invites/{raw}")
+    assert preview.status_code == 200, preview.text
+    assert preview.json()["has_account"] is True
+
+
+async def test_anonymous_redemption_creates_the_account_and_joins(client: AsyncClient):
+    """The closed-signup flow: the link is the capability, the password makes
+    the account private, and the org membership is created in one step."""
+    owner_token, org_id = await _owner_with_org(client)
+    raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
+        "invite_token"
+    ]
+
+    accepted = await client.post(
+        "/api/invites/accept", json={"token": raw, "password": "password123"}
+    )
+    assert accepted.status_code == 200, accepted.text
+    assert accepted.json()["organization_id"] == org_id
+    assert accepted.json()["role"] == "member"
+
+    # The account exists for exactly the invited email, with that password, as
+    # a plain platform user — an invite must never mint a manager.
+    login = await client.post(
+        "/api/auth/login", json={"email": "newbie@example.com", "password": "password123"}
+    )
+    assert login.status_code == 200, login.text
+    token = login.json()["access_token"]
+    assert login.json()["user"]["platform_role"] == "user"
+
+    me = await client.get("/api/auth/me", headers=auth_header(token))
+    joined = [o for o in me.json()["organizations"] if o["id"] == org_id]
+    assert joined and joined[0]["role"] == "member"
+
+
+async def test_anonymous_redemption_without_a_password_is_rejected(client: AsyncClient):
     owner_token, org_id = await _owner_with_org(client)
     raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
         "invite_token"
     ]
 
     response = await client.post("/api/invites/accept", json={"token": raw})
-    assert response.status_code == 401
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "validation_error"
+
+
+async def test_anonymous_redemption_rejects_a_short_password(client: AsyncClient):
+    owner_token, org_id = await _owner_with_org(client)
+    raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
+        "invite_token"
+    ]
+
+    response = await client.post(
+        "/api/invites/accept", json={"token": raw, "password": "123"}
+    )
+    assert response.status_code == 422
+
+
+async def test_a_link_cannot_reset_an_existing_account(client: AsyncClient):
+    """A leaked invite must not become a password reset for the real account.
+
+    Whoever holds the link can create an account *only if none exists* for the
+    invited email; otherwise they must authenticate first.
+    """
+    owner_token, org_id = await _owner_with_org(client)
+    raw = (await _invite(client, owner_token, org_id, "taken@example.com", "member"))[
+        "invite_token"
+    ]
+    await signup(client, "taken@example.com", password="password123")
+
+    response = await client.post(
+        "/api/invites/accept", json={"token": raw, "password": "attacker-password"}
+    )
+    assert response.status_code == 403
+    assert "وارد شوید" in response.json()["error"]["message"]
+
+    # The real account's password is untouched, and the invite is not burned:
+    # the invitee can still redeem it after signing in.
+    login = await client.post(
+        "/api/auth/login", json={"email": "taken@example.com", "password": "password123"}
+    )
+    assert login.status_code == 200
+    accepted = await client.post(
+        "/api/invites/accept",
+        json={"token": raw},
+        headers=auth_header(login.json()["access_token"]),
+    )
+    assert accepted.status_code == 200, accepted.text

apps/api/tests/test_platform_hierarchy.py

@@ -0,0 +1,519 @@
+"""The platform hierarchy: one owner, owner-created managers, invited tenants.
+
+This is the security core of the identity model, so every rule gets both sides
+tested — the allow and the deny:
+
+1. **Exactly one owner.** The database refuses a second owner row outright;
+   the API refuses to mint or demote one. The seat moves only through a
+   transactional transfer that demotes the previous owner in the same step.
+2. **Only the owner and managers create organizations.** A plain user may not.
+3. **Only the owner creates managers.** No manager can mint another manager,
+   and no manager can enter the admin console.
+4. **A departing manager's organizations move to the owner** — never orphaned,
+   never deleted with their data.
+"""
+
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from app.domain.identity.models import Role
+from app.infrastructure.db.models import MembershipRow, OrganizationRow, UserRow
+from tests.conftest import auth_header
+
+
+async def _create_org(client: AsyncClient, token: str, name: str) -> dict:
+    response = await client.post(
+        "/api/organizations", json={"name": name}, headers=auth_header(token)
+    )
+    assert response.status_code == 201, response.text
+    return response.json()
+
+
+async def _login(client: AsyncClient, email: str, password: str) -> str:
+    response = await client.post(
+        "/api/auth/login", json={"email": email, "password": password}
+    )
+    assert response.status_code == 200, response.text
+    return response.json()["access_token"]
+
+
+async def _owner_creates_user(
+    client: AsyncClient,
+    owner_token: str,
+    email: str,
+    *,
+    platform_role: str = "manager",
+    name: str = "",
+) -> dict:
+    response = await client.post(
+        "/api/admin/users",
+        json={"email": email, "name": name, "platform_role": platform_role},
+        headers=auth_header(owner_token),
+    )
+    assert response.status_code == 201, response.text
+    return response.json()
+
+
+# ---------------------------------------------------------------------------
+# Exactly one owner
+# ---------------------------------------------------------------------------
+
+
+async def test_only_one_platform_owner_may_exist(
+    client: AsyncClient, platform_owner_token: str, session_factory
+):
+    """The invariant is enforced by the database, not by habit.
+
+    Even a direct INSERT — the kind a future code path or a migration would
+    perform — cannot produce a second owner row.
+    """
+    from sqlalchemy.exc import IntegrityError as _IntegrityError
+
+    async with session_factory() as session:
+        session.add(
+            UserRow(
+                id="second-owner-row",
+                email="second-owner@example.com",
+                password_hash="x",
+                platform_role="owner",
+            )
+        )
+        with pytest.raises(_IntegrityError):
+            await session.flush()
+        await session.rollback()
+
+
+async def test_owner_cannot_be_created_through_the_api(
+    client: AsyncClient, platform_owner_token: str
+):
+    response = await client.post(
+        "/api/admin/users",
+        json={"email": "wannabe@example.com", "platform_role": "owner"},
+        headers=auth_header(platform_owner_token),
+    )
+    assert response.status_code == 422
+    assert "مالکیت" in response.json()["error"]["message"]
+
+
+async def test_owner_account_is_immutable_from_the_api(
+    client: AsyncClient, platform_owner_token: str
+):
+    """No API path may demote, deactivate or delete the single owner seat."""
+    me = await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
+    owner_id = me.json()["user"]["id"]
+
+    demote = await client.patch(
+        f"/api/admin/users/{owner_id}",
+        json={"platform_role": "manager"},
+        headers=auth_header(platform_owner_token),
+    )
+    assert demote.status_code == 403
+
+    deactivate = await client.patch(
+        f"/api/admin/users/{owner_id}",
+        json={"is_active": False},
+        headers=auth_header(platform_owner_token),
+    )
+    assert deactivate.status_code == 403
+
+    delete = await client.delete(
+        f"/api/admin/users/{owner_id}", headers=auth_header(platform_owner_token)
+    )
+    assert delete.status_code == 403
+
+
+async def test_transfer_ownership_moves_the_seat_and_keeps_exactly_one(
+    client: AsyncClient, platform_owner_token: str, session_factory
+):
+    """The safety valve for a lost owner account — transactional by design."""
+    manager = await _owner_creates_user(
+        client, platform_owner_token, "successor@example.com", name="جانشین"
+    )
+    successor_token = await _login(
+        client, "successor@example.com", manager["temp_password"]
+    )
+
+    response = await client.post(
+        "/api/admin/transfer-ownership",
+        json={"user_id": manager["user"]["id"]},
+        headers=auth_header(platform_owner_token),
+    )
+    assert response.status_code == 200, response.text
+    assert response.json()["platform_role"] == "owner"
+
+    async with session_factory() as session:
+        owners = (
+            (await session.execute(select(UserRow).where(UserRow.platform_role == "owner")))
+            .scalars()
+            .all()
+        )
+    assert [o.email for o in owners] == ["successor@example.com"]
+
+    # The previous owner is demoted to manager in the same step — a manager,
+    # not a powerless user, so they keep the ability to run organizations.
+    previous = await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
+    assert previous.json()["user"]["platform_role"] == "manager"
+
+    # And the successor really is the owner: the console opens for them.
+    console = await client.get("/api/admin/users", headers=auth_header(successor_token))
+    assert console.status_code == 200
+
+
+# ---------------------------------------------------------------------------
+# Organization creation is gated by platform role
+# ---------------------------------------------------------------------------
+
+
+async def test_plain_user_cannot_create_an_organization(
+    client: AsyncClient, platform_owner_token: str, plain_user_token: str
+):
+    response = await client.post(
+        "/api/organizations",
+        json={"name": "سازمان ممنوعه"},
+        headers=auth_header(plain_user_token),
+    )
+    assert response.status_code == 403
+    assert "دعوت" in response.json()["error"]["message"]
+
+
+async def test_manager_can_create_an_organization(client: AsyncClient, manager_token: str):
+    org = await _create_org(client, manager_token, "سازمان مدیر")
+    assert org["owner_user_id"] == (
+        await client.get("/api/auth/me", headers=auth_header(manager_token))
+    ).json()["user"]["id"]
+
+
+async def test_owner_can_create_an_organization(client: AsyncClient, platform_owner_token: str):
+    org = await _create_org(client, platform_owner_token, "سازمان مالک")
+    assert org["name"] == "سازمان مالک"
+
+
+# ---------------------------------------------------------------------------
+# Only the owner creates managers
+# ---------------------------------------------------------------------------
+
+
+async def test_manager_cannot_create_accounts(client: AsyncClient, manager_token: str):
+    response = await client.post(
+        "/api/admin/users",
+        json={"email": "sneaky@example.com", "platform_role": "manager"},
+        headers=auth_header(manager_token),
+    )
+    assert response.status_code == 403
+
+
+async def test_plain_user_cannot_create_accounts(client: AsyncClient, plain_user_token: str):
+    response = await client.post(
+        "/api/admin/users",
+        json={"email": "sneaky2@example.com", "platform_role": "manager"},
+        headers=auth_header(plain_user_token),
+    )
+    assert response.status_code == 403
+
+
+async def test_manager_cannot_reach_the_admin_console(client: AsyncClient, manager_token: str):
+    for path in ("/api/admin/users", "/api/admin/models", "/api/admin/organizations"):
+        response = await client.get(path, headers=auth_header(manager_token))
+        assert response.status_code == 403, path
+
+
+async def test_owner_creates_a_manager_whose_temp_password_works(
+    client: AsyncClient, platform_owner_token: str
+):
+    created = await _owner_creates_user(
+        client, platform_owner_token, "manager1@example.com", name="مدیر یک"
+    )
+    assert created["user"]["platform_role"] == "manager"
+    assert created["user"]["is_manager"] is True
+    assert created["user"]["owned_organizations"] == 0
+    temp = created["temp_password"]
+    assert temp
+
+    login = await client.post(
+        "/api/auth/login",
+        json={"email": "manager1@example.com", "password": temp},
+    )
+    assert login.status_code == 200, login.text
+    assert login.json()["user"]["platform_role"] == "manager"
+
+
+async def test_duplicate_email_is_rejected(client: AsyncClient, platform_owner_token: str):
+    await _owner_creates_user(client, platform_owner_token, "twin@example.com")
+    response = await client.post(
+        "/api/admin/users",
+        json={"email": "twin@example.com", "platform_role": "manager"},
+        headers=auth_header(platform_owner_token),
+    )
+    assert response.status_code == 409
+
+
+async def test_owner_can_create_plain_users(client: AsyncClient, platform_owner_token: str):
+    created = await _owner_creates_user(
+        client,
+        platform_owner_token,
+        "plain@example.com",
+        platform_role="user",
+        name="کاربر عادی",
+    )
+    assert created["user"]["platform_role"] == "user"
+    assert created["user"]["is_manager"] is False
+
+
+async def test_owner_resets_a_forgotten_temp_password(
+    client: AsyncClient, platform_owner_token: str
+):
+    created = await _owner_creates_user(
+        client, platform_owner_token, "forgetful@example.com", name="فراموشکار"
+    )
+    first = created["temp_password"]
+
+    reset = await client.post(
+        f"/api/admin/users/{created['user']['id']}/reset-password",
+        headers=auth_header(platform_owner_token),
+    )
+    assert reset.status_code == 200, reset.text
+    second = reset.json()["temp_password"]
+    assert second and second != first
+
+    stale = await client.post(
+        "/api/auth/login", json={"email": "forgetful@example.com", "password": first}
+    )
+    assert stale.status_code == 401
+    fresh = await client.post(
+        "/api/auth/login", json={"email": "forgetful@example.com", "password": second}
+    )
+    assert fresh.status_code == 200
+
+
+# ---------------------------------------------------------------------------
+# A departing manager's organizations move to the owner
+# ---------------------------------------------------------------------------
+
+
+async def test_demoting_a_manager_transfers_organizations_to_the_owner(
+    client: AsyncClient, platform_owner_token: str, session_factory
+):
+    manager = await _owner_creates_user(
+        client, platform_owner_token, "leaving@example.com", name="خداحافظ"
+    )
+    manager_token = await _login(
+        client, "leaving@example.com", manager["temp_password"]
+    )
+    org = await _create_org(client, manager_token, "سازمان خداحافظ")
+
+    demote = await client.patch(
+        f"/api/admin/users/{manager['user']['id']}",
+        json={"platform_role": "user"},
+        headers=auth_header(platform_owner_token),
+    )
+    assert demote.status_code == 200, demote.text
+    assert demote.json()["platform_role"] == "user"
+    assert demote.json()["owned_organizations"] == 0
+
+    async with session_factory() as session:
+        row = await session.get(OrganizationRow, org["id"])
+        assert row is not None
+        owner_id = (
+            await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
+        ).json()["user"]["id"]
+        assert row.owner_user_id == owner_id
+
+        memberships = (
+            (
+                await session.execute(
+                    select(MembershipRow).where(MembershipRow.organization_id == org["id"])
+                )
+            )
+            .scalars()
+            .all()
+        )
+        by_user = {m.user_id: m for m in memberships}
+        # The new owner can actually see the tenant...
+        assert by_user[owner_id].role == Role.OWNER.value
+        # ...and the demoted manager keeps working access without ownership.
+        assert by_user[manager["user"]["id"]].role == Role.ADMIN.value
+
+
+async def test_deleting_a_manager_transfers_organizations_and_memberships(
+    client: AsyncClient, platform_owner_token: str, session_factory
+):
+    manager = await _owner_creates_user(
+        client, platform_owner_token, "doomed@example.com", name="حذف‌شونده"
+    )
+    manager_token = await _login(client, "doomed@example.com", manager["temp_password"])
+    org = await _create_org(client, manager_token, "سازمان حذف‌شونده")
+
+    response = await client.delete(
+        f"/api/admin/users/{manager['user']['id']}", headers=auth_header(platform_owner_token)
+    )
+    assert response.status_code == 200, response.text
+    body = response.json()
+    assert body["deleted_user_id"] == manager["user"]["id"]
+    assert [t["organization_id"] for t in body["transferred_organizations"]] == [org["id"]]
+
+    async with session_factory() as session:
+        assert await session.get(UserRow, manager["user"]["id"]) is None
+        row = await session.get(OrganizationRow, org["id"])
+        assert row is not None, "the organization must survive its owner's deletion"
+        owner_id = (
+            await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
+        ).json()["user"]["id"]
+        assert row.owner_user_id == owner_id
+        stale = (
+            await session.execute(
+                select(MembershipRow).where(
+                    MembershipRow.user_id == manager["user"]["id"]
+                )
+            )
+        ).scalars().all()
+        assert stale == [], "the deleted account's memberships must be gone"
+
+
+# ---------------------------------------------------------------------------
+# Managers create and join organizations
+# ---------------------------------------------------------------------------
+
+
+async def test_manager_can_join_another_managers_organization_by_invite(
+    client: AsyncClient, platform_owner_token: str
+):
+    """The invited manager keeps their own organization and gains a second."""
+    first = await _owner_creates_user(
+        client, platform_owner_token, "first@example.com", name="مدیر یکم"
+    )
+    second = await _owner_creates_user(
+        client, platform_owner_token, "second@example.com", name="مدیر دوم"
+    )
+    first_token = await _login(client, "first@example.com", first["temp_password"])
+    second_token = await _login(client, "second@example.com", second["temp_password"])
+
+    own_org = await _create_org(client, first_token, "سازمان مدیر یکم")
+    # The second manager has a tenant of their own before being invited.
+    second_own_org = await _create_org(client, second_token, "سازمان مدیر دوم")
+
+    invite = await client.post(
+        f"/api/organizations/{own_org['id']}/invites",
+        json={"email": "second@example.com", "role": "admin"},
+        headers=auth_header(first_token),
+    )
+    assert invite.status_code == 201, invite.text
+
+    accepted = await client.post(
+        "/api/invites/accept",
+        json={"token": invite.json()["invite_token"]},
+        headers=auth_header(second_token),
+    )
+    assert accepted.status_code == 200, accepted.text
+
+    me = await client.get("/api/auth/me", headers=auth_header(second_token))
+    roles = {o["id"]: o["role"] for o in me.json()["organizations"]}
+    # Joined the other manager's organization with the invited role...
+    assert roles[own_org["id"]] == "admin"
+    # ...and still owns their own, untouched.
+    assert roles[second_own_org["id"]] == "owner"
+    assert len(me.json()["organizations"]) == 2
+
+
+async def test_promoting_a_plain_user_to_manager_is_possible(
+    client: AsyncClient, platform_owner_token: str
+):
+    created = await _owner_creates_user(
+        client, platform_owner_token, "promoted@example.com", platform_role="user"
+    )
+    promoted = await client.patch(
+        f"/api/admin/users/{created['user']['id']}",
+        json={"platform_role": "manager"},
+        headers=auth_header(platform_owner_token),
+    )
+    assert promoted.status_code == 200, promoted.text
+    assert promoted.json()["platform_role"] == "manager"
+
+    # The new tier takes effect immediately: organizations are now allowed.
+    promoted_token = await _login(
+        client, "promoted@example.com", created["temp_password"]
+    )
+    org = await _create_org(client, promoted_token, "سازمان ارتقایافته")
+    assert org["name"] == "سازمان ارتقایافته"
+
+
+async def test_deactivating_an_account_blocks_login(
+    client: AsyncClient, platform_owner_token: str
+):
+    created = await _owner_creates_user(
+        client, platform_owner_token, "frozen@example.com", name="مسدود"
+    )
+    response = await client.patch(
+        f"/api/admin/users/{created['user']['id']}",
+        json={"is_active": False},
+        headers=auth_header(platform_owner_token),
+    )
+    assert response.status_code == 200, response.text
+    assert response.json()["is_active"] is False
+
+    login = await client.post(
+        "/api/auth/login",
+        json={"email": "frozen@example.com", "password": created["temp_password"]},
+    )
+    assert login.status_code == 401
+
+
+async def test_user_list_reports_roles_and_owned_organization_counts(
+    client: AsyncClient, platform_owner_token: str
+):
+    manager = await _owner_creates_user(
+        client, platform_owner_token, "listed@example.com", name="فهرست"
+    )
+    manager_token = await _login(client, "listed@example.com", manager["temp_password"])
+    await _create_org(client, manager_token, "سازمان فهرست")
+
+    listing = await client.get("/api/admin/users", headers=auth_header(platform_owner_token))
+    assert listing.status_code == 200
+    rows = {u["email"]: u for u in listing.json()}
+    assert rows["listed@example.com"]["platform_role"] == "manager"
+    assert rows["listed@example.com"]["owned_organizations"] == 1
+    me = await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
+    owner_email = me.json()["user"]["email"]
+    assert rows[owner_email]["platform_role"] == "owner"
+    assert rows[owner_email]["is_platform_owner"] is True
+
+
+async def test_admin_user_endpoints_require_authentication(client: AsyncClient):
+    assert (await client.get("/api/admin/users")).status_code == 401
+    assert (
+        await client.post("/api/admin/users", json={"email": "x@example.com"})
+    ).status_code == 401
+
+
+async def test_every_seat_move_is_audited(
+    client: AsyncClient, platform_owner_token: str, session_factory
+):
+    """The transfer and the deletion are the two irreversible moves; both must
+    leave a trail that names the actor and the moved organizations."""
+    from app.domain.audit.models import AuditAction
+
+    manager = await _owner_creates_user(
+        client, platform_owner_token, "audited@example.com", name="ممیزی"
+    )
+    manager_token = await _login(client, "audited@example.com", manager["temp_password"])
+    org = await _create_org(client, manager_token, "سازمان ممیزی")
+
+    response = await client.delete(
+        f"/api/admin/users/{manager['user']['id']}", headers=auth_header(platform_owner_token)
+    )
+    assert response.status_code == 200
+
+    from app.infrastructure.repositories.audit import AuditRepository
+
+    async with session_factory() as session:
+        events = await AuditRepository(session).list_platform(limit=100)
+    actions = {e.action: e for e in events}
+    assert AuditAction.USER_CREATE in actions
+    assert AuditAction.USER_DELETE in actions
+    delete_event = actions[AuditAction.USER_DELETE]
+    assert delete_event.metadata_json["transferred_organizations"] == [org["id"]]
+    assert delete_event.actor_user_id == (
+        await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
+    ).json()["user"]["id"]

apps/api/tests/test_realtime.py

@@ -25,13 +25,14 @@ from sqlalchemy.pool import NullPool
 import app.infrastructure.db.models as _orm_models  # noqa: F401 - registers metadata
 from app.application.ports.realtime import RealtimeEvent, set_publisher
 from app.core.config import settings
-from app.core.database import Base, get_session
+from app.core.database import Base, enable_sqlite_foreign_keys, get_session
 from app.core.security import create_access_token
 from app.domain.base import new_id
 from app.infrastructure.db.models import UserRow
 from app.interfaces.realtime.authz import authorize_channel
 from app.interfaces.realtime.bus import get_bus
 from app.interfaces.realtime.manager import Connection, ConnectionManager
+
 # The internal "full" app holds every HTTP handler; the public app serves only
 # /health + /api/ws. The socket and the RPC surface both live on the full app.
 from app.interfaces.realtime.rpc_app import get_rpc_app
@@ -85,7 +86,7 @@ async def _user_with_org(session: AsyncSession, *, platform_owner: bool = False)
             id=user_id,
             email=f"{user_id}@example.com",
             password_hash="x",
-            is_platform_owner=platform_owner,
+            platform_role=("owner" if platform_owner else "user"),
         )
     )
     await session.flush()
@@ -277,6 +278,7 @@ class _NullPublisher:
 
 
 async def _create_schema(engine) -> None:
+    enable_sqlite_foreign_keys(engine)
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
 
@@ -299,6 +301,9 @@ def ws_client(monkeypatch):
     monkeypatch.setattr("app.main.engine", engine)
     monkeypatch.setattr("app.interfaces.realtime.router.SessionFactory", factory)
     monkeypatch.setattr(settings, "realtime_transport", "local")
+    # Published for ``_create_account`` below: signup is closed, so the socket
+    # tests that need an identity mint it directly against this database.
+    _ws_factory_holder["factory"] = factory
 
     async def _override_session():
         async with factory() as session:
@@ -335,11 +340,45 @@ def _frames_until(websocket, wanted_type: str, *, max_frames: int = 20) -> dict:
     raise AssertionError(f"no {wanted_type} event arrived")
 
 
-def test_socket_allows_anonymous_then_restricts_actions(ws_client):
-    """No token opens an anonymous socket, but only login/signup may be called.
+#: Set by the ``ws_client`` fixture; read by ``_create_account``.
+_ws_factory_holder: dict[str, async_sessionmaker] = {}
 
-    A visitor who has not logged in can reach ``/api/auth/login`` and
-    ``/api/auth/signup`` over RPC; every other RPC and any subscribe is refused.
+
+def _create_account(email: str, *, name: str = "", platform_role: str = "manager") -> str:
+    """Create an account directly in the throwaway database; return its token.
+
+    Signup is closed — accounts come from the platform owner or from an
+    invite — so a socket test that needs an identity writes the row the way
+    those flows would and signs a token for it.
+    """
+    from app.core.security import create_access_token, hash_password
+    from app.domain.identity.models import PlatformRole, User
+    from app.infrastructure.repositories.identity import UserRepository
+
+    factory = _ws_factory_holder["factory"]
+
+    async def _create():
+        async with factory() as session:
+            user = await UserRepository(session).create(
+                User(
+                    email=email,
+                    password_hash=hash_password("password123"),
+                    name=name,
+                    platform_role=PlatformRole(platform_role),
+                )
+            )
+            await session.commit()
+            return user
+
+    user = asyncio.run(_create())
+    return create_access_token(user.id)
+
+
+def test_socket_allows_anonymous_then_restricts_actions(ws_client):
+    """No token opens an anonymous socket, but only the bootstrap RPCs run.
+
+    A visitor who has not logged in can reach ``/api/auth/login`` and redeem an
+    invite over RPC; every other RPC and any subscribe is refused.
     """
     with ws_client.websocket_connect("/api/ws") as socket:
         hello = socket.receive_json()
@@ -347,7 +386,9 @@ def test_socket_allows_anonymous_then_restricts_actions(ws_client):
         assert hello["user_id"] is None
 
         # A protected RPC is refused for an anonymous connection.
-        socket.send_json({"action": "rpc", "id": "1", "method": "GET", "path": "/api/organizations"})
+        socket.send_json(
+            {"action": "rpc", "id": "1", "method": "GET", "path": "/api/organizations"}
+        )
         result = socket.receive_json()
         assert result["type"] == "rpc_result"
         assert result["status"] == 401
@@ -378,13 +419,9 @@ def test_anonymous_may_call_login_over_rpc(ws_client):
 
 
 def test_task_creation_is_pushed_live(ws_client):
-    # Sign up, create an organization — the tenant the socket subscribes to.
-    signup = ws_client.post(
-        "/api/auth/signup",
-        json={"email": "live@example.com", "password": "password123", "name": "زنده"},
-    )
-    assert signup.status_code == 201, signup.text
-    token = signup.json()["access_token"]
+    # Create a manager account and its organization — the tenant the socket
+    # subscribes to. (Signup is closed; this is the owner-created path.)
+    token = _create_account("live@example.com", name="زنده")
     org = ws_client.post(
         "/api/organizations",
         json={"name": "سازمان زنده"},
@@ -425,12 +462,7 @@ def test_second_user_cannot_read_another_org(ws_client):
     tokens = []
     org_ids = []
     for index in range(2):
-        signup = ws_client.post(
-            "/api/auth/signup",
-            json={"email": f"user{index}@example.com", "password": "password123"},
-        )
-        assert signup.status_code == 201, signup.text
-        token = signup.json()["access_token"]
+        token = _create_account(f"user{index}@example.com")
         org = ws_client.post(
             "/api/organizations",
             json={"name": f"org {index}"},

apps/api/tests/test_seed.py

@@ -71,28 +71,59 @@ async def test_seeded_platform_owner_can_log_in(client: AsyncClient, seeded_inst
     assert me.json()["user"]["is_platform_owner"] is True
 
 
-async def test_seeded_accounts_belong_to_the_sample_organization(
+async def test_seeded_demo_manager_owns_the_sample_organization(
     client: AsyncClient, seeded_install
 ):
     """A login that lands on an empty dashboard is a broken demo.
 
-    Both seeded accounts must see the sample organization, otherwise every
-    screen behind the login renders nothing.
+    The demo account is a *manager* — the tier that creates organizations — so
+    it must own the sample organization and land on a populated dashboard.
     """
-    for email, password in (
-        (seeded_install.DEMO_EMAIL, seeded_install.DEMO_PASSWORD),
-        (seeded_install.PLATFORM_OWNER_EMAIL, seeded_install.PLATFORM_OWNER_PASSWORD),
-    ):
-        login = await client.post("/api/auth/login", json={"email": email, "password": password})
-        assert login.status_code == 200, login.text
+    login = await client.post(
+        "/api/auth/login",
+        json={
+            "email": seeded_install.DEMO_EMAIL,
+            "password": seeded_install.DEMO_PASSWORD,
+        },
+    )
+    assert login.status_code == 200, login.text
+    assert login.json()["user"]["platform_role"] == "manager"
+    assert login.json()["user"]["is_manager"] is True
+    assert login.json()["user"]["is_platform_owner"] is False
 
-        me = await client.get(
-            "/api/auth/me", headers=auth_header(login.json()["access_token"])
-        )
-        organizations = me.json()["organizations"]
-        assert organizations, f"{email} belongs to no organization"
-        assert organizations[0]["name"] == seeded_install.DEMO_ORG_NAME
-        assert organizations[0]["role"] == "owner"
+    me = await client.get(
+        "/api/auth/me", headers=auth_header(login.json()["access_token"])
+    )
+    organizations = me.json()["organizations"]
+    assert organizations, "the demo manager belongs to no organization"
+    assert organizations[0]["name"] == seeded_install.DEMO_ORG_NAME
+    assert organizations[0]["role"] == "owner"
+
+
+async def test_platform_owner_is_not_joined_to_any_organization(
+    client: AsyncClient, seeded_install
+):
+    """The product rule: the owner observes tenants from the console.
+
+    Auto-joining the owner to a tenant would put the single owner seat inside
+    customer data, which is exactly what the hierarchy is designed to avoid.
+    """
+    login = await client.post(
+        "/api/auth/login",
+        json={
+            "email": seeded_install.PLATFORM_OWNER_EMAIL,
+            "password": seeded_install.PLATFORM_OWNER_PASSWORD,
+        },
+    )
+    assert login.status_code == 200, login.text
+
+    me = await client.get(
+        "/api/auth/me", headers=auth_header(login.json()["access_token"])
+    )
+    assert me.status_code == 200, me.text
+    assert me.json()["user"]["is_platform_owner"] is True
+    assert me.json()["user"]["platform_role"] == "owner"
+    assert me.json()["organizations"] == []
 
 
 async def test_seeded_install_can_use_the_product(
@@ -216,4 +247,5 @@ async def test_seed_is_idempotent(client: AsyncClient, seeded_install, session_f
 
     assert after == before, f"re-seeding changed row counts: {before} -> {after}"
     assert after["organizations"] == 1
-    assert after["memberships"] == 2, "one per seeded account, and no duplicates"
+    # Only the demo manager: the platform owner stays out of tenants.
+    assert after["memberships"] == 1