apartment_mapping.json

@@ -807,25 +807,25 @@
       "floor": 7
     },
     {
-      "apartment_code": "PH1E-",
+      "apartment_code": "PH1E",
       "property": "East",
       "building": "Pavilion East",
       "floor": null
     },
     {
-      "apartment_code": "PH1W-",
+      "apartment_code": "PH1W",
       "property": "West",
       "building": "Pavilion West",
       "floor": null
     },
     {
-      "apartment_code": "PH2E-",
+      "apartment_code": "PH2E",
       "property": "East",
       "building": "Pavilion East",
       "floor": null
     },
     {
-      "apartment_code": "PH2W-",
+      "apartment_code": "PH2W",
       "property": "West",
       "building": "Pavilion West",
       "floor": null

app/core/roles.py

@@ -35,6 +35,7 @@ CANONICAL_ROLES: tuple[str, ...] = (
     "CS Manager",
     "FM Dispatcher",
     "Tech",
+    "Sub-contractor",
     "CEO",
     "Director",
 )
@@ -45,6 +46,18 @@ ADMIN_ROLES: tuple[str, ...] = ("Admin/Jerome", "Admin/Wahab")
 # Roles shown to the frontend nav/assignment helpers as "technician" pool.
 TECHNICIAN_ROLE = "Tech"
 
+# External subcontractor labour (client directive 2026-09: Wahab). Sub-contractors
+# are assignable work resources tracked "under tech" — the FM coordinator owns
+# their tickets — so they must surface in the Assign-to picker next to Techs.
+SUBCONTRACTOR_ROLE = "Sub-contractor"
+
+# The exact roles the assignment pickers offer (active users only).
+# Client directive 2026-09-28: sub-contractors appear in the "Assign to"
+# list alongside the technicians — external labour tracked under the FM
+# coordinator, so they are a sub-tier of the Tech pool, not a separate
+# workflow lane.
+ASSIGNEE_POOL_ROLES: tuple[str, ...] = (TECHNICIAN_ROLE, SUBCONTRACTOR_ROLE)
+
 # ── Legacy alias → canonical mapping (case-insensitive) ───────────────
 # Keys are lowercased. Unambiguous nicknames from legacy/early seeds and the
 # brief's role model ("technician/cs/fm/ceo") converge onto canonical roles.
@@ -60,6 +73,10 @@ ROLE_ALIASES: dict[str, str] = {
     "fm": "FM Dispatcher",
     "fm dispatcher": "FM Dispatcher",
     "fm_dispatcher": "FM Dispatcher",
+    "sub-contractor": SUBCONTRACTOR_ROLE,
+    "sub contractor": SUBCONTRACTOR_ROLE,
+    "subcontractor": SUBCONTRACTOR_ROLE,
+    "sub_contractor": SUBCONTRACTOR_ROLE,
     "ceo": "CEO",
     "director": "Director",
 }

app/main.py

@@ -14,6 +14,7 @@ from sqlalchemy import text
 from app.core.config import settings
 from app.core.database import Base, async_session_factory, engine
 from app.routers import auth, health, pages, tickets, whatsapp
+from app.services.roster import converge_tech_roster
 from app.services.seed import (
     normalize_legacy_user_emails,
     normalize_legacy_user_roles,
@@ -118,6 +119,9 @@ async def lifespan(app: FastAPI):
         # ``technician``/``cs``/``fm``) onto the canonical taxonomy at startup.
         await normalize_legacy_user_roles(session)
         await normalize_legacy_user_emails(session)
+        # Client directive 2026-09-28: the active Tech pool is exactly the
+        # five confirmed names; any other active Tech is deactivated here.
+        await converge_tech_roster(session)
         await session.commit()
         await seed_units(session, json_path=str(settings.BASE_DIR / "apartment_mapping.json"))
         await session.commit()

app/routers/auth.py

@@ -14,7 +14,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import get_db
 from app.core.ratelimit import login_rate_limiter
-from app.core.roles import ADMIN_ROLES
+from app.core.roles import ADMIN_ROLES, ASSIGNEE_POOL_ROLES
 from app.core.security import get_current_user, require_roles
 from app.models.user import User
 from app.schemas.auth import (
@@ -88,6 +88,25 @@ async def list_users(
     return list(result.scalars().all())
 
 
+@router.get("/assignees", response_model=list[UserOut])
+async def list_assignees(
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(get_current_user)],
+) -> list[User]:
+    """Active users selectable in the "Assign to" pickers.
+
+    The pool is the technician roster plus sub-contractors (client
+    directive 2026-09-28): external labour assigned directly, tracked
+    under the FM coordinator. Deactivated accounts never appear.
+    """
+    result = await db.execute(
+        select(User)
+        .where(User.role.in_(ASSIGNEE_POOL_ROLES), User.active.is_(True))
+        .order_by(User.full_name)
+    )
+    return list(result.scalars().all())
+
+
 # ── Admin user management ────────────────────────────────────────────
 @router.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
 async def create_user(

app/services/roster.py

@@ -0,0 +1,82 @@
+"""Client-confirmed technician roster enforcement (Pavilion Denya OneCare).
+
+Implements the FM's (Wahab Abdul) directive of 2026-09-28:
+
+    "i want the system to have these 5 names and nothing else, so replace
+     the current names with the 5 i sent to you. I'll confirm their email
+     address to you."
+
+The five are seeded directly (``SEED_USERS_DATA``); this module is the
+startup self-heal that keeps a *live* database honest, following the same
+idempotent pattern as ``normalize_legacy_user_roles``:
+
+* The active Tech pool is EXACTLY the five confirmed accounts (keyed on the
+  roster emails). Any other active Tech — legacy nickname accounts like
+  Prosper/Sam/Steven/Junior, or a stale same-name account the seed has
+  replaced — is deactivated. Never deleted: ``tickets.assigned_to`` FKs and
+  ticket history must stay intact. Deactivation fails closed at login
+  (``auth.authenticate_user``) and drops the user from every picker.
+* A second active account carrying a roster member's full name but a
+  different email is deactivated as a duplicate; the roster-email account
+  wins (it is what admins will manage once the client confirms emails).
+* Roster accounts that exist but were deactivated get re-activated (the
+  client put the name back on the list).
+
+Sub-contractors (role ``Sub-contractor``) are NOT part of this roster —
+the client has not named them yet. They belong to the assignable pool
+(``app.core.roles.ASSIGNEE_POOL_ROLES``) alongside Techs and will be seeded
+as soon as names/emails arrive. Convergence never touches them.
+"""
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.roles import TECHNICIAN_ROLE
+from app.models.user import User
+from app.services.seed import CONFIRMED_TECH_EMAILS
+
+logger = logging.getLogger(__name__)
+
+_ROSTER_EMAILS = frozenset(CONFIRMED_TECH_EMAILS)
+_ROSTER_NAMES = {
+    "samuel shang", "desmond afful", "desmond odekyi",
+    "francis norgbey", "nicholas nartey",
+}
+
+
+async def converge_tech_roster(session: AsyncSession) -> dict:
+    """Enforce the client-confirmed 5-name active Tech pool (idempotent)."""
+    users = list((await session.execute(select(User))).scalars().all())
+
+    roster_by_email = {u.email: u for u in users if u.email in _ROSTER_EMAILS}
+    roster_ids = {u.id for u in roster_by_email.values()}
+
+    deactivated: list[str] = []
+    reactivated: list[str] = []
+
+    for u in users:
+        if u.role != TECHNICIAN_ROLE:
+            continue
+        name_key = u.full_name.strip().lower()
+        on_roster = u.email in _ROSTER_EMAILS
+        duplicate_name = (not on_roster) and name_key in _ROSTER_NAMES
+        if on_roster:
+            if not u.active:
+                u.active = True
+                reactivated.append(u.full_name)
+        elif u.active:
+            # Off-roster Tech, or a same-name shadow of a roster account:
+            # deactivate (never delete) so the picker shows exactly the five.
+            u.active = False
+            deactivated.append(
+                f"{u.full_name} <{u.email}>" + (" (duplicate name)" if duplicate_name else "")
+            )
+
+    stats = {"deactivated": deactivated, "reactivated": reactivated,
+             "roster_present": len(roster_ids)}
+    if deactivated or reactivated:
+        logger.info("Tech roster converged: %s", stats)
+    return stats

app/services/seed.py

@@ -20,7 +20,25 @@ from app.models.category import Category
 
 logger = logging.getLogger(__name__)
 
+# Penthouse (Pavilion pent-house) units. Client directive 2026-09-28
+# (Wahab Abdul): "PHE1 & PHW2 means Pent House East/West, they must be
+# selectable when raising a ticket. they should be added to the units".
+# The mapping file shipped with dangling-dash codes (PH1E-…) and null floors;
+# they are cleaned here and self-healed in any existing database.
+_PENTHOUSE_FLOORS = {"PH1E": 1, "PH1W": 1, "PH2E": 2, "PH2W": 2}
+_LEGACY_PENTHOUSE_RENAMES = {
+    "PH1E-": "PH1E", "PH1W-": "PH1W", "PH2E-": "PH2E", "PH2W-": "PH2W",
+}
+
 # ── Seed user data (dicts to avoid module-level model instantiation) ─
+# TECH POOL (client directive 2026-09-28, Wahab Abdul — "these 5 names and
+# nothing else"): Samuel Shang, Desmond Afful, Desmond Odekyi, Francis
+# Norgbey, Nicholas Nartey. Emails are placeholders until the client confirms
+# them; the picker shows full names, and login matching is by email only.
+# The old nickname Techs (Prosper/Sam/Steven/Junior/Francis/Desmond/Afful)
+# are removed from the seed; ``roster.converge_tech_roster`` deactivates any
+# such account still present in a live database (never deletes — ticket
+# history holds ``users.id`` FKs).
 SEED_USERS_DATA = [
     {"email": "jerome@denya.com", "full_name": "Jerome Tabiri", "phone": "+233000000001", "role": "Admin/Jerome"},
     {"email": "wahab@denya.com", "full_name": "Wahab", "phone": "+233000000002", "role": "Admin/Wahab"},
@@ -30,17 +48,24 @@ SEED_USERS_DATA = [
     {"email": "ama@denya.com", "full_name": "Ama", "phone": "+233000000006", "role": "CS Rep"},
     {"email": "nicholas@denya.com", "full_name": "Nicholas", "phone": "+233000000007", "role": "FM Dispatcher"},
     {"email": "collins@denya.com", "full_name": "Collins", "phone": "+233000000008", "role": "FM Dispatcher"},
-    {"email": "prosper@denya.com", "full_name": "Prosper", "phone": "+233000000010", "role": "Tech"},
-    {"email": "sam@denya.com", "full_name": "Sam", "phone": "+233000000011", "role": "Tech"},
-    {"email": "steven@denya.com", "full_name": "Steven", "phone": "+233000000012", "role": "Tech"},
-    {"email": "junior@denya.com", "full_name": "Junior (Samuel)", "phone": "+233000000013", "role": "Tech"},
-    {"email": "francis@denya.com", "full_name": "Francis", "phone": "+233000000014", "role": "Tech"},
-    {"email": "desmond@denya.com", "full_name": "Desmond Afful", "phone": "+233000000015", "role": "Tech"},
-    {"email": "afful@denya.com", "full_name": "Afful", "phone": "+233000000016", "role": "Tech"},
+    {"email": "samuel.shang@denya.com", "full_name": "Samuel Shang", "phone": "+233000000010", "role": "Tech"},
+    {"email": "desmond.afful@denya.com", "full_name": "Desmond Afful", "phone": "+233000000015", "role": "Tech"},
+    {"email": "desmond.odekyi@denya.com", "full_name": "Desmond Odekyi", "phone": "+233000000011", "role": "Tech"},
+    {"email": "francis.norgbey@denya.com", "full_name": "Francis Norgbey", "phone": "+233000000014", "role": "Tech"},
+    {"email": "nicholas.nartey@denya.com", "full_name": "Nicholas Nartey", "phone": "+233000000013", "role": "Tech"},
     {"email": "scott@denya.com", "full_name": "Scott Murray", "phone": "+233000000020", "role": "CEO"},
     {"email": "director@denya.com", "full_name": "Director", "phone": "+233000000021", "role": "Director"},
 ]
 
+# The confirmed Tech pool, single source of truth for roster convergence.
+CONFIRMED_TECH_EMAILS = [
+    "samuel.shang@denya.com",
+    "desmond.afful@denya.com",
+    "desmond.odekyi@denya.com",
+    "francis.norgbey@denya.com",
+    "nicholas.nartey@denya.com",
+]
+
 
 async def normalize_legacy_user_roles(db: AsyncSession) -> int:
     """Converge legacy role strings onto the unified canonical taxonomy.
@@ -178,6 +203,15 @@ async def seed_units(db: AsyncSession, json_path: str | Path | None = None) -> l
                         "building": f"Pavilion {wing}",
                         "floor": floor,
                     })
+            # Penthouse units (client directive 2026-09-28: must be
+            # selectable when raising a ticket)
+            for level in (1, 2):
+                units_data.append({
+                    "apartment_code": f"PH{level}{wing[0]}",
+                    "property": wing,
+                    "building": f"Pavilion {wing}",
+                    "floor": _PENTHOUSE_FLOORS[f"PH{level}{wing[0]}"],
+                })
 
     created: list[Unit] = []
     for entry in units_data:
@@ -194,6 +228,35 @@ async def seed_units(db: AsyncSession, json_path: str | Path | None = None) -> l
             )
             db.add(unit)
             created.append(unit)
+
+    # Self-heal malformed penthouse codes from earlier mappings
+    # (``PH1E-`` with a dangling dash). Client directive 2026-09-28: the
+    # penthouse units (Pent House East/West) must be selectable when raising
+    # a ticket — they were present but mangled. Rename in place so any
+    # ``tickets.unit_id`` FKs stay intact; only rename when the clean code
+    # is free, and deactivate (never delete) a row that would collide.
+    renamed = 0
+    for legacy_code, clean_code in _LEGACY_PENTHOUSE_RENAMES.items():
+        legacy_row = (
+            await db.execute(select(Unit).where(Unit.apartment_code == legacy_code))
+        ).scalar_one_or_none()
+        if legacy_row is None or legacy_row.apartment_code == clean_code:
+            continue
+        clash = (
+            await db.execute(select(Unit).where(Unit.apartment_code == clean_code))
+        ).scalar_one_or_none()
+        if clash is not None:
+            # A clean-code row already exists; leave the legacy row as an
+            # inert historical alias (its tickets keep pointing at it).
+            continue
+        legacy_row.apartment_code = clean_code
+        if legacy_row.floor is None:
+            legacy_row.floor = _PENTHOUSE_FLOORS.get(clean_code)
+        renamed += 1
+    if renamed:
+        await db.flush()
+        logger.info("Repaired %d legacy penthouse unit codes", renamed)
+
     if created:
         await db.flush()
         for u in created:

app/services/ticket.py

@@ -10,6 +10,7 @@ from sqlalchemy import delete as sa_delete, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
+from app.core.roles import ASSIGNEE_POOL_ROLES
 from app.models.ticket import Escalation, Ticket, TicketPhoto, TicketTimeline
 from app.models.unit import Unit
 from app.models.user import User
@@ -105,6 +106,26 @@ async def _get_ticket_or_404(db: AsyncSession, ticket_id: int) -> Ticket:
 
 
 # ── CRUD ─────────────────────────────────────────────────────────────
+async def _validate_assignee(db: AsyncSession, assigned_to: int | None) -> None:
+    """Assignment must land on an active member of the pool (Tech roster +
+    sub-contractors, client directive 2026-09-28). Deactivated accounts —
+    replaced technicians included — can no longer receive new work."""
+    if assigned_to is None:
+        return
+    user = (
+        await db.execute(select(User).where(User.id == assigned_to))
+    ).scalar_one_or_none()
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail="Assignee not found")
+    if not user.active or user.role not in ASSIGNEE_POOL_ROLES:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"{user.full_name} is not in the assignable pool "
+                   f"({'/'.join(ASSIGNEE_POOL_ROLES)}, active only)",
+        )
+
+
 async def create_ticket(
     db: AsyncSession,
     data: dict[str, Any],
@@ -118,6 +139,7 @@ async def create_ticket(
     # when omitted the ticket is considered reported right now. The SLA clock
     # is unchanged — deadlines run from creation time, not the reported date.
     reported_at = data.get("reported_at") or datetime.now(timezone.utc)
+    await _validate_assignee(db, data.get("assigned_to"))
 
     ticket = Ticket(
         ticket_number=ticket_number,
@@ -158,6 +180,20 @@ async def create_ticket(
         user_id=user.id if user else None,
     )
 
+    # Assigning at creation advances Logged → Assigned (same rule as PATCH
+    # /tickets/{id} — client directive 2026-09-28 added the "Assign to"
+    # picker on the new-ticket form).
+    if ticket.assigned_to:
+        ticket.status = "Assigned"
+        await _log_status_change(
+            db,
+            ticket.id,
+            from_status="Logged",
+            to_status="Assigned",
+            note="Assigned at creation",
+            user_id=user.id if user else None,
+        )
+
     await db.flush()
     await db.refresh(ticket)
     return ticket
@@ -237,6 +273,7 @@ async def update_ticket(
 ) -> Ticket:
     """Update a ticket. Status changes are validated and logged."""
     ticket = await _get_ticket_or_404(db, ticket_id)
+    await _validate_assignee(db, data.get("assigned_to"))
 
     # Handle status transitions separately
     new_status = data.get("status")

app/templates/base.html

@@ -183,6 +183,10 @@
                 // Role helpers
                 get isCS() { return ['CS Rep', 'CS Manager'].includes(this.user.role) },
                 get isFM() { return ['FM Dispatcher', 'Admin/Jerome', 'Admin/Wahab'].includes(this.user.role) },
+                // Sub-contractors are external labour tracked under the FM
+                // coordinator — client directive 2026-09-28 puts them in the
+                // Tech pool ("under tech"), so they share the technician UX.
+                get isTech() { return ['Tech', 'Sub-contractor'].includes(this.user.role) },
                 get isExecutive() { return ['CEO', 'Director'].includes(this.user.role) },
                 get isAdmin() { return ['Admin/Jerome', 'Admin/Wahab'].includes(this.user.role) },
 

app/templates/tickets/detail.html

@@ -342,9 +342,11 @@
 
             async loadTechnicians() {
                 try {
-                    const users = await app().apiGet('/api/auth/users');
-                    // Assign dropdown should only offer Tech-role staff
-                    this.technicians = (users || []).filter(u => u.role === 'Tech');
+                    // Assignable pool: active Techs + sub-contractors
+                    // (client directive 2026-09-28). Deactivated, replaced
+                    // technicians are excluded server-side.
+                    const users = await app().apiGet('/api/auth/assignees');
+                    this.technicians = users || [];
                 } catch (e) {
                     console.error('Technicians load error', e);
                     this.technicians = [];

app/templates/tickets/new.html

@@ -154,6 +154,20 @@
                 </div>
             </div>
 
+            <!-- Assign to (optional at creation; Tech roster + sub-contractors) -->
+            <div class="grid grid-cols-1 md:grid-cols-2 gap-4">
+                <div>
+                    <label class="block text-sm font-medium text-gray-700 mb-1">Assign To</label>
+                    <select x-model="form.assigned_to" class="w-full px-4 py-2.5 rounded-lg border border-gray-300 focus:ring-2 focus:ring-denya-500 focus:border-transparent outline-none">
+                        <option value="">Unassigned (dispatch later)</option>
+                        <template x-for="tech in technicians" :key="tech.id">
+                            <option :value="tech.id" x-text="tech.role === 'Sub-contractor' ? `${tech.full_name} (sub-contractor)` : tech.full_name"></option>
+                        </template>
+                    </select>
+                    <p class="mt-1 text-xs text-gray-400">Technicians and sub-contractors. Leaving blank dispatches via the FM queue.</p>
+                </div>
+            </div>
+
             <!-- Reported date (backdating support) -->
             <div class="grid grid-cols-1 md:grid-cols-2 gap-4">
                 <div>
@@ -215,7 +229,8 @@
                 description: '',
                 reporter: '',
                 reported_via: '',
-                reported_date: ''
+                reported_date: '',
+                assigned_to: ''
             },
             todayStr: '',
             priorityAuto: false,
@@ -230,10 +245,12 @@
             photoPreviews: [],
             submitting: false,
             error: '',
+            technicians: [],
 
             async init() {
                 await this.loadCategories();
                 await this.loadUnits();
+                await this.loadTechnicians();
                 // Default reported date to today (local), allow backdating via the date picker
                 this.todayStr = this.localDateStr(new Date());
                 if (!this.form.reported_date) this.form.reported_date = this.todayStr;
@@ -279,6 +296,15 @@
                 } catch (e) { console.error('Units load error', e); }
             },
 
+            async loadTechnicians() {
+                try {
+                    // "Assign to" pool: the 5 client-confirmed technicians
+                    // plus sub-contractors (client directive 2026-09-28).
+                    const data = await app().apiGet('/api/auth/assignees');
+                    this.technicians = data || [];
+                } catch (e) { console.error('Assignees load error', e); }
+            },
+
             // ── Location cascade ─────────────────────────────────────
             onPropertyChange() {
                 this.form.building = '';
@@ -404,6 +430,7 @@
                         customer_name: this.form.customer_name || null,
                         phone: this.form.phone || null,
                         reported_at: this.form.reported_date || null,
+                        assigned_to: this.form.assigned_to ? parseInt(this.form.assigned_to) : null,
                     };
 
                     const ticket = await app().apiPost('/api/tickets', payload);

tests/test_categories_and_units.py

@@ -192,12 +192,19 @@ async def test_units_grouped_shape(client):
     assert "East" in grouped and "West" in grouped
     east = grouped["East"]
     assert "Pavilion East" in east
-    assert len(east["Pavilion East"]) == 60
+    # 10 floors x 6 apartments + 2 penthouse units (PH1E, PH2E) —
+    # client directive 2026-09-28: penthouses must be selectable.
+    assert len(east["Pavilion East"]) == 62
     unit = east["Pavilion East"][0]
     assert {"id", "property", "apartment_code", "building", "floor"} <= set(unit.keys())
     # Distinct apartment codes
     codes = [u["apartment_code"] for u in east["Pavilion East"]]
     assert len(set(codes)) == len(codes)
+    # Penthouse units (East/West) are present and selectable
+    assert {"PH1E", "PH2E"} <= set(codes)
+    west = grouped["West"]
+    west_codes = [u["apartment_code"] for u in west["Pavilion West"]]
+    assert {"PH1W", "PH2W"} <= set(west_codes)
 
 
 async def test_units_grouped_property_filter(client):

tests/test_tech_performance.py

@@ -44,6 +44,7 @@ async def _user_id(email: str) -> int:
         return user.id
 
 
+
 async def _seed_tickets(specs: list[dict]) -> None:
     """Insert tickets directly so tests control status/timestamps exactly."""
     async with async_session_factory() as session:
@@ -84,7 +85,7 @@ async def test_workload_sources_report_real_technician_names(client):
     ticket both must expose the technician's real full name, never a
     ``Tech #<id>`` placeholder.
     """
-    prosper = await _user_id("prosper@denya.com")
+    prosper = await _make_user("prosper@denya.com", "Prosper")
     await _seed_tickets([_ticket("In Progress", prosper), _ticket("Escalated", prosper)])
 
     token = await _login(client)
@@ -156,8 +157,8 @@ async def test_tech_performance_empty_state(client):
 
 async def test_tech_performance_aggregates_by_real_name(client):
     """Counts, completion rate, aging, sorting and totals per technician."""
-    prosper = await _user_id("prosper@denya.com")
-    sam = await _user_id("sam@denya.com")
+    prosper = await _make_user("prosper@denya.com", "Prosper")
+    sam = await _make_user("sam@denya.com", "Sam")
 
     await _seed_tickets(
         [
@@ -249,7 +250,7 @@ async def test_tech_performance_aggregates_by_real_name(client):
 
 async def test_tech_performance_reports_counts_when_timestamps_absent(client):
     """Finished tasks without ``closed_at`` yield no average but a count."""
-    afful = await _user_id("afful@denya.com")
+    afful = await _make_user("afful@denya.com", "Afful")
     await _seed_tickets([
         _ticket("Completed", afful),
         _ticket("Closed", afful),

tests/test_wahab_directives_20260928.py

@@ -0,0 +1,197 @@
+"""Regression tests for Wahab Abdul's 2026-09-28 directives.
+
+Covers the three client decisions:
+1. Technician roster converges to exactly 5 named techs (others deactivated,
+   never deleted — ticket history must survive).
+2. Sub-contractors appear in the "Assign to" pool alongside technicians.
+3. Penthouse units (East/West) are selectable when raising a ticket.
+"""
+
+import pytest
+import pytest_asyncio
+
+from app.core.database import async_session_factory
+from app.core.roles import (
+    ASSIGNEE_POOL_ROLES,
+    SUBCONTRACTOR_ROLE,
+    TECHNICIAN_ROLE,
+)
+from app.models.ticket import Ticket
+from app.models.user import User
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest_asyncio.fixture
+async def db():
+    """Direct DB session bound to the same test database as `client`.
+
+    conftest.py only exposes `client`, so DB-level assertions open their own
+    session against the same engine.
+    """
+    async with async_session_factory() as session:
+        yield session
+
+CONFIRMED_TECHS = (
+    "Samuel Shang",
+    "Desmond Afful",
+    "Desmond Odekyi",
+    "Francis Norgbey",
+    "Nicholas Nartey",
+)
+
+
+async def _make_user(db, email, full_name, role, active=True):
+    """Insert a user row directly (no HTTP admin round-trip)."""
+    user = User(
+        email=email,
+        full_name=full_name,
+        role=role,
+        password_hash="not-a-real-hash",
+        active=active,
+    )
+    db.add(user)
+    await db.flush()
+    return user
+
+
+def _grouped_codes(grouped):
+    """Flatten {property: {building: [units]}} to a set of apartment codes."""
+    return {
+        u["apartment_code"]
+        for wing in grouped.values()
+        for units in wing.values()
+        for u in units
+    }
+
+
+# ── 1. Technician roster ───────────────────────────────────────────────
+
+
+async def test_confirmed_techs_are_active_and_assignable(client, db):
+    """All 5 client-confirmed technicians exist, active, role=Tech."""
+    for name in CONFIRMED_TECHS:
+        user = await _make_user(
+            db, f"{name.lower().replace(' ', '.')}@denya.test", name, TECHNICIAN_ROLE
+        )
+        assert user.active is True
+        assert user.role in ASSIGNEE_POOL_ROLES
+
+
+async def test_roster_converge_is_idempotent(client, db):
+    """Running convergence twice leaves exactly the 5 confirmed techs."""
+    from sqlalchemy import select
+
+    from app.services.roster import converge_tech_roster
+
+    await converge_tech_roster(db)
+    await db.flush()
+    await converge_tech_roster(db)
+    await db.flush()
+
+    result = await db.execute(
+        select(User).where(User.role == TECHNICIAN_ROLE, User.active.is_(True))
+    )
+    active_techs = result.scalars().all()
+    assert {t.full_name for t in active_techs} == set(CONFIRMED_TECHS)
+
+
+async def test_roster_converge_deactivates_not_deletes(client, db):
+    """Off-roster techs are deactivated, never hard-deleted.
+
+    Deleting would orphan ticket history. The account row must survive so
+    previously-raised tickets keep a valid assignee reference.
+    """
+    from sqlalchemy import select
+
+    from app.services.roster import converge_tech_roster
+
+    stale = await _make_user(db, "stale.tech@denya.test", "Stale Tech", TECHNICIAN_ROLE)
+    stale_id = stale.id
+    await db.flush()
+
+    await converge_tech_roster(db)
+    await db.flush()
+
+    row = (
+        await db.execute(select(User).where(User.id == stale_id))
+    ).scalar_one_or_none()
+    assert row is not None, "off-roster tech must NOT be deleted"
+    assert row.active is False, "off-roster tech must be deactivated"
+
+
+# ── 2. Sub-contractors in the assign-to pool ───────────────────────────
+
+
+async def test_subcontractor_role_is_in_assignee_pool():
+    """Sub-contractor is a canonical role inside the assignee pool."""
+    assert SUBCONTRACTOR_ROLE in ASSIGNEE_POOL_ROLES
+    assert TECHNICIAN_ROLE in ASSIGNEE_POOL_ROLES
+
+
+async def test_subcontractor_can_own_a_ticket(client, db):
+    """A ticket can reference an active sub-contractor as assignee.
+
+    Sub-contractors are external, tracked "under tech" by the FM
+    coordinator, but they must be assignable.
+
+    The ticket is created in this session rather than via the `seed_tickets`
+    fixture: that fixture opens its own session against the same file-backed
+    SQLite DB and the two concurrent writers deadlock it.
+    """
+    from sqlalchemy import select
+
+    sub = await _make_user(
+        db, "sub.assign@denya.test", "External Sub", SUBCONTRACTOR_ROLE
+    )
+    ticket = Ticket(
+        ticket_number="PAV-SUB-00001",
+        status="Logged",
+        priority="medium",
+        description="Sub-contractor assignment check",
+    )
+    db.add(ticket)
+    await db.flush()
+
+    ticket.assigned_to = sub.id
+    await db.flush()
+
+    ticket_id = ticket.id
+
+    reloaded = (
+        await db.execute(select(Ticket).where(Ticket.id == ticket_id))
+    ).scalar_one()
+    assert reloaded.assigned_to == sub.id
+
+
+# ── 3. Penthouse units selectable ──────────────────────────────────────
+
+
+async def test_penthouse_units_present_and_selectable(client):
+    """All 4 penthouse units are selectable when raising a ticket."""
+    resp = await client.get("/api/tickets/units/grouped")
+    assert resp.status_code == 200
+    codes = _grouped_codes(resp.json())
+    for code in ("PH1E", "PH1W", "PH2E", "PH2W"):
+        assert code in codes, f"{code} must be selectable"
+
+
+async def test_penthouse_codes_have_no_dangling_dash(client):
+    """Legacy malformed codes (PH1E-) are renamed to clean form."""
+    resp = await client.get("/api/tickets/units/grouped")
+    codes = _grouped_codes(resp.json())
+    assert not any(c.endswith("-") for c in codes), "no dangling-dash codes"
+
+
+async def test_penthouse_units_have_floors(client):
+    """Penthouse units carry a floor so the 3-level picker can group them."""
+    resp = await client.get("/api/tickets/units/grouped")
+    penthouses = [
+        u
+        for wing in resp.json().values()
+        for units in wing.values()
+        for u in units
+        if u["apartment_code"].startswith("PH")
+    ]
+    assert len(penthouses) == 4
+    assert all(u["floor"] is not None for u in penthouses)