feat(connect): Phase 5 — stats snapshot push to cloud

Adds _push_stats_snapshot() to cloud_sync.py. Every 5 minutes
(piggybacked on the existing _connect_loop push tick alongside the
menu snapshot) it queries the local DB and POSTs a JSON stats blob
to POST /api/remote/snapshot (site API key auth).

Stats collected:
  - open_tables: count of open/partially_paid POS orders
  - today_revenue: sum of active+paid item prices on orders
    closed today
  - today_orders: count of paid/closed POS orders today
  - online_orders_pending: online orders awaiting acceptance
  - online_orders_today: all online orders opened today
  - current_shift: active waiter shift info (waiter_id, started_at)
    if a business day is open; null otherwise
  - as_of: UTC timestamp of the snapshot

The remote manager dashboard reads this via
GET /api/remote/sites/{id}/snapshot (manager JWT).

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

local_backend/services/cloud_sync.py

@@ -351,17 +351,111 @@ async def _sync_loop():
         await asyncio.sleep(SYNC_INTERVAL_SECONDS)
 
 
+async def _push_stats_snapshot():
+    """Collect live operational stats and POST to cloud for the remote manager dashboard."""
+    if not settings.SITE_ID or not settings.CLOUD_URL:
+        return
+
+    site_numeric_id = license_state.get("site_numeric_id")
+    if not site_numeric_id:
+        return
+
+    try:
+        from database import SessionLocal
+        from models.order import Order, OrderItem
+        from models.business_day import BusinessDay
+        from models.shift import WaiterShift
+        from sqlalchemy import func
+
+        db = SessionLocal()
+        try:
+            now = datetime.now(timezone.utc)
+            today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
+
+            # Open tables = orders currently in status open or partially_paid
+            open_tables = db.query(Order).filter(
+                Order.status.in_(["open", "partially_paid"]),
+                Order.source == "pos",
+            ).count()
+
+            # Today's completed POS orders (paid or closed today)
+            today_pos_orders = db.query(Order).filter(
+                Order.status.in_(["paid", "closed"]),
+                Order.source == "pos",
+                Order.closed_at >= today_start,
+            ).all()
+            today_order_count = len(today_pos_orders)
+
+            # Today's revenue: sum of paid items on those orders
+            today_revenue = 0.0
+            for o in today_pos_orders:
+                for item in o.items:
+                    if item.status in ("active", "paid"):
+                        today_revenue += item.unit_price * item.quantity
+
+            # Online orders today
+            online_orders_today = db.query(Order).filter(
+                Order.source == "online",
+                Order.opened_at >= today_start,
+            ).count()
+
+            online_orders_pending = db.query(Order).filter(
+                Order.source == "online",
+                Order.online_status == "pending_acceptance",
+            ).count()
+
+            # Active business day
+            open_day = db.query(BusinessDay).filter(BusinessDay.status == "open").first()
+            current_shift = None
+            if open_day:
+                active_shift = db.query(WaiterShift).filter(
+                    WaiterShift.business_day_id == open_day.id,
+                    WaiterShift.ended_at == None,
+                ).first()
+                if active_shift:
+                    current_shift = {
+                        "waiter_id": active_shift.waiter_id,
+                        "started_at": active_shift.started_at.isoformat(),
+                    }
+
+        finally:
+            db.close()
+
+        snapshot = {
+            "open_tables": open_tables,
+            "today_revenue": round(today_revenue, 2),
+            "today_orders": today_order_count,
+            "online_orders_pending": online_orders_pending,
+            "online_orders_today": online_orders_today,
+            "current_shift": current_shift,
+            "as_of": now.isoformat(),
+        }
+
+        async with httpx.AsyncClient(timeout=10) as client:
+            resp = await client.post(
+                f"{settings.CLOUD_URL}/api/remote/snapshot",
+                headers={"X-Site-ID": settings.SITE_ID, "X-Site-Key": settings.SITE_KEY},
+                json={"site_id": site_numeric_id, "snapshot_json": json.dumps(snapshot)},
+            )
+            resp.raise_for_status()
+        logger.info("Stats snapshot pushed: %s", snapshot)
+
+    except Exception as e:
+        logger.warning("Stats snapshot push failed: %s", e)
+
+
 async def _connect_loop():
     """Faster loop: pulls pending online orders every CONNECT_SYNC_INTERVAL_SECONDS.
-    Also pushes the menu snapshot every 5 minutes (piggybacking on this loop)."""
-    menu_push_every = max(1, SYNC_INTERVAL_SECONDS and (5 * 60) // ORDER_POLL_INTERVAL)
+    Also pushes menu snapshot and stats every 5 minutes (piggybacking on this loop)."""
+    push_every = max(1, (5 * 60) // ORDER_POLL_INTERVAL)
     tick = 0
     while True:
         await asyncio.sleep(ORDER_POLL_INTERVAL)
         await _pull_pending_orders()
         tick += 1
-        if tick % menu_push_every == 0:
+        if tick % push_every == 0:
             await _push_menu_snapshot()
+            await _push_stats_snapshot()
 
 
 async def start_cloud_sync() -> asyncio.Task: