Dev

backend/app/api/shifts.py

@@ -0,0 +1,99 @@
+"""Shift endpoints (spec §7.5): the manager marks an absence, which opens a rescue.
+
+The manager action is authoritative — no WhatsApp confirmation round trip:
+the API validates ownership and state, enqueues the domain change and answers
+202 (same shape as the close/approval writes). The worker applies it through
+`RescueOrchestrator.mark_absence`, which reuses the confirmed-absence path.
+No demo gate here: this is production behaviour (unlike `/dev/*`).
+"""
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from structlog import get_logger
+
+from app.api.dependencies import ManagerPrincipal, current_manager, get_db
+from app.db.models import Manager, RescueCase, Shift
+from app.observability.redaction import redact_if_health
+from app.workers.tasks import mark_shift_absence_task
+
+router = APIRouter(prefix="/api/shifts", tags=["shifts"])
+
+logger = get_logger(__name__)
+
+# Case statuses where a rescue is still running on the shift (must match the
+# orchestrator's redelivery guard in `RescueOrchestrator.mark_absence`).
+_LIVE_CASE_STATUSES = ("OPEN", "OFFERING", "AWAITING_APPROVAL", "ESCALATED")
+
+
+class MarkAbsenceIn(BaseModel):
+    """Optional context for the audit trail.
+
+    The reason rides only inside the `ABSENCE_MARKED` audit payload,
+    health-redacted before storage (spec §10) and capped; nothing else is
+    persisted, and it never becomes a message to anyone.
+    """
+
+    reason: str | None = Field(default=None, max_length=200)
+
+
+async def _shift_or_404(
+    session: AsyncSession, shift_id: str, manager_id: str
+) -> Shift:
+    """The shift when it exists at one of the manager's locations, else 404.
+
+    A shift from another location is indistinguishable from a missing one:
+    no existence leak across locations (spec §7.5).
+    """
+    shift = (
+        await session.execute(select(Shift).where(Shift.id == shift_id))
+    ).scalar_one_or_none()
+    if shift is not None:
+        manager = (
+            await session.execute(select(Manager).where(Manager.id == manager_id))
+        ).scalar_one_or_none()
+        if manager is not None and shift.location_id in (manager.location_ids or []):
+            return shift
+    raise HTTPException(status_code=404, detail="Shift not found")
+
+
+@router.post("/{shift_id}/absence", status_code=202)
+async def mark_shift_absence(
+    shift_id: str,
+    body: MarkAbsenceIn | None = None,
+    principal: ManagerPrincipal = Depends(current_manager),
+    session: AsyncSession = Depends(get_db),
+) -> dict[str, str]:
+    """Enqueue the manager-marked absence; the worker owns the domain change.
+
+    202 after the enqueue, 404 when the shift does not exist or is not at the
+    manager's location, 409 when the shift is already absent or already has a
+    live rescue — the detail says which, so the dashboard can show it.
+    """
+    shift = await _shift_or_404(session, shift_id, principal.manager_id)
+    if shift.status == "absent":
+        raise HTTPException(status_code=409, detail="Shift is already marked absent")
+    live = (
+        await session.execute(
+            select(RescueCase.id).where(
+                RescueCase.shift_id == shift_id,
+                RescueCase.status.in_(_LIVE_CASE_STATUSES),
+            )
+        )
+    ).first()
+    if live is not None:
+        raise HTTPException(
+            status_code=409, detail="A rescue is already running for this shift"
+        )
+    reason = redact_if_health(body.reason) if body is not None and body.reason else None
+    try:
+        mark_shift_absence_task.delay(shift_id, principal.manager_id, reason)
+    except Exception as error:
+        logger.error(
+            "shift_absence_enqueue_failed", shift_id=shift_id, error=str(error)[:200]
+        )
+        raise HTTPException(
+            status_code=500, detail="Failed to enqueue shift absence"
+        ) from error
+    return {"status": "queued", "id": shift_id}

backend/app/api/ws.py

