diff --git a/starpilot/system/the_galaxy/tests/test_version_history.py b/starpilot/system/the_galaxy/tests/test_version_history.py
index 4eb55225da..5e50368bfd 100644
--- a/starpilot/system/the_galaxy/tests/test_version_history.py
+++ b/starpilot/system/the_galaxy/tests/test_version_history.py
@@ -313,17 +313,22 @@ def test_invalid_version_metadata_is_retryable(history, repo, api, monkeypatch,
   URLError('offline'), IncompleteRead(b'partial', 100),
 ])
 def test_version_network_failure_is_not_cached_as_missing(history, repo, api, monkeypatch, failure):
+  clock = [1000.0]
+  monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
+  monkeypatch.setattr(history.time, 'time', lambda: clock[0])
   api.heads['StarPilot'] = 1
   def fail(*args, **kwargs):
     raise failure
   monkeypatch.setattr(history, '_open_raw_url', fail, raising=False)
   with pytest.raises(history.VersionHistoryError, match='retry'):
     history.list_versions(repo, 'StarPilot')
+  clock[0] += 61
   monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
   assert history.list_versions(repo, 'StarPilot')['commits'][0]['version'] == '6.7.7'
 
 
 def test_version_cache_outlives_branch_cache_and_expires(history, repo, api, monkeypatch):
+  monkeypatch.setattr(history, '_saved_display_versions', lambda *a: {})  # Exercise memory-cache expiry alone.
   api.heads['StarPilot'] = 1
   clock = [100.0]
   monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
@@ -488,7 +493,7 @@ def test_rate_limit_reports_retry_time(history, monkeypatch, headers, expected):
 
 
 @pytest.mark.parametrize('code', [403, 429])
-def test_rate_limit_short_backoff_is_shared_and_then_retries(history, monkeypatch, code):
+def test_rate_limit_full_backoff_is_shared_and_then_retries(history, monkeypatch, code):
   clock = [1000.0]
   monkeypatch.setattr(history.time, 'time', lambda: clock[0])
   monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
