fix(roster): bench outgoing unit on swap / unassign / deploy-classify

The legacy RosterUnit.deployed flag drives heartbeat polling and
benched-vs-deployed roster filters.  Three workflows ended an
assignment without flipping it, so the outgoing unit kept being
polled and showed up as "deployed" forever:

  - swap endpoint            (POST /locations/{loc}/swap)
  - unassign endpoint        (POST /assignments/{aid}/unassign)
  - promote-pending endpoint (POST /deployments/pending/{id}/promote)

All three now: close the previous active assignment, break the
outgoing unit's modem pairing (both directions), and set
`deployed = False` on the outgoing unit.  Unassign and swap also
clear the modem's back-reference.

The promote-pending path additionally handles the case where the
target location already has an active assignment — that previously
silently created two active assignments at the same location.  Now
the old one is closed (assigned_until = pending capture time, status
= completed), the old unit is benched and unpaired, and an
"assignment_swapped" history row is written.  Incoming unit gets
`deployed = True` if it was on the bench.

Verified live: triggered a swap via the existing endpoint and saw
the outgoing unit flip True → False while the incoming flipped
False → True.  Test mutations rolled back.

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

backend/routers/pending_deployments.py

@@ -336,6 +336,36 @@ async def promote_pending(
         db.add(location)
         db.flush()
 
+    # If this location already has an active assignment, the /deploy
+    # capture means someone replaced that unit in the field — close the
+    # old assignment, break the outgoing unit's modem pairing, and bench
+    # it so the heartbeat / polling subsystem stops chasing it.
+    existing_active = db.query(UnitAssignment).filter(
+        UnitAssignment.location_id == location.id,
+        UnitAssignment.assigned_until == None,  # noqa: E711
+    ).first()
+    if existing_active and existing_active.unit_id != pd.unit_id:
+        existing_active.assigned_until = pd.captured_at
+        existing_active.status = "completed"
+        old_unit = db.query(RosterUnit).filter_by(id=existing_active.unit_id).first()
+        if old_unit:
+            if old_unit.deployed_with_modem_id:
+                old_modem = db.query(RosterUnit).filter_by(
+                    id=old_unit.deployed_with_modem_id, device_type="modem"
+                ).first()
+                if old_modem and old_modem.deployed_with_unit_id == old_unit.id:
+                    old_modem.deployed_with_unit_id = None
+                old_unit.deployed_with_modem_id = None
+            if old_unit.deployed:
+                old_unit.deployed = False
+        _record_history(
+            db, unit_id=existing_active.unit_id,
+            change_type="assignment_swapped",
+            old_value=location.name,
+            new_value=f"superseded by /deploy capture → {pd.unit_id}",
+            notes=notes,
+        )
+
     # Create the assignment.  assigned_at = pending capture time (so
     # events emitted after the install are correctly attributed back
     # to this location).
@@ -354,6 +384,12 @@ async def promote_pending(
     db.add(assignment)
     db.flush()
 
+    # Incoming unit is in the field again — flip it back to deployed
+    # if it was on the bench (mirrors the swap endpoint).
+    incoming_unit = db.query(RosterUnit).filter_by(id=pd.unit_id).first()
+    if incoming_unit and not incoming_unit.deployed:
+        incoming_unit.deployed = True
+
     # Promote the pending row.
     pd.status                   = "assigned"
     pd.promoted_at              = datetime.utcnow()

backend/routers/project_locations.py

@@ -876,6 +876,20 @@ async def unassign_unit(
     assignment.status = "completed"
     assignment.assigned_until = datetime.utcnow()
 
+    # Unit is leaving the field — bench it so the heartbeat / polling
+    # subsystem stops chasing it.  Also break the modem pairing both ways.
+    unit = db.query(RosterUnit).filter_by(id=assignment.unit_id).first()
+    if unit:
+        if unit.deployed_with_modem_id:
+            modem = db.query(RosterUnit).filter_by(
+                id=unit.deployed_with_modem_id, device_type="modem"
+            ).first()
+            if modem and modem.deployed_with_unit_id == unit.id:
+                modem.deployed_with_unit_id = None
+            unit.deployed_with_modem_id = None
+        if unit.deployed:
+            unit.deployed = False
+
     location = db.query(MonitoringLocation).filter_by(id=assignment.location_id).first()
     _record_assignment_history(
         db,
@@ -1320,13 +1334,18 @@ async def swap_unit_on_location(
             # deployed_with_modem_id / deployed_with_unit_id back-reference
             # doesn't orphan onto the unit that just left the field.
             old_unit = db.query(RosterUnit).filter_by(id=current.unit_id).first()
-            if old_unit and old_unit.deployed_with_modem_id:
-                old_modem = db.query(RosterUnit).filter_by(
-                    id=old_unit.deployed_with_modem_id, device_type="modem"
-                ).first()
-                if old_modem and old_modem.deployed_with_unit_id == current.unit_id:
-                    old_modem.deployed_with_unit_id = None
-                old_unit.deployed_with_modem_id = None
+            if old_unit:
+                if old_unit.deployed_with_modem_id:
+                    old_modem = db.query(RosterUnit).filter_by(
+                        id=old_unit.deployed_with_modem_id, device_type="modem"
+                    ).first()
+                    if old_modem and old_modem.deployed_with_unit_id == current.unit_id:
+                        old_modem.deployed_with_unit_id = None
+                    old_unit.deployed_with_modem_id = None
+                # Bench the outgoing unit — it's no longer in the field, so
+                # the heartbeat / polling subsystem should stop chasing it.
+                if old_unit.deployed:
+                    old_unit.deployed = False
 
     # Create new assignment
     new_assignment = UnitAssignment(