fix(stats): guard daily mileage source selection marking

.superpowers/sdd/task-3-report.md

@@ -80,3 +80,31 @@ Corrected the remaining `projectDailyMileageSQL` contract issue so `trusted_sour
   - explicitly verify no fallback expression using `s.source_endpoint` remains.
 
 This enforces the strict source-of-truth rule that endpoint data for final projection must come from latest connection metadata only, even when it is null/empty.
+
+## Fix (reviewer findings #3)
+
+Addressed stale candidate selection re-flagging by narrowing source marking to the same eligibility rules used by projection:
+
+- In `go/vehicle-gateway/internal/stats/source_mileage.go`, `markSelectedSourceSQL` now:
+  - No longer joins `vehicle_daily_mileage` to discover the selected source.
+  - Selects the same single winning candidate directly from `vehicle_daily_mileage_source`, constrained by:
+    - VIN/date/protocol
+    - `quality_status = QualityOK`
+    - mileage bounds (`0..maxSelectedDailyMileageKM`)
+    - enabled source metadata
+    - same trust/sample/time ordering
+  - Marks only that candidate as `is_selected = 1`.
+- In `go/vehicle-gateway/internal/stats/source_mileage_test.go`, `TestProjectDailyMileageSelectsCandidateAndMarksSource` now verifies:
+  - the mark query includes candidate filters and ordering, including `s2.quality_status`, `s2.daily_mileage_km`, and `COALESCE(ds.enabled, 1) = 1`.
+  - it no longer references `vehicle_daily_mileage` as a join source.
+  - it passes the selection window arg through to marking (`maxSelectedDailyMileageKM`) and uses seven bind args.
+
+Validation commands:
+
+```bash
+cd go/vehicle-gateway && go test ./internal/stats -run TestProjectDailyMileage -count=1
+# PASS: lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats
+
+cd go/vehicle-gateway && go test ./internal/stats -count=1
+# PASS: lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats
+```

go/vehicle-gateway/internal/stats/source_mileage.go

@@ -118,7 +118,15 @@ func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate
 	); err != nil {
 		return err
 	}
-	_, err := exec.ExecContext(ctx, markSelectedSourceSQL, vin, statDate, string(protocol))
+	_, err := exec.ExecContext(ctx, markSelectedSourceSQL,
+		vin,
+		statDate,
+		string(protocol),
+		maxSelectedDailyMileageKM,
+		vin,
+		statDate,
+		string(protocol),
+	)
 	return err
 }
 
@@ -221,11 +229,31 @@ ON DUPLICATE KEY UPDATE
 
 const markSelectedSourceSQL = `
 UPDATE vehicle_daily_mileage_source s
-JOIN vehicle_daily_mileage m
+JOIN (
-  ON m.vin = s.vin
+  SELECT
- AND m.stat_date = s.stat_date
+    s2.source_key,
- AND m.protocol = s.protocol
+    s2.vin,
- AND m.trusted_source_key = s.source_key
+    s2.stat_date,
+    s2.protocol
+  FROM vehicle_daily_mileage_source s2
+  LEFT JOIN vehicle_data_source ds
+    ON ds.protocol = s2.protocol AND ds.source_ip = s2.source_ip
+  WHERE s2.vin = ?
+    AND s2.stat_date = ?
+    AND s2.protocol = ?
+    AND s2.quality_status = '` + QualityOK + `'
+    AND s2.daily_mileage_km BETWEEN 0 AND ?
+    AND COALESCE(ds.enabled, 1) = 1
+  ORDER BY COALESCE(ds.trust_priority, 100),
+           s2.sample_count DESC,
+           s2.latest_event_time DESC,
+           s2.source_key ASC
+  LIMIT 1
+) selected_source
+  ON selected_source.source_key = s.source_key
+ AND selected_source.vin = s.vin
+ AND selected_source.stat_date = s.stat_date
+ AND selected_source.protocol = s.protocol
 SET s.is_selected = 1
 WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ?
 `

go/vehicle-gateway/internal/stats/source_mileage_test.go

@@ -119,4 +119,27 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) {
 	if !strings.Contains(markSelected, "UPDATE vehicle_daily_mileage_source s") || !strings.Contains(markSelected, "SET s.is_selected = 1") {
 		t.Fatalf("mark query should flag the elected source: %s", markSelected)
 	}
+	if strings.Contains(markSelected, "JOIN vehicle_daily_mileage m") {
+		t.Fatalf("mark query should not rejoin final mileage for candidate selection: %s", markSelected)
+	}
+	for _, want := range []string{
+		"JOIN (",
+		"SELECT",
+		"s2.source_key",
+		"s2.quality_status = '" + QualityOK + "'",
+		"s2.daily_mileage_km BETWEEN 0 AND",
+		"COALESCE(ds.enabled, 1) = 1",
+		"ORDER BY COALESCE(ds.trust_priority, 100)",
+		"LIMIT 1",
+	} {
+		if !strings.Contains(markSelected, want) {
+			t.Fatalf("mark query missing %q: %s", want, markSelected)
+		}
+	}
+	if len(exec.calls[2].args) != 7 {
+		t.Fatalf("mark query args = %d, want 7", len(exec.calls[2].args))
+	}
+	if got := exec.calls[2].args[3]; got != maxSelectedDailyMileageKM {
+		t.Fatalf("mark query max mileage arg = %#v, want %d", got, maxSelectedDailyMileageKM)
+	}
 }