Restore in-car Force Offroad controls

openpilot/starpilot/drive_state/tests/test_galaxy_drive_eligibility.py

@@ -127,11 +127,13 @@ def test_authenticated_remote_force_drive_modes_use_shared_owner():
     result = case.drive_action('onroad', remote=True)
     assert result[0] == 200 and result[1]['mode'] == 'onroad'
     assert result[1]['effective'] == 'offroad'
-    assert case.drive_action('offroad', remote=True)[0] == 409
+    assert case.drive_action('offroad', remote=True)[0] == 403
     assert case.drive.snapshot().mode == Mode.ONROAD
     case.allowed = True
     result = case.drive_action('offroad', remote=True)
-    assert result[0] == 200 and result[1]['mode'] == 'offroad'
+    assert result[0] == 403
+    case.control.change('offroad', case.drive.snapshot().revision, lambda: True)
+    assert case.drive.snapshot().mode == Mode.OFFROAD
 
 
 def test_update_snapshot_keeps_one_admission_and_matching_reason():

openpilot/starpilot/drive_state/tests/test_http.py

@@ -4,7 +4,8 @@ from pathlib import Path
 from types import SimpleNamespace as NS
 import tempfile
 from openpilot.starpilot.galaxy.tests import test_navigation_http as helpers
-from openpilot.starpilot.drive_state.owner import DriveStateOwner, Mode
+from openpilot.starpilot.drive_state.owner import DriveStateOwner, Mode, Rejected
+import pytest
 from openpilot.starpilot.drive_state.control import DriveStateControl
 from openpilot.starpilot.drive_state.tests.test_owner import Params, BOOT
 
@@ -41,15 +42,17 @@ class ForceHttpTest(helpers.NavigationHttpTest):
     self.assertEqual((result[1]['mode'], result[1]['effective']), ('onroad', 'offroad'))
     status = self.request('/api/drive-state/status', remote=True, cookie=self.remote_cookie)
     self.assertEqual(status[1]['revision'], result[1]['revision'])
-    self.assertEqual(self.drive_action('offroad', remote=True)[0], 200)
+    self.assertEqual(self.drive_action('offroad', remote=True)[0], 403)
+    self.control.change('offroad', self.drive.snapshot().revision, lambda: True)
     self.assertEqual(self.drive_action('auto', revision=result[1]['revision'])[0], 409)
 
   def test_physical_loss_blocks_offroad_but_onroad_and_auto_remain_recoverable(self):
-    self.assertEqual(self.drive_action('offroad')[0], 200)
+    self.assertEqual(self.drive_action('offroad')[0], 403)
+    self.control.change('offroad', self.drive.snapshot().revision, lambda: True)
     self.allowed = False
     self.assertEqual(self.drive_action('onroad')[0], 200)
     self.assertEqual(self.drive.snapshot().mode, Mode.ONROAD)
-    self.assertEqual(self.drive_action('offroad')[0], 409)
+    self.assertEqual(self.drive_action('offroad')[0], 403)
     self.assertEqual(self.drive.snapshot().mode, Mode.ONROAD)
     self.assertEqual(self.drive_action('auto', remote=True)[0], 200)
     self.assertEqual(self.drive.snapshot().mode, Mode.AUTO)
@@ -60,18 +63,28 @@ class ForceHttpTest(helpers.NavigationHttpTest):
     self.assertEqual(self.drive_action('onroad', origin='https://unrelated.invalid')[0], 403)
     self.assertEqual(self.drive_action('badmode')[0], 409)
     self.assertEqual(self.params.writes, before)
-    calls = []
+    original_change = self.control.change
 
-    def revoke():
-      calls.append(1)
-      if len(calls) == 2:
-        self.pairing.unpair()
-      return True
+    def revoke_remote(mode, revision, authorized):
+      self.assertEqual(mode, 'onroad')
+      self.assertTrue(authorized())
+      self.pairing.unpair()  # Real HTTP credential changes inside the route's effect lock.
+      self.assertFalse(authorized())
+      return original_change(mode, revision, authorized)
 
-    self.physical.allowed = revoke
-    self.assertEqual(self.drive_action('offroad', remote=True)[0], 409)
+    with patch.object(self.control, 'change', side_effect=revoke_remote):
+      self.assertEqual(self.drive_action('onroad', remote=True)[0], 409)
     self.assertEqual(self.params.writes, before)
 
