diff --git a/openpilot/starpilot/flm/offroad_processor.py b/openpilot/starpilot/flm/offroad_processor.py index ed067a088b..511405118f 100644 --- a/openpilot/starpilot/flm/offroad_processor.py +++ b/openpilot/starpilot/flm/offroad_processor.py @@ -189,6 +189,9 @@ def collect_route_inputs(root, route, *, permitted, walker=None): raw_windows = {} context_metadata = {} counts = { + 'recorded_lateral_request_rows': 0, + 'active_lateral_request_rows': 0, + 'invalid_lateral_request_rows': 0, 'recorded_tracking': 0, 'joined_tracking': 0, 'missing_join': 0, @@ -226,6 +229,10 @@ def collect_route_inputs(root, route, *, permitted, walker=None): if index % 128 == 0 and not permitted(): raise ValueError('Parked analysis cancelled') kind = event.which() + if kind == 'carControl': + counts['recorded_lateral_request_rows'] += 1 + counts['active_lateral_request_rows'] += bool(event.carControl.latActive) + counts['invalid_lateral_request_rows'] += not event.valid if kind == 'carParams': # Own the actual CP reader independently of this segment's byte buffer. cp = event.carParams.as_builder().as_reader() @@ -352,9 +359,52 @@ def collect_route_inputs(root, route, *, permitted, walker=None): return RecordedInputs(receipt, state, windows, stable_windows, baseline_tracking, bookends, raw_windows, context_metadata, counts, quality_stops) -def analyze_latest(params, root, previous, *, permitted): +def collect_latest_steering_inputs(params, root, *, permitted): route = latest_route(params, root, permitted=permitted) - collected = collect_route_inputs(root, route, permitted=permitted) + import time + from openpilot.starpilot.flm.offroad_route import MAX_ANALYSIS_SECONDS, MAX_SEGMENTS, LOCAL_ID + from openpilot.starpilot.galaxy.drive_history import DriveHistory + + started = time.monotonic() + + def allowed(): + return permitted() and time.monotonic() - started <= MAX_ANALYSIS_SECONDS + + collected = collect_route_inputs(root, route, permitted=allowed) + current_cp = collected.receipt.cp_sha256 + skipped = 0 + + def inactive(inputs): + c = inputs.counts + return ( + c['recorded_lateral_request_rows'] >= 40 and c['active_lateral_request_rows'] == 0 and c['invalid_lateral_request_rows'] == 0 and not inputs.quality_stops + ) + + if inactive(collected) and LOCAL_ID.fullmatch(route['routeId']): + snapshot = DriveHistory(root, max_segments_per_route=MAX_SEGMENTS).snapshot() + if snapshot['scanIncomplete']: + raise ValueError('Latest steering drive inventory incomplete') + counter = int(route['routeId'].split('--')[0], 16) + candidates = [] + for candidate in snapshot['routes']: + if LOCAL_ID.fullmatch(candidate['routeId']): + distance = (counter - int(candidate['routeId'].split('--')[0], 16)) % 2**32 + if 1 <= distance <= 3 and candidate['segments'] and all(s['files']['rlog'] for s in candidate['segments']): + candidates.append((distance, candidate['routeId'], candidate)) + for _, _, candidate in sorted(candidates): + earlier = collect_route_inputs(root, candidate, permitted=allowed) + if earlier.receipt.cp_sha256 != current_cp: + continue + skipped += 1 + if not inactive(earlier): + route, collected = candidate, earlier + collected.counts['inactive_recordings_skipped'] = skipped + break + return route, collected + + +def analyze_latest(params, root, previous, *, permitted): + route, collected = collect_latest_steering_inputs(params, root, permitted=permitted) receipt, state = collected.receipt, collected.state if type(receipt) is not RouteReceipt: raise ValueError('A whole completed-route assessment requires all original segments') diff --git a/openpilot/starpilot/flm/offroad_quality.py b/openpilot/starpilot/flm/offroad_quality.py index b42bfd076f..7cff03f524 100644 --- a/openpilot/starpilot/flm/offroad_quality.py +++ b/openpilot/starpilot/flm/offroad_quality.py @@ -6,7 +6,9 @@ Only complete, input-matched original windows can trigger a withdrawal. """ from collections import deque +from collections.abc import Callable import math +import os from openpilot.starpilot.flm.offroad_surface import ROUTE_NAMES as NAMES from openpilot.starpilot.flm.offroad_analysis import cohort @@ -38,6 +40,7 @@ class TrialGuard: self.last_ns = 0 self.override_ns = 0 self.stopped = None + self.start_trial: Callable[[], bool] | None = None self.cohort = None self.baseline = () try: @@ -106,6 +109,15 @@ class TrialGuard: ): self.rows.clear() return None + if self.start_trial is not None: + start, self.start_trial = self.start_trial, None + try: + claimed = start() + except (OSError, ValueError): + claimed = False + if not claimed: + self.stopped = 'baseline_unavailable' + return self.marker point = read_point(event, companions, route='current-route-revocation-only', segment=0, context_key=self.scope_key) if point is None: self.rows.clear() @@ -167,8 +179,11 @@ def from_configuration(cp, params, cache_directory, *, vehicle_session): cache = AssessmentCache(cache_directory) trial = trial_document(cache, row['analysis_sha256']) guard = TrialGuard(scope.key(), row['values'], trial) - if row['state'] == 'ready_for_next_drive' and guard.stopped is None and not cache.claim_trial(row['analysis_sha256'], vehicle_session): - guard.stopped = 'baseline_unavailable' + if row['state'] == 'ready_for_next_drive' and guard.stopped is None: + if os.path.lexists(cache.directory / ('trial-' + row['analysis_sha256'] + '.json')): + guard.stopped = 'baseline_unavailable' + else: + guard.start_trial = lambda digest=row['analysis_sha256'], session=vehicle_session: cache.claim_trial(digest, session) except (OSError, KeyError, TypeError, ValueError, OverflowError): guard = TrialGuard(scope.key(), row['values'], {}) return guard diff --git a/openpilot/starpilot/flm/offroad_runtime.py b/openpilot/starpilot/flm/offroad_runtime.py index a83e9a0244..27983281da 100644 --- a/openpilot/starpilot/flm/offroad_runtime.py +++ b/openpilot/starpilot/flm/offroad_runtime.py @@ -4,6 +4,7 @@ No permissions, response witnesses, ACKs or source clocks are loaded from disk. """ import time +import os from openpilot.starpilot.flm.offroad_surface import ROUTE_NAMES as NAMES from openpilot.starpilot.flm.offroad_store import RouteScope, startup_row @@ -273,6 +274,11 @@ def configure(cp, ci, controller, params, *, manual_status): row, status = startup_row(params, scope) if row is None: return None, status + if row['state'] == 'ready_for_next_drive': + from openpilot.starpilot.flm.offroad_cache import directory_for_params + + if os.path.lexists(directory_for_params(params) / ('trial-' + row['analysis_sha256'] + '.json')): + return None, 'provisional_trial_already_started' values = tuple((n, float(row['values'][n])) for n in NAMES) previous = tuple((n, float(row['previous'][n])) for n in NAMES) fixed = None if not fixed_manual else fixed_manual_numeric(controller, fixed_manual) diff --git a/openpilot/starpilot/flm/offroad_worker.py b/openpilot/starpilot/flm/offroad_worker.py index 8fed8500ca..f9df3f59c4 100644 --- a/openpilot/starpilot/flm/offroad_worker.py +++ b/openpilot/starpilot/flm/offroad_worker.py @@ -8,6 +8,7 @@ from dataclasses import replace import hashlib import json import math +import os from pathlib import Path from openpilot.starpilot.flm.offroad_surface import ROUTE_NAMES as NAMES @@ -148,7 +149,7 @@ def quality_withdrawn(assessment, scope, values): return any(stop_marker(scope.key(), values, r) in reasons for r in STOP_REASONS) -def assess_next_drive(trial, assessment, scope): +def assess_next_drive(trial, assessment, scope, *, trial_started=True): """Compare independently held-out actual observations; no raw-RMS reward.""" if trial.get('policy') != POLICY or trial.get('kind') != 'next_drive_trial': raise ValueError('Exact provisional assessment provenance required') @@ -200,6 +201,14 @@ def assess_next_drive(trial, assessment, scope): else: results.append(compare_observed(old, new)) comparisons.append({'original_context_sha256': group['original_context_sha256'], 'comparison_contract_sha256': contract, 'independent': results}) + if not comparisons and not trial_started: + return { + 'outcome': 'pending', + 'reason': 'trial_not_applied_waiting_for_eligible_drive', + 'observed_only': True, + 'causal_road_benefit_proven': False, + 'comparisons': [], + } improved = bool(comparisons and all(all(r['improved'] for r in c['independent']) for c in comparisons)) return { 'outcome': 'accepted_after_drive' if improved else 'rolled_back', @@ -327,11 +336,17 @@ class OffroadWorker: scope, row, expected = state['scope'], state['row'], state['raw'] def authorized(): - return ( - self.permitted() - and self.scope_for(assessment.cp, self.params) == scope - and latest_route(self.params, self.root, permitted=self.permitted) == assessment.source_inventory - ) + if not ( + self.permitted() and self.scope_for(assessment.cp, self.params) == scope and latest_route(self.params, self.root, permitted=self.permitted) == route + ): + return False + if assessment.source_inventory == route: + return True + from openpilot.starpilot.flm.offroad_route import MAX_SEGMENTS + from openpilot.starpilot.galaxy.drive_history import DriveHistory + + inventory = DriveHistory(self.root, max_segments_per_route=MAX_SEGMENTS).snapshot() + return not inventory['scanIncomplete'] and sum(r == assessment.source_inventory for r in inventory['routes']) == 1 if not authorized(): raise ValueError('Source or settings changed after fitting') @@ -360,7 +375,7 @@ class OffroadWorker: observed_checks.append( {'qualified': actual['qualified'] and contribution['qualified'], 'actual_recorded_map': actual, 'compared_with_prior_map': contribution} ) - outcome = assess_next_drive(trial, assessment, scope) + outcome = assess_next_drive(trial, assessment, scope, trial_started=os.path.lexists(self.cache.directory / ('trial-' + row['analysis_sha256'] + '.json'))) if outcome['outcome'] == 'accepted_after_drive' and (not observed_checks or not all(r['qualified'] for r in observed_checks)): outcome = { 'outcome': 'rolled_back', @@ -369,28 +384,33 @@ class OffroadWorker: 'causal_road_benefit_proven': False, 'comparisons': outcome['comparisons'], } - record = { - 'kind': 'next_drive_outcome', - 'policy': POLICY, - 'trial_analysis_sha256': row['analysis_sha256'], - 'assessment': report, - 'all_actual_next_drive_checks': observed_checks, - **outcome, - } - digest = self.cache.put(record) - saved = RouteStorage(self.params).resolve( - scope=scope, - expected=expected, - outcome=outcome['outcome'], - route_sha256=assessment.receipt.source_sha256, - completed_wall_ns=assessment.receipt.completed_wall_ns, - analysis_sha256=digest, - authorized=authorized, - ) - if not saved: - raise ValueError('Next-drive outcome compare-and-swap refused') - status = outcome['outcome'] - reason = outcome['reason'] + if outcome['outcome'] == 'pending': + status, reason = 'ready_for_next_drive', outcome['reason'] + report['unused_recording_review'] = outcome + report['recorded_tracking'] = trial.get('assessment', {}).get('recorded_tracking') + else: + record = { + 'kind': 'next_drive_outcome', + 'policy': POLICY, + 'trial_analysis_sha256': row['analysis_sha256'], + 'assessment': report, + 'all_actual_next_drive_checks': observed_checks, + **outcome, + } + digest = self.cache.put(record) + saved = RouteStorage(self.params).resolve( + scope=scope, + expected=expected, + outcome=outcome['outcome'], + route_sha256=assessment.receipt.source_sha256, + completed_wall_ns=assessment.receipt.completed_wall_ns, + analysis_sha256=digest, + authorized=authorized, + ) + if not saved: + raise ValueError('Next-drive outcome compare-and-swap refused') + status = outcome['outcome'] + reason = outcome['reason'] else: current = dict(ZERO) if row is None else dict(row['values']) choices = [] diff --git a/openpilot/starpilot/flm/tests/test_offroad_backend.py b/openpilot/starpilot/flm/tests/test_offroad_backend.py index 02ad0d3f44..063a70d9f4 100644 --- a/openpilot/starpilot/flm/tests/test_offroad_backend.py +++ b/openpilot/starpilot/flm/tests/test_offroad_backend.py @@ -123,6 +123,34 @@ class TestOffroadCache(unittest.TestCase): first = from_configuration(None, None, self.cache.directory, vehicle_session='a' * 32) self.assertIsNotNone(first) self.assertIsNone(first.stopped) + claim = self.cache.directory / ('trial-' + row['analysis_sha256'] + '.json') + self.assertFalse(claim.exists(), 'A parked constructor does not consume an untested trial') + from types import SimpleNamespace + + stamp = 1_000_000_000 + control = SimpleNamespace( + valid=True, + enabled=True, + active=False, + manual=False, + override=False, + fault=False, + limited=False, + saturated=False, + control_ns=stamp, + route_effect_revision=REVISION, + route_configuration_scope=scope().key(), + route_applied_coordinates=tuple(values(0.01).items()), + ) + event = SimpleNamespace(logMonoTime=stamp) + first.observe(event, {}, control, now_ns=stamp) + self.assertFalse(claim.exists()) + stamp += 10_000_000 + event.logMonoTime = control.control_ns = stamp + control.active = True + with patch('openpilot.starpilot.flm.offroad_quality.read_point', return_value=None): + first.observe(event, {}, control, now_ns=stamp) + self.assertTrue(claim.exists(), 'The first valid current active route frame consumes the trial once') restarted = from_configuration(None, None, self.cache.directory, vehicle_session='b' * 32) self.assertEqual(restarted.stopped, 'baseline_unavailable') owner = RouteEffects(scope(), tuple(values(0.01).items()), (), previous=tuple(values(0.005).items())) @@ -133,6 +161,118 @@ class TestOffroadCache(unittest.TestCase): row['state'] = 'accepted_after_drive' self.assertIsNone(from_configuration(None, None, self.cache.directory, vehicle_session='c' * 32).stopped) + def test_unused_session_waits_but_started_or_withdrawn_trials_still_roll_back(self): + from types import SimpleNamespace + from openpilot.starpilot.flm.offroad_worker import assess_next_drive, POLICY + from openpilot.starpilot.flm.offroad_surface import stop_marker + + declared = dict(trial(), policy=POLICY, values=values(0.01), original_metadata={}) + empty = SimpleNamespace(groups=(), stable_groups=(), quality_stop_markers=()) + self.assertEqual(assess_next_drive(declared, empty, scope(), trial_started=False)['outcome'], 'pending') + self.assertEqual(assess_next_drive(declared, empty, scope(), trial_started=True)['outcome'], 'rolled_back') + marker = stop_marker(scope().key(), declared['values'], 'gross_tracking_or_oscillation_worse') + stopped = SimpleNamespace(groups=(), stable_groups=(), quality_stop_markers=({'reason': marker},)) + self.assertEqual(assess_next_drive(declared, stopped, scope(), trial_started=False)['outcome'], 'rolled_back') + + def test_actual_start_is_blocked_before_controller_reuse_and_rollback_restores_prior_map(self): + from types import SimpleNamespace + from openpilot.starpilot.flm.offroad_runtime import configure + + params = Params(str(self.root / 'startup')) + cache = AssessmentCache(directory_for_params(params)) + row = {'state': 'ready_for_next_drive', 'values': values(0.01), 'previous': values(), 'analysis_sha256': cache.put(trial())} + with ( + patch('openpilot.starpilot.flm.offroad_runtime.configuration_data', return_value=(scope(), (), ())), + patch('openpilot.starpilot.flm.offroad_runtime.startup_row', return_value=(row, 'loaded')), + ): + owner, status = configure(None, None, SimpleNamespace(), params, manual_status='not_configured') + self.assertIsNotNone(owner) + self.assertEqual(status, 'configured_pending_actual_stage') + self.assertTrue(cache.claim_trial(row['analysis_sha256'], 'a' * 32)) + owner, status = configure(None, None, SimpleNamespace(), params, manual_status='not_configured') + self.assertIsNone(owner, 'A genuine consumed trial cannot apply even one frame after restart') + self.assertEqual(status, 'provisional_trial_already_started') + row['state'] = 'rolled_back' + row['values'] = values(0.005) + owner, status = configure(None, None, SimpleNamespace(), params, manual_status='not_configured') + self.assertIsNotNone(owner, 'Rollback must restore the previous known map, including a nonzero accepted tune') + assert owner is not None + self.assertEqual(dict(owner.values), row['values']) + self.assertEqual(status, 'configured_pending_actual_stage') + + def test_last_steering_drive_selection_ignores_only_recorded_inactive_sessions(self): + from types import SimpleNamespace + from openpilot.starpilot.flm import offroad_processor as module + + latest = {'routeId': '00000012--aaaaaaaaaa', 'segments': [{'files': {'rlog': True}}]} + earlier = {'routeId': '00000011--bbbbbbbbbb', 'segments': [{'files': {'rlog': True}}]} + counts = {'recorded_lateral_request_rows': 100, 'active_lateral_request_rows': 0, 'invalid_lateral_request_rows': 0} + parked = SimpleNamespace(receipt=SimpleNamespace(cp_sha256='a' * 64), counts=dict(counts), quality_stops=[]) + driving = SimpleNamespace(receipt=SimpleNamespace(cp_sha256='a' * 64), counts=dict(counts, active_lateral_request_rows=70), quality_stops=[]) + calls = [] + + def collect(root, route, *, permitted): + self.assertTrue(permitted()) + calls.append(route['routeId']) + return parked if route is latest else driving + + history = SimpleNamespace(snapshot=lambda: {'scanIncomplete': False, 'routes': [latest, earlier]}) + with ( + patch.object(module, 'latest_route', return_value=latest), + patch.object(module, 'collect_route_inputs', new=collect), + patch('openpilot.starpilot.galaxy.drive_history.DriveHistory', return_value=history), + ): + route, collected = module.collect_latest_steering_inputs(None, '.', permitted=lambda: True) + self.assertIs(route, earlier) + self.assertIs(collected, driving) + self.assertEqual(calls, [latest['routeId'], earlier['routeId']]) + calls.clear() + parked.counts['active_lateral_request_rows'] = 1 + route, _ = module.collect_latest_steering_inputs(None, '.', permitted=lambda: True) + self.assertIs(route, latest, 'A failed or weak active drive must never cause a search for favorable older results') + self.assertEqual(calls, [latest['routeId']]) + calls.clear() + parked.counts['active_lateral_request_rows'] = 0 + driving.receipt.cp_sha256 = 'b' * 64 + route, _ = module.collect_latest_steering_inputs(None, '.', permitted=lambda: True) + self.assertIs(route, latest, 'An older car cannot be substituted') + + def test_selected_previous_drive_keeps_both_source_inventories_guarded(self): + from types import SimpleNamespace + from openpilot.starpilot.flm import offroad_worker as module + from openpilot.starpilot.flm.offroad_processor import CompletedAssessment + from openpilot.starpilot.flm.offroad_route import RouteReceipt + + parked = {'routeId': '00000012--aaaaaaaaaa'} + selected = {'routeId': '00000011--bbbbbbbbbb'} + cp = SimpleNamespace(carFingerprint='declared-unit-only') + receipt = RouteReceipt(selected['routeId'], 'd' * 64, (), 'a' * 64, 1, 2, 3, ()) + assessment = CompletedAssessment(receipt, (), {}, cp, (), selected) + + def assessed(params, root, previous, *, permitted): + previous(cp) + return assessment + + for intact in (True, False): + with self.subTest(selected_source_intact=intact), tempfile.TemporaryDirectory() as directory: + history = SimpleNamespace(snapshot=lambda intact=intact: {'scanIncomplete': False, 'routes': [parked, selected] if intact else [parked]}) + with ( + patch.object(module, 'latest_route', return_value=parked), + patch('openpilot.starpilot.galaxy.drive_history.DriveHistory', return_value=history), + ): + params = Params(directory) + worker = module.OffroadWorker( + params, directory, Path(directory) / 'cache', permitted=lambda: True, assessor=assessed, scope_for=lambda cp, params: scope() + ) + if intact: + result = worker.run_once() + self.assertEqual(result['completed_route'], selected['routeId']) + self.assertFalse(result['configuration_changed']) + else: + with self.assertRaisesRegex(ValueError, 'Source or settings changed'): + worker.run_once() + self.assertEqual(read_saved(params, KEY, MAX_BYTES), (None, True)) + class TestOffroadSurface(unittest.TestCase): def test_real_controller_route_replay_preserves_limiter_and_requires_exact_hooks(self): diff --git a/prebuilt.json b/prebuilt.json index 345bdaf278..7d833c67c7 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "89a56ed49aa9b917f6284ab4d50ebeb52b659220a49bd13ec2f0b90a49ae0299", "schema_version": 1} +{"protected_inventory_sha256": "fb95974ed48cf6ef0178e5090be20cde2e4a10104fd9d72abcbf71c493fb3c0f", "schema_version": 1}