fix(monitor): hold conflicting terminal coordinates

go/vehicle-gateway/internal/realtime/snapshot_writer.go

@@ -717,7 +717,7 @@ const electJT808RealtimeLocationSQL = `
 INSERT INTO vehicle_realtime_location
   (protocol, vin, plate, source_key, event_time, latitude, longitude, speed_kmh, total_mileage_km, total_mileage_event_time,
    soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, location_conflict, location_conflict_distance_m,
-   received_at, event_id)
+   received_at, event_id, updated_at)
 SELECT s.protocol, s.vin, s.plate, s.source_key, s.event_time, s.latitude, s.longitude, s.speed_kmh,
        s.total_mileage_km, s.total_mileage_event_time, s.soc_percent, s.altitude_m, s.direction_deg,
        s.alarm_flag, s.status_flag,
@@ -729,7 +729,7 @@ SELECT s.protocol, s.vin, s.plate, s.source_key, s.event_time, s.latitude, s.lon
           FROM vehicle_realtime_location_source other
          WHERE other.vin = s.vin AND other.protocol = s.protocol AND other.source_key <> s.source_key
            AND other.quality_status = 'OK' AND other.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE)),
-       s.received_at, s.event_id
+       s.received_at, s.event_id, COALESCE(s.received_at, s.event_time, CURRENT_TIMESTAMP)
 FROM vehicle_realtime_location_source s
 LEFT JOIN vehicle_location_source_policy policy
   ON policy.vin = s.vin AND policy.protocol = s.protocol AND policy.source_key = s.source_key
@@ -737,8 +737,22 @@ LEFT JOIN vehicle_realtime_location cur
 	ON cur.vin = s.vin AND cur.protocol = s.protocol
 WHERE s.vin = ? AND s.protocol = 'JT808' AND s.quality_status = 'OK' AND COALESCE(policy.enabled, 1) = 1
 ORDER BY
-  CASE WHEN s.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) THEN 0 ELSE 1 END ASC,
+  CASE WHEN s.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE)
+         OR (cur.source_key = s.source_key AND EXISTS(
+           SELECT 1 FROM vehicle_realtime_location_source conflicting
+            WHERE conflicting.vin = s.vin AND conflicting.protocol = s.protocol
+              AND conflicting.source_key <> s.source_key AND conflicting.quality_status = 'OK'
+              AND conflicting.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE)
+              AND ST_Distance_Sphere(POINT(s.longitude, s.latitude), POINT(conflicting.longitude, conflicting.latitude)) > 200
+         )) THEN 0 ELSE 1 END ASC,
   (COALESCE(policy.priority, CASE s.source_kind WHEN 'DIRECT' THEN 20 WHEN 'PLATFORM' THEN 30 ELSE 40 END)
+    + CASE WHEN cur.source_key = s.source_key AND EXISTS(
+        SELECT 1 FROM vehicle_realtime_location_source conflicting
+         WHERE conflicting.vin = s.vin AND conflicting.protocol = s.protocol
+           AND conflicting.source_key <> s.source_key AND conflicting.quality_status = 'OK'
+           AND conflicting.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE)
+           AND ST_Distance_Sphere(POINT(s.longitude, s.latitude), POINT(conflicting.longitude, conflicting.latitude)) > 200
+      ) THEN -10000 ELSE 0 END
     + CASE WHEN cur.source_key = s.source_key AND cur.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) THEN -5 ELSE 0 END
     + CASE WHEN cur.source_key <> '' AND cur.source_key <> s.source_key AND s.consecutive_good_samples < 3 THEN 1000 ELSE 0 END) ASC,
   s.received_at DESC, s.source_key ASC
@@ -751,7 +765,7 @@ ON DUPLICATE KEY UPDATE
   altitude_m = VALUES(altitude_m), direction_deg = VALUES(direction_deg), alarm_flag = VALUES(alarm_flag),
   status_flag = VALUES(status_flag), location_conflict = VALUES(location_conflict),
   location_conflict_distance_m = VALUES(location_conflict_distance_m), received_at = VALUES(received_at),
-  event_id = VALUES(event_id), updated_at = CURRENT_TIMESTAMP
+  event_id = VALUES(event_id), updated_at = VALUES(updated_at)
 `
 
 var cleanupInvalidRealtimeTotalMileageSQL = []string{

go/vehicle-gateway/internal/realtime/snapshot_writer_test.go

@@ -317,7 +317,7 @@ func TestJT808LocationArbitrationGuardsJumpsAndSourceFlapping(t *testing.T) {
 			t.Fatalf("JT808 source guard missing %q: %s", want, upsertJT808RealtimeLocationSourceSQL)
 		}
 	}
-	for _, want := range []string{"vehicle_location_source_policy", "cur.source_key = s.source_key", "consecutive_good_samples < 3", "INTERVAL 2 MINUTE", "> 200"} {
+	for _, want := range []string{"vehicle_location_source_policy", "cur.source_key = s.source_key", "consecutive_good_samples < 3", "INTERVAL 2 MINUTE", "> 200", "THEN -10000", "updated_at = VALUES(updated_at)"} {
 		if !strings.Contains(electJT808RealtimeLocationSQL, want) {
 			t.Fatalf("JT808 canonical election missing %q: %s", want, electJT808RealtimeLocationSQL)
 		}

vehicle-data-platform/apps/api/internal/platform/mysql_queries.go

@@ -402,8 +402,8 @@ func buildVehicleRealtimeSQL(query url.Values) SQLQuery {
 	// Ordering only by updated_at makes the selected point flap whenever protocols
 	// report at different cadences. When every source is stale, recency remains the
 	// safest fallback so an old high-priority source cannot mask newer evidence.
-	orderExpr := `CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) THEN 0 ELSE 1 END ASC, ` +
+	orderExpr := `CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 10 MINUTE) THEN 0 ELSE 1 END ASC, ` +
-		`CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) THEN CASE l.protocol ` +
+		`CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 10 MINUTE) THEN CASE l.protocol ` +
 		`WHEN 'GB32960' THEN 10 WHEN 'YUTONG_MQTT' THEN 20 WHEN 'JT808' THEN 30 ELSE 100 END ELSE 100 END ASC, ` +
 		`l.updated_at DESC, l.protocol ASC`
 	return SQLQuery{

vehicle-data-platform/apps/api/internal/platform/query_builders_test.go

@@ -209,7 +209,7 @@ func TestBuildVehicleRealtimeSQL(t *testing.T) {
 		t.Fatalf("SQL should keep vehicle-level stable pagination order: %s", built.Text)
 	}
 	for _, want := range []string{
-		"INTERVAL 2 MINUTE",
+		"INTERVAL 10 MINUTE",
 		"WHEN 'GB32960' THEN 10",
 		"WHEN 'YUTONG_MQTT' THEN 20",
 		"WHEN 'JT808' THEN 30",