+  def test_native_park_permission_revoked_during_commit_writes_nothing(self):
+    before = list(self.params.writes)
+    permission = iter((True, False))
+    self.physical.allowed = lambda: next(permission)
+    with pytest.raises(Rejected):
+      self.control.change('offroad', self.drive.snapshot().revision, lambda: True)
+    self.assertEqual(self.params.writes, before)
+    self.assertEqual(self.drive.snapshot().mode, Mode.AUTO)
+
   def test_authenticated_force_offroad_from_live_onroad_state(self):
     self._force_offroad_from_live_onroad_state(True)
 
@@ -98,12 +111,14 @@ class ForceHttpTest(helpers.NavigationHttpTest):
     self.assertEqual(status[1]['effective'], 'onroad')
     self.assertEqual(status[1]['overrideAllowed'], can_valid)
     result = self.drive_action('offroad', remote=True)
-    self.assertEqual(result[0], 200 if can_valid else 409)
+    self.assertEqual(result[0], 403)
     if can_valid:
-      self.assertEqual(result[1]['mode'], 'offroad')
+      self.control.change('offroad', self.drive.snapshot().revision, lambda: True)
+      self.assertEqual(self.drive.snapshot().mode, Mode.OFFROAD)
     else:
+      with pytest.raises(Rejected, match='Park'):
+        self.control.change('offroad', self.drive.snapshot().revision, lambda: True)
       self.assertEqual(self.drive.snapshot().mode, Mode.AUTO)
-      self.assertIn('Park', result[1]['error'])
     # hardwared consumes this same owner request through the actual resolver.
     messages.values['deviceState'].started = should_start(self.drive.snapshot().mode,
       {'ignition': True}, {'ready': True}, already_started=True)

openpilot/starpilot/galaxy/batch_routes.py

@@ -109,6 +109,9 @@ def change_drive_state(self, payload, effect_lock, drive_owner):
       len(payload['revision']) != 32):
     self.json(400, {'error': 'Invalid drive state request'})
     return
+  if payload['mode'] == 'offroad':
+    self.json(403, {'error': 'Use the in-car Force Off-road control'})
+    return
   from openpilot.starpilot.drive_state.owner import Rejected
   try:
     with effect_lock:

openpilot/starpilot/galaxy/tests/test_drive_state.mjs

@@ -36,10 +36,6 @@ await onroad.change('offroad');assert.equal(requests.length,before)
 window.confirm=()=>true
 await onroad.change('offroad');assert.equal(requests.length,before)
 onroad.state.overrideAllowed=true
-const stop=onroad.change('offroad')
-assert.equal(requests.length,before+1)
-assert.deepEqual(JSON.parse(requests.at(-1).options.body),{mode:'offroad',revision:'a'.repeat(32)})
-requests.at(-1).resolve(response({...snapshot('offroad','d'.repeat(32)),effective:'onroad'}));await flush()
-assert.equal(DriveStatePanel.computed.pending.call(onroad),true)
-requests.at(-1).resolve(response({...snapshot('offroad','d'.repeat(32)),effective:'offroad'}));await stop;await flush()
-assert.equal(DriveStatePanel.computed.pending.call(onroad),false)
+await onroad.change('offroad');assert.equal(requests.length,before)
+assert.match(DriveStatePanel.template, /\['onroad', 'auto'\]/)
+assert.doesNotMatch(DriveStatePanel.template, /v-for="mode in \['offroad'/)

openpilot/starpilot/galaxy/web/js/drive-state.js

@@ -35,9 +35,7 @@ export const DriveStatePanel = {
     },
     async change(mode) {
       if (this.busy || !this.state?.available) return
-      if (mode === "offroad" && !this.state.overrideAllowed) { this.error = "Park the vehicle and disengage before forcing Offroad."; return }
-      if (mode === "offroad" && this.state.effective === "onroad" &&
-          !window.confirm("Switch to Offroad and stop driving services? Disengage and stop safely first. Stay stopped until you return to Auto.")) return
+      if (!["onroad", "auto"].includes(mode)) return
       const revision = this.state.revision
       this.busy = true
       this.loading = false
@@ -61,11 +59,10 @@ export const DriveStatePanel = {
     <GxNotice tone="danger" v-if="state && !state.available">The drive-state manager is unavailable. Reconnect to check again.</GxNotice>
     <GxNotice v-if="pending" tone="info">Waiting for the device to switch {{ state.mode }}…</GxNotice>
     <p>Device: {{ state?.effective === "onroad" ? "Onroad" : state?.effective === "offroad" ? "Offroad" : "Checking…" }}</p>
-    <p>Auto follows the car. Force Onroad starts driving services at any time. Force Offroad stops them when the vehicle is parked and disengaged. Return to Auto restores normal operation.</p>
-    <p v-if="state?.available && !state.overrideAllowed" role="status">{{ state.reason || "Park the vehicle and disengage to use Force Offroad." }}</p>
-    <div class="gx-force-drive__actions gx-actions"><button v-for="mode in ['offroad', 'onroad', 'auto']" :key="mode" type="button"
+    <p>Auto follows the car. Force Onroad starts driving services at any time. Return to Auto restores normal operation.</p>
+    <div class="gx-force-drive__actions gx-actions"><button v-for="mode in ['onroad', 'auto']" :key="mode" type="button"
       class="gx-btn" :class="state?.mode === mode ? '' : 'gx-btn--tonal'"
-      :disabled="!state?.available || (mode === 'offroad' && !state.overrideAllowed)"
-      :aria-pressed="state?.mode === mode" @click="change(mode)">{{ mode === 'auto' ? 'Return to Auto' : mode === 'onroad' ? 'Onroad' : 'Offroad' }}</button></div>
+      :disabled="!state?.available"
+      :aria-pressed="state?.mode === mode" @click="change(mode)">{{ mode === 'auto' ? 'Return to Auto' : 'Onroad' }}</button></div>
     </section>`,
 }

openpilot/starpilot/ui/runtime_app.py

@@ -812,7 +812,7 @@ class StarShellSession:
       drive_row = FeatureRow("drive_state", "Force Drive State", LABELS[drive["mode"]],
                              source=drive["revision"].encode() if drive["revision"] is not None else None,
                              available=drive["available"], reason=f"Device: {drive['effective'] or 'unavailable'}",
-                             actions=tuple((label, drive["mode"] != mode and (mode != "onroad" or drive["overrideAllowed"]))
+                             actions=tuple((label, drive["mode"] != mode and (mode != "offroad" or drive["overrideAllowed"]))
                                            for mode, label in LABELS.items()))
       power_available = getattr(self, "system_power_action", None) is not None and not ui_state.engaged
       device_power = FeatureRow("device_power", "Device Power", "", available=power_available,
@@ -1239,10 +1239,12 @@ class StarShellSession:
                                    (item.destination == Destination.FORCE_DRIVE and drive["available"]) or
                                    snapshot.device.offroad and item.destination == Destination.PAIR)
                            for item in snapshot.settings.availability)
-      next_mode = DriveStateControl.next_mode(drive)
+      from openpilot.starpilot.drive_state.control import CYCLE
+      next_mode = CYCLE[(CYCLE.index(next(mode for mode in CYCLE if mode.value == drive["mode"])) + 1) % len(CYCLE)]
       label = ("unavailable" if not drive["available"] else
                {"offroad": "Force Off-road", "onroad": "Force On-road", "auto": "Return to Auto"}[next_mode.value])
-      availability = tuple(replace(item, request_value=next_mode.value, request_revision=drive["revision"])
+      availability = tuple(replace(item, request_value=next_mode.value, request_revision=drive["revision"],
+                                   reason=drive.get("reason") if next_mode.value == "offroad" and not drive["overrideAllowed"] else None)
                            if item.destination == Destination.FORCE_DRIVE else item for item in availability)
       snapshot = replace(snapshot, settings=replace(snapshot.settings, availability=availability, nav_bar_alpha=0,
                                                    force_drive_label=label))

openpilot/starpilot/ui/settings.py

@@ -104,6 +104,9 @@ class SettingsView:
         rl.draw_texture_ex(background, rl.Vector2(x, top), 0, 1, rl.WHITE)
         font_size = 40 if destination == Destination.FORCE_DRIVE else 42 if destination == Destination.DRIVING_MODEL else (
           48 if destination in (Destination.GALAXY, Destination.PAIR) else 64)
+        if destination == Destination.FORCE_DRIVE and state.destination(destination).reason:
+          reason = state.destination(destination).reason
+          self.fonts.draw(reason[:46], FontRole.MEDIUM, 18, x + 20, top + 18, rl.Color(255, 190, 140, 255))
         measured = self.fonts.measure(text, FontRole.BOLD, font_size)
         self.fonts.draw(text, FontRole.BOLD, font_size, x + 40, top + 180 - 23 - measured.height,
                         rl.Color(255, 255, 255, 229))

openpilot/starpilot/ui/tests/test_compact_settings_lifecycle.py

@@ -101,6 +101,7 @@ def test_hidden_settings_still_updates_navigation_but_does_not_project(lifecycle
 
 @pytest.mark.parametrize("mode,allowed,label,target", [
   ("auto", True, "Force Off-road", "offroad"),
+  ("auto", False, "Force Off-road", "offroad"),
   ("offroad", True, "Force On-road", "onroad"),
   ("onroad", True, "Return to Auto", "auto"),
   ("offroad", False, "Force On-road", "onroad"),
@@ -121,3 +122,30 @@ def test_force_menu_names_requested_action_with_supported_glyphs(lifecycle, mode
   glyphs = {int(value) for value in re.findall(r"char id=(\d+)", font.read_text())}
   assert all(ord(char) in glyphs for char in label)
   assert ord("→") not in glyphs
+
+
+@pytest.mark.parametrize('allowed,target', [(True, 'offroad'), (False, 'offroad')])
+def test_actual_compact_force_card_hitbox_dispatches_captured_owner_request(lifecycle, allowed, target):
+  from dataclasses import replace
+  from openpilot.starpilot.ui.settings_state import SettingsInput
+  from openpilot.starpilot.ui.shell import ShellRequest
+  _, session, _, _ = lifecycle
+  session._mode = runtime_app.ShellMode.SETTINGS
+  session._drive_change = Mock()
+  revision = 'current-owner-revision'
+  session.drive_state.snapshot = lambda: {'mode': 'auto', 'revision': revision, 'available': True,
+                                        'effective': 'onroad', 'overrideAllowed': allowed, 'reason': None if allowed else 'Current Panda status is stale'}
+  snapshot = session.snapshot(runtime_app.ShellMode.SETTINGS)
+  assert snapshot.settings.destination(runtime_app.Destination.FORCE_DRIVE).reason == (None if allowed else "Current Panda status is stale")
+  index = next(i for i, item in enumerate(compact_menu(snapshot.settings)) if item[0] == runtime_app.Destination.FORCE_DRIVE)
+  state = replace(snapshot.settings, compact_scroll_x=-422 * index, compact_y=0)
+  inputs = SettingsInput(session.profile, lambda action: session._emit(ShellRequest('settings', action)))
+  inputs.press(221, 120, state)
+  session._drive_change.assert_not_called()
+  inputs.release(221, 120, state)
+  session._drive_change.assert_called_once_with(target, revision)
+  session._drive_change.reset_mock()
+  inputs.press(221, 120, state)
+  inputs.move(230, 120, state)
+  inputs.release(230, 120, state)
+  session._drive_change.assert_not_called()

openpilot/starpilot/ui/tests/test_display_settings.py

@@ -133,10 +133,14 @@ class DisplaySettingsTests(unittest.TestCase):
     session._drive_change.assert_not_called()
     status["overrideAllowed"] = False
     fresh = session.system_snapshot()
-    self.assertIsNone(inputs.target(*point, fresh))
-    session._display_ui(FeatureUiAction("action", row, 2))
+    self.assertIsNotNone(inputs.target(*point, fresh))
+    session._drive_change.reset_mock()
+    session._display_ui(FeatureUiAction("action", fresh.rows[0], 2))
+    session._drive_change.assert_called_once_with("onroad", "shown-revision")
+    self.assertFalse(fresh.rows[0].actions[1][1])  # Only Offroad requires parked admission.
+    session._drive_change.reset_mock()
+    session._display_ui(FeatureUiAction("action", fresh.rows[0], 1))
     session._drive_change.assert_not_called()
-    self.assertTrue(fresh.rows[0].actions[1][1])  # Offroad remains available when Onroad admission fails.
 
   def test_display_submenu_opens_while_off_enabling_enters_and_back_keeps_the_hierarchy(self):
     with patch("openpilot.starpilot.ui.runtime_app.native_device.invalidate_display_preferences"):

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "08f89bc4d521155ab31354a99634e73173838ecc5350ed9bb1fde7aa04bbeab7", "schema_version": 1}
+{"protected_inventory_sha256": "954e1e15518212b2ac6b07ebf3bd576c13b57818156029ce7618438ed5dbb7ec", "schema_version": 1}