@@ -0,0 +1,133 @@
+"""Live event channel for the dashboard (spec §7.5): a thin authenticated relay.
+
+`WS /ws/locations/{location_id}?token=...` validates the manager's JWT before
+the socket streams anything, checks the manager's right to that location, and
+then forwards every event published on the location's Redis channel as JSON
+text. It is a relay only: no database access per event, no business logic —
+the screens keep rendering from the API, the events just tell them what to
+refetch (spec §7.6).
+
+Close codes (sent right after `accept` so the browser can actually read them:
+closing before the accept degenerates into an opaque HTTP 403 handshake
+rejection under uvicorn, which would hide the reason from the client):
+
+- 4401: missing, malformed, expired or wrong-role token
+- 4403: a manager without rights on that location
+- 4503: the event broker (Redis) is unavailable
+"""
+
+import asyncio
+import contextlib
+
+import structlog
+from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.core.config import Settings, get_settings
+from app.db.models import Manager
+from app.db.session import create_engine_and_session
+from app.events import EventSubscription, channel_for, subscribe
+from app.security.tokens import verify_token
+
+router = APIRouter()
+
+logger = structlog.get_logger(__name__)
+
+ALLOWED_ROLES = ("manager", "operator")
+
+
+def ws_session_factory(
+    settings: Settings = Depends(get_settings),
+) -> async_sessionmaker[AsyncSession]:
+    """One short-lived session factory per socket (the manager check is the
+    only DB access in the whole connection — never per event)."""
+    _, factory = create_engine_and_session(settings.database_url)
+    return factory
+
+
+async def _reject(websocket: WebSocket, code: int) -> None:
+    """Answer the handshake with a close code the client can read."""
+    await websocket.accept()
+    await websocket.close(code=code)
+
+
+async def _forward(websocket: WebSocket, subscription: EventSubscription) -> None:
+    """Relay loop: publish-side frames in, socket text frames out."""
+    async for event in subscription.events():
+        await websocket.send_text(event.to_json())
+
+
+async def _drain(websocket: WebSocket) -> None:
+    """Consume client frames until the socket closes.
+
+    The relay and this loop run together and the connection ends when EITHER
+    finishes. Awaiting `websocket.receive()` alone deadlocks against a client
+    that is waiting for the server to finish (the browser and the test client
+    both do), which is how this hung the suite the first time.
+    """
+    try:
+        while True:
+            await websocket.receive()
+    except WebSocketDisconnect:
+        return
+
+
+@router.websocket("/ws/locations/{location_id}")
+async def location_events(
+    websocket: WebSocket,
+    location_id: str,
+    token: str = "",
+    session_factory: async_sessionmaker[AsyncSession] = Depends(ws_session_factory),
+    settings: Settings = Depends(get_settings),
+) -> None:
+    """Stream one location's dashboard events to an authenticated manager."""
+    claims = verify_token(token, settings) if token else None
+    if claims is None or claims.role not in ALLOWED_ROLES:
+        await _reject(websocket, 4401)
+        return
+
+    # The manager may only watch their own locations; the operator role sees
+    # everything (spec §7.6). Seeded managers carry `location_ids`.
+    async with session_factory() as session:
+        manager = (
+            await session.execute(select(Manager).where(Manager.id == claims.manager_id))
+        ).scalar_one_or_none()
+    if manager is None or (
+        claims.role == "manager" and location_id not in (manager.location_ids or [])
+    ):
+        await _reject(websocket, 4403)
+        return
+
+    try:
+        # Subscribe BEFORE accepting: a down broker must answer a clear close
+        # code instead of a socket that hangs open and never forwards (§9.3).
+        subscription = await subscribe(settings.redis_url, channel_for(location_id))
+    except Exception as error:
+        logger.warning(
+            "ws_event_broker_unavailable",
+            location_id=location_id,
+            error=str(error)[:200],
+        )
+        await _reject(websocket, 4503)
+        return
+
+    await websocket.accept()
+    relay = asyncio.create_task(_forward(websocket, subscription))
+    listener = asyncio.create_task(_drain(websocket))
+    try:
+        # Whichever ends first ends the connection: the client going away, the
+        # subscription finishing (the broker closed), or a socket error.
+        await asyncio.wait({relay, listener}, return_when=asyncio.FIRST_COMPLETED)
+    finally:
+        # Close BEFORE cancelling: cancelling a `receive()` that is blocked in
+        # the ASGI portal deadlocks the test client, while closing the socket
+        # lets the listener raise its disconnect and finish on its own.
+        with contextlib.suppress(RuntimeError):
+            await websocket.close()
+        relay.cancel()
+        await asyncio.gather(relay, listener, return_exceptions=True)
+        await subscription.close()
+
+
+__all__ = ["router", "ws_session_factory"]