@@ -504,6 +509,10 @@ def test_rate_limit_short_backoff_is_shared_and_then_retries(history, monkeypatc
       history._get_json('https://api.github.com/repos/a/b/' + path)
   assert len(calls) == 1
   clock[0] += 61
+  with pytest.raises(history.HistoryUnavailable):
+    history._get_json('https://api.github.com/repos/a/b/branches/main')
+  assert len(calls) == 1
+  clock[0] = 3280
   assert history._get_json('https://api.github.com/repos/a/b/branches/main') == {'ok': True}
   assert len(calls) == 2
 
@@ -713,6 +722,7 @@ def test_starpilot_snapshot_fallback_only_for_version_transport_failure(history,
   monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
   first = history.list_versions(repo, 'StarPilot')
   offline = offline_reload(history, monkeypatch)
+  monkeypatch.setattr(offline, '_saved_display_versions', lambda *a: {})  # Force transport to test failure policy.
   monkeypatch.setattr(offline, '_get_json', api)
   def raw(*args, **kwargs):
     if failure_kind == 'offline':
@@ -773,6 +783,7 @@ def test_http_failure_snapshot_fallback_boundary(history, repo, api, monkeypatch
   monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
   first = history.list_versions(repo, branch)
   offline = offline_reload(history, monkeypatch)
+  monkeypatch.setattr(offline, '_saved_display_versions', lambda *a: {})  # Force transport to test failure policy.
   def fail(*args, **kwargs):
     raise HTTPError('https://github.com/', code, 'Unavailable', headers, io.BytesIO())
   if transport == 'api':
@@ -792,3 +803,58 @@ def test_http_failure_snapshot_fallback_boundary(history, repo, api, monkeypatch
 def test_empty_snapshot_preload_timestamp_is_rejected(history, repo, api):
   first = history.list_versions(repo, 'main')
   assert not history._save_history_snapshot('https://api.github.com/repos/firestar5683/openpilot', 'main', 1, None, first, saved_at='')
+
+def test_saved_release_labels_survive_restart_without_repeating_raw_downloads(history, repo, api, monkeypatch):
+  api.heads['StarPilot'] = 3
+  monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
+  first = history.list_versions(repo, 'StarPilot')
+  restarted = offline_reload(history, monkeypatch)
+  monkeypatch.setattr(restarted, '_get_json', api)
+  def forbidden(*args, **kwargs):
+    pytest.fail('A saved immutable release label was downloaded again')
+  monkeypatch.setattr(restarted, '_open_raw_url', forbidden)
+  assert restarted.list_versions(repo, 'StarPilot')['commits'] == first['commits']
+  # A changed head cannot reuse the previous first-page snapshot blindly.
+  api.heads['StarPilot'] = 4
+  restarted._cache.clear()
+  calls = []
+  def raw(*args, **kwargs):
+    calls.append(1)
+    return io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.8"\n')
+  monkeypatch.setattr(restarted, '_open_raw_url', raw)
+  result = restarted.list_versions(repo, 'StarPilot')
+  assert result['head'] == f'{4:040x}' and result['commits'][0]['version'] == '6.7.8'
+  assert calls
+
+
+def test_saved_labels_never_bypass_live_history_validation(history, repo, api, monkeypatch):
+  api.heads['StarPilot'] = 3
+  monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
+  first = history.list_versions(repo, 'StarPilot')
+  restarted = offline_reload(history, monkeypatch)
+  def invalid(url):
+    if '/commits?' in url: return {'invalid': True}
+    return api(url)
+  monkeypatch.setattr(restarted, '_get_json', invalid)
+  with pytest.raises(restarted.VersionHistoryError, match='invalid commit history'):
+    restarted.list_versions(repo, 'StarPilot')
+
+def test_nine_page_release_browse_reuses_saved_labels_after_restart(history, repo, api, monkeypatch):
+  api.heads['StarPilot'] = 240
+  calls = []
+  def raw(*args, **kwargs):
+    calls.append(1)
+    return io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n')
+  monkeypatch.setattr(history, '_open_raw_url', raw)
+  head = f'{240:040x}'
+  for page in range(1, 10):
+    history.list_versions(repo, 'StarPilot', page=page, head=head if page > 1 else None)
+  assert len(calls) == 225
+  assert len(api.requests) == 4  # One branch head and three 100-commit blocks.
+  restarted = offline_reload(history, monkeypatch)
+  monkeypatch.setattr(restarted, '_get_json', api)
+  monkeypatch.setattr(restarted, '_open_raw_url', raw)
+  for page in range(1, 10):
+    restarted.list_versions(repo, 'StarPilot', page=page, head=head if page > 1 else None)
+  assert len(calls) == 225  # No repeat raw-file requests for those 225 saved labels.
+  assert len(api.requests) == 8  # The live branch and history are still validated.
diff --git a/starpilot/system/the_galaxy/tests/test_version_install_hidden.py b/starpilot/system/the_galaxy/tests/test_version_install_hidden.py
new file mode 100644
index 0000000000..e7e62c0e7d
--- /dev/null
+++ b/starpilot/system/the_galaxy/tests/test_version_install_hidden.py
@@ -0,0 +1,109 @@
+"""Refuse destructive checkout when ordinary patches cannot preserve local files."""
+import pytest
+
+from test_version_install_index import checkout, git, install, installer, repository_snapshot
+
+
+@pytest.mark.parametrize('flag', ['assume-unchanged', 'skip-worktree'])
+def test_hidden_index_flags_refuse_install_without_mutation(checkout, flag):
+  repo, data, old, latest = checkout
+  git(repo, 'update-index', '--' + flag, 'tracked.txt')
+  (repo / 'tracked.txt').write_bytes(b'valuable hidden working edit\n')
+  before = repository_snapshot(repo, data)
+  flags = git(repo, 'ls-files', '-v', '-z')
+  with pytest.raises(installer.InstallError, match='assume-unchanged|skip-worktree'):
+    install(checkout)
+  assert repository_snapshot(repo, data) == before
+  assert git(repo, 'ls-files', '-v', '-z') == flags
+  assert not (data / 'starpilot/version-backups').exists()
+
+
+@pytest.mark.parametrize('flag', ['assume-unchanged', 'skip-worktree'])
+def test_hidden_index_flags_refuse_restore_without_mutation(checkout, flag):
+  repo, data, old, latest = checkout
+  backup = install(checkout)
+  git(repo, 'update-index', '--' + flag, 'tracked.txt')
+  (repo / 'tracked.txt').write_bytes(b'valuable hidden post-install edit\n')
+  before = repository_snapshot(repo, data)
+  flags = git(repo, 'ls-files', '-v', '-z')
+  with pytest.raises(installer.InstallError, match='assume-unchanged|skip-worktree'):
+    installer.restore(backup, check_parked=lambda: installer.require_parked(data))
+  assert repository_snapshot(repo, data) == before
+  assert git(repo, 'ls-files', '-v', '-z') == flags
+
+
+def ignored_checkout(checkout, shape):
+  repo, data, old, latest = checkout
+  target_path = 'hidden-local/child.txt' if shape == 'ancestor' else 'hidden-local'
+  target = repo / target_path
+  target.parent.mkdir(parents=True, exist_ok=True)
+  target.write_bytes(b'target committed content\n')
+  git(repo, 'add', target_path)
+  git(repo, 'commit', '-m', 'Target containing collision path')
+  target_sha = git(repo, 'rev-parse', 'HEAD').decode().strip()
+  git(repo, 'rm', target_path)
+  git(repo, 'commit', '-m', 'Current without collision path')
+  current = git(repo, 'rev-parse', 'HEAD').decode().strip()
+  (repo / '.git/info/exclude').write_text('hidden-local\n')
+  local = repo / ('hidden-local/child.txt' if shape == 'descendant' else 'hidden-local')
+  local.parent.mkdir(parents=True, exist_ok=True)
+  outside = data / 'outside-sentinel'
+  outside.write_bytes(b'valuable local ignored content\n')
+  if shape == 'symlink':
+    local.symlink_to(outside)
+  else:
+    local.write_bytes(outside.read_bytes())
+  return (repo, data, target_sha, current), local, outside
+
+
+@pytest.mark.parametrize('shape', ['exact', 'ancestor', 'descendant', 'symlink'])
+def test_ignored_conflicts_refuse_install_before_backup(checkout, shape):
+  state, local, outside = ignored_checkout(checkout, shape)
+  repo, data, old, latest = state
+  before = repository_snapshot(repo, data)
+  with pytest.raises(installer.InstallError, match='[Ii]gnored'):
+    install(state)
+  assert repository_snapshot(repo, data) == before
+  assert local.read_bytes() == outside.read_bytes() == b'valuable local ignored content\n'
+  assert local.is_symlink() == (shape == 'symlink')
+  assert not (data / 'starpilot/version-backups').exists()
+
+
+@pytest.mark.parametrize('shape', ['exact', 'ancestor', 'descendant', 'symlink'])
+def test_ignored_conflicts_refuse_restore_without_mutation(checkout, shape):
+  state, local, outside = ignored_checkout(checkout, shape)
+  repo, data, target, current = state
+  # Save source that tracks the conflicting path, then mimic a later checkout
+  # with a newly created ignored file. Restore must inspect that live state.
+  if local.is_symlink() or local.is_file():
+    local.unlink()
+  else:
+    raise AssertionError('Expected a file or symlink')
+  if shape == 'descendant':
+    local.parent.rmdir()
+  git(repo, 'checkout', '--force', '-B', 'Dom', target)
+  backup = installer._backup(repo, {'branch': 'Dom', 'commit': current, 'pinned': False}, data)
+  git(repo, 'checkout', '--force', '-B', 'Dom', current)
+  local.parent.mkdir(parents=True, exist_ok=True)
+  if shape == 'symlink':
+    local.symlink_to(outside)
+  else:
+    local.write_bytes(outside.read_bytes())
+  before = repository_snapshot(repo, data)
+  with pytest.raises(installer.InstallError, match='[Ii]gnored'):
+    installer.restore(backup, check_parked=lambda: installer.require_parked(data))
+  assert repository_snapshot(repo, data) == before
+  assert local.read_bytes() == outside.read_bytes() == b'valuable local ignored content\n'
+  assert local.is_symlink() == (shape == 'symlink')
+
+
+def test_nonconflicting_ignored_files_allow_install_and_restore(checkout):
+  repo, data, old, latest = checkout
+  (repo / '.git/info/exclude').write_text('local-cache/\ntracked.txt.extra\n')
+  paths = [repo / 'local-cache/data.bin', repo / 'tracked.txt.extra']
+  for path in paths:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(b'kept ignored content\n')
+  backup = install(checkout)
+  installer.restore(backup, check_parked=lambda: installer.require_parked(data))
+  assert all(path.read_bytes() == b'kept ignored content\n' for path in paths)
diff --git a/starpilot/system/the_galaxy/tests/test_version_rate_limits.py b/starpilot/system/the_galaxy/tests/test_version_rate_limits.py
new file mode 100644
index 0000000000..8753b1139a
--- /dev/null
+++ b/starpilot/system/the_galaxy/tests/test_version_rate_limits.py
@@ -0,0 +1,191 @@
+"""Offline quota and request-coalescing regressions."""
+import importlib.util
+import io
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from urllib.error import HTTPError
+
+import pytest
+
+
+@pytest.fixture
+def history():
+  path = Path(__file__).resolve().parents[1] / 'version_history.py'
+  spec = importlib.util.spec_from_file_location('version_rate_limits_under_test', path)
+  module = importlib.util.module_from_spec(spec)
+  spec.loader.exec_module(module)
+  return module
+
+
+@pytest.mark.parametrize('transport', ['api', 'raw'])
+@pytest.mark.parametrize('headers,deadline', [({'Retry-After': '180'}, 1180),
+  ({'X-RateLimit-Remaining': '0', 'X-RateLimit-Reset': '4600'}, 4600), ({}, 1060)])
+def test_full_server_backoff_blocks_all_urls_until_deadline(history, monkeypatch, transport, headers, deadline):
+  clock = [1000.0]
+  monkeypatch.setattr(history.time, 'time', lambda: clock[0])
+  monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
+  calls = []
+  def limited(request, **kwargs):
+    calls.append(request.full_url)
+    if len(calls) == 1:
+      raise HTTPError(request.full_url, 429, 'Limited', headers, io.BytesIO())
+    return io.BytesIO(b'{}' if transport == 'api' else b'STARPILOT_DISPLAY_VERSION = "6.7.7"')
+  monkeypatch.setattr(history, '_open_url' if transport == 'api' else '_open_raw_url', limited)
+  fetch = history._get_json if transport == 'api' else history._get_display_version
+  with pytest.raises(history.HistoryUnavailable):
+    fetch('https://example.test/first')
+  clock[0] = deadline - 1
+  with pytest.raises(history.HistoryUnavailable):
+    fetch('https://example.test/second')
+  assert len(calls) == 1
+  clock[0] = deadline
+  fetch('https://example.test/second')
+  assert len(calls) == 2
+
+
+@pytest.mark.parametrize('transport', ['api', 'raw'])
+def test_concurrent_identical_cache_misses_download_once(history, monkeypatch, transport):
+  start = threading.Barrier(3)
+  entered, duplicate, release = threading.Event(), threading.Event(), threading.Event()
+  calls = []
+  def response(*args, **kwargs):
+    calls.append(1)
+    entered.set()
+    if len(calls) > 1:
+      duplicate.set()
+    assert release.wait(timeout=0.8)
+    return {} if transport == 'api' else '6.7.7'
+  monkeypatch.setattr(history, '_get_json' if transport == 'api' else '_get_display_version', response)
+  def caller():
+    start.wait(timeout=0.8)
+    if transport == 'api':
+      return history._json('https://api.github.com/repos/a/b/commits')
+    return history._display_version('https://api.github.com/repos/a/b', 'a' * 40)
+  with ThreadPoolExecutor(max_workers=2) as executor:
+    futures = [executor.submit(caller) for _ in range(2)]
+    start.wait(timeout=0.8)
+    try:
+      assert entered.wait(timeout=0.8)
+      duplicate.wait(timeout=0.1)
+    finally:
+      release.set()
+    assert futures[0].result(timeout=0.8) == futures[1].result(timeout=0.8)
+  assert len(calls) == 1
+
+
+def test_raw_quota_does_not_block_api_or_cached_raw(history, monkeypatch):
+  base, sha = 'https://api.github.com/repos/a/b', 'a' * 40
+  monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"'))
+  assert history._display_version(base, sha) == '6.7.7'
+  def limited(request, **kwargs):
+    raise HTTPError(request.full_url, 429, 'Limited', {'Retry-After': '180'}, io.BytesIO())
+  monkeypatch.setattr(history, '_open_raw_url', limited)
+  with pytest.raises(history.HistoryUnavailable):
+    history._display_version(base, 'b' * 40)
+  assert history._display_version(base, sha) == '6.7.7'
+  monkeypatch.setattr(history, '_open_url', lambda *a, **kw: io.BytesIO(b'{}'))
+  assert history._get_json(base) == {}
+
+
+def test_fresh_api_calls_do_not_reuse_cached_head(history, monkeypatch):
+  calls = []
+  def response(url):
+    calls.append(url)
+    return {'call': len(calls)}
+  monkeypatch.setattr(history, '_get_json', response)
+  assert history._json('head') == {'call': 1}
+  assert history._json('head', fresh=True) == {'call': 2}
+  assert history._json('head', fresh=True) == {'call': 3}
+
+
+def test_later_shorter_raw_response_cannot_shorten_active_backoff(history, monkeypatch):
+  clock = [1000.0]
+  monkeypatch.setattr(history.time, 'time', lambda: clock[0])
+  monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
+  started, long_finished = threading.Barrier(2), threading.Event()
+  calls = []
+  def limited(request, **kwargs):
+    calls.append(request.full_url)
+    started.wait(timeout=0.8)
+    if request.full_url.endswith('short'):
+      assert long_finished.wait(timeout=0.8)
+    wait = '180' if request.full_url.endswith('long') else '60'
+    raise HTTPError(request.full_url, 429, 'Limited', {'Retry-After': wait}, io.BytesIO())
+  monkeypatch.setattr(history, '_open_raw_url', limited)
+  def caller(suffix):
+    with pytest.raises(history.HistoryUnavailable):
+      history._get_display_version('https://raw.githubusercontent.com/' + suffix)
+    if suffix == 'long':
+      long_finished.set()
+  with ThreadPoolExecutor(max_workers=2) as executor:
+    futures = [executor.submit(caller, suffix) for suffix in ['long', 'short']]
+    for future in futures:
+      future.result(timeout=0.8)
+  clock[0] = 1179
+  with pytest.raises(history.HistoryUnavailable):
+    history._get_display_version('https://raw.githubusercontent.com/third')
+  assert len(calls) == 2
+
+
+def test_raw_worker_waiting_for_slot_observes_new_quota(history, monkeypatch):
+  entered, queued, release = threading.Event(), threading.Event(), threading.Event()
+  slots, gate_lock = threading.Semaphore(1), threading.Lock()
+  counts = {'attempts': 0, 'http': 0}
+  class Gate:
+    def __enter__(self):
+      with gate_lock:
+        counts['attempts'] += 1
+        if counts['attempts'] == 2:
+          queued.set()
+      assert slots.acquire(timeout=0.8)
+    def __exit__(self, *args):
+      slots.release()
+  monkeypatch.setattr(history, '_version_slots', Gate())
+  def limited(request, **kwargs):
+    counts['http'] += 1
+    entered.set()
+    assert release.wait(timeout=0.8)
+    raise HTTPError(request.full_url, 429, 'Limited', {'Retry-After': '180'}, io.BytesIO())
+  monkeypatch.setattr(history, '_open_raw_url', limited)
+  with ThreadPoolExecutor(max_workers=2) as executor:
+    first = executor.submit(history._display_version, 'https://api.github.com/repos/a/b', 'a' * 40)
+    assert entered.wait(timeout=0.8)
+    second = executor.submit(history._display_version, 'https://api.github.com/repos/a/b', 'b' * 40)
+    try:
+      assert queued.wait(timeout=0.8)
+    finally:
+      release.set()
+    for future in (first, second):
+      with pytest.raises(history.HistoryUnavailable):
+        future.result(timeout=0.8)
+  assert counts['http'] == 1
+
+
+def test_distinct_api_requests_serialize_and_observe_first_quota(history, monkeypatch):
+  start = threading.Barrier(3)
+  entered, duplicate, release = threading.Event(), threading.Event(), threading.Event()
+  calls = []
+  def limited(request, **kwargs):
+    calls.append(request.full_url)
+    entered.set()
+    if len(calls) > 1:
+      duplicate.set()
+    assert release.wait(timeout=0.8)
+    raise HTTPError(request.full_url, 429, 'Limited', {'Retry-After': '180'}, io.BytesIO())
+  monkeypatch.setattr(history, '_open_url', limited)
+  def caller(suffix):
+    start.wait(timeout=0.8)
+    return history._get_json('https://api.github.com/repos/a/b/' + suffix)
+  with ThreadPoolExecutor(max_workers=2) as executor:
+    futures = [executor.submit(caller, suffix) for suffix in ['branches/main', 'commits']]
+    start.wait(timeout=0.8)
+    try:
+      assert entered.wait(timeout=0.8)
+      duplicate.wait(timeout=0.1)
+    finally:
+      release.set()
+    for future in futures:
+      with pytest.raises(history.HistoryUnavailable):
+        future.result(timeout=0.8)
+  assert len(calls) == 1
diff --git a/starpilot/system/the_galaxy/version_history.py b/starpilot/system/the_galaxy/version_history.py
index d9f59e9776..af4437dc19 100644
--- a/starpilot/system/the_galaxy/version_history.py
+++ b/starpilot/system/the_galaxy/version_history.py
@@ -10,6 +10,7 @@ import subprocess
 import threading
 import time
 from collections import OrderedDict
+from contextlib import contextmanager
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
 from http.client import HTTPException
@@ -31,6 +32,10 @@ _cache = OrderedDict()
 _cache_lock = threading.Lock()
 _api_backoff_until = 0.0
 _api_retry_at = 0.0
+_api_request_lock = threading.Lock()
+_raw_backoff_until = 0.0
+_raw_retry_at = 0.0
+_request_locks = {}
 VERSION_CACHE_TTL = 6 * 60 * 60
 MAX_VERSION_CACHE_ENTRIES = 1024
 MAX_VERSION_BYTES = 64 * 1024
@@ -82,8 +87,50 @@ def _is_quota_failure(error):
                               (headers.get('X-RateLimit-Remaining') == '0' or (wait.isdigit() and len(wait) <= 6)))
 
 
+@contextmanager
+def _request_lock(kind, url):
+  """Coalesce identical misses without retaining locks after callers finish."""
+  key = (kind, url)
+  with _cache_lock:
+    entry = _request_locks.setdefault(key, [threading.Lock(), 0])
+    entry[1] += 1
+  try:
+    with entry[0]:
+      yield
+  finally:
+    with _cache_lock:
+      entry[1] -= 1
+      if entry[1] == 0:
+        del _request_locks[key]
+
+
+def _retry_at(headers):
+  wait = headers.get('Retry-After', '').strip()
+  reset = headers.get('X-RateLimit-Reset', '').strip()
+  if wait.isdigit() and len(wait) <= 6:
+    return time.time() + int(wait)
+  if reset.isdigit() and len(reset) <= 12 and int(reset) > time.time():
+    return float(reset)
+  return 0.0
+
+
+def _record_quota_backoff(retry_at, raw=False):
+  global _api_backoff_until, _api_retry_at, _raw_backoff_until, _raw_retry_at
+  delay = max(1, retry_at - time.time()) if retry_at else 60
+  deadline = time.monotonic() + delay
+  with _cache_lock:
+    if raw:
+      if deadline > _raw_backoff_until:
+        _raw_backoff_until, _raw_retry_at = deadline, retry_at
+    elif deadline > _api_backoff_until:
+      _api_backoff_until, _api_retry_at = deadline, retry_at
+
+
 def _get_display_version(url):
   """Read only the numeric version literal; never import or execute remote code."""
+  with _cache_lock:
+    if time.monotonic() < _raw_backoff_until:
+      raise HistoryUnavailable(_rate_limit_message(_raw_retry_at))
   request = Request(url, headers={'User-Agent': 'StarPilot-Galaxy-VersionPicker'})
   try:
     with _open_raw_url(request, timeout=HTTP_TIMEOUT) as response:
@@ -103,8 +150,11 @@ def _get_display_version(url):
     if error.code == 404:
       return None  # Older builds can predate the display-version file.
     if error.code in (403, 429):
-      failure = HistoryUnavailable if _is_quota_failure(error) else VersionHistoryError
-      raise failure('GitHub version metadata access is limited; retry later.') from None
+      if _is_quota_failure(error):
+        retry_at = _retry_at(error.headers or {})
+        _record_quota_backoff(retry_at, raw=True)
+        raise HistoryUnavailable(_rate_limit_message(retry_at)) from None
+      raise VersionHistoryError('GitHub version metadata access is limited; retry later.') from None
     failure = HistoryUnavailable if 500 <= error.code < 600 else VersionHistoryError
     raise failure('GitHub version metadata is unavailable; retry later.') from None
   except (URLError, TimeoutError, OSError, HTTPException):
@@ -117,14 +167,16 @@ def _display_version(base, sha):
   # The validated origin and full SHA make this an immutable public file URL.
   repository = base.removeprefix('https://api.github.com/repos/')
   url = f'https://raw.githubusercontent.com/{repository}/{sha}/selfdrive/ui/lib/starpilot_version.py'
-  # Share the limit across requests, and recheck the cache after acquiring a slot.
-  with _version_slots:
+  # Duplicate callers share one miss; cached values do not consume download slots.
+  with _request_lock('raw', url):
     with _cache_lock:
       entry = _version_cache.get(url)
       if entry is not None and time.monotonic() - entry[0] < VERSION_CACHE_TTL:
         _version_cache.move_to_end(url)
         return entry[1]
-    value = _get_display_version(url)
+    with _version_slots:
+      # Check raw quota inside the slot, including workers queued by other callers.
+      value = _get_display_version(url)
     with _cache_lock:
       _version_cache[url] = (time.monotonic(), value)
       _version_cache.move_to_end(url)
@@ -145,8 +197,13 @@ def _rate_limit_message(retry_at):
 
 
 def _get_json(url):
+  """Serialize public API requests and recheck quota after waiting for a turn."""
+  with _api_request_lock:
+    return _request_json(url)
+
+
+def _request_json(url):
   """Fetch one bounded public API response. Kept separate for offline tests."""
-  global _api_backoff_until, _api_retry_at
   with _cache_lock:
     if time.monotonic() < _api_backoff_until:
       raise HistoryUnavailable(_rate_limit_message(_api_retry_at))
@@ -162,21 +219,10 @@ def _get_json(url):
   except HTTPError as error:
     error.close()
     if error.code in (403, 429):
-      headers = error.headers or {}
-      wait = headers.get('Retry-After', '').strip()
-      reset = headers.get('X-RateLimit-Reset', '').strip()
-      retry_at = 0.0
-      if wait.isdigit() and len(wait) <= 6:
-        retry_at = time.time() + int(wait)
-      elif reset.isdigit() and len(reset) <= 12 and int(reset) > time.time():
-        retry_at = float(reset)
-      # Share only a short quota backoff; unrelated 403 responses do not block
-      # other lookups. Valid cached responses remain available during backoff.
+      retry_at = _retry_at(error.headers or {})
+      # Honor the complete server deadline. Unrelated 403s do not block lookups.
       if _is_quota_failure(error):
-        delay = max(1, min(60, retry_at - time.time())) if retry_at else 60
-        with _cache_lock:
-          _api_backoff_until = time.monotonic() + delay
-          _api_retry_at = retry_at
+        _record_quota_backoff(retry_at)
       failure = HistoryUnavailable if _is_quota_failure(error) else VersionHistoryError
       raise failure(_rate_limit_message(retry_at)) from None
     if error.code == 404:
@@ -190,20 +236,20 @@ def _get_json(url):
 
 
 def _json(url, fresh=False, ttl=CACHE_TTL):
-  now = time.monotonic()
-  if not fresh:
+  with _request_lock('api', url):
+    if not fresh:
+      with _cache_lock:
+        entry = _cache.get(url)
+        if entry is not None and time.monotonic() - entry[0] < ttl:
+          _cache.move_to_end(url)
+          return copy.deepcopy(entry[1])
+    value = _get_json(url)
     with _cache_lock:
-      entry = _cache.get(url)
-      if entry is not None and now - entry[0] < ttl:
-        _cache.move_to_end(url)
-        return copy.deepcopy(entry[1])
-  value = _get_json(url)
-  with _cache_lock:
-    _cache[url] = (time.monotonic(), copy.deepcopy(value))
-    _cache.move_to_end(url)
-    while len(_cache) > MAX_CACHE_ENTRIES:
-      _cache.popitem(last=False)
-  return value
+      _cache[url] = (time.monotonic(), copy.deepcopy(value))
+      _cache.move_to_end(url)
+      while len(_cache) > MAX_CACHE_ENTRIES:
+        _cache.popitem(last=False)
+    return value
 
 
 def _git(repo_path, *args):
@@ -444,7 +490,17 @@ def list_versions(repo_path, branch, page=1, head=None):
   return result
 
 
+def _saved_display_versions(base, branch, page, requested_head, resolved_head):
+  # A version literal belongs to an immutable SHA. Reuse its validated saved
+  # value only after the live branch head and requested commit page are checked.
+  snapshot = _load_history_snapshot(base, branch, page, requested_head)
+  if snapshot is None or snapshot['head'] != resolved_head:
+    return {}
+  return {row['sha']: row['version'] for row in snapshot['commits'] if 'version' in row}
+
+
 def _list_versions(base, branch, page, pinned_head):
+  requested_head = pinned_head
   current = _head(base, branch)
   if pinned_head is not None:
     _ancestor(base, pinned_head, current)
@@ -458,9 +514,18 @@ def _list_versions(base, branch, page, pinned_head):
     has_more = bool(_block(base, pinned_head, block_page + 2))
   selected = rows[offset:end]
   if branch.lower() == 'starpilot' and selected:
+    saved_versions = _saved_display_versions(base, branch, page, requested_head, pinned_head)
+    pending = []
+    for index, row in enumerate(selected):
+      if row['sha'] in saved_versions:
+        selected[index] = dict(row, version=saved_versions[row['sha']])
+      else:
+        pending.append(index)
+    if not pending:
+      return {'branch': branch, 'head': pinned_head, 'page': page, 'hasMore': has_more, 'commits': selected}
     executor = ThreadPoolExecutor(max_workers=_VERSION_WORKERS)
     try:
-      futures = {executor.submit(_display_version, base, row['sha']): index for index, row in enumerate(selected)}
+      futures = {executor.submit(_display_version, base, selected[index]['sha']): index for index in pending}
       for future in as_completed(futures):
         index = futures[future]
         selected[index] = dict(selected[index], version=future.result())
diff --git a/starpilot/system/the_galaxy/version_install.py b/starpilot/system/the_galaxy/version_install.py
index da57c0ff68..234a015c67 100644
--- a/starpilot/system/the_galaxy/version_install.py
+++ b/starpilot/system/the_galaxy/version_install.py
@@ -208,6 +208,31 @@ def check_repository_idle(repo):
   # Check independently of operation markers, which may be missing or stale.
   if git(repo, 'ls-files', '--unmerged'):
     raise InstallError('Repository has unmerged index entries; resolve them before installing or restoring')
+  # These flags can hide working edits from ordinary patches, and the patches
+  # cannot restore the flags themselves. Leave both source and index untouched.
+  for entry in git(repo, 'ls-files', '-v', '-z', binary=True).split(b'\0'):
+    if entry[:1].islower() or entry[:1] == b'S':
+      name = entry[2:].decode(errors='replace')
+      raise InstallError(f'Repository uses assume-unchanged or skip-worktree on {name}; save its contents and clear the flag before installing or restoring')
+
+
+def _check_ignored_collisions(repo, commit):
+  # Force checkout also replaces ignored files, including file/directory
+  # collisions. They are deliberately absent from the ordinary recovery tar.
+  ignored = [name for name in git(repo, 'ls-files', '--others', '--ignored', '--exclude-standard', '-z', binary=True).split(b'\0') if name]
+  if not ignored:
+    return
+  tracked = set(git(repo, 'ls-tree', '-r', '-z', '--name-only', commit, binary=True).split(b'\0')) - {b''}
+  directories = set()
+  for name in tracked:
+    parts = name.split(b'/')
+    directories.update(b'/'.join(parts[:index]) for index in range(1, len(parts)))
+  for name in ignored:
+    parts = name.split(b'/')
+    if (name in tracked or name in directories or
+        any(b'/'.join(parts[:index]) in tracked for index in range(1, len(parts)))):
+      label = name.decode(errors='replace')
+      raise InstallError(f'Ignored local path conflicts with the selected source: {label}; move or save it outside the checkout before installing or restoring')
 
 
 def _check_submodules(repo):
@@ -282,6 +307,7 @@ def restore(folder, *, check_parked, restore_data=False):
   validate_target(old)
   check_parked()
   check_repository_idle(repo)
+  _check_ignored_collisions(repo, old['commit'])
   # Validate the whole archive before changing the checkout.
   with tarfile.open(folder / 'untracked.tar') as archive:
     for member in archive.getmembers():
@@ -337,6 +363,7 @@ def install(repo, target, *, data_root=Path('/data'), check_parked, progress, re
   progress(2, 'Checking compatibility', 100, sha[:10])
   check_repository_idle(repo)
   _check_submodules(repo)
+  _check_ignored_collisions(repo, sha)
   preflight(repo, sha, require_device_binaries, data_root)
   check_parked()
   progress(3, 'Saving recovery backup', 0, 'Preserving local changes, settings and model statistics')