diff --git a/openpilot/common/api.py b/openpilot/common/api.py
index 3e3733c3ad..3a16521319 100644
--- a/openpilot/common/api.py
+++ b/openpilot/common/api.py
@@ -5,6 +5,8 @@ from datetime import datetime, timedelta, UTC
 from openpilot.common.hardware.hw import Paths
 from openpilot.common.version import get_version
 
+from openpilot.starpilot.connect.provider import active_provider, konik_key_pair
+
 API_HOST = os.getenv('API_HOST', 'https://api.commadotai.com')
 
 # name: jwt signature algorithm
@@ -45,6 +47,9 @@ class Api:
 
 
 def api_get(endpoint, method='GET', timeout=None, access_token=None, session=None, **params):
+  provider = active_provider()
+  if provider.name == 'offline' or os.getenv('STARPILOT_CLOUD_DISABLED') == '1':
+    raise RuntimeError('Cloud provider is unavailable')
   headers = {}
   if access_token is not None:
     headers['Authorization'] = "JWT " + access_token
@@ -53,10 +58,15 @@ def api_get(endpoint, method='GET', timeout=None, access_token=None, session=Non
 
   # TODO: add session to Api
   req = requests if session is None else session
-  return req.request(method, API_HOST + "/" + endpoint, timeout=timeout, headers=headers, params=params)
+  return req.request(method, (API_HOST if provider.name == "comma" else provider.api) + "/" + endpoint, timeout=timeout, headers=headers, params=params)
 
 
 def get_key_pair() -> tuple[str, str, str] | tuple[None, None, None]:
+  provider = active_provider()
+  if provider.name == 'offline' or os.getenv('STARPILOT_CLOUD_DISABLED') == '1':
+    return None, None, None
+  if provider.name == 'konik':
+    return konik_key_pair()
   for key in KEYS:
     if os.path.isfile(Paths.persist_root() + f'/comma/{key}') and os.path.isfile(Paths.persist_root() + f'/comma/{key}.pub'):
       with open(Paths.persist_root() + f'/comma/{key}') as private, open(Paths.persist_root() + f'/comma/{key}.pub') as public:
diff --git a/openpilot/starpilot/connect/settings.py b/openpilot/starpilot/connect/settings.py
new file mode 100644
index 0000000000..d6e5b7bee0
--- /dev/null
+++ b/openpilot/starpilot/connect/settings.py
@@ -0,0 +1,50 @@
+"""Developer-only, offroad selection of the next-boot cloud provider."""
+
+from openpilot.selfdrive.ui.mici.widgets.button import BigButton, GreyBigButton
+from openpilot.selfdrive.ui.mici.widgets.dialog import BigDialog, BigConfirmationCircleButton
+from openpilot.selfdrive.ui.mici.widgets.qr import QR
+from openpilot.selfdrive.ui.ui_state import ui_state
+from openpilot.system.ui.lib.application import gui_app
+from openpilot.system.ui.widgets.scroller import NavScroller
+from openpilot.starpilot.connect.provider import PROVIDERS, active_provider, select_provider, status
+
+
+class CloudProviderConfirmation(NavScroller):
+  def __init__(self, name, revision, completed):
+    super().__init__()
+    def confirm():
+      try:
+        select_provider(name, revision, ui_state.is_offroad)
+      except (OSError, ValueError, TypeError):
+        gui_app.push_widget(BigDialog('', 'Cloud selection changed or the vehicle is on. Refresh and try again.'))
+        return
+      self.dismiss(completed)
+    accept = BigConfirmationCircleButton('use after\nnext reboot', gui_app.texture('icons_mici/setup/driver_monitoring/dm_check.png', 64, 64), confirm)
+    accept.set_enabled(ui_state.is_offroad)
+    self._scroller.add_widgets([
+      GreyBigButton('Cloud Provider', f'Use {PROVIDERS[name].label} after the next device reboot. No automatic reboot.'),
+      GreyBigButton('', 'Cloud accounts stay separate. Returning to comma restores its saved identity.'),
+      accept,
+    ])
+
+
+class CloudProviderPage(NavScroller):
+  def __init__(self):
+    super().__init__()
+    try:
+      current = status()
+    except (OSError, ValueError, TypeError):
+      self._scroller.add_widgets([GreyBigButton('Cloud Unavailable', 'Repair the saved provider configuration before choosing a cloud.')])
+      return
+    active = active_provider()
+    self._scroller.add_widgets([
+      GreyBigButton('Developer: Cloud Provider', f'Active: {active.label}. Next boot: {PROVIDERS[current["selected"]].label}.'),
+      GreyBigButton('', 'Changes apply after a device reboot. Turn off the vehicle first.'),
+      QR(active.web),
+    ])
+    for provider in PROVIDERS.values():
+      button = BigButton(f'Use {provider.label}', 'Next boot only')
+      button.set_enabled(lambda name=provider.name: ui_state.is_offroad() and name != current['selected'])
+      button.set_click_callback(lambda name=provider.name: gui_app.push_widget(
+        CloudProviderConfirmation(name, current['revision'], lambda: self.dismiss(lambda: gui_app.push_widget(CloudProviderPage())))))
+      self._scroller.add_widgets([button])
diff --git a/openpilot/starpilot/connect/test_consumers.py b/openpilot/starpilot/connect/test_consumers.py
new file mode 100644
index 0000000000..dd552c1362
--- /dev/null
+++ b/openpilot/starpilot/connect/test_consumers.py
@@ -0,0 +1,80 @@
+"""Actual cloud consumers exclude recordings owned by another provider."""
+
+from pathlib import Path
+from types import SimpleNamespace
+from queue import Queue
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from openpilot.starpilot.connect import provider
+from openpilot.system.athena import athenad
+from openpilot.system.loggerd import uploader
+
+
+class TestCloudConsumers(unittest.TestCase):
+  def setUp(self):
+    temporary = tempfile.TemporaryDirectory()
+    self.addCleanup(temporary.cleanup)
+    self.root = Path(temporary.name)
+    self.enterContext(patch.object(provider, 'active_provider', return_value=provider.PROVIDERS['konik']))
+    self.enterContext(patch.object(athenad.Paths, 'log_root', return_value=str(self.root)))
+    self.enterContext(patch.object(athenad, 'CLOUD_PROVIDER', provider.PROVIDERS['konik']))
+    self.konik = self.root/'aaaaaaaa--bbbbbbbbbb--0'
+    self.comma = self.root/'cccccccc--dddddddddd--0'
+    for directory in (self.konik,self.comma):
+      directory.mkdir()
+      (directory/'qlog.zst').write_bytes(b'log')
+      (directory/'fcamera.hevc').write_bytes(b'camera')
+    provider.mark_recording(self.konik,'konik')
+
+  def test_actual_uploader_and_athena_listing_keep_legacy_comma_out(self):
+    with patch.object(uploader, 'Api', return_value=SimpleNamespace()):
+      instance = uploader.Uploader('0'*16, str(self.root))
+    with patch.object(uploader,'getxattr',return_value=None):
+      files = list(instance.list_upload_files(False))
+    self.assertTrue(files)
+    self.assertTrue(all(Path(path).parent == self.konik for _,_,path in files))
+    listed = athenad.scan_dir(str(self.root),'')
+    self.assertIn(self.konik.name+'/qlog.zst',listed)
+    self.assertNotIn(self.comma.name+'/qlog.zst',listed)
+
+  def test_actual_athena_upload_queue_rejects_foreign_source(self):
+    with (patch.object(athenad,'upload_queue',Queue()),
+          patch.object(athenad.UploadQueueCache,'cache'),
+          patch.object(athenad,'listUploadQueue',return_value=[])):
+      result=athenad.uploadFilesToUrls([
+        {'fn':self.comma.name+'/qlog.zst','url':'https://upload.invalid/comma','headers':{}},
+        {'fn':self.konik.name+'/qlog.zst','url':'https://upload.invalid/konik','headers':{}},
+      ])
+    self.assertEqual(result['enqueued'],1)
+    self.assertEqual(result['failed'],[self.comma.name+'/qlog.zst'])
+
+  def clips(self):
+    with patch.object(athenad.threading.Thread,'start'):
+      return athenad.VideoClips()
+
+  def test_clip_source_and_derived_directory_are_provider_owned(self):
+    clips=self.clips()
+    self.assertEqual(Path(clips.clip_path),self.root/'clips-konik')
+    self.assertEqual(clips._available_ranges('cccccccc--dddddddddd'),{})
+    own=clips._source_inputs('aaaaaaaa--bbbbbbbbbb','fcamera.hevc',0,1)
+    self.assertEqual(own,[str(self.konik/'fcamera.hevc')])
+    with self.assertRaises(ValueError):
+      clips._source_inputs('cccccccc--dddddddddd','fcamera.hevc',0,1)
+    provider.mark_recording(clips.clip_path,'konik')
+    (Path(clips.clip_path)/'test.mp4').write_bytes(b'video')
+    self.assertEqual(clips.getClipChunk('test.mp4',0)['size'],5)
+    with patch.object(provider,'active_provider',return_value=provider.PROVIDERS['comma']):
+      self.assertFalse(provider.owns_recording(Path(clips.clip_path)/'test.mp4',self.root))
+
+  def test_existing_comma_clips_keep_legacy_path_and_foreign_marker_denied(self):
+    with patch.object(athenad,'CLOUD_PROVIDER',provider.PROVIDERS['comma']):
+      clips=self.clips()
+    self.assertEqual(Path(clips.clip_path),self.root/'clips')
+    self.assertEqual(clips._source_inputs('cccccccc--dddddddddd','fcamera.hevc',0,1),
+                     [str(self.comma/'fcamera.hevc')])
+    directory=self.root/'clips-konik'
+    provider.mark_recording(directory,'konik')
+    with self.assertRaises(ValueError):
+      provider.mark_recording(directory,'comma')
diff --git a/openpilot/starpilot/connect/test_provider.py b/openpilot/starpilot/connect/test_provider.py
new file mode 100644
index 0000000000..dcb523137d
--- /dev/null
+++ b/openpilot/starpilot/connect/test_provider.py
@@ -0,0 +1,235 @@
+import base64
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from openpilot.starpilot.connect import provider as p
+
+
+class ParamsFiles:
+  def __init__(self, root):
+    self.root = root
+    root.mkdir()
+
+  def get_param_path(self, key):
+    return str(self.root / key)
+
+  def check_key(self, key):
+    assert key in (*p.STATE_KEYS, *p.TRANSIENT_KEYS, 'ConnectProvider')
+
+  def put(self, key, value, block=True):
+    self.check_key(key)
+    (self.root / key).write_bytes(value.encode() if isinstance(value, str) else value)
+
+  def remove(self, key):
+    (self.root / key).unlink(missing_ok=True)
+
+  def get(self, key):
+    path = self.root / key
+    return path.read_text() if path.exists() else None
+
+
+@pytest.fixture
+def environment(tmp_path):
+  return tmp_path / 'cloud', ParamsFiles(tmp_path / 'params')
+
+
+def choose(root, name, boot='first'):
+  return p.select_provider(name, p.status(root)['revision'], lambda: True, root=root, boot=lambda: boot)
+
+
+def test_selection_waits_for_reboot_and_restores_exact_previous_identity(environment):
+  root, params = environment
+  params.put('DongleId', '0123456789abcdef')
+  params.put('PairingEmail', 'owner@example.test')
+  params.put('AthenadUploadQueue', b'[ {"url":"https://comma.test/item"} ]')
+  params.put('AccessToken', 'transient')
+  original = {key: Path(params.get_param_path(key)).read_bytes() if Path(params.get_param_path(key)).exists() else None for key in p.STATE_KEYS}
+  choose(root, 'konik')
+  p.activate_at_boot(params, root=root, boot=lambda: 'first')
+  assert params.get('DongleId') == '0123456789abcdef'
+  p.activate_at_boot(params, root=root, boot=lambda: 'second')
+  assert params.get('DongleId') is None and params.get('AccessToken') is None
+  assert params.get('ConnectProvider') == 'konik'
+  params.put('DongleId', 'fedcba9876543210')
+  params.put('PairingEmail', 'separate@example.test')
+  choose(root, 'comma', 'second')
+  p.activate_at_boot(params, root=root, boot=lambda: 'third')
+  assert p._capture(params) == {key: None if raw is None else base64.b64encode(raw).decode() for key, raw in original.items()}
+  assert params.get('ConnectProvider') == 'comma'
+  choose(root, 'konik', 'third')
+  p.activate_at_boot(params, root=root, boot=lambda: 'fourth')
+  assert params.get('DongleId') == 'fedcba9876543210'
+  assert params.get('PairingEmail') == 'separate@example.test'
+
+
+def test_failed_restore_is_replayed_from_durable_journal(environment):
+  root, params = environment
+  params.put('DongleId', '0123456789abcdef')
+  choose(root, 'konik')
+  def interrupted(params, values):
+    params.remove('DongleId')
+    raise OSError('interrupted')
+  with patch.object(p, '_restore', side_effect=interrupted), pytest.raises(OSError):
+    p.activate_at_boot(params, root=root, boot=lambda: 'second')
+  assert (root / 'switch.json').exists()
+  p.activate_at_boot(params, root=root, boot=lambda: 'second')
+  assert not (root / 'switch.json').exists()
+  assert p.configuration(root)['active'] == 'konik'
+  choose(root, 'comma', 'second')
+  p.activate_at_boot(params, root=root, boot=lambda: 'third')
+  assert params.get('DongleId') == '0123456789abcdef'
+
+
+def test_selection_requires_fresh_revision_and_authority(environment):
+  root, _ = environment
+  rev = p.status(root)['revision']
+  with pytest.raises(ValueError):
+    p.select_provider('konik', rev, lambda: False, root=root, boot=lambda: 'one')
+  choose(root, 'konik')
+  with pytest.raises(ValueError):
+    p.select_provider('comma', rev, lambda: True, root=root, boot=lambda: 'one')
+  calls = iter((True, False))
+  with pytest.raises(ValueError):
+    p.select_provider('comma', p.status(root)['revision'], lambda: next(calls), root=root, boot=lambda: 'one')
+  assert p.configuration(root)['selected'] == 'konik'
+
+
+def test_malformed_journal_does_not_partially_restore(environment):
+  root, params = environment
+  params.put('DongleId', '0123456789abcdef')
+  choose(root, 'konik')
+  bad = dict.fromkeys(p.STATE_KEYS)
+  bad['DongleId'] = 'not base64!'
+  p._write(root / 'switch.json', {'target': 'konik', 'values': bad})
+  with pytest.raises(ValueError):
+    p.activate_at_boot(params, root=root, boot=lambda: 'second')
+  assert params.get('DongleId') == '0123456789abcdef'
+
+
+def test_default_boot_never_discards_existing_registration(environment):
+  root, params = environment
+  params.put('DongleId', '0123456789abcdef')
+  p.activate_at_boot(params, root=root)
+  assert params.get('DongleId') == '0123456789abcdef'
+  assert params.get('ConnectProvider') == 'comma'
+  assert not root.exists()
+
+
+def test_corrupt_provider_goes_offline_without_selecting_other_account(environment):
+  root, _ = environment
+  root.mkdir()
+  (root / 'provider.json').write_text('{')
+  with patch.object(p, 'root_path', return_value=root):
+    assert p.active_provider() == p.OFFLINE
+
+
+def test_konik_identity_is_separate_private_and_stable(environment):
+  root, _ = environment
+  with patch.object(p, 'root_path', return_value=root):
+    first = p.konik_key_pair()
+    assert p.konik_key_pair() == first
+  assert first[0] == 'RS256' and 'RSA PRIVATE KEY' in first[1]
+  assert (root / 'konik/identity.json').stat().st_mode & 0o777 == 0o600
+
+
+def test_recordings_are_never_uploaded_under_another_provider(tmp_path):
+  old = tmp_path / 'old--0'
+  old.mkdir()
+  konik = tmp_path / 'new--0'
+  konik.mkdir()
+  (konik / p.ROUTE_MARKER).write_text('konik')
+  assert p.owns_recording(old / 'qlog.zst', tmp_path, 'comma')
+  assert not p.owns_recording(old / 'qlog.zst', tmp_path, 'konik')
+  assert p.owns_recording(konik / 'qlog.zst', tmp_path, 'konik')
+  assert not p.owns_recording(konik / 'qlog.zst', tmp_path, 'comma')
+  assert not p.owns_recording(tmp_path / '../elsewhere', tmp_path, 'comma')
+  (konik / p.ROUTE_MARKER).write_text('unknown')
+  assert not p.owns_recording(konik / 'qlog.zst', tmp_path, 'comma')
+  (konik / p.ROUTE_MARKER).unlink()
+  (konik / p.ROUTE_MARKER).symlink_to('/dev/null')
+  assert not p.owns_recording(konik / 'qlog.zst', tmp_path, 'comma')
+
+
+def test_actual_params_registration_switch_and_cache_isolation(tmp_path):
+  from openpilot.common.params import Params
+  params = Params(str(tmp_path / 'real-params'))
+  root = tmp_path / 'real-cloud'
+  params.put('DongleId', '0123456789abcdef', block=True)
+  params.put('ApiCache_Device', '{"owner":"original"}', block=True)
+  choose(root, 'konik')
+  p.activate_at_boot(params, root=root, boot=lambda: 'next')
+  assert params.get('ConnectProvider') == 'konik'
+  assert params.get('DongleId') is None and params.get('ApiCache_Device') is None
+  choose(root, 'comma', 'next')
+  p.activate_at_boot(params, root=root, boot=lambda: 'last')
+  assert params.get('DongleId') == '0123456789abcdef'
+  assert params.get('ApiCache_Device') == '{"owner":"original"}'
+
+
+def test_registration_uses_konik_key_and_query_contract(environment):
+  from openpilot.system.athena import registration
+  from types import SimpleNamespace
+  import jwt
+  root, params = environment
+  with patch.object(p, 'root_path', return_value=root):
+    algorithm, private, public = p.konik_key_pair()
+  response = SimpleNamespace(status_code=200, json=lambda: {'dongle_id': 'fedcba9876543210', 'access_token': ''})
+  with patch.object(registration, 'get_key_pair', return_value=(algorithm, private, public)), \
+       patch.object(registration, 'api_get', return_value=response) as request, \
+       patch.object(registration.HARDWARE, 'get_imei', return_value='test-imei'), \
+       patch.object(registration.HARDWARE, 'get_serial', return_value='test-serial'):
+    assert registration.register_konik(params) == 'fedcba9876543210'
+  args = request.call_args.kwargs
+  assert args['public_key'] == public and args['timeout'] == 10
+  assert args['serial'] == 'test-serial' and args['imei'] == 'test-imei'
+  assert jwt.decode(args['register_token'], public, algorithms=['RS256'])['register'] is True
+  assert params.get('DongleId') == 'fedcba9876543210'
+
+
+def test_api_never_sends_credentials_when_configuration_invalid(environment):
+  from openpilot.common import api
+  root, _ = environment
+  root.mkdir()
+  (root / 'provider.json').write_text('{')
+  with patch.object(p, 'root_path', return_value=root), patch.object(api.requests, 'request') as request:
+    assert api.get_key_pair() == (None, None, None)
+    with pytest.raises(RuntimeError):
+      api.api_get('v1/me', access_token='private')
+    request.assert_not_called()
+
+
+def test_cloudlogs_stay_with_their_provider(environment):
+  root, _ = environment
+  with patch.object(p, 'root_path', return_value=root), patch.object(p.Paths, 'swaglog_root', return_value='/tmp/cloudlogs'):
+    assert p.cloudlog_root() == '/tmp/cloudlogs'
+    p._write(root / 'provider.json', {'active': 'konik', 'selected': 'konik', 'requestedBoot': ''})
+    assert p.cloudlog_root() == '/tmp/cloudlogs/konik'
+
+
+def test_bootlogs_are_owned_per_file_without_retagging_legacy(tmp_path):
+  boot = tmp_path / 'boot'
+  boot.mkdir()
+  (boot / 'old.zst').touch()
+  (boot / 'new.zst').touch()
+  (boot / '.cloud-new.zst').write_text('konik')
+  assert p.owns_recording(boot / 'old.zst', tmp_path, 'comma')
+  assert not p.owns_recording(boot / 'old.zst', tmp_path, 'konik')
+  assert p.owns_recording(boot / 'new.zst', tmp_path, 'konik')
+  assert not p.owns_recording(boot / 'new.zst', tmp_path, 'comma')
+  (boot / '.cloud-new.zst').write_text('')
+  assert not p.owns_recording(boot / 'new.zst', tmp_path, 'comma')
+  assert not p.owns_recording(boot / 'new.zst', tmp_path, 'konik')
+
+
+def test_galaxy_identity_stays_physical_when_konik_config_is_corrupt(environment):
+  root, params = environment
+  params.put('DongleId', '0123456789abcdef')
+  choose(root, 'konik')
+  p.activate_at_boot(params, root=root, boot=lambda: 'second')
+  params.put('DongleId', 'fedcba9876543210')
+  with patch.object(p, 'root_path', return_value=root):
+    assert p.galaxy_device_id(params) == '0123456789abcdef'
+    (root / 'provider.json').write_text('{')
+    assert p.galaxy_device_id(params) == '0123456789abcdef'
diff --git a/openpilot/system/athena/athenad.py b/openpilot/system/athena/athenad.py
index 2b67638776..9ddc67b953 100755
--- a/openpilot/system/athena/athenad.py
+++ b/openpilot/system/athena/athenad.py
@@ -37,6 +37,7 @@ from openpilot.common.api import Api, get_key_pair
 from openpilot.common.basedir import BASEDIR
 from openpilot.common.utils import CallbackReader, get_upload_stream
 from openpilot.common.params import Params
+from openpilot.starpilot.schema_cache import get_cache
 from openpilot.common.realtime import set_core_affinity
 from openpilot.common.hardware import HARDWARE, PC
 from openpilot.system.loggerd.config import CAMERA_FPS, SEGMENT_LENGTH
@@ -48,11 +49,14 @@ from openpilot.common.hardware.hw import Paths
 from openpilot.system.athena.rpc import dispatcher, dumps_call, handle, is_call, is_response, loads
 
 
-ATHENA_HOST = os.getenv('ATHENA_HOST', 'wss://athena.comma.ai')
+from openpilot.starpilot.connect.provider import active_provider, owns_recording, cloudlog_root, mark_recording
+
+CLOUD_PROVIDER = active_provider()
+ATHENA_HOST = os.getenv('ATHENA_HOST', CLOUD_PROVIDER.athena) if CLOUD_PROVIDER.name == 'comma' else CLOUD_PROVIDER.athena
 HANDLER_THREADS = int(os.getenv('HANDLER_THREADS', "4"))
 LOCAL_PORT_WHITELIST = {22, }  # SSH
 
-LOG_ATTR_NAME = 'user.upload'
+LOG_ATTR_NAME = CLOUD_PROVIDER.upload_attribute
 LOG_ATTR_VALUE_MAX_UNIX_TIME = int.to_bytes(2147483647, 4, sys.byteorder)
 RECONNECT_TIMEOUT_S = 70
 
@@ -261,7 +265,7 @@ def cb(sm, item, tid, end_event: threading.Event, sz: int, cur: int) -> None:
   if not item.allow_cellular:
     if (time.monotonic() - sm.recv_time['deviceState']) > DEVICE_STATE_UPDATE_INTERVAL:
       sm.update(0)
-      if sm['deviceState'].networkMetered:
+      if sm['deviceState'].networkMetered and not Params().get_bool("AlwaysAllowUploads"):
         raise AbortTransferException
 
   if end_event.is_set():
@@ -294,7 +298,7 @@ def upload_handler(end_event: threading.Event) -> None:
       sm.update(0)
       metered = sm['deviceState'].networkMetered
       network_type = sm['deviceState'].networkType.raw
-      if metered and (not item.allow_cellular):
+      if metered and not item.allow_cellular and not Params().get_bool("AlwaysAllowUploads"):
         retry_upload(tid, end_event, False)
         continue
 
@@ -330,6 +334,8 @@ def upload_handler(end_event: threading.Event) -> None:
 
 def _do_upload(upload_item: UploadItem, callback: Callable | None = None) -> requests.Response:
   path = upload_item.path
+  if not owns_recording(path, Paths.log_root()):
+    raise ValueError('Recording belongs to another cloud provider')
   compress = False
 
   # If file does not exist, but does exist without the .zst extension we will compress on the fly
@@ -387,6 +393,8 @@ def scan_dir(path: str, prefix: str) -> list[str]:
   # (glob and friends traverse entire dir tree)
   with os.scandir(path) as i:
     for e in i:
+      if e.name.startswith(".") or (e.name != "boot" and not owns_recording(e.path, Paths.log_root())):
+        continue
       rel_path = os.path.relpath(e.path, Paths.log_root())
       if e.is_dir(follow_symlinks=False):
         # add trailing slash
@@ -418,7 +426,8 @@ class VideoClips:
     requested_at: float
 
   def __init__(self):
-    self.clip_path = os.path.join(Paths.log_root(), "clips")
+    self.clip_provider = CLOUD_PROVIDER.name
+    self.clip_path = os.path.join(Paths.log_root(), "clips" if self.clip_provider == "comma" else "clips-" + self.clip_provider)
     self.lock = threading.Condition()
     self.clips: dict[str, VideoClips.Clip] = {}
     self.transcode_proc: tuple[str, subprocess.Popen] | None = None
@@ -482,11 +491,10 @@ class VideoClips:
           if self.clips.get(clip.filename) is not clip:
             continue
         first_segment = math.floor(clip.source_start_time / SEGMENT_LENGTH)
-        inputs = (
-          os.path.join(Paths.log_root(), f"{clip.route}--{segment}", clip.camera)
-          for segment in range(first_segment, math.ceil(clip.source_end_time / SEGMENT_LENGTH))
-        )
+        inputs = self._source_inputs(clip.route, clip.camera, clip.source_start_time, clip.source_end_time)
         os.makedirs(self.clip_path, exist_ok=True)
+        if self.clip_provider != "comma":
+          mark_recording(self.clip_path, self.clip_provider)
         temporary_path = os.path.join(self.clip_path, f".{clip.filename}")
         output_path = os.path.join(self.clip_path, clip.filename)
         self._encode(clip, inputs, temporary_path, clip.source_start_time - first_segment * SEGMENT_LENGTH,
@@ -538,7 +546,8 @@ class VideoClips:
       with os.scandir(Paths.log_root()) as entries:
         for entry in entries:
           entry_route, _, segment = entry.name.rpartition("--")
-          if entry_route != route or not segment.isdigit() or not entry.is_dir():
+          if (entry_route != route or not segment.isdigit() or not entry.is_dir() or
+              not owns_recording(entry.path, Paths.log_root(), self.clip_provider)):
             continue
           with os.scandir(entry.path) as files:
             for camera in files:
@@ -558,6 +567,16 @@ class VideoClips:
       available[camera] = {"available_ranges": ranges}
     return available
 
+  def _source_inputs(self, route, camera, start, end):
+    if (self.clip_provider not in ("comma", "konik") or not math.isfinite(start) or not math.isfinite(end) or
+        start < 0 or end <= start):
+      raise ValueError("Invalid clip source range")
+    inputs = [os.path.join(Paths.log_root(), f"{route}--{segment}", camera)
+              for segment in range(math.floor(start / SEGMENT_LENGTH), math.ceil(end / SEGMENT_LENGTH))]
+    if not all(os.path.isfile(path) and owns_recording(path, Paths.log_root(), self.clip_provider) for path in inputs):
+      raise ValueError("Clip source belongs to another cloud provider or is unavailable")
+    return inputs
+
   def createClip(self, route: str, source_start_time: float, source_end_time: float, clip: dict):
     if not PC and not Params().get_bool("IsOffroad"):
       raise RuntimeError("video clips can only be created while offroad")
@@ -568,6 +587,7 @@ class VideoClips:
     filename = clip["filename"]
     assert camera == os.path.basename(camera) and camera.endswith("camera.hevc"), "invalid camera filename"
     assert filename == os.path.basename(filename), "invalid filename"
+    self._source_inputs(route_name, camera, source_start_time, source_end_time)
     with self.lock:
       self.clips[filename] = self.Clip(route_name, camera, source_start_time, source_end_time, clip["bitrate"], clip["speedup"],
                                         filename, datetime.now().timestamp())
@@ -601,6 +621,8 @@ class VideoClips:
     assert filename == os.path.basename(filename) and not filename.startswith("."), "invalid filename"
     assert isinstance(offset, int) and offset >= 0, "invalid offset"
     path = os.path.join(self.clip_path, filename)
+    if not owns_recording(path, Paths.log_root(), self.clip_provider):
+      raise ValueError("Clip belongs to another cloud provider")
     size = os.path.getsize(path)
     assert offset <= size, "offset past end of file"
     with open(path, "rb") as f:
@@ -639,7 +661,7 @@ def uploadFilesToUrls(files_data: list[UploadFileDict]) -> UploadFilesToUrlRespo
       continue
 
     path = os.path.join(Paths.log_root(), file.fn)
-    if not os.path.exists(path) and not os.path.exists(strip_zst_extension(path)):
+    if not owns_recording(path, Paths.log_root()) or (not os.path.exists(path) and not os.path.exists(strip_zst_extension(path))):
       failed.append(file.fn)
       continue
 
@@ -765,7 +787,7 @@ def getGithubUsername() -> str:
 
 @dispatcher.add_method
 def getNotCar() -> bool:
-  cp_bytes = Params().get("CarParamsPersistent")
+  cp_bytes = get_cache(Params(), "CarParamsPersistent")
   if cp_bytes is not None:
     with car.CarParams.from_bytes(cp_bytes) as CP:
       return CP.notCar
@@ -795,7 +817,7 @@ def startStream(sdp: str, enabled: bool) -> dict:
   bridge_services_in = []
 
   # stale car params case taken care of by webrtcd being shut off on ignition
-  cp_bytes = params.get("CarParamsPersistent")
+  cp_bytes = get_cache(params, "CarParamsPersistent")
   if cp_bytes is not None:
     with car.CarParams.from_bytes(cp_bytes) as CP:
       if CP.notCar:
@@ -819,8 +841,10 @@ def get_logs_to_send_sorted() -> list[str]:
   # TODO: scan once then use inotify to detect file creation/deletion
   curr_time = int(time.time())  # noqa: TID251
   logs = []
-  for log_entry in os.listdir(Paths.swaglog_root()):
-    log_path = os.path.join(Paths.swaglog_root(), log_entry)
+  for log_entry in os.listdir(cloudlog_root()):
+    if not log_entry.startswith("swaglog."):
+      continue
+    log_path = os.path.join(cloudlog_root(), log_entry)
     time_sent = 0
     try:
       value = getxattr(log_path, LOG_ATTR_NAME)
@@ -855,7 +879,7 @@ def log_handler(end_event: threading.Event) -> None:
         cloudlog.debug(f"athena.log_handler.forward_request {log_entry}")
         try:
           curr_time = int(time.time())  # noqa: TID251
-          log_path = os.path.join(Paths.swaglog_root(), log_entry)
+          log_path = os.path.join(cloudlog_root(), log_entry)
           setxattr(log_path, LOG_ATTR_NAME, int.to_bytes(curr_time, 4, sys.byteorder))
           with open(log_path) as f:
             send_queue_push(dumps_call("forwardLogs", {"logs": f.read()}, request_id=log_entry), SEND_PRIORITY_LOW)
@@ -874,7 +898,7 @@ def log_handler(end_event: threading.Event) -> None:
           log_success = "result" in log_resp and log_resp["result"].get("success")
           cloudlog.debug(f"athena.log_handler.forward_response {log_entry} {log_success}")
           if log_entry and log_success:
-            log_path = os.path.join(Paths.swaglog_root(), log_entry)
+            log_path = os.path.join(cloudlog_root(), log_entry)
             try:
               setxattr(log_path, LOG_ATTR_NAME, LOG_ATTR_VALUE_MAX_UNIX_TIME)
             except OSError:
diff --git a/openpilot/system/athena/manage_athenad.py b/openpilot/system/athena/manage_athenad.py
index f0cd124f37..03d80764b1 100755
--- a/openpilot/system/athena/manage_athenad.py
+++ b/openpilot/system/athena/manage_athenad.py
@@ -27,6 +27,12 @@ def main():
 
   try:
     while 1:
+      from openpilot.starpilot.connect.provider import active_provider
+      if active_provider().name == 'konik':
+        from openpilot.system.athena.registration import register, UNREGISTERED_DONGLE_ID
+        if register() == UNREGISTERED_DONGLE_ID:
+          time.sleep(15)
+          continue
       cloudlog.info("starting athena daemon")
       proc = Process(name='athenad', target=launcher, args=('openpilot.system.athena.athenad', 'athenad'))
       proc.start()
diff --git a/openpilot/system/athena/registration.py b/openpilot/system/athena/registration.py
index 35051c0659..918ad272d3 100755
--- a/openpilot/system/athena/registration.py
+++ b/openpilot/system/athena/registration.py
@@ -7,6 +7,9 @@ from pathlib import Path
 
 from datetime import datetime, timedelta, UTC
 from openpilot.common.api import api_get, get_key_pair
+from openpilot.starpilot.connect.provider import active_provider
+import os
+import re
 from openpilot.common.params import Params
 from openpilot.common.spinner import Spinner
 from openpilot.selfdrive.selfdrived.alertmanager import set_offroad_alert
@@ -33,6 +36,10 @@ def register(show_spinner=False) -> str | None:
   entirely.
   """
   params = Params()
+  if os.getenv('STARPILOT_CLOUD_DISABLED') == '1':
+    return UNREGISTERED_DONGLE_ID
+  if active_provider().name == 'konik':
+    return register_konik(params)
 
   dongle_id: str | None = params.get("DongleId")
   if dongle_id is None and Path(Paths.persist_root()+"/comma/dongle_id").is_file():
@@ -102,5 +109,27 @@ def register(show_spinner=False) -> str | None:
   return dongle_id
 
 
+
+def register_konik(params):
+  known = params.get('DongleId')
+  if known and re.fullmatch(r'[a-f0-9]{16}', known):
+    return known
+  try:
+    algorithm, private, public = get_key_pair()
+    if private is None or public is None:
+      return UNREGISTERED_DONGLE_ID
+    token = jwt.encode({'register': True, 'exp': datetime.now(UTC) + timedelta(hours=1)}, private, algorithm=algorithm)
+    response = api_get('v2/pilotauth/', method='POST', timeout=10, imei=HARDWARE.get_imei() or '', imei2='',
+                       serial=HARDWARE.get_serial(), public_key=public, register_token=token)
+    identity = response.json().get('dongle_id') if response.status_code == 200 else None
+    if not isinstance(identity, str) or not re.fullmatch(r'[a-f0-9]{16}', identity):
+      return UNREGISTERED_DONGLE_ID
+    params.put('DongleId', identity, block=True)
+    return identity
+  except Exception:
+    cloudlog.warning('Konik registration unavailable; cloud services will retry')
+    return UNREGISTERED_DONGLE_ID
+
+
 if __name__ == "__main__":
   print(register())
diff --git a/openpilot/system/loggerd/bootlog.cc b/openpilot/system/loggerd/bootlog.cc
index b2e3abfcf1..f692c5bce0 100644
--- a/openpilot/system/loggerd/bootlog.cc
+++ b/openpilot/system/loggerd/bootlog.cc
@@ -55,6 +55,12 @@ int main(int argc, char** argv) {
   bool r = util::create_directories(Path::log_root() + "/boot/", 0775);
   assert(r);
 
+  const std::string provider = Params().get("ConnectProvider");
+  const std::string cloud = provider.empty() ? "comma" : provider;
+  const std::string marker = Path::log_root() + "/boot/.cloud-" + id + ".zst";
+  const bool tagged = logger_write_cloud_marker(marker, cloud);
+  assert(tagged);
+
   ZstdFileWriter file(path, LOG_COMPRESSION_LEVEL);
   // Write initdata
   file.write(logger_build_init_data().asBytes());
diff --git a/openpilot/system/loggerd/logger.cc b/openpilot/system/loggerd/logger.cc
index f9371314d3..639b6b9e66 100644
--- a/openpilot/system/loggerd/logger.cc
+++ b/openpilot/system/loggerd/logger.cc
@@ -183,6 +183,12 @@ bool LoggerState::next() {
   bool ret = util::create_directories(segment_path, 0775);
   assert(ret == true);
 
+  const std::string provider = Params().get("ConnectProvider");
+  const std::string cloud = provider.empty() ? "comma" : provider;
+  const std::string cloud_file = segment_path + "/.cloud-provider";
+  const bool tagged = logger_write_cloud_marker(cloud_file, cloud);
+  assert(tagged);
+
   lock_file = segment_path + "/rlog.lock";
   std::ofstream{lock_file};
 
diff --git a/openpilot/system/loggerd/logger.h b/openpilot/system/loggerd/logger.h
index 17c29d1a02..3d3e598f93 100644
--- a/openpilot/system/loggerd/logger.h
+++ b/openpilot/system/loggerd/logger.h
@@ -1,6 +1,9 @@
 #pragma once
 
 #include <cassert>
+#include <cerrno>
+#include <fcntl.h>
+#include <unistd.h>
 #include <memory>
 #include <string>
 
@@ -9,6 +12,36 @@
 #include "common/hardware/hw.h"
 #include "system/loggerd/zstd_writer.h"
 
+// Persist ownership before any recording bytes can become visible.
+inline bool logger_write_cloud_marker(const std::string &path, const std::string &provider) {
+  if (provider != "comma" && provider != "konik" && provider != "offline") return false;
+  const size_t slash = path.find_last_of('/');
+  if (slash == std::string::npos) return false;
+  const int directory = open(path.substr(0, slash).c_str(), O_RDONLY | O_DIRECTORY | O_CLOEXEC);
+  if (directory < 0) return false;
+  const int marker = open(path.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC, 0644);
+  if (marker < 0) {
+    close(directory);
+    return false;
+  }
+  size_t offset = 0;
+  bool ok = true;
+  while (offset < provider.size()) {
+    const ssize_t written = write(marker, provider.data() + offset, provider.size() - offset);
+    if (written < 0 && errno == EINTR) continue;
+    if (written <= 0) {
+      ok = false;
+      break;
+    }
+    offset += static_cast<size_t>(written);
+  }
+  if (ok) ok = fsync(marker) == 0;
+  if (close(marker) != 0) ok = false;
+  if (ok) ok = fsync(directory) == 0;
+  if (close(directory) != 0) ok = false;
+  return ok;
+}
+
 constexpr int LOG_COMPRESSION_LEVEL = 10;
 
 typedef cereal::Sentinel::SentinelType SentinelType;
diff --git a/openpilot/system/loggerd/tests/test_uploader_preferences.py b/openpilot/system/loggerd/tests/test_uploader_preferences.py
new file mode 100644
index 0000000000..fb696ccc80
--- /dev/null
+++ b/openpilot/system/loggerd/tests/test_uploader_preferences.py
@@ -0,0 +1,82 @@
+"""Run the uploader's actual main loop with inert transport and live preferences."""
+import ast
+from enum import IntEnum
+from pathlib import Path
+from types import SimpleNamespace as NS
+import unittest
+
+
+class Network(IntEnum):
+  none = 0
+  wifi = 1
+  cell = 2
+
+  @property
+  def raw(self):
+    return int(self)
+
+
+class TestUploadPreferences(unittest.TestCase):
+  def run_loop(self, observations, *, force_wifi=False):
+    main = next(node for node in ast.parse(Path(__file__).parents[1].joinpath('uploader.py').read_text()).body
+                if isinstance(node, ast.FunctionDef) and node.name == 'main')
+    state = NS(index=-1, calls=[], reads=[], cleared=False)
+
+    class Saved:
+      def get(self, key):
+        return 'test-local-id' if key == 'DongleId' else None
+
+      def get_bool(self, key):
+        state.reads.append(key)
+        current = observations[max(0, state.index)]
+        return current.get('allow', False) if key == 'AlwaysAllowUploads' else current.get('offroad', False)
+
+    class Master:
+      def update(self, _):
+        state.index += 1
+
+      def __getitem__(self, _):
+        row = observations[state.index]
+        return NS(networkType=row.get('network', Network.cell), networkMetered=row.get('metered', True))
+
+    class InertUploader:
+      def __init__(self, *_):
+        pass
+
+      def step(self, network, metered):
+        state.calls.append((network, metered))
+        return None
+
+    namespace: dict = {'threading': NS(Event=object), 'Params': Saved, 'Uploader': InertUploader,
+                 'Paths': NS(log_root=lambda: '/unused'), 'clear_locks': lambda _: None,
+                 'set_core_affinity': lambda _: None, 'messaging': NS(SubMaster=lambda _: Master()),
+                 'NetworkType': Network, 'force_wifi': force_wifi, 'allow_sleep': False,
+                 'cloudlog': NS(info=lambda *_: None, exception=lambda *_: None),
+                 'active_provider': lambda: NS(name='comma')}
+    exec(compile(ast.Module(body=[main], type_ignores=[]), 'uploader.py', 'exec'), namespace)
+    namespace['main'](NS(is_set=lambda: state.index + 1 >= len(observations)))
+    return state
+
+  def test_metered_default_off_and_live_on_off_refresh(self):
+    state = self.run_loop([{}, {'allow': True}, {'allow': False}, {'allow': True, 'metered': False},
+                           {'allow': False, 'metered': False}])
+    self.assertEqual(state.calls, [(2, True), (2, False), (2, True), (2, False), (2, False)])
+    self.assertEqual(state.reads.count('AlwaysAllowUploads'), 5)
+
+  def test_no_network_and_offroad_policy_are_preserved(self):
+    state = self.run_loop([{'allow': True, 'network': Network.none}, {'offroad': True},
+                           {'offroad': True, 'allow': True}])
+    self.assertEqual(state.calls, [(2, True), (2, False)])
+    self.assertEqual(state.reads.count('IsOffroad'), 3)
+
+  def test_force_wifi_preserves_actual_network_metadata_and_metering(self):
+    state = self.run_loop([{'network': Network.none}, {'network': Network.cell, 'allow': True}], force_wifi=True)
+    self.assertEqual(state.calls, [(0, True), (2, False)])
+
+  def test_typed_persistent_default_is_off(self):
+    keys = Path(__file__).parents[3].joinpath('common/params_keys.h').read_text()
+    self.assertIn('{"AlwaysAllowUploads", {PERSISTENT, BOOL, "0"}}', keys)
+
+
+if __name__ == '__main__':
+  unittest.main()
diff --git a/openpilot/system/loggerd/uploader.py b/openpilot/system/loggerd/uploader.py
index 81f8ed1ba3..173430c14b 100755
--- a/openpilot/system/loggerd/uploader.py
+++ b/openpilot/system/loggerd/uploader.py
@@ -20,7 +20,9 @@ from openpilot.system.loggerd.xattr_cache import getxattr, setxattr
 from openpilot.common.swaglog import cloudlog
 
 NetworkType = log.DeviceState.NetworkType
-UPLOAD_ATTR_NAME = 'user.upload'
+from openpilot.starpilot.connect.provider import active_provider, owns_recording
+
+UPLOAD_ATTR_NAME = active_provider().upload_attribute
 UPLOAD_ATTR_VALUE = b'1'
 
 MAX_UPLOAD_SIZES = {
@@ -91,6 +93,8 @@ class Uploader:
 
     for logdir in listdir_by_creation(self.root):
       path = os.path.join(self.root, logdir)
+      if logdir != "boot" and not owns_recording(path, self.root):
+        continue
       try:
         names = os.listdir(path)
       except OSError:
@@ -100,8 +104,12 @@ class Uploader:
         continue
 
       for name in sorted(names, key=lambda n: self.immediate_priority.get(n, 1000)):
+        if name.startswith("."):
+          continue
         key = os.path.join(logdir, name)
         fn = os.path.join(path, name)
+        if not owns_recording(fn, self.root):
+          continue
         # skip files already uploaded
         try:
           ctime = os.path.getctime(fn)
@@ -238,6 +246,11 @@ def main(exit_event: threading.Event | None = None) -> None:
   params = Params()
   dongle_id = params.get("DongleId")
 
+  while active_provider().name == 'konik' and dongle_id in (None, 'UnregisteredDevice') and not exit_event.is_set():
+    exit_event.wait(10)
+    dongle_id = params.get('DongleId')
+  if exit_event.is_set():
+    return
   if dongle_id is None:
     cloudlog.info("uploader missing dongle_id")
     raise Exception("uploader can't start without dongle id")
@@ -248,6 +261,7 @@ def main(exit_event: threading.Event | None = None) -> None:
   backoff = 0.1
   while not exit_event.is_set():
     sm.update(0)
+    always_allow_uploads = params.get_bool("AlwaysAllowUploads")
     offroad = params.get_bool("IsOffroad")
     network_type = sm['deviceState'].networkType if not force_wifi else NetworkType.wifi
     if network_type == NetworkType.none:
@@ -255,7 +269,7 @@ def main(exit_event: threading.Event | None = None) -> None:
         time.sleep(60 if offroad else 5)
       continue
 
-    success = uploader.step(sm['deviceState'].networkType.raw, sm['deviceState'].networkMetered)
+    success = uploader.step(sm['deviceState'].networkType.raw, sm['deviceState'].networkMetered and not always_allow_uploads)
     if success is None:
       backoff = 60 if offroad else 5
     elif success: