feat(dashboard): reorder top row, move schedule below map, source call-ins from SFM

- Top row left→right: Recent Alerts | Recent Call-Ins (2 cols) | Fleet Summary
- Today's Schedule becomes a horizontal collapsible card below Fleet Map.
  Collapsed by default; auto-expands when pending actions are detected in
  the rendered partial; manual toggle sticks via localStorage.
- New /api/recent-event-callins proxies SFM /db/events and bulk-joins each
  serial against RosterUnit for in-roster annotation. Phases the
  heartbeat-derived /api/recent-callins out of the UI while keeping it as
  a backup endpoint for now.
- Call-ins card renders a dense 2-column grid (last 10 events) showing
  PVS, sensor_location, false-trigger badge, event timestamp, and
  links to the unit page when rostered.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

backend/routers/activity.py

@@ -4,11 +4,29 @@ from sqlalchemy import desc
 from pathlib import Path
 from datetime import datetime, timedelta, timezone
 from typing import List, Dict, Any
+import os
+import logging
+import httpx
 from backend.database import get_db
 from backend.models import UnitHistory, Emitter, RosterUnit
 
+log = logging.getLogger(__name__)
+
 router = APIRouter(prefix="/api", tags=["activity"])
 
+SFM_BASE_URL = os.getenv("SFM_BASE_URL", "http://localhost:8200")
+
+
+def _humanize_age(seconds: float) -> str:
+    if seconds < 60:
+        return "just now"
+    if seconds < 3600:
+        return f"{int(seconds / 60)}m ago"
+    if seconds < 86400:
+        hrs = seconds / 3600
+        return f"{int(hrs)}h {int((hrs % 1) * 60)}m ago"
+    return f"{int(seconds / 86400)}d ago"
+
 PHOTOS_BASE_DIR = Path("data/photos")
 
 
@@ -144,3 +162,86 @@ def get_recent_callins(hours: int = 6, limit: int = None, db: Session = Depends(
         "hours": hours,
         "time_threshold": time_threshold.isoformat()
     }
+
+
+@router.get("/recent-event-callins")
+async def get_recent_event_callins(limit: int = 10, db: Session = Depends(get_db)):
+    """
+    Recent unit call-ins derived from SFM event forwards.
+
+    Architecture context: the live ACH replacement is on hold, so call-homes
+    arrive as Blastware ACH event files forwarded by series3-watcher and
+    landed in the SFM events store.  One event ≈ one call-in.  This is the
+    forward-looking source of "recent call-ins" that will eventually replace
+    the heartbeat-based /recent-callins endpoint entirely.
+
+    Each row represents one event; multiple consecutive events from the same
+    serial are intentionally NOT collapsed — each one is a distinct call-home.
+    """
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.get(
+                f"{SFM_BASE_URL}/db/events",
+                params={"limit": limit},
+            )
+            resp.raise_for_status()
+            payload = resp.json()
+    except httpx.HTTPError as e:
+        log.warning("SFM /db/events failed for recent-event-callins: %s", e)
+        return {"call_ins": [], "total": 0, "error": str(e)}
+
+    events = payload.get("events", []) or []
+
+    # Bulk-resolve serials → roster (single query, no N+1)
+    serials = list({ev.get("serial") for ev in events if ev.get("serial")})
+    roster_map: Dict[str, RosterUnit] = {}
+    if serials:
+        roster_map = {
+            r.id: r
+            for r in db.query(RosterUnit).filter(RosterUnit.id.in_(serials)).all()
+        }
+
+    now = datetime.now(timezone.utc)
+    call_ins: List[Dict[str, Any]] = []
+
+    for ev in events:
+        serial = ev.get("serial")
+        if not serial:
+            continue
+
+        roster = roster_map.get(serial)
+
+        # created_at = when SFM received the forward.  Falls back to the event
+        # timestamp if the SFM payload didn't carry created_at (older rows).
+        created_at_str = ev.get("created_at") or ev.get("timestamp")
+        time_ago = "—"
+        if created_at_str:
+            try:
+                ts = datetime.fromisoformat(created_at_str.replace("Z", "+00:00"))
+                if ts.tzinfo is None:
+                    ts = ts.replace(tzinfo=timezone.utc)
+                time_ago = _humanize_age((now - ts).total_seconds())
+            except ValueError:
+                pass
+
+        call_ins.append({
+            "unit_id":         serial,
+            "serial":          serial,
+            "event_id":        ev.get("id"),
+            "event_timestamp": ev.get("timestamp"),
+            "created_at":      ev.get("created_at"),
+            "time_ago":        time_ago,
+            "peak_vector_sum": ev.get("peak_vector_sum"),
+            "false_trigger":   bool(ev.get("false_trigger")),
+            "sensor_location": ev.get("sensor_location") or "",
+            "project":         ev.get("project") or "",
+            "device_type":     roster.device_type if roster else "seismograph",
+            "in_roster":       roster is not None,
+            "note":            (roster.note if roster else "") or "",
+        })
+
+    return {
+        "call_ins": call_ins,
+        "total":    len(call_ins),
+        "source":   "sfm-events",
+    }