lateral tuning migration and dashcam mode

system/manager/manager.py

@@ -136,6 +136,15 @@ def _load_first_available_param_value(params: Params, params_cache: Params, sour
   return None
 
 
+def _has_persisted_param_file(params: Params, key: str | bytes) -> bool:
+  try:
+    path = params.get_param_path(key)
+  except Exception:
+    return False
+
+  return bool(path) and os.path.isfile(path)
+
+
 def migrate_starpilot_param_renames(params: Params, params_cache: Params) -> None:
   if STARPILOT_PARAM_RENAME_MIGRATION_FLAG.exists():
     return
@@ -320,6 +329,7 @@ def migrate_starpilot_default_parity(params: Params, params_cache: Params) -> No
   if STARPILOT_DEFAULTS_PARITY_MIGRATION_FLAG.exists():
     return
 
+  seeded_keys: list[str] = []
   desired_bool_values = {
     "AdvancedLateralTune": True,
     "ForceAutoTuneOff": True,
@@ -330,27 +340,34 @@ def migrate_starpilot_default_parity(params: Params, params_cache: Params) -> No
   }
 
   for key, value in desired_bool_values.items():
+    if _has_persisted_param_file(params, key) or _has_persisted_param_file(params_cache, key):
+      continue
     params.put_bool(key, value)
     params_cache.put_bool(key, value)
+    seeded_keys.append(key)
 
-  params.put_float("CEModelStopTime", 7.0)
-  params_cache.put_float("CEModelStopTime", 7.0)
+  if not _has_persisted_param_file(params, "CEModelStopTime") and not _has_persisted_param_file(params_cache, "CEModelStopTime"):
+    params.put_float("CEModelStopTime", 7.0)
+    params_cache.put_float("CEModelStopTime", 7.0)
+    seeded_keys.append("CEModelStopTime")
 
   # Rebase default regression fix:
   # EVTuning must default to enabled on EV/direct-drive platforms to preserve
-  # StarPilot acceleration profile behavior.
+  # StarPilot acceleration profile behavior, but existing user overrides win.
   carparams_blob = params.get("CarParamsPersistent") or params.get("CarParams")
   if carparams_blob is not None:
     try:
       with car.CarParams.from_bytes(carparams_blob) as cp:
         is_ev_platform = cp.transmissionType == car.CarParams.TransmissionType.direct
-      if is_ev_platform and not params.get_bool("TruckTuning"):
+      if is_ev_platform and not params.get_bool("TruckTuning") and not _has_persisted_param_file(params, "EVTuning") and not _has_persisted_param_file(params_cache, "EVTuning"):
         params.put_bool("EVTuning", True)
         params_cache.put_bool("EVTuning", True)
+        seeded_keys.append("EVTuning")
     except Exception:
       cloudlog.exception("Failed EVTuning EV default parity migration")
 
-  cloudlog.warning("Applied one-time StarPilot default parity migration for lateral/longitudinal toggles")
+  if seeded_keys:
+    cloudlog.warning(f"Applied one-time StarPilot default parity migration for {seeded_keys}")
 
   try:
     STARPILOT_DEFAULTS_PARITY_MIGRATION_FLAG.parent.mkdir(parents=True, exist_ok=True)
@@ -524,6 +541,7 @@ def manager_init() -> None:
   # Canonicalize legacy string encodings (e.g. INT params stored as "26.000000")
   # before bulk reads below to avoid repeated cast warnings and UI-side churn.
   migrate_param_type_canonicalization(params)
+  migrate_starpilot_default_parity(params, params_cache)
 
   # set unset params to their default value
   for k in params.all_keys():
@@ -558,7 +576,6 @@ def manager_init() -> None:
 
   # Branch migration: rename legacy Bolt fingerprint persisted in CarParams.
   migrate_legacy_bolt_fingerprint(params)
-  migrate_starpilot_default_parity(params, params_cache)
 
   # set dongle id
   reg_res = register(show_spinner=True)

system/manager/test/test_manager.py

@@ -2,6 +2,8 @@ import os
 import pytest
 import signal
 import time
+from pathlib import Path
+import json
 
 from cereal import car
 from openpilot.common.params import Params
@@ -16,6 +18,59 @@ MAX_STARTUP_TIME = 3
 BLACKLIST_PROCS = ['manage_athenad', 'pandad', 'pigeond']
 
 
+class FileBackedFakeParams:
+  def __init__(self, root: Path, values: dict[str, object] | None = None):
+    self.root = root
+    self.root.mkdir(parents=True, exist_ok=True)
+    for key, value in (values or {}).items():
+      self.put(key, value)
+
+  def get_param_path(self, key):
+    return str(self.root / (key.decode() if isinstance(key, bytes) else str(key)))
+
+  def get(self, key):
+    path = Path(self.get_param_path(key))
+    if not path.is_file():
+      return None
+
+    raw = path.read_bytes()
+    try:
+      return raw.decode("utf-8")
+    except UnicodeDecodeError:
+      return raw
+
+  def get_bool(self, key):
+    value = self.get(key)
+    if value is None:
+      return False
+    if isinstance(value, bytes):
+      value = value.decode("utf-8", errors="ignore")
+    return str(value).strip().lower() in ("1", "true", "yes", "on")
+
+  def put(self, key, value):
+    path = Path(self.get_param_path(key))
+    path.parent.mkdir(parents=True, exist_ok=True)
+
+    if isinstance(value, bytes):
+      raw = value
+    elif isinstance(value, bool):
+      raw = b"1" if value else b"0"
+    elif isinstance(value, float):
+      raw = str(float(value)).encode("utf-8")
+    elif isinstance(value, (dict, list)):
+      raw = json.dumps(value, separators=(",", ":")).encode("utf-8")
+    else:
+      raw = str(value).encode("utf-8")
+
+    path.write_bytes(raw)
+
+  def put_bool(self, key, value):
+    self.put(key, bool(value))
+
+  def put_float(self, key, value):
+    self.put(key, float(value))
+
+
 class TestManager:
   def setup_method(self):
     HARDWARE.set_power_save(False)
@@ -85,6 +140,30 @@ class TestManager:
     assert params.get("ExperimentalLongitudinalEnabled") is None
     assert params_cache.get("ExperimentalLongitudinalEnabled") is None
 
+  def test_migrate_starpilot_default_parity_preserves_existing_values(self, tmp_path, monkeypatch):
+    monkeypatch.setattr(manager, "STARPILOT_DEFAULTS_PARITY_MIGRATION_FLAG", tmp_path / "starpilot_defaults_parity_v1")
+
+    params = FileBackedFakeParams(tmp_path / "params", {
+      "AdvancedLateralTune": False,
+      "ForceAutoTuneOff": False,
+      "HumanAcceleration": True,
+      "CEModelStopTime": 3.5,
+    })
+    params_cache = FileBackedFakeParams(tmp_path / "cache", {
+      "NNFF": True,
+    })
+
+    manager.migrate_starpilot_default_parity(params, params_cache)
+
+    assert not params.get_bool("AdvancedLateralTune")
+    assert not params.get_bool("ForceAutoTuneOff")
+    assert params.get_bool("HumanAcceleration")
+    assert params.get("CEModelStopTime") == "3.5"
+    assert params_cache.get_bool("NNFF")
+
+    assert Path(params.get_param_path("HumanFollowing")).is_file()
+    assert not params.get_bool("HumanFollowing")
+
   @pytest.mark.skip("this test is flaky the way it's currently written, should be moved to test_onroad")
   def test_clean_exit(self, subtests):
     """