Qualify split HKG angle EPS publications

.github/workflows/hkg-angle-recovery-check.yaml

@@ -0,0 +1,64 @@
+name: HKG angle recovery qualification
+on:
+  push:
+    branches: [codex/hkg-angle-retention-20261106]
+permissions:
+  contents: read
+jobs:
+  recovery:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 30
+    env:
+      PYTHONPATH: ${{ github.workspace }}
+      SCALE: '1'
+    steps:
+      - uses: actions/checkout@v7
+      - uses: actions/setup-go@v7
+        with:
+          go-version: '1.25.1'
+          cache-dependency-path: mapd_repo/go.sum
+      - name: Fetch pinned map provider dependencies
+        working-directory: mapd_repo
+        env:
+          GOTOOLCHAIN: local
+        run: go mod download
+      - run: ./tools/op.sh setup
+      - name: Build matching Linux dependencies
+        run: .venv/bin/scons -j2
+      - name: Require the original split-delivery failure
+        shell: bash
+        run: |
+          .venv/bin/python - <<'PYCODE'
+          from pathlib import Path
+          import subprocess, xml.etree.ElementTree as ET, json
+          path=Path('openpilot/starpilot/aol/temporary_steering.py')
+          shipping=path.read_bytes()
+          original=subprocess.check_output(['git','show','68434cf39c501604fe962e9488a03bde0fed0d73:openpilot/starpilot/aol/temporary_steering.py'])
+          try:
+            path.write_bytes(original)
+            result=subprocess.run(['.venv/bin/python','-m','pytest','-q','openpilot/starpilot/aol/tests/test_hkg_angle_eps_retention.py','-k','split_publications','--junitxml=baseline.xml'])
+            root=ET.parse('baseline.xml').getroot()
+            cases=list(root.iter('testcase')); failed=[c for c in cases if c.find('failure') is not None]
+            assert result.returncode==1 and len(failed)==4 and not list(root.iter('error'))
+            assert all('NATIVE_FIRST_EPS_DISARMED' in ''.join(c.find('failure').itertext()) for c in failed)
+            Path('baseline-result.json').write_text(json.dumps({'cases':len(cases),'expectedFailures':len(failed),'confirmed':True}))
+          finally:
+            path.write_bytes(shipping)
+          assert path.read_bytes()==shipping
+          PYCODE
+      - name: Run steering retention and independent safety veto regressions
+        run: |
+          .venv/bin/python -m pytest -q --junitxml=qualification.xml             openpilot/starpilot/aol/tests/test_hkg_angle_eps_retention.py             openpilot/starpilot/aol/tests/test_pe_combined_profile_contract.py             openpilot/starpilot/aol/tests/test_stock_eps_recovery.py             openpilot/starpilot/aol/tests/test_runtime.py             openpilot/starpilot/aol/tests/test_source_pause_driver_monitoring.py
+      - name: Check owned Python files
+        run: |
+          .venv/bin/ruff check openpilot/starpilot/aol/temporary_steering.py openpilot/starpilot/aol/tests/test_hkg_angle_eps_retention.py
+          .venv/bin/ty check openpilot/starpilot/aol/temporary_steering.py openpilot/starpilot/aol/tests/test_hkg_angle_eps_retention.py
+      - name: Save exact results
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: hkg-angle-recovery-results
+          path: |
+            baseline.xml
+            baseline-result.json
+            qualification.xml

openpilot/starpilot/aol/temporary_steering.py

@@ -150,7 +150,7 @@ def temporary_steering_intent_current(self, sm, cp, cs, axis, stamp, now_ns, *,
   if (native is None or not native.pandaInventory or
       stock_eps_user and (native.requestedLongitudinal or native.longitudinalAllowed) or
       held_original and not (native.requestedLateral and (native.lateralAllowed or
-        restricted_eps and (ford_alpha_eps or not native.requestedLongitudinal and not native.longitudinalAllowed))) or
+        restricted_eps and (ford_alpha_eps or hkg_angle_eps or not native.requestedLongitudinal and not native.longitudinalAllowed))) or
       axis.longitudinalActive and not native.longitudinalAllowed or
       native.observedMonoTime < self.lateral_native_ns or
       (native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory)) != self.lateral_inventory):

openpilot/starpilot/aol/tests/test_hkg_angle_eps_retention.py

@@ -121,3 +121,46 @@ def test_angle_eps_retention_preserves_independent_vetoes(fault):
       sm.update_msgs(now / 1e9, frames)
       rejected = feedback.observe(sm, sd.CP, state, now, latched=True, intent_producer_session='authority-card')
       assert rejected == (tick == 3)
+
+
+@pytest.mark.parametrize('identity', ('pe_long', 'ev9_long'))
+@pytest.mark.parametrize('full', (False, True))
+@pytest.mark.parametrize('steering', (False, True))
+@pytest.mark.parametrize('delivery', ('native_first', 'event_first', 'together'))
+def test_angle_eps_split_publications_preserve_only_armed_intent(identity, full, steering, delivery):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    cleanup.enter_context(OpenpilotPrefix())
+    cp = profile(identity)
+    assert cp is not None
+    sd, _, drive = case._authority_callers(full, cp=cp, genuine_events=True)
+    feedback = TransportPauseFeedback(sd.CP)
+    sm = messaging.SubMaster(['aolAxisState', 'onroadEvents', 'aolSafetyWire', 'pandaStates'])
+    state = fixture.car_state()
+    state.vEgoRaw = state.vEgo
+    state.cruiseState.enabled = full and cp.pcmCruise
+    for tick in range(9):
+      fault = 3 <= tick < 8
+      state.steerFaultTemporary = fault
+      axis, _, command = drive(tick, observed_state=state, steering=steering if fault else False,
+        allowed=not fault, native_request_lat=True, native_request_long=full)
+      now = 10_000_000_000 + tick * 10_000_000
+      frames = [sd.pm.messages['aolAxisState'], sd.pm.messages['onroadEvents']]
+      for name in ('aolSafetyWire', 'pandaStates'):
+        message = messaging.new_message(name, len(sd.sm[name]), valid=sd.sm.valid[name],
+          logMonoTime=sd.sm.logMonoTime[name])
+        setattr(message, name, sd.sm[name])
+        frames.append(message.as_reader())
+      assert bool(axis.lateralActive and command.latActive) == (not fault)
+      if tick == 3 and delivery != 'together':
+        sm.update_msgs(now / 1e9, frames[2:] if delivery == 'native_first' else frames[:2])
+        assert not feedback.observe(sm, sd.CP, state, now, latched=True,
+          intent_producer_session='authority-card'), 'NATIVE_FIRST_EPS_DISARMED'
+        assert not feedback.failed
+      sm.update_msgs(now / 1e9, frames)
+      assert not feedback.observe(sm, sd.CP, state, now, latched=True,
+        intent_producer_session='authority-card'), 'EPS_JOIN_DISARMED'
+      assert not feedback.failed
+      if fault:
+        assert not axis.lateralActive and not command.latActive