backend/app/core/config.py

@@ -66,7 +66,9 @@ class Settings(BaseSettings):
     langfuse_public_key: str = ""
     langfuse_secret_key: str = ""
     langfuse_host: str = "https://cloud.langfuse.com"
-    sentry_dsn: str = ""  # declared for .env parity; Sentry init not wired yet
+    # Sentry (spec §9.3): initialized in the API lifespan and the worker
+    # bootstrap when set; unset stays a silent no-op.
+    sentry_dsn: str = ""
 
     @property
     def cors_origin_list(self) -> list[str]:

backend/app/db/seed.py

@@ -121,13 +121,69 @@ def _shift(
     )
 
 
-def _shifts() -> list[Shift]:
+def _demo_anchors(shifts: list[Shift], now: datetime) -> list[Shift]:
+    """Two extra shifts that keep a demo day usable at any hour.
+
+    The rotation has fixed windows (07:00, 15:00, 23:00...), so a demo late in
+    the evening finds today's shifts already finished and tomorrow's outside the
+    roster's "today": the Simulator then has nobody who can report an absence and
+    the board looks dead. When the day is dry, two shifts are anchored to the
+    current hour instead — one in progress, one starting soon — assigned to
+    employees who are free that day. During working hours nothing is added, so
+    the normal case stays exactly as before.
+    """
+    day_zero = [
+        shift for shift in shifts if shift.starts_at.date() == SEED_START_DATE.date()
+    ]
+    in_progress = [s for s in day_zero if s.starts_at <= now <= s.ends_at]
+    starting_later = [s for s in day_zero if s.starts_at > now]
+    if in_progress and starting_later:
+        return []
+
+    # The pool is exactly the employees the rotation uses: inventing ids here
+    # would schedule a shift for somebody who does not exist.
+    pool = sorted(
+        {shift.employee_id for shift in shifts if shift.employee_id is not None}
+    )
+    assigned = {shift.employee_id for shift in day_zero if shift.employee_id is not None}
+    available = [employee for employee in pool if employee not in assigned]
+
+    date = SEED_START_DATE.date().isoformat()
+    anchors: list[Shift] = []
+    for slot, (starts, ends) in enumerate(
+        (
+            (now - timedelta(hours=1), now + timedelta(hours=6)),
+            (now + timedelta(hours=2), now + timedelta(hours=10)),
+        )
+    ):
+        if (slot == 0 and in_progress) or (slot == 1 and starting_later):
+            continue
+        if not available:
+            break
+        employee_id = available.pop(0)
+        role = employee_id.split("_", 2)[-1].rsplit("_", 1)[0] if "_" in employee_id else "floor"
+        anchors.append(
+            Shift(
+                id=f"shift_lt_{date}_{role}_anchor{slot + 1}",
+                location_id=DEMO_LOCATION_ID,
+                role=role,
+                starts_at=starts,
+                ends_at=ends,
+                employee_id=employee_id,
+                status="scheduled",
+            )
+        )
+    return anchors
+
+
+def _shifts(now: datetime | None = None) -> list[Shift]:
     shifts: list[Shift] = []
     zone = SEED_START_DATE.replace(tzinfo=UTC)
 
     def at(day_offset: int, hour: int) -> datetime:
         return zone + timedelta(days=day_offset, hours=hour)
 
+    moment = now if now is not None else datetime.now(UTC)
     for d in range(SEED_DAYS):
         weekend = (SEED_START_DATE + timedelta(days=d)).weekday() >= 5
 
@@ -184,7 +240,7 @@ def at(day_offset: int, hour: int) -> datetime:
         supervisor_id = f"emp_{d % 2 + 25:02d}_supervisor"
         shifts.append(_shift("supervisor", d, "main", at(d, 11), at(d, 19), supervisor_id))
 
-    return shifts
+    return shifts + _demo_anchors(shifts, moment)
 
 
 def _availability_blocks() -> list[AvailabilityBlock]: