feat(sfm): unified deployment timeline (deprecate deployment_records)

Phase 4.  Rebuilds the seismograph "Deployment History" + "Timeline"
sections on the unit detail page as a single derived view computed from
three sources: unit_assignments (authoritative project/location windows),
unit_history (calibration/retirement/deployed state changes), and SFM
events overlaid per assignment window (count + peak PVS + last event).

Fixes the wonky-timeline symptoms: missing entries, duplicate/contradictory
rows, and no visibility into what the unit was actually doing during each
deployment window.

Backend:
- backend/services/deployment_timeline.py: new deployment_timeline_for_unit()
  helper.  Merges UnitAssignment rows (with SFM event overlay fetched
  concurrently via httpx), UnitHistory state-change rows (filtered to
  meaningful change_types and de-noised by dropping rows where
  old_value == new_value — there's noise in legacy audit log from
  record_history() being called on every save), and synthetic "gap"
  entries between assignments >= 1 day apart.  Sorts newest first.

- backend/routers/units.py: new GET /api/units/{unit_id}/deployment_timeline
  endpoint with optional include_events=false flag.

- backend/routers/project_locations.py: assign / unassign / swap /
  update endpoints now write UnitHistory rows on every assignment
  lifecycle event.  New change_types: assignment_created,
  assignment_ended, assignment_swapped, assignment_updated.  These
  surface in the unified timeline (where the assignment row itself
  shows the structural data; the audit row is filtered out to avoid
  double-rendering).  Closes a real gap — assignment changes were
  previously invisible to any audit consumer.

- backend/migrate_deprecate_deployment_records.py: non-destructive
  migration.  Adds deployment_records.deprecated_at column.  For each
  legacy row without a matching UnitAssignment, best-effort
  synthesizes one (with the free-text location_name preserved in
  notes).  Marks every processed row.  Idempotent.  DROP TABLE
  deferred to a follow-up release.

Frontend (templates/unit_detail.html):
- Removed legacy "Deployment History" card (with Log Deployment button)
  and the separate "Timeline" card.  Replaced with a single
  "Deployment Timeline" section.
- Three entry visual styles: assignment rows (orange dot, location +
  project link, event-overlay summary), gap rows (dashed outline, idle
  day count), and state_change rows (navy dot, friendly label, old →
  new value).  Active assignments get a green dot + "active" badge.
- Existing loadUnitHistory() and loadDeploymentHistory() functions kept
  as shims that delegate to loadDeploymentTimeline(), so modal-save
  callbacks that referenced them still trigger a refresh of the visible
  section.  Legacy function bodies preserved under _legacy_*_unused
  names for archeology; not called by anything.

Verified end-to-end:
- BE11529 timeline now shows 2 entries (active assignment with 24-event
  overlay + the deployed→benched state change), compared to the previous
  noisy mix that included 6 no-op state-change rows.
- Migration ran against real DB: 1 legacy row processed (had no
  project_id, marked deprecated without backfill).
- Assign / unassign / swap / edit now leave a paper trail in
  unit_history.

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

backend/migrate_deprecate_deployment_records.py

@@ -0,0 +1,209 @@
+"""
+Migration: deprecate the `deployment_records` table.
+
+Why:
+  The deployment-history view on the unit detail page used to render
+  from `deployment_records` — a manually-maintained table that drifted
+  out of sync with `unit_assignments` (the auto-written project/location
+  assignment table).  That caused the "wonky timeline" symptom: missing
+  entries, duplicate / contradictory rows, and a UI that couldn't tell
+  the operator what the unit was actually doing during each window.
+
+  Phase 4 of the SFM integration replaces the deployment-history view
+  with a derived timeline computed from `unit_assignments` +
+  `unit_history` + SFM event overlay.  This migration is the cleanup:
+
+  1. Adds a `deprecated_at` timestamp column to `deployment_records` so
+     we can mark rows that have been migrated.
+  2. For every `deployment_records` row that does NOT have a matching
+     `unit_assignments` row (matched by unit_id + overlapping date
+     range), synthesizes a best-effort UnitAssignment row.  The
+     free-text `location_name` from the legacy table is preserved on
+     the new row's `notes` field (we do NOT try to fuzzy-match it to a
+     MonitoringLocation id; too error-prone — operators will need to
+     reattach those manually if they want).
+  3. Marks every migrated deployment_records row with `deprecated_at`.
+
+  This migration is non-destructive: deployment_records rows stay in
+  the DB.  The actual `DROP TABLE` happens in a follow-up release after
+  one operator cycle confirms nothing relies on the legacy data.
+
+Idempotent: re-running the script is a no-op if the column already
+exists and all migratable rows have already been processed.
+
+Run with:
+  docker exec terra-view-terra-view-1 python3 /app/backend/migrate_deprecate_deployment_records.py
+"""
+
+import os
+import sqlite3
+import uuid
+from datetime import datetime
+
+DB_PATH = "./data/seismo_fleet.db"
+
+
+def migrate_database():
+    if not os.path.exists(DB_PATH):
+        print(f"Database not found at {DB_PATH}")
+        return
+
+    print(f"Migrating database: {DB_PATH}")
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    cur = conn.cursor()
+
+    # 1. Add deprecated_at column if not present.
+    cur.execute("PRAGMA table_info(deployment_records)")
+    cols = {row["name"] for row in cur.fetchall()}
+    if "deprecated_at" not in cols:
+        print("Adding deployment_records.deprecated_at column ...")
+        cur.execute("ALTER TABLE deployment_records ADD COLUMN deprecated_at TEXT")
+        conn.commit()
+    else:
+        print("deployment_records.deprecated_at column already exists — skipping ADD COLUMN")
+
+    # 2. Find candidate rows: not-yet-deprecated deployment_records that
+    #    have no matching unit_assignments row.
+    cur.execute("""
+        SELECT id, unit_id, deployed_date, estimated_removal_date,
+               actual_removal_date, project_id, project_ref, location_name, notes
+          FROM deployment_records
+         WHERE deprecated_at IS NULL
+    """)
+    rows = cur.fetchall()
+    print(f"\nFound {len(rows)} deployment_records rows not yet deprecated.")
+
+    backfilled = 0
+    skipped_no_match_attempted = 0
+    skipped_already_in_assignments = 0
+    skipped_missing_unit = 0
+
+    for row in rows:
+        unit_id = row["unit_id"]
+        if not unit_id:
+            print(f"  ⚠ row {row['id']!r}: no unit_id, marking deprecated without backfill")
+            cur.execute(
+                "UPDATE deployment_records SET deprecated_at=? WHERE id=?",
+                (datetime.utcnow().isoformat(), row["id"]),
+            )
+            skipped_missing_unit += 1
+            continue
+
+        # Does the unit still exist?  If not, skip — we don't synthesize
+        # assignments for ghost units.
+        cur.execute("SELECT id, device_type FROM roster WHERE id=?", (unit_id,))
+        roster = cur.fetchone()
+        if not roster:
+            print(f"  ⚠ row {row['id']!r}: unit_id {unit_id!r} not in roster, marking deprecated without backfill")
+            cur.execute(
+                "UPDATE deployment_records SET deprecated_at=? WHERE id=?",
+                (datetime.utcnow().isoformat(), row["id"]),
+            )
+            skipped_missing_unit += 1
+            continue
+
+        # Check if a UnitAssignment already covers this window (any overlap).
+        # We don't try to be clever — just see if a row exists for this unit
+        # whose [assigned_at, assigned_until] overlaps the deployment window.
+        cur.execute("""
+            SELECT id FROM unit_assignments
+             WHERE unit_id=?
+               AND (assigned_at <= COALESCE(?, '9999')
+                    AND COALESCE(assigned_until, '9999') >= COALESCE(?, '0000'))
+             LIMIT 1
+        """, (
+            unit_id,
+            row["actual_removal_date"] or row["estimated_removal_date"] or row["deployed_date"],
+            row["deployed_date"],
+        ))
+        if cur.fetchone():
+            cur.execute(
+                "UPDATE deployment_records SET deprecated_at=? WHERE id=?",
+                (datetime.utcnow().isoformat(), row["id"]),
+            )
+            skipped_already_in_assignments += 1
+            continue
+
+        # No matching UnitAssignment — synthesize one.  We can't FK to a
+        # MonitoringLocation because the legacy `location_name` is free
+        # text.  Backfilled rows go in with location_id = "" (empty) and
+        # the original location_name dropped into notes for operator
+        # context.
+        if not row["project_id"]:
+            print(f"  ⚠ row {row['id']!r}: no project_id, can't synthesize unit_assignment, marking deprecated")
+            cur.execute(
+                "UPDATE deployment_records SET deprecated_at=? WHERE id=?",
+                (datetime.utcnow().isoformat(), row["id"]),
+            )
+            skipped_no_match_attempted += 1
+            continue
+
+        synthesized_id = str(uuid.uuid4())
+        synth_notes_parts = []
+        if row["location_name"]:
+            synth_notes_parts.append(f"Legacy location: {row['location_name']}")
+        if row["project_ref"]:
+            synth_notes_parts.append(f"Legacy project_ref: {row['project_ref']}")
+        if row["notes"]:
+            synth_notes_parts.append(f"Original notes: {row['notes']}")
+        synth_notes_parts.append(f"(Synthesized from deployment_records row {row['id']})")
+        synth_notes = " | ".join(synth_notes_parts)
+
+        assigned_until = row["actual_removal_date"]
+        # Don't auto-close active deployments based on estimated_removal_date.
+        status = "completed" if assigned_until else "active"
+
+        # Need a location_id to satisfy NOT NULL constraint.  Use a
+        # placeholder UUID so the FK can be cleaned up later if the
+        # operator decides to retarget the assignment to a real location.
+        # We tag this with the synthesized notes so it's discoverable.
+        placeholder_loc_id = ""
+
+        try:
+            cur.execute("""
+                INSERT INTO unit_assignments (
+                    id, unit_id, location_id, project_id, device_type,
+                    assigned_at, assigned_until, status, notes, created_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """, (
+                synthesized_id,
+                unit_id,
+                placeholder_loc_id,
+                row["project_id"],
+                roster["device_type"] or "seismograph",
+                row["deployed_date"] or datetime.utcnow().isoformat(),
+                assigned_until,
+                status,
+                synth_notes,
+                datetime.utcnow().isoformat(),
+            ))
+            cur.execute(
+                "UPDATE deployment_records SET deprecated_at=? WHERE id=?",
+                (datetime.utcnow().isoformat(), row["id"]),
+            )
+            backfilled += 1
+            print(
+                f"  ✓ row {row['id']!r}: synthesized unit_assignment {synthesized_id} "
+                f"for unit={unit_id} project={row['project_id'][:8]}… "
+                f"({row['deployed_date']} → {assigned_until or 'present'})"
+            )
+        except Exception as e:
+            print(f"  ✗ row {row['id']!r}: failed to synthesize — {e}")
+
+    conn.commit()
+    conn.close()
+
+    print("\n────────────────────────────────────────────────────────")
+    print(f"Backfilled new unit_assignments:      {backfilled}")
+    print(f"Already covered (deprecated only):    {skipped_already_in_assignments}")
+    print(f"No project_id (deprecated only):      {skipped_no_match_attempted}")
+    print(f"Missing/orphaned unit (deprecated):   {skipped_missing_unit}")
+    print(f"\nNOTE: synthesized rows have an empty location_id and the legacy")
+    print(f"      free-text location is preserved in notes.  An operator should")
+    print(f"      retarget them to real MonitoringLocation rows if they want")
+    print(f"      events to show up on a location detail page.")
+
+
+if __name__ == "__main__":
+    migrate_database()

backend/routers/project_locations.py

@@ -30,6 +30,7 @@ from backend.models import (
     RosterUnit,
     MonitoringSession,
     DataFile,
+    UnitHistory,
 )
 from backend.templates_config import templates
 from backend.utils.timezone import local_to_utc
@@ -37,6 +38,42 @@ from backend.utils.timezone import local_to_utc
 router = APIRouter(prefix="/api/projects/{project_id}", tags=["project-locations"])
 
 
+# ── Audit log helper ──────────────────────────────────────────────────────────
+# Mirrors record_history() in roster_edit.py.  Kept local to avoid cross-router
+# imports.  The four assignment endpoints below use this to write UnitHistory
+# rows that the unit-detail deployment timeline (Phase 4) renders.
+
+def _record_assignment_history(
+    db: Session,
+    unit_id: str,
+    change_type: str,
+    *,
+    old_value: Optional[str] = None,
+    new_value: Optional[str] = None,
+    notes: Optional[str] = None,
+) -> None:
+    """Append a UnitHistory row for an assignment-lifecycle event.
+
+    change_type values used:
+      - assignment_created   — unit assigned to a location (new assignment)
+      - assignment_ended     — unit unassigned / removed (assigned_until set)
+      - assignment_swapped   — unit replaced by another at the same location
+      - assignment_updated   — assignment dates / notes edited via PATCH
+
+    Caller is responsible for db.commit().
+    """
+    db.add(UnitHistory(
+        unit_id=unit_id,
+        change_type=change_type,
+        field_name="unit_assignment",
+        old_value=old_value,
+        new_value=new_value,
+        changed_at=datetime.utcnow(),
+        source="manual",
+        notes=notes,
+    ))
+
+
 # ============================================================================
 # Shared helpers
 # ============================================================================
@@ -403,6 +440,13 @@ async def assign_unit_to_location(
     )
 
     db.add(assignment)
+    _record_assignment_history(
+        db,
+        unit_id=unit_id,
+        change_type="assignment_created",
+        new_value=f"{location.name} (project: {location.project_id})",
+        notes=form_data.get("notes"),
+    )
     db.commit()
     db.refresh(assignment)
 
@@ -448,6 +492,15 @@ async def unassign_unit(
     assignment.status = "completed"
     assignment.assigned_until = datetime.utcnow()
 
+    location = db.query(MonitoringLocation).filter_by(id=assignment.location_id).first()
+    _record_assignment_history(
+        db,
+        unit_id=assignment.unit_id,
+        change_type="assignment_ended",
+        old_value=location.name if location else assignment.location_id,
+        new_value="unassigned",
+    )
+
     db.commit()
 
     return {"success": True, "message": "Unit unassigned successfully"}
@@ -558,12 +611,28 @@ async def update_assignment(
                 ),
             )
 
+    # Capture change description for audit log BEFORE mutating.
+    old_start = assignment.assigned_at.isoformat() if assignment.assigned_at else None
+    old_end   = assignment.assigned_until.isoformat() if assignment.assigned_until else "active"
+    new_start = new_assigned_at.isoformat() if new_assigned_at else None
+    new_end   = new_assigned_until.isoformat() if new_assigned_until else "active"
+
     # Apply.
     assignment.assigned_at = new_assigned_at
     assignment.assigned_until = new_assigned_until
     assignment.notes = new_notes
     assignment.status = "completed" if new_assigned_until is not None else "active"
 
+    if old_start != new_start or old_end != new_end:
+        _record_assignment_history(
+            db,
+            unit_id=assignment.unit_id,
+            change_type="assignment_updated",
+            old_value=f"{old_start} → {old_end}",
+            new_value=f"{new_start} → {new_end}",
+            notes=new_notes,
+        )
+
     db.commit()
     db.refresh(assignment)
 
@@ -631,6 +700,16 @@ async def swap_unit_on_location(
     if current:
         current.assigned_until = datetime.utcnow()
         current.status = "completed"
+        # If the swap is replacing a different unit, that unit's deployment ended.
+        if current.unit_id != unit_id:
+            _record_assignment_history(
+                db,
+                unit_id=current.unit_id,
+                change_type="assignment_swapped",
+                old_value=location.name,
+                new_value=f"swapped out → {unit_id}",
+                notes=notes,
+            )
 
     # Create new assignment
     new_assignment = UnitAssignment(
@@ -644,6 +723,13 @@ async def swap_unit_on_location(
         notes=notes,
     )
     db.add(new_assignment)
+    _record_assignment_history(
+        db,
+        unit_id=unit_id,
+        change_type="assignment_swapped" if (current and current.unit_id != unit_id) else "assignment_created",
+        new_value=f"{location.name} (project: {location.project_id})",
+        notes=notes,
+    )
 
     # Update modem pairing on the seismograph if modem provided
     if modem_id:

backend/routers/units.py

@@ -136,3 +136,37 @@ async def get_unit_events(
     )
     result["device_type"] = unit.device_type
     return result
+
+
+@router.get("/units/{unit_id}/deployment_timeline")
+async def get_unit_deployment_timeline(
+    unit_id: str,
+    include_events: bool = Query(True),
+    db: Session = Depends(get_db),
+):
+    """
+    Return a chronological deployment timeline for a unit.
+
+    Merges three sources:
+      1. unit_assignments — authoritative project/location deployments
+      2. unit_history — state changes (calibration, retirement, etc.)
+      3. SFM events — per-assignment overlay (count + peak PVS + last event)
+
+    Replaces the legacy /api/deployments/{unit_id} (which read the
+    deprecated `deployment_records` table) and the
+    /api/roster/history/{unit_id} timeline endpoint, unifying them into
+    a single derived view.
+
+    Gaps >= 1 day between consecutive assignments are surfaced as
+    synthetic "gap" entries.
+
+    Pass include_events=false to skip the SFM event overlay (saves N
+    HTTP calls; useful for fast text-only history dumps).
+    """
+    from backend.services.deployment_timeline import deployment_timeline_for_unit
+
+    return await deployment_timeline_for_unit(
+        db,
+        unit_id,
+        include_event_overlay=include_events,
+    )

backend/services/deployment_timeline.py

@@ -0,0 +1,256 @@
+"""
+Deployment timeline service — replaces the legacy `deployment_records`-driven
+timeline on the seismograph unit detail page.
+
+Architecture:
+  - `unit_assignments` is the authoritative source for "where was this unit"
+    (one row per location/time-window).  Auto-written by the project location
+    swap/assign/unassign/update workflows.
+  - `unit_history` is the audit log for non-location state changes
+    (calibration toggles, retirement, allocation, etc.).
+  - SFM events are overlaid per assignment window to show "what was the unit
+    actually doing during this deployment" (count + peak PVS + last-event).
+
+Gaps between assignments are emitted as synthetic "gap" entries so operators
+can see when the unit was idle vs out-of-service.
+
+`deployment_records` is being deprecated; this module does not read it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import datetime, timedelta
+from typing import Optional
+
+import httpx
+from sqlalchemy.orm import Session
+
+from backend.models import (
+    UnitAssignment,
+    UnitHistory,
+    MonitoringLocation,
+    Project,
+    RosterUnit,
+)
+from backend.services.sfm_events import (
+    SFM_BASE_URL,
+    _fetch_events_for_serial,
+    _iso_utc,
+)
+
+log = logging.getLogger("backend.services.deployment_timeline")
+
+# Don't emit synthetic gap entries shorter than this (seconds).  Avoids visual
+# clutter from a sub-second handoff during a swap workflow.
+_MIN_GAP_SECONDS = 24 * 3600  # 1 day
+
+# Per-call timeout when querying SFM for the event overlay.
+_SFM_TIMEOUT = 10.0
+_SFM_FETCH_CEILING = 5000
+
+
+# ── Public API ────────────────────────────────────────────────────────────────
+
+
+async def deployment_timeline_for_unit(
+    db: Session,
+    unit_id: str,
+    *,
+    include_event_overlay: bool = True,
+) -> dict:
+    """Build a chronological timeline for a unit.
+
+    Returns:
+      {
+        "unit_id":       str,
+        "device_type":   str,
+        "entries": [
+          {
+            "kind":        "assignment" | "gap" | "state_change",
+            "starts_at":   ISO timestamp,
+            "ends_at":     ISO timestamp | None,
+            "duration_days": float | None,
+            # — assignment-only fields —
+            "assignment_id":  str,
+            "location_id":    str,
+            "location_name":  str,
+            "project_id":     str,
+            "project_name":   str,
+            "is_active":      bool,
+            "event_overlay":  {event_count, peak_pvs, peak_pvs_at, last_event}
+                              or None if include_event_overlay=False,
+            "notes":          str | None,
+            # — gap-only fields —
+            "context":        "between assignments" | None,
+            # — state_change-only fields —
+            "change_type":    str,
+            "field_name":     str | None,
+            "old_value":      str | None,
+            "new_value":      str | None,
+            "source":         str,
+            "history_notes":  str | None,
+          },
+          ...                            # newest first
+        ],
+      }
+    """
+    unit = db.query(RosterUnit).filter_by(id=unit_id).first()
+    if not unit:
+        return {"unit_id": unit_id, "device_type": None, "entries": []}
+
+    # 1. Load assignments + their location/project lookups in bulk.
+    assignments = (
+        db.query(UnitAssignment)
+        .filter(UnitAssignment.unit_id == unit_id)
+        .order_by(UnitAssignment.assigned_at.asc())
+        .all()
+    )
+
+    loc_ids = {a.location_id for a in assignments}
+    proj_ids = {a.project_id for a in assignments}
+    loc_map = {
+        l.id: l for l in db.query(MonitoringLocation).filter(
+            MonitoringLocation.id.in_(loc_ids)
+        ).all()
+    } if loc_ids else {}
+    proj_map = {
+        p.id: p for p in db.query(Project).filter(
+            Project.id.in_(proj_ids)
+        ).all()
+    } if proj_ids else {}
+
+    # 2. Load relevant unit_history rows.  We surface state changes that
+    #    operators care about on a deployment timeline: calibration status,
+    #    retirement, deployed flag, allocation, calibration date, and the
+    #    assignment_* events we just added (those are redundant with the
+    #    assignment rows themselves, so we skip them to avoid double-rendering).
+    interesting_change_types = (
+        "calibration_status_change",
+        "retired_change",
+        "deployed_change",
+        "allocation_change",
+        "last_calibrated_change",
+        "next_calibration_due_change",
+    )
+    history = (
+        db.query(UnitHistory)
+        .filter(UnitHistory.unit_id == unit_id)
+        .filter(UnitHistory.change_type.in_(interesting_change_types))
+        .order_by(UnitHistory.changed_at.asc())
+        .all()
+    )
+
+    now = datetime.utcnow()
+
+    # 3. Optionally fetch SFM event overlay for each assignment window.
+    #    Concurrent fan-out via httpx + asyncio.gather.
+    overlays: dict[str, dict] = {}
+    if include_event_overlay and assignments and unit.device_type == "seismograph":
+        async with httpx.AsyncClient(timeout=_SFM_TIMEOUT) as client:
+            results = await asyncio.gather(
+                *(
+                    _fetch_events_for_serial(
+                        client,
+                        serial=unit_id,
+                        from_dt=a.assigned_at,
+                        to_dt=a.assigned_until or now,
+                        false_trigger=None,
+                        limit=_SFM_FETCH_CEILING,
+                    )
+                    for a in assignments
+                ),
+                return_exceptions=False,
+            )
+        for a, events in zip(assignments, results):
+            peak = None
+            peak_at = None
+            last_ev = None
+            for ev in events:
+                pvs = ev.get("peak_vector_sum")
+                if pvs is not None and (peak is None or pvs > peak):
+                    peak = pvs
+                    peak_at = ev.get("timestamp")
+                ts = ev.get("timestamp")
+                if ts and (last_ev is None or ts > last_ev):
+                    last_ev = ts
+            overlays[a.id] = {
+                "event_count": len(events),
+                "peak_pvs":    peak,
+                "peak_pvs_at": peak_at,
+                "last_event":  last_ev,
+            }
+
+    # 4. Build entries.  Start by emitting assignment rows + gap rows between
+    #    consecutive assignments, then add state-change rows from unit_history.
+    entries: list[dict] = []
+
+    for idx, a in enumerate(assignments):
+        loc = loc_map.get(a.location_id)
+        proj = proj_map.get(a.project_id)
+        is_active = a.assigned_until is None
+        ends_at = a.assigned_until or now
+        duration_days = (ends_at - a.assigned_at).total_seconds() / 86400 if a.assigned_at else None
+
+        entry = {
+            "kind":          "assignment",
+            "starts_at":     _iso_utc(a.assigned_at),
+            "ends_at":       _iso_utc(a.assigned_until),
+            "duration_days": round(duration_days, 1) if duration_days is not None else None,
+            "assignment_id": a.id,
+            "location_id":   a.location_id,
+            "location_name": loc.name if loc else None,
+            "project_id":    a.project_id,
+            "project_name":  proj.name if proj else None,
+            "is_active":     is_active,
+            "notes":         a.notes,
+            "event_overlay": overlays.get(a.id),
+        }
+        entries.append(entry)
+
+        # Gap detection: from the end of this assignment to the start of the
+        # next one.  Only emit gaps that are at least _MIN_GAP_SECONDS long
+        # so trivial sub-second handoffs during swaps don't clutter the view.
+        if idx + 1 < len(assignments):
+            next_a = assignments[idx + 1]
+            gap_start = a.assigned_until or now
+            gap_end = next_a.assigned_at
+            gap_seconds = (gap_end - gap_start).total_seconds() if gap_end and gap_start else 0
+            if gap_seconds >= _MIN_GAP_SECONDS:
+                entries.append({
+                    "kind":          "gap",
+                    "starts_at":     _iso_utc(gap_start),
+                    "ends_at":       _iso_utc(gap_end),
+                    "duration_days": round(gap_seconds / 86400, 1),
+                    "context":       "between assignments",
+                })
+
+    # 5. State changes — interleaved by timestamp.  Skip no-op rows where
+    #    old_value == new_value (an artifact of the legacy record_history()
+    #    being called on every save regardless of whether the field changed).
+    for h in history:
+        if h.old_value == h.new_value:
+            continue
+        entries.append({
+            "kind":          "state_change",
+            "starts_at":     _iso_utc(h.changed_at),
+            "ends_at":       None,
+            "duration_days": None,
+            "change_type":   h.change_type,
+            "field_name":    h.field_name,
+            "old_value":     h.old_value,
+            "new_value":     h.new_value,
+            "source":        h.source,
+            "history_notes": h.notes,
+        })
+
+    # 6. Sort newest first.  Active assignments (no end) sort by start time,
+    #    same as everything else.
+    entries.sort(key=lambda e: e.get("starts_at") or "", reverse=True)
+
+    return {
+        "unit_id":     unit.id,
+        "device_type": unit.device_type,
+        "entries":     entries,
+    }

templates/unit_detail.html

@@ -278,19 +278,17 @@
                 <p id="viewNote" class="mt-1 text-gray-900 dark:text-white whitespace-pre-wrap">--</p>
             </div>
 
-            <!-- Deployment History -->
+            <!-- Deployment Timeline (Phase 4 unified view — derived from
+                 unit_assignments + unit_history + SFM event overlay) -->
             <div class="border-t border-gray-200 dark:border-gray-700 pt-6">
                 <div class="flex justify-between items-center mb-4">
-                    <h3 class="text-lg font-semibold text-gray-900 dark:text-white">Deployment History</h3>
-                    <button onclick="openNewDeploymentModal()" class="px-3 py-1.5 bg-seismo-orange hover:bg-orange-600 text-white rounded-lg text-sm transition-colors flex items-center gap-1.5">
-                        <svg class="w-4 h-4" fill="none" stroke="currentColor" viewBox="0 0 24 24">
-                            <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M12 4v16m8-8H4"></path>
-                        </svg>
-                        Log Deployment
+                    <h3 class="text-lg font-semibold text-gray-900 dark:text-white">Deployment Timeline</h3>
+                    <button onclick="loadDeploymentTimeline()" class="px-3 py-1.5 text-sm rounded-lg border border-gray-300 dark:border-gray-600 hover:bg-gray-50 dark:hover:bg-gray-700 text-gray-700 dark:text-gray-300">
+                        ↻ Refresh
                     </button>
                 </div>
-                <div id="deploymentHistory" class="space-y-3">
-                    <p class="text-sm text-gray-500 dark:text-gray-400">Loading...</p>
+                <div id="deploymentTimeline" class="space-y-3">
+                    <p class="text-sm text-gray-500 dark:text-gray-400">Loading timeline…</p>
                 </div>
             </div>
 
@@ -379,13 +377,6 @@
                 </div>
             </div>
 
-            <!-- Unit History Timeline -->
-            <div class="border-t border-gray-200 dark:border-gray-700 pt-6">
-                <h3 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">Timeline</h3>
-                <div id="historyTimeline" class="space-y-3">
-                    <p class="text-sm text-gray-500 dark:text-gray-400">Loading history...</p>
-                </div>
-            </div>
 
             <!-- Photos -->
             <div class="border-t border-gray-200 dark:border-gray-700 pt-6">
@@ -1632,8 +1623,15 @@ async function uploadPhoto(file) {
     }
 }
 
-// Load and display unit history timeline
+// Legacy timeline loader — Phase 4 unified the timeline view.  Now a shim
+// that delegates to loadDeploymentTimeline() so existing callers from modal
+// save handlers still trigger a refresh of the visible section.
 async function loadUnitHistory() {
+    if (typeof loadDeploymentTimeline === 'function') {
+        return loadDeploymentTimeline();
+    }
+}
+async function _legacy_loadUnitHistory_unused() {
     try {
         const response = await fetch(`/api/roster/history/${unitId}`);
         if (!response.ok) {
@@ -1805,10 +1803,18 @@ async function pingModem() {
 }
 
 // ============================================================
-// Deployment History
+// Deployment History (legacy — Phase 4 superseded by deployment_timeline)
 // ============================================================
 
+// Phase 4 shim: delegate to the unified timeline loader so existing modal
+// save handlers (legacy "Log Deployment" form, edit-save callbacks) still
+// trigger a refresh of the visible Deployment Timeline section.
 async function loadDeploymentHistory() {
+    if (typeof loadDeploymentTimeline === 'function') {
+        return loadDeploymentTimeline();
+    }
+}
+async function _legacy_loadDeploymentHistory_unused() {
     try {
         const res = await fetch(`/api/deployments/${unitId}`);
         const data = await res.json();
@@ -1969,14 +1975,163 @@ loadCalibrationInterval();
 setupCalibrationAutoCalc();
 loadUnitData().then(() => {
     loadPhotos();
-    loadUnitHistory();
-    loadDeploymentHistory();
+    loadDeploymentTimeline();
     if (currentUnit && currentUnit.device_type === 'seismograph') {
         document.getElementById('sfmEventsSection').classList.remove('hidden');
         loadUnitEvents();
     }
 });
 
+// ── Unified Deployment Timeline (Phase 4) ────────────────────────────────────
+// Replaces the legacy loadDeploymentHistory() + loadUnitHistory() pair.
+// Derives entries from unit_assignments + unit_history + SFM event overlay.
+
+async function loadDeploymentTimeline() {
+    const container = document.getElementById('deploymentTimeline');
+    container.innerHTML = '<p class="text-sm text-gray-500 dark:text-gray-400">Loading timeline…</p>';
+
+    try {
+        const r = await fetch(`/api/units/${currentUnit.id}/deployment_timeline`);
+        if (!r.ok) throw new Error('HTTP ' + r.status);
+        const d = await r.json();
+        renderDeploymentTimeline(d.entries || [], container);
+    } catch (e) {
+        container.innerHTML = `<p class="text-sm text-red-500">Failed to load timeline: ${e.message}</p>`;
+    }
+}
+
+function _dtFmtDate(iso) {
+    if (!iso) return '—';
+    return iso.slice(0, 10);
+}
+
+function _dtFmtDateTime(iso) {
+    if (!iso) return '—';
+    return iso.slice(0, 19).replace('T', ' ');
+}
+
+function _dtEsc(s) {
+    if (s == null) return '';
+    return String(s).replace(/&/g, '&amp;').replace(/</g, '&lt;').replace(/>/g, '&gt;').replace(/"/g, '&quot;');
+}
+
+function _dtPpvClass(v) {
+    if (v == null) return 'text-gray-400';
+    if (v < 0.5) return 'text-green-600 dark:text-green-400';
+    if (v < 2.0) return 'text-amber-600 dark:text-amber-400';
+    return 'text-red-600 dark:text-red-400 font-semibold';
+}
+
+function _dtRenderAssignment(e) {
+    const start = _dtFmtDate(e.starts_at);
+    const end   = e.is_active ? 'present' : _dtFmtDate(e.ends_at);
+    const dur   = (e.duration_days != null)
+        ? `<span class="text-xs text-gray-500 dark:text-gray-400 ml-2">(${e.duration_days.toFixed(1)} day${e.duration_days === 1 ? '' : 's'})</span>`
+        : '';
+    const ov = e.event_overlay || {};
+    const evCount = ov.event_count ?? 0;
+    const peak    = ov.peak_pvs;
+
+    const locLink = e.location_id
+        ? `<a href="/projects/${_dtEsc(e.project_id)}/nrl/${_dtEsc(e.location_id)}" class="text-seismo-orange hover:text-seismo-navy font-medium">📍 ${_dtEsc(e.location_name || 'unnamed location')}</a>`
+        : `<span class="text-gray-500 dark:text-gray-400 italic">📍 (no location FK — synthesized from legacy deployment_records)</span>`;
+
+    const projLine = e.project_name
+        ? `<div class="text-xs text-gray-500 dark:text-gray-400 mt-0.5">${_dtEsc(e.project_name)}</div>`
+        : '';
+
+    const activeBadge = e.is_active
+        ? '<span class="px-2 py-0.5 rounded text-xs bg-green-100 text-green-800 dark:bg-green-900/30 dark:text-green-300">active</span>'
+        : '';
+
+    const overlay = evCount > 0
+        ? `<div class="mt-2 flex items-center gap-4 text-xs text-gray-600 dark:text-gray-400">
+               <span><strong class="text-gray-900 dark:text-white">${evCount.toLocaleString()}</strong> event${evCount === 1 ? '' : 's'}</span>
+               ${peak != null ? `<span>peak <strong class="${_dtPpvClass(peak)}">${peak.toFixed(4)} in/s</strong></span>` : ''}
+               ${ov.last_event ? `<span>last ${_dtFmtDateTime(ov.last_event)}</span>` : ''}
+           </div>`
+        : `<div class="mt-2 text-xs text-gray-500 dark:text-gray-400 italic">No events recorded during this window.</div>`;
+
+    const notes = e.notes
+        ? `<div class="mt-2 text-xs text-gray-600 dark:text-gray-400 italic">${_dtEsc(e.notes)}</div>`
+        : '';
+
+    return `<div class="flex gap-3">
+        <div class="flex flex-col items-center pt-1">
+            <span class="w-3 h-3 rounded-full ${e.is_active ? 'bg-green-500' : 'bg-seismo-orange'}"></span>
+        </div>
+        <div class="flex-1 bg-gray-50 dark:bg-slate-900/40 rounded-lg p-3">
+            <div class="flex items-center justify-between flex-wrap gap-2">
+                <div class="text-sm text-gray-700 dark:text-gray-300">
+                    <strong>${start}</strong> → <strong>${end}</strong>${dur}
+                </div>
+                ${activeBadge}
+            </div>
+            <div class="mt-1">${locLink}</div>
+            ${projLine}
+            ${overlay}
+            ${notes}
+        </div>
+    </div>`;
+}
+
+function _dtRenderGap(e) {
+    return `<div class="flex gap-3">
+        <div class="flex flex-col items-center pt-1">
+            <span class="w-3 h-3 rounded-full border-2 border-gray-400 dark:border-gray-500"></span>
+        </div>
+        <div class="flex-1 bg-gray-50/40 dark:bg-slate-900/20 rounded-lg p-3 border border-dashed border-gray-300 dark:border-gray-700">
+            <div class="text-sm text-gray-600 dark:text-gray-400">
+                <strong>${_dtFmtDate(e.starts_at)}</strong> → <strong>${_dtFmtDate(e.ends_at)}</strong>
+                <span class="text-xs ml-2">(${e.duration_days.toFixed(1)} day${e.duration_days === 1 ? '' : 's'} idle)</span>
+            </div>
+            <div class="text-xs text-gray-500 dark:text-gray-400 mt-0.5">No active assignment</div>
+        </div>
+    </div>`;
+}
+
+function _dtRenderStateChange(e) {
+    // Friendly labels for known change_types.
+    const labels = {
+        deployed_change:           'Deployed status changed',
+        retired_change:            'Retired status changed',
+        calibration_status_change: 'Calibration status changed',
+        last_calibrated_change:    'Last calibrated updated',
+        next_calibration_due_change: 'Next calibration due updated',
+        allocation_change:         'Allocation changed',
+    };
+    const label = labels[e.change_type] || e.change_type;
+
+    return `<div class="flex gap-3">
+        <div class="flex flex-col items-center pt-1">
+            <span class="w-3 h-3 rounded-full bg-seismo-navy"></span>
+        </div>
+        <div class="flex-1 bg-gray-50 dark:bg-slate-900/30 rounded-lg p-3">
+            <div class="text-sm text-gray-700 dark:text-gray-300">
+                📅 <strong>${_dtFmtDateTime(e.starts_at)}</strong> — ${_dtEsc(label)}
+            </div>
+            <div class="text-xs text-gray-500 dark:text-gray-400 mt-0.5">
+                ${_dtEsc(e.old_value || '—')} → <strong>${_dtEsc(e.new_value || '—')}</strong>
+            </div>
+            ${e.history_notes ? `<div class="text-xs text-gray-500 dark:text-gray-400 mt-1 italic">${_dtEsc(e.history_notes)}</div>` : ''}
+        </div>
+    </div>`;
+}
+
+function renderDeploymentTimeline(entries, container) {
+    if (!entries.length) {
+        container.innerHTML = '<p class="text-sm text-gray-500 dark:text-gray-400">No deployment history yet. Assign this unit to a project location to start a deployment record.</p>';
+        return;
+    }
+    const html = entries.map(e => {
+        if (e.kind === 'assignment')   return _dtRenderAssignment(e);
+        if (e.kind === 'gap')          return _dtRenderGap(e);
+        if (e.kind === 'state_change') return _dtRenderStateChange(e);
+        return '';
+    }).join('');
+    container.innerHTML = html;
+}
+
 // ── SFM Events section ──────────────────────────────────────────────────────
 function clearUnitEventFilters() {
     document.getElementById('ue-filter-bucket').value = 'all';