Galaxy: Add remote keys and app Sentry notifications

Serve the Galaxy app key and its QR code from /api/galaxy/session to direct local connections so the Android app can import the tunnel key on its own. Let the app and the iOS app register for Sentry push alerts.

openpilot/starpilot/galaxy/sentry_events.py

@@ -44,6 +44,7 @@ class SentryEvents:
           raise ValueError("Invalid motion event")
         events.append({"eventId": value["eventId"], "kind": value["kind"],
                        "systemTimeMs": value["wallTimeNs"] // 1_000_000,
+                       **({"simulated": True} if value.get("simulated") else {}),
                        "images": cast(Callable[[str], list], self.store.images)(value["eventId"]) if hasattr(self.store, "images") else []})
       result = {"schemaVersion": 1, "source": "local", "scanIncomplete": source["incomplete"],
                 "capacity": source["capacity"], "events": events,

openpilot/starpilot/galaxy/server.py

@@ -13,7 +13,7 @@ import socket
 from pathlib import Path
 import threading
 import time
-from urllib.parse import parse_qs, unquote, urlsplit
+from urllib.parse import parse_qs, quote, unquote, urlsplit
 
 from openpilot.common.swaglog import cloudlog
 from openpilot.starpilot.galaxy.batch_routes import change_drive_state, delete_records, non_default_settings, preview_layout, preview_settings_recovery
@@ -158,6 +158,7 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
   summary_lock = threading.Lock()
   motion_events = sentry_events if sentry_events is not None else SentryEvents()
   sentry_events_lock = threading.Lock()
+  last_simulation = [float("-inf")]
   notification_owner = notifications
   snapshots = camera_snapshot if camera_snapshot is not None else CameraSnapshot()
   access = owner if owner is not None else default_owner()
@@ -766,6 +767,30 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
         else:
           if self.require_session():
             self.json(200, result)
+      elif path in ('/api/galaxy/session', '/api/galaxy/session/qr.svg'):
+        record = pairing.read()
+        remote = getattr(self.server, 'remote_transport', False)
+        if not self.direct_local() and not (remote and record and gateway_cookie_valid(self.session_token(), record)):
+          self.json(403, {'error': 'A direct connection or matching gateway cookie is required'})
+          return
+        if not self.require_session():
+          return
+        if record is None:
+          self.json(200, {'paired': False})
+          return
+        token = quote(f"{record['slug']}:{record['session']}", safe='')
+        if path.endswith('/qr.svg'):
+          from openpilot.common.qrcode import _Qr, _capacity
+          raw = ('starpilot-remote://add?key=' + quote(token, safe='')).encode()
+          version = next(version for version in range(1, 21) if 4 + 8 + len(raw) * 8 <= _capacity(version) * 8)
+          matrix = _Qr(version, raw).modules
+          size = len(matrix) + 8
+          cells = ''.join(f'<path d="M{x + 4} {y + 4}h1v1h-1z"/>' for y, row in enumerate(matrix) for x, active in enumerate(row) if active)
+          svg = f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {size} {size}"><rect width="100%" height="100%" fill="white"/><g fill="black">{cells}</g></svg>'
+          self.respond(200, svg.encode(), 'image/svg+xml', cache_control='no-store')
+        else:
+          self.json(200, {'paired': True, 'slug': record['slug'], 'url': pairing.url(record['slug']),
+                          'cookieName': 'galaxy_session', 'sessionToken': token})
       elif path == '/api/galaxy/status':
         if not self.require_session():
           return
@@ -1688,7 +1713,24 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
           self.json(401, {'error': 'Sign in to Galaxy'})
           return
         try:
-          result = notification_owner.action(payload, permitted=lambda: self.settings_session() == identity)
+          if payload.get('action') == 'simulate':
+            with sentry_events_lock:
+              if set(payload) != {'action'}:
+                raise ValueError('Invalid simulation request')
+              if not parked():
+                self.json(403, {'error': 'Park the car before simulating an event'})
+                return
+              if clock() - last_simulation[0] < 10:
+                self.json(429, {'error': 'Wait ten seconds before simulating another event'})
+                return
+              notification_owner.store.record('warning', time.monotonic_ns(), simulated=True,
+                                               permitted=lambda: self.settings_session() == identity and parked())
+              last_simulation[0] = clock()
+              result = notification_owner.snapshot()
+          else:
+            result = notification_owner.action(payload, permitted=lambda: self.settings_session() == identity)
+        except LookupError:
+          self.json(404, {'error': 'Phone subscription not found'})
         except ValueError as error:
           self.json(400, {'error': str(error)})
         except PermissionError:

openpilot/starpilot/galaxy/tests/test_install_app.mjs

@@ -59,7 +59,7 @@ const android = { ...InstallApp.data() }
 InstallApp.methods.installAndroid.call(android)
 assert.equal(apk, './app.apk')
 assert.match(InstallApp.template, /Install web app/)
-assert.match(InstallApp.template, /Download Android APK/)
+assert.match(InstallApp.template, /Direct \(advanced\) · Download APK/)
 assert.match(InstallApp.template, /Update offline app \(Android\)/)
 assert.match(InstallApp.template, /android && !ios/)
 location.protocol = 'https:'

openpilot/starpilot/galaxy/tests/test_remote_pairing.py

@@ -58,6 +58,30 @@ class RemotePairingTest(unittest.TestCase):
     connection.close()
     return result
 
+  def test_app_key_export_requires_direct_session_or_matching_gateway_cookie(self):
+    for path in ('/api/galaxy/session', '/api/galaxy/session/qr.svg'):
+      self.assertEqual(self.request(self.local, path)[0], 401)
+    cookie = self.request(self.local, '/api/auth/session')[2]['Set-Cookie'].split(';', 1)[0]
+    code, body, headers = self.request(self.local, '/api/galaxy/session', cookie=cookie)
+    self.assertEqual((code, json.loads(body)), (200, {'paired': False}))
+    self.assertEqual(headers['Cache-Control'], 'no-store')
+    slug = self.pairing.pair(hashlib.sha256(b'password123').hexdigest())
+    record = self.pairing.read()
+    token = f"{slug}%3A{record['session']}"
+    for path in ('/api/galaxy/session', '/api/galaxy/session/qr.svg'):
+      self.assertEqual(self.request(self.local, path, cookie=cookie)[0], 200)
+      with mock.patch('openpilot.starpilot.galaxy.server.direct_local_connection', return_value=False):
+        self.assertEqual(self.request(self.local, path, cookie=cookie)[0], 403)
+      self.assertEqual(self.request(self.remote, path, host='galaxy.firestar.link', cookie=cookie)[0], 403)
+      self.assertEqual(self.request(self.remote, path, host='galaxy.firestar.link', cookie='galaxy_session=wrong')[0], 403)
+      code, body, headers = self.request(self.remote, f'/{slug}' + path, host='galaxy.firestar.link', cookie='galaxy_session=' + token)
+      self.assertEqual(code, 200)
+      self.assertEqual(headers['Cache-Control'], 'no-store')
+      if not path.endswith('.svg'):
+        self.assertEqual(json.loads(body)['sessionToken'], token)
+      else:
+        self.assertIn(b'<svg', body)
+
   def test_gateway_session_cannot_access_routes(self):
     self.assertTrue(self.owner.configure('password123', lambda: True))
     slug = self.pairing.pair(hashlib.sha256(b'password123').hexdigest())

openpilot/starpilot/galaxy/tests/test_sentry_notifications_http.py

@@ -12,6 +12,41 @@ from openpilot.starpilot.sentry_mode.notifications import NotificationOwner
 
 
 class NotificationHTTPTests(unittest.TestCase):
+  def test_app_subscription_test_and_simulation(self):
+    with tempfile.TemporaryDirectory() as directory:
+      root = Path(directory).resolve()
+      owner = NotificationOwner(root / 'notifications')
+      from openpilot.starpilot.sentry_mode.storage import EventStore
+      owner.store = EventStore(root / 'events')
+      parked = [True]
+      server = make_server(port=0, owner=GalaxyAccessOwner(root / 'access'), notifications=owner, parked=lambda: parked[0])
+      thread = threading.Thread(target=server.serve_forever, kwargs={'poll_interval': .01}, daemon=True)
+      thread.start()
+      def request(payload=None, cookie=''):
+        connection = http.client.HTTPConnection('127.0.0.1', server.server_port, timeout=3)
+        headers = {'Cookie': cookie, 'Origin': f'http://127.0.0.1:{server.server_port}', 'Content-Type': 'application/json'}
+        connection.request('POST' if payload is not None else 'GET', '/api/sentry/notifications' if payload is not None else '/api/auth/session',
+                           json.dumps(payload) if payload is not None else None, headers)
+        reply = connection.getresponse()
+        result = reply.status, json.loads(reply.read()), dict(reply.getheaders())
+        connection.close()
+        return result
+      try:
+        cookie = request()[2]['Set-Cookie'].split(';', 1)[0]
+        code, result, _ = request({'action': 'appSubscribe', 'token': 'test-token', 'appId': 'link.firestar.galaxy', 'label': 'Phone'}, cookie)
+        self.assertEqual(code, 200)
+        self.assertNotIn('test-token', json.dumps(result))
+        self.assertEqual(request({'action': 'test', 'target': 'app', 'id': 'missing'}, cookie)[0], 404)
+        self.assertEqual(request({'action': 'test', 'target': 'app', 'id': result['appSubscriptions'][0]['id']}, cookie)[0], 200)
+        parked[0] = False
+        self.assertEqual(request({'action': 'simulate'}, cookie)[0], 403)
+        parked[0] = True
+        self.assertEqual(request({'action': 'simulate'}, cookie)[0], 200)
+        self.assertTrue(owner.store.snapshot()['events'][0]['simulated'])
+        self.assertEqual(request({'action': 'simulate'}, cookie)[0], 429)
+      finally:
+        server.shutdown(); thread.join(timeout=2); server.server_close()
+
   def test_auth_origin_secret_redaction_and_session_change(self):
     with tempfile.TemporaryDirectory() as directory:
       root = Path(directory).resolve()

openpilot/starpilot/galaxy/web/js/companion.js

@@ -23,7 +23,7 @@ export const GalaxyCompanion = {
   components: { GxNotice, BluetoothNotice },
   emits: ['status'],
   props: { mode: String, unauthorized: Function },
-  data: () => ({ status: null, error: '', native: !!window.StarPilotApp, sharing: false, autoStart: false,
+  data: () => ({ status: null, error: '', native: !!window.StarPilotApp?.uplinkState, sharing: false, autoStart: false,
     bluetooth: bluetoothConnection(), nativeBytes: 0, busy: false, generation: 0, alive: false }),
   computed: { windowAutoStart() { return !!window.StarPilotApp?.setAutoStart } },
   mounted() { this.start() },
@@ -66,11 +66,12 @@ export const GalaxyCompanion = {
       this.busy = true
       try {
         const enabled = this.native ? !this.sharing : !this.status.enabled
+        if (this.native && window.StarPilotApp.setUplinkEnabled(enabled) === false) { this.sharing = false; return }
         const status = await requestJson('./api/companion/settings', { request: { method: 'POST',
           headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ enabled }) } })
         if (!this.alive || generation !== this.generation) return
         this.adopt(status)
-        if (this.native) { window.StarPilotApp.setUplinkEnabled(enabled); this.sharing = enabled }
+        if (this.native) this.sharing = enabled
         this.error = ''
       } catch (e) {
         if (!this.alive || generation !== this.generation) return

openpilot/starpilot/galaxy/web/js/galaxy.js

@@ -15,14 +15,15 @@ export const GalaxyPage = {
   components: { GxState, GxNotice, BluetoothNotice, InstallApp, GalaxyHotspot, GalaxyCompanion, LocalAccess },
   props: { mode: { type: String, required: true }, unauthorized: { type: Function, required: true } },
   data: () => ({ loading: true, paired: false, url: "", tunnelClientAvailable: false, legacyPassword: false,
-    legacyPairingAvailable: false, password: "", busy: false, error: "", companionStatus: null,
+    appKey: '', keyVisible: false, legacyPairingAvailable: false, password: "", busy: false, error: "", companionStatus: null,
     bluetooth: bluetoothConnection(), nativeVersion: window.StarPilotApp?.version?.() || '', dismissedUpdateVersion: dismissedUpdate(),
     ios: isIosDevice() }),
   computed: {
     availableAppVersion() { return this.companionStatus?.appAvailable ? this.companionStatus.shippedVersion : '' },
     installedAppVersion() { return this.nativeVersion || this.companionStatus?.version || '' },
+    playDistribution() { return window.StarPilotApp?.distribution?.() === 'play' },
     appUpdateAvailable() {
-      return this.mode === 'local' && newerCompanionVersion(this.availableAppVersion, this.installedAppVersion)
+      return !this.playDistribution && this.mode === 'local' && newerCompanionVersion(this.availableAppVersion, this.installedAppVersion)
     },
     nativeInstallerAvailable() { return typeof window.StarPilotApp?.installUpdate === 'function' },
     showAppUpdate() {
@@ -31,6 +32,13 @@ export const GalaxyPage = {
   },
   mounted() { this.load() },
   methods: {
+    async showAppKey() {
+      try { const data = await this.request('./api/galaxy/session'); this.appKey = data.paired ? decodeURIComponent(data.sessionToken) : ''; this.keyVisible = !!this.appKey }
+      catch (error) { this.error = error.message }
+    },
+    async copyAppKey() {
+      try { await navigator.clipboard.writeText(this.appKey) } catch { this.error = 'Select and copy the displayed key.' }
+    },
     installAppUpdate() { window.StarPilotApp?.installUpdate?.() },
     dismissAppUpdate() {
       this.dismissedUpdateVersion = this.availableAppVersion
@@ -84,6 +92,7 @@ export const GalaxyPage = {
   template: `
     <div class="gx-settings gx-tunnel" :inert="busy" :aria-busy="busy || undefined"><header class="gx-settings__header gx-page-header"><div><h2>Galaxy Access & Install</h2>
       <p>Pair your comma, then install Galaxy from your remote link.</p></div></header>
+      <p v-if="playDistribution" class="gx-note">Updates come from Google Play.</p>
       <BluetoothNotice v-if="showAppUpdate && bluetooth && !ios" :feature="'Android app update ' + availableAppVersion">
         Installed: {{ installedAppVersion }}.
         <button type="button" class="gx-btn gx-btn--tonal" aria-label="Dismiss app update notice" @click="dismissAppUpdate">Dismiss</button>
@@ -106,6 +115,11 @@ export const GalaxyPage = {
               <BluetoothNotice v-if="bluetooth" feature="Galaxy QR code">Your Galaxy link remains available below.</BluetoothNotice>
               <img v-else src="./api/galaxy/qr.svg" alt="QR code for your Galaxy link" class="gx-tunnel__qr" />
               <p><a :href="url" target="_blank" rel="noopener" class="gx-btn gx-btn--tonal gx-wrap">Open your Galaxy link</a></p>
+              <section class="gx-card"><h3>App key</h3><p>Keep this key private. It grants remote access to this comma.</p>
+                <button v-if="!keyVisible" class="gx-btn" @click="showAppKey">Show</button>
+                <template v-else><code class="gx-wrap">{{ appKey }}</code><button class="gx-btn" @click="copyAppKey">Copy</button>
+                  <img src="./api/galaxy/session/qr.svg" class="gx-tunnel__qr" alt="Add remote access to Galaxy" /></template>
+              </section>
               <div class="gx-actions"><button type="button" class="gx-btn gx-btn--danger" @click="unpair">{{ busy ? 'Unpairing…' : 'Unpair' }}</button></div>
             </template>
             <template v-else>

openpilot/starpilot/galaxy/web/js/install-app.js

@@ -27,7 +27,7 @@ export const InstallApp = {
   props: { url: { type: String, default: '' }, updateAvailable: { type: Boolean, default: false } },
   data: () => ({ prompt: null, installed: false, help: false, message: '',
     android: isAndroidDevice(), ios: isIosDevice() }),
-  computed: { onPublicGalaxy() { return window.location?.hostname === 'galaxy.firestar.link' } },
+  computed: { playDistribution() { return window.StarPilotApp?.distribution?.() === 'play' }, onPublicGalaxy() { return window.location?.hostname === 'galaxy.firestar.link' } },
   mounted() {
     this.prompt = deferredPrompt
     this.displayMode = window.matchMedia('(display-mode: standalone)')
@@ -75,7 +75,9 @@ export const InstallApp = {
     <a v-else-if="url" class="gx-btn" :href="url" target="_blank" rel="noopener"><i class="bi bi-box-arrow-up-right" aria-hidden="true"></i> Install web app</a>
     <p v-else class="gx-note">Pair in step 1 to get your install link.</p>
     <a v-if="ios" class="gx-btn" href="https://apps.apple.com/ca/app/starpilot-bt/id6821039119" target="_blank" rel="noopener"><i class="bi bi-apple" aria-hidden="true"></i> Download StarPilot BT on the App Store</a>
-    <button v-if="android && !ios && (!installed || updateAvailable)" type="button" class="gx-btn" @click="installAndroid"><i class="bi bi-download" aria-hidden="true"></i><span>{{ updateAvailable ? 'Update offline app (Android)' : 'Download Android APK' }}</span></button>
+    <button v-if="android && !ios && !playDistribution && (!installed || updateAvailable)" type="button" class="gx-btn" @click="installAndroid"><i class="bi bi-download" aria-hidden="true"></i><span>{{ updateAvailable ? 'Update offline app (Android)' : 'Direct (advanced) · Download APK' }}</span></button>
+    <a v-if="android" class="gx-btn" href="https://play.google.com/store/apps/details?id=link.firestar.galaxy" target="_blank" rel="noopener">Galaxy on Google Play</a>
+    <p v-if="playDistribution" class="gx-note">Updates come from Google Play.</p>
     <p v-if="help" class="gx-note gx-install-help">{{ message }}</p>
   </div>`,
 }

openpilot/starpilot/galaxy/web/js/sentry-notifications.js

@@ -61,7 +61,7 @@ export const SentryNotifications = {
   components: { GxNotice },
   name: "SentryNotifications",
   props: { mode: { type: String, required: true }, unauthorized: { type: Function, required: true } },
-  data: () => ({ channels, labels, notifications: null, notificationBusy: false, notificationError: "",
+  data: () => ({ nativePush: !!window.StarPilotApp?.sentryPush, pushState: {}, testPoll: null, channels, labels, notifications: null, notificationBusy: false, notificationError: "",
     drafts: Object.fromEntries(channels.map(name => [name, { url: "", token: "" }])) }),
   created() { this.client = new NotificationClient({ publish: update => Object.assign(this.$data, update), unauthorized: this.unauthorized }) },
   mounted() {
@@ -69,11 +69,23 @@ export const SentryNotifications = {
     document.addEventListener("visibilitychange", this.visibility)
     this.start()
   },
-  beforeUnmount() { this.client.stop(); clearInterval(this.poll); document.removeEventListener("visibilitychange", this.visibility) },
+  beforeUnmount() { clearInterval(this.testPoll); this.client.stop(); clearInterval(this.poll); document.removeEventListener("visibilitychange", this.visibility) },
   methods: {
     reportSave,
+    refreshPush() { try { this.pushState = JSON.parse(window.StarPilotApp?.sentryPush?.() || '{}') } catch { this.pushState = {} } },
+    togglePhone() { window.StarPilotApp?.setSentryPush?.(!this.pushState.enabled); this.refreshPush(); setTimeout(() => this.client.run(), 1000) },
+    async testPhone(id) {
+      if (!await this.client.run({ action: 'test', target: 'app', id })) return
+      clearInterval(this.testPoll)
+      let attempts = 0
+      this.testPoll = setInterval(() => { if (++attempts >= 15) { clearInterval(this.testPoll); this.testPoll = null } if (!this.client.pending) this.client.run() }, 1000)
+    },
+    removePhone(id) { return this.client.run({ action: 'appUnsubscribe', id }) },
+    simulate() { return this.client.run({ action: 'simulate' }) },
+    canSimulate() { return !!window.StarPilotApp || location.protocol === 'http:' },
     start() {
       clearInterval(this.poll)
+      this.refreshPush()
       if (this.mode !== "local" || document.hidden) return
       this.client.run()
       this.poll = setInterval(() => { if (!this.notificationBusy && !this.client.pending) this.client.run() }, 5000)
@@ -120,6 +132,18 @@ export const SentryNotifications = {
         <GxNotice tone="danger" v-if="notifications?.queueFull">The notification queue is full. Some events could not be queued.</GxNotice>
         <template v-if="notifications">
           <p class="gx-note">{{ notifications.deliverySemantics }}</p>
+          <section v-if="nativePush || notifications.appSubscriptions?.length" class="gx-card gx-panel gx-stack">
+            <h3>This phone (Galaxy app)</h3>
+            <button v-if="nativePush" class="gx-btn" @click="togglePhone">{{ pushState.enabled ? 'Disable phone alerts' : 'Enable phone alerts' }}</button>
+            <button v-if="pushState.registered && pushState.id" class="gx-btn" @click="testPhone(pushState.id)">Send test to this phone</button>
+            <p v-if="!notifications.appPushAvailable" class="gx-note">Push not set up — the app will pick up real alerts by polling.</p>
+            <p v-for="phone in (notifications.appSubscriptions || [])" :key="phone.id">{{ phone.label }} · {{ phone.lastResult || 'Not tested' }} · {{ phone.lastAt ? new Date(phone.lastAt * 1000).toLocaleString() : '' }}
+              <button class="gx-btn" @click="testPhone(phone.id)">Test</button><button class="gx-btn" @click="removePhone(phone.id)">Remove</button>
+            </p>
+            <p v-for="phone in (notifications.removedAppSubscriptions || [])" :key="phone.id">{{ phone.lastResult }}</p>
+            <button v-if="canSimulate()" class="gx-btn" @click="simulate">Simulate event</button>
+          </section>
+          <button class="gx-btn" @click="client.run({action: 'test', target: 'all'})">Test all channels</button>
           <section v-for="name in channels" :key="name" class="gx-card gx-sentry-channel">
             <div class="gx-section__header"><i class="bi" :class="{webPush: 'bi-bell', discord: 'bi-discord', webhook: 'bi-link-45deg', ntfy: 'bi-broadcast'}[name]" aria-hidden="true"></i><span class="gx-section__title">{{ labels[name] }}</span></div>
             <div class="gx-panel gx-stack"><div class="gx-row gx-row--borderless"><div class="gx-row__info"><span class="gx-row__label">Enabled</span>

openpilot/starpilot/galaxy/web/js/sentry.js

@@ -16,7 +16,8 @@ export function validSentryEvents(value) {
   if (value.storageIncomplete !== undefined && typeof value.storageIncomplete !== "boolean") return false
   const seen = new Set()
   return value.events.every((event) => {
-    if (event === null || typeof event !== "object" || Object.keys(event).some((key) => !["eventId", "kind", "systemTimeMs", "images"].includes(key)) ||
+    if (event === null || typeof event !== "object" || Object.keys(event).some((key) => !["eventId", "kind", "systemTimeMs", "images", "simulated"].includes(key)) ||
+        (event.simulated !== undefined && event.simulated !== true) ||
         (event.images !== undefined && (!Array.isArray(event.images) || event.images.length > 2 ||
           new Set(event.images).size !== event.images.length || event.images.some((camera) => !["wide", "cabin"].includes(camera)))) ||
         typeof event.eventId !== "string" || !ID.test(event.eventId) || seen.has(event.eventId) ||

openpilot/starpilot/sentry_mode/README.md

@@ -1,13 +1,13 @@
 # Sentry motion events
 
 This is the parked motion-event part of Sentry. Galaxy can list local event
-metadata at `#/cameras/events`. Camera capture, image galleries, notifications
-and remote access are not implemented here. Motion detection has
+metadata and captured images at `#/cameras/events`. Galaxy supports browser push,
+Discord, webhooks, ntfy, and optional Firebase app push through the companion relay. Motion detection has
 not yet been qualified on a physical device or vehicle.
 
 The manager runs it only on device hardware when both
-`STARPILOT_SENTRY_DEVELOPMENT=1` and a valid enabled `SentryMotionPreferences`
-document are present. Ordinary startup is unchanged. The same manager-owned
+the vehicle is parked and a valid enabled `SentryMotionPreferences`
+document is present. The same manager-owned
 `sensord` supplies acceleration onroad and, under that explicit opt-in, while
 parked. Sentry never starts another sensor, camera or power service.
 

openpilot/starpilot/sentry_mode/notifications.py

@@ -23,7 +23,9 @@ from openpilot.starpilot.storage import starpilot_storage_root
 from openpilot.starpilot.sentry_mode.storage import EventStore
 from openpilot.starpilot.sentry_mode import web_push
 
-CHANNELS = ('webPush', 'discord', 'webhook', 'ntfy')
+CHANNELS = ('webPush', 'discord', 'webhook', 'ntfy', 'appPush')
+APP_PUSH_RELAY_URL = ''
+APP_IDS = ('link.firestar.companion', 'link.firestar.galaxy', 'link.galaxy.bluetooth.natehuby76')
 MAX_STATE_BYTES = 1024 * 1024
 MAX_JOBS = 512
 MAX_AGE = 86400
@@ -52,7 +54,7 @@ def endpoint(value: object, *, allow_http: bool = False) -> str:
 def fresh_state() -> dict:
   return {'version': 1, 'channels': {channel: {'enabled': False, 'url': '', 'token': '', 'generation': uuid.uuid4().hex,
                                               'since': 0} for channel in CHANNELS},
-          'subscriptions': [], 'vapid': '', 'jobs': {}, 'queueFull': False}
+          'appSubscriptions': [], 'subscriptions': [], 'vapid': '', 'jobs': {}, 'queueFull': False}
 
 
 def wall_seconds() -> float:
@@ -136,6 +138,10 @@ class NotificationOwner:
           result[key] = value
         return result
       state = json.loads(raw, object_pairs_hook=unique)
+      if type(state) is dict and 'appSubscriptions' not in state:
+        state['appSubscriptions'] = []
+        if type(state.get('channels')) is dict:
+          state['channels']['appPush'] = fresh_state()['channels']['appPush']
       self._validate_state(state)
       self.state = state
       self.saved_bytes = json.dumps(state, separators=(',', ':'), allow_nan=False).encode()
@@ -161,8 +167,18 @@ class NotificationOwner:
         raise ValueError('Invalid notification channel')
       if channel['url']:
         endpoint(channel['url'], allow_http=self.allow_http)
-      if name != 'webPush' and channel['enabled'] and not channel['url']:
+      if name not in ('webPush', 'appPush') and channel['enabled'] and not channel['url']:
         raise ValueError('Missing notification endpoint')
+    if type(state['appSubscriptions']) is not list or len(state['appSubscriptions']) > 8:
+      raise ValueError('Invalid app subscriptions')
+    for sub in state['appSubscriptions']:
+      if (type(sub) is not dict or set(sub) != {'id', 'token', 'appId', 'label', 'addedAt', 'since', 'lastResult', 'lastAt'} or
+          type(sub['token']) is not str or not 1 <= len(sub['token']) <= 4096 or any(ord(c) <= 32 for c in sub['token']) or
+          sub['id'] != hashlib.sha256(sub['token'].encode()).hexdigest()[:12] or sub['appId'] not in APP_IDS or
+          type(sub['label']) is not str or len(sub['label']) > 64 or
+          type(sub['lastResult']) is not str or len(sub['lastResult']) > 64 or
+          any(type(sub[key]) not in (int, float) or not 0 <= sub[key] < 2**63 for key in ('addedAt', 'since', 'lastAt'))):
+        raise ValueError('Invalid app subscription')
     for sub in state['subscriptions']:
       if type(sub) is not dict or set(sub) != {'id', 'subscription', 'since'} or type(sub['id']) is not str or ID.fullmatch(sub['id']) is None:
         raise ValueError('Invalid browser subscription')
@@ -176,7 +192,7 @@ class NotificationOwner:
           type(key) is not str or len(key) != 64 or type(job['eventId']) is not str or ID.fullmatch(job['eventId']) is None or
           job['kind'] not in ('warning', 'alarm', 'test') or job['channel'] not in CHANNELS or
           type(job['generation']) is not str or ID.fullmatch(job['generation']) is None or
-          (type(job['subscriptionId']) is not str or (job['subscriptionId'] and ID.fullmatch(job['subscriptionId']) is None)) or
+          (type(job['subscriptionId']) is not str or (job['subscriptionId'] and re.fullmatch(r'[0-9a-f]{12}' if job['channel'] == 'appPush' else r'[0-9a-f]{32}', job['subscriptionId']) is None)) or
           type(job['attempts']) is not int or not 0 <= job['attempts'] <= MAX_ATTEMPTS or
           job['state'] not in ('queued', 'sending', 'sent', 'failed', 'cancelled') or
           job['error'] not in ('', 'http', 'expired', 'timeout', 'transport', 'delivery_unknown', 'stale') or
@@ -219,6 +235,9 @@ class NotificationOwner:
                 'subscriptionCount': len(self.state['subscriptions']),
                 'subscriptions': [{'id': sub['id']} for sub in self.state['subscriptions']],
                 'deliverySemantics': 'Retries may repeat delivery after an interrupted remote response.'}
+      result['appSubscriptions'] = [{key: sub[key] for key in ('id', 'label', 'appId', 'addedAt', 'lastResult', 'lastAt')} for sub in self.state['appSubscriptions']]
+      result['removedAppSubscriptions'] = [{'id': job['subscriptionId'], 'lastResult': 'Phone unregistered — removed', 'lastAt': job['next']} for job in jobs.values() if job['channel'] == 'appPush' and job['error'] == 'expired'][-8:]
+      result['appPushAvailable'] = bool(self.state['channels']['appPush']['url'] or APP_PUSH_RELAY_URL)
       for name, cfg in self.state['channels'].items():
         relevant = [j for j in jobs.values() if j['channel'] == name and j['generation'] == cfg['generation']]
         latest = max(relevant, key=lambda j: j['created'], default=None)
@@ -262,7 +281,7 @@ class NotificationOwner:
         # Empty fields preserve configured credentials; explicit disable does not delete them.
         url = url or old['url']
         token = payload['token'] or old['token']
-        if payload['enabled'] and name != 'webPush' and not url:
+        if payload['enabled'] and name not in ('webPush', 'appPush') and not url:
           raise ValueError('Enter a notification URL')
         self.state['channels'][name] = {'enabled': payload['enabled'], 'url': url, 'token': token,
                                         'generation': uuid.uuid4().hex, 'since': now}
@@ -272,6 +291,8 @@ class NotificationOwner:
         self.state['channels'][payload['channel']] = fresh_state()['channels'][payload['channel']]
         if payload['channel'] == 'webPush':
           self.state['subscriptions'] = []
+        if payload['channel'] == 'appPush':
+          self.state['appSubscriptions'] = []
       elif action == 'pushKey':
         if set(payload) != {'action'}:
           raise ValueError('Invalid push key request')
@@ -295,6 +316,41 @@ class NotificationOwner:
         if set(payload) != {'action', 'id'} or type(payload['id']) is not str or ID.fullmatch(payload['id']) is None:
           raise ValueError('Choose a browser subscription')
         self.state['subscriptions'] = [s for s in self.state['subscriptions'] if s['id'] != payload['id']]
+      elif action == 'appSubscribe':
+        if (set(payload) != {'action', 'token', 'appId', 'label'} or type(payload['token']) is not str or
+            not 1 <= len(payload['token']) <= 4096 or any(ord(c) <= 32 for c in payload['token']) or
+            payload['appId'] not in APP_IDS or type(payload['label']) is not str or len(payload['label']) > 64):
+          raise ValueError('Invalid app subscription')
+        current = self.state['appSubscriptions']
+        existing = next((sub for sub in current if sub['token'] == payload['token']), None)
+        if existing:
+          existing.update(appId=payload['appId'], label=payload['label'])
+        elif len(current) < 8:
+          current.append({'id': hashlib.sha256(payload['token'].encode()).hexdigest()[:12], 'token': payload['token'],
+                          'appId': payload['appId'], 'label': payload['label'], 'addedAt': now, 'since': now,
+                          'lastResult': '', 'lastAt': 0})
+        else:
+          raise ValueError('Remove a phone before adding another')
+        cfg = self.state['channels']['appPush']
+        cfg['enabled'] = True
+        cfg['url'] = cfg['url'] or APP_PUSH_RELAY_URL
+      elif action == 'appUnsubscribe':
+        if set(payload) not in ({'action', 'token'}, {'action', 'id'}):
+          raise ValueError('Choose a phone subscription')
+        key = 'token' if 'token' in payload else 'id'
+        if type(payload[key]) is not str:
+          raise ValueError('Choose a phone subscription')
+        self.state['appSubscriptions'] = [sub for sub in self.state['appSubscriptions'] if sub[key] != payload[key]]
+      elif action == 'test' and 'channel' not in payload:
+        if set(payload) - {'action', 'target', 'id'} or payload.get('target', 'all') not in ('all', 'app'):
+          raise ValueError('Choose a test target')
+        target_id = payload.get('id')
+        if payload.get('target') == 'app' and not any(sub['id'] == target_id for sub in self.state['appSubscriptions']):
+          raise LookupError('Phone subscription not found')
+        if sum(j['kind'] == 'test' and j['created'] > now - 60 for j in self.state['jobs'].values()) >= 32:
+          raise ValueError('Wait a minute before sending more tests')
+        self._enqueue({'eventId': uuid.uuid4().hex, 'kind': 'test', 'wallTimeNs': int(now * 1e9)},
+                      only='appPush' if payload.get('target') == 'app' else None, target_id=target_id)
       elif action == 'test':
         if set(payload) != {'action', 'channel'} or payload['channel'] not in CHANNELS:
           raise ValueError('Choose a test channel')
@@ -314,15 +370,17 @@ class NotificationOwner:
       self._save()
       return self.snapshot()
 
-  def _enqueue(self, event: dict, *, only: str | None = None) -> None:
+  def _enqueue(self, event: dict, *, only: str | None = None, target_id: str | None = None) -> None:
     if event['kind'] == 'selfie':
       return
     now = self.clock()
     for name, cfg in self.state['channels'].items():
       if not cfg['enabled'] or (only is not None and only != name):
         continue
-      targets = self.state['subscriptions'] if name == 'webPush' else [{'id': '', 'since': cfg['since']}]
+      targets = self.state['subscriptions'] if name == 'webPush' else self.state['appSubscriptions'] if name == 'appPush' else [{'id': '', 'since': cfg['since']}]
       for target in targets:
+        if target_id is not None and target['id'] != target_id:
+          continue
         if event['kind'] != 'test' and event['wallTimeNs'] / 1e9 <= max(cfg['since'], target['since']):
           continue
         key = hashlib.sha256((event['eventId'] + name + cfg['generation'] + target['id']).encode()).hexdigest()
@@ -372,7 +430,9 @@ class NotificationOwner:
           continue
         cfg = self.state['channels'][job['channel']]
         sub = next((s['subscription'] for s in self.state['subscriptions'] if s['id'] == job['subscriptionId']), None)
-        if not cfg['enabled'] or cfg['generation'] != job['generation'] or (job['channel'] == 'webPush' and sub is None):
+        if job['channel'] == 'appPush':
+          sub = next((s for s in self.state['appSubscriptions'] if s['id'] == job['subscriptionId']), None)
+        if not cfg['enabled'] or cfg['generation'] != job['generation'] or (job['channel'] in ('webPush', 'appPush') and sub is None):
           job['state'] = 'cancelled'
           continue
         if now < job['created'] or now - job['created'] > MAX_AGE:
@@ -398,8 +458,18 @@ class NotificationOwner:
           if current_job is None:
             continue
           job = current_job
+          if job['channel'] == 'appPush':
+            current_sub = next((s for s in self.state['appSubscriptions'] if s['id'] == job['subscriptionId']), None)
+            if current_sub is not None:
+              current_sub['lastAt'] = self.clock()
+              current_sub['lastResult'] = ('Delivered to relay' if 200 <= code < 300 else 'Relay not configured' if code == 0 else
+                                           'Phone unregistered — removed' if code == 410 else 'Rate limited' if code == 429 else f'Relay error {code}')
+            if code == 410:
+              self.state['appSubscriptions'] = [s for s in self.state['appSubscriptions'] if s['id'] != job['subscriptionId']]
           if 200 <= code < 300:
             job['state'], job['error'] = 'sent', ''
+          elif code == 410 and job['channel'] == 'appPush':
+            job['state'], job['error'], job['next'] = 'failed', 'expired', now
           elif code in (404, 410) and job['channel'] == 'webPush':
             job['state'], job['error'] = 'failed', 'expired'
             self.state['subscriptions'] = [s for s in self.state['subscriptions']
@@ -408,9 +478,13 @@ class NotificationOwner:
             job['state'], job['error'] = ('queued' if code in (408, 429) or code >= 500 else 'failed'), 'http'
         except (requests.Timeout, TimeoutError):
           job['state'], job['error'] = 'queued', 'timeout'
+          if job['channel'] == 'appPush' and sub is not None:
+            sub['lastAt'], sub['lastResult'] = self.clock(), 'Relay timeout'
         except Exception:
           # Never log response bodies, URLs, exception text, subscription secrets or auth headers.
           job['state'], job['error'] = 'queued', 'transport'
+          if job['channel'] == 'appPush' and sub is not None:
+            sub['lastAt'], sub['lastResult'] = self.clock(), 'Relay unavailable'
         if job['state'] == 'queued':
           if job['attempts'] >= MAX_ATTEMPTS:
             job['state'] = 'failed'
@@ -440,7 +514,17 @@ class NotificationOwner:
     message = 'This is a test StarPilot Sentry notification.' if job['kind'] == 'test' else f"Parked motion detected ({job['kind']})."
     event = {'eventId': job['eventId'], 'kind': job['kind'], 'systemTimeMs': int(job['created'] * 1000)}
     headers = {'Idempotency-Key': key}
-    if name == 'webPush':
+    if name == 'appPush':
+      if subscription is None:
+        raise ValueError('Missing app subscription')
+      relay = cfg['url'] or APP_PUSH_RELAY_URL
+      if not relay:
+        return 0
+      response = requests.post(relay.rstrip('/') + '/v1/sentry', json={
+        'token': subscription['token'], 'appId': subscription['appId'], 'kind': job['kind'], 'eventId': job['eventId'],
+        'wallTimeNs': int(job['created'] * 1e9), 'label': subscription['label']}, headers=headers,
+        timeout=(3, 10), allow_redirects=False, stream=True)
+    elif name == 'webPush':
       if subscription is None:
         raise ValueError('Missing browser subscription')
       body, push_headers = web_push.request(subscription, {'title': 'StarPilot Sentry Mode', 'body': message,

openpilot/starpilot/sentry_mode/storage.py

@@ -46,7 +46,7 @@ def _valid(value: object, event_id: str) -> bool:
   if any(type(key) is not str for key in value):
     return False
   fields = {key: item for key, item in value.items() if type(key) is str}
-  if set(fields) != FIELDS or ID.fullmatch(event_id) is None:
+  if set(fields) not in (FIELDS, FIELDS | {"simulated"}) or ("simulated" in fields and fields["simulated"] is not True) or ID.fullmatch(event_id) is None:
     return False
   version = fields["version"]
   session_id = fields["sessionId"]
@@ -117,10 +117,12 @@ class EventStore:
     finally:
       os.close(fd)
 
-  def record(self, kind: str, mono_time_ns: int, *, permitted: Callable[[], bool], images: dict | None = None) -> RecordReceipt:
+  def record(self, kind: str, mono_time_ns: int, *, permitted: Callable[[], bool], images: dict | None = None, simulated: bool = False) -> RecordReceipt:
     event_id = uuid.uuid4().hex
     value = {"version": 1, "eventId": event_id, "sessionId": self.session_id, "kind": kind,
              "monoTimeNs": mono_time_ns, "wallTimeNs": time.time_ns()}
+    if simulated:
+      value["simulated"] = True
     if not _valid(value, event_id):
       raise ValueError("Invalid Sentry event")
     raw = json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
@@ -262,6 +264,9 @@ class EventStore:
               value = json.loads(raw, object_pairs_hook=_unique)
               if not _valid(value, event_id):
                 raise ValueError("Invalid event metadata")
+              if value.get('simulated') and time.time_ns() - value['wallTimeNs'] > 600_000_000_000:
+                os.unlink(entry.name, dir_fd=directory)
+                continue
               result["events"].append(value)
             except (OSError, ValueError, TypeError, RecursionError):
               result["incomplete"] = True

openpilot/starpilot/sentry_mode/tests/test_notifications.py

@@ -120,7 +120,7 @@ def test_real_http_contracts_and_independent_channels(tmp_path):
     assert paths['/push'][0]['Authorization'].startswith('vapid t=')
     assert paths['/push'][0]['Content-Encoding'] == 'aes128gcm'
     assert len(paths['/push'][1]) > 100 and b'This is a test' not in paths['/push'][1]
-    assert all(owner.snapshot()['channels'][name]['lastState'] == 'sent' for name in CHANNELS)
+    assert all(owner.snapshot()['channels'][name]['lastState'] == 'sent' for name in ('webPush', 'discord', 'webhook', 'ntfy'))
   finally:
     owner.close()
     server.shutdown()
@@ -312,3 +312,54 @@ def test_missing_saved_images_send_original_text_contract(tmp_path, monkeypatch,
       assert json.loads(sent['data']['event']) == {'eventId': 'a' * 32, 'kind': 'warning', 'systemTimeMs': 1000000}
   finally:
     owner.close()
+
+
+def test_app_subscription_dispatch_redaction_and_expired_removal(tmp_path):
+  calls = []
+  clock = [100.]
+  owner = NotificationOwner(tmp_path / 'notifications', store(), clock=lambda: clock[0], transport=lambda *args: calls.append(args) or 410)
+  try:
+    snapshot = owner.action({'action': 'appSubscribe', 'token': 'private-fcm-token', 'appId': 'link.firestar.galaxy', 'label': 'Phone'})
+    sub_id = snapshot['appSubscriptions'][0]['id']
+    assert len(sub_id) == 12 and 'private-fcm-token' not in json.dumps(snapshot)
+    with pytest.raises(LookupError):
+      owner.action({'action': 'test', 'target': 'app', 'id': 'missing'})
+    owner.action({'action': 'test', 'target': 'app', 'id': sub_id})
+    owner.tick()
+    assert calls[0][0] == 'appPush' and calls[0][2]['token'] == 'private-fcm-token'
+    assert owner.snapshot()['appSubscriptions'] == []
+  finally:
+    owner.close()
+
+
+def test_app_success_dedupe_selfie_and_migration(tmp_path):
+  calls = []
+  clock = [100.]
+  events = []
+  root = tmp_path / 'notifications'
+  owner = NotificationOwner(root, store(events), clock=lambda: clock[0], transport=lambda *args: calls.append(args) or 200)
+  owner.action({'action': 'appSubscribe', 'token': 'token', 'appId': 'link.firestar.companion', 'label': 'Phone'})
+  events.extend([{'eventId': 'a' * 32, 'kind': 'selfie', 'wallTimeNs': int(101e9)}, {'eventId': 'b' * 32, 'kind': 'warning', 'wallTimeNs': int(101e9)}])
+  clock[0] = 102.
+  owner.tick(); owner.tick()
+  assert len(calls) == 1
+  assert owner.snapshot()['appSubscriptions'][0]['lastResult'] == 'Delivered to relay'
+  owner.close()
+  reopened = NotificationOwner(root, store(events), clock=lambda: clock[0])
+  assert reopened.available and len(reopened.snapshot()['appSubscriptions']) == 1
+  reopened.close()
+  state = json.loads((root / 'state.json').read_text())
+  state.pop('appSubscriptions'); state['channels'].pop('appPush'); state['jobs'] = {}
+  (root / 'state.json').write_text(json.dumps(state))
+  migrated = NotificationOwner(root, store())
+  assert migrated.available and migrated.snapshot()['appSubscriptions'] == []
+  migrated.close()
+
+
+def test_ios_bundle_id_can_subscribe(tmp_path):
+  owner = NotificationOwner(tmp_path / 'notifications', store(), transport=lambda *args: 200)
+  try:
+    snapshot = owner.action({'action': 'appSubscribe', 'token': 'ios-fcm-token', 'appId': 'link.galaxy.bluetooth.natehuby76', 'label': 'iPhone'})
+    assert snapshot['appSubscriptions'][0]['appId'] == 'link.galaxy.bluetooth.natehuby76'
+  finally:
+    owner.close()

openpilot/starpilot/sentry_mode/tests/test_storage.py

@@ -140,3 +140,17 @@ class EventStoreTest(unittest.TestCase):
 
 if __name__ == "__main__":
   unittest.main()
+
+
+def test_simulated_events_expire_without_removing_real_events(tmp_path, monkeypatch):
+  import time
+  from openpilot.starpilot.sentry_mode.storage import EventStore
+  store = EventStore(tmp_path / 'events')
+  clock = [1000000000000]
+  monkeypatch.setattr(time, 'time_ns', lambda: clock[0])
+  fake = store.record('warning', 1, permitted=lambda: True, simulated=True)
+  real = store.record('warning', 2, permitted=lambda: True)
+  assert len(store.snapshot()['events']) == 2
+  clock[0] += 601000000000
+  assert [event['eventId'] for event in store.snapshot()['events']] == [real.event_id]
+  assert not (store.root / (fake.event_id + '.json')).exists()