From 261e2ca45d5800a77570666337a0491b5db9e220 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Mon, 26 Oct 2026 13:31:42 -0500 Subject: [PATCH] Preserve completed FLM windows during inline probe input gaps --- .../starpilot/flm/live_tracking_actor.py | 13 ++- .../flm/tests/test_inline_probe_retention.py | 86 +++++++++++++++++++ prebuilt.json | 2 +- 3 files changed, 99 insertions(+), 2 deletions(-) create mode 100644 openpilot/starpilot/flm/tests/test_inline_probe_retention.py diff --git a/openpilot/starpilot/flm/live_tracking_actor.py b/openpilot/starpilot/flm/live_tracking_actor.py index ac485b9dfc..2fd8fe38e8 100644 --- a/openpilot/starpilot/flm/live_tracking_actor.py +++ b/openpilot/starpilot/flm/live_tracking_actor.py @@ -730,6 +730,17 @@ class LiveTrackingActor: same_source = collector is not None and self._same_bootstrap_source(source, now_ns) retain = same_source and self.ingress is None retain_probe = same_source and self.ingress is not None and self.certificate is not None and self.probe_token is not None + preparation = self.probe_preparation + # Inline numerical work owns the same original window as an external + # probe. Retain only that matching, uncanceled, unexpired frozen job; + # the interruption below still withdraws every live witness and offer. + retain_preparation = bool( + same_source and self.ingress is not None and self.certificate is not None + and self.prepared_probe_context is not None and preparation.pid is not None + and not preparation.canceled and 0 < preparation.started <= now_ns <= preparation.started + 30_000_000_000 + and preparation.token == (self.token, self.certificate.model_revision, + full_profile_sha256(self.certificate.binding.baseline_revision, self.ingress.optimizer.accepted)) + ) if self.process is not None: result = self.process.poll(token=self.token if retain else None, now_ns=now_ns) if retain and type(result) is dict: @@ -750,7 +761,7 @@ class LiveTrackingActor: self.probe_queue_key = None if self.ingress is not None: self.ingress.interrupt('current_evidence_delayed' if evidence_delay_only else 'current_source_or_native_permission_unavailable', - preserve_completed=retain_probe) + preserve_completed=retain_probe or retain_preparation) return None try: seed = self.binding_for(collector, source) diff --git a/openpilot/starpilot/flm/tests/test_inline_probe_retention.py b/openpilot/starpilot/flm/tests/test_inline_probe_retention.py new file mode 100644 index 0000000000..a80589b928 --- /dev/null +++ b/openpilot/starpilot/flm/tests/test_inline_probe_retention.py @@ -0,0 +1,86 @@ +"""Completed-number/worker ownership regression; no native or road proof. + +The established numerical fixture supplies synthetic endpoints and derivatives. +Actual actor/ingress/window ownership, source expiry and withdrawal execute. +""" +from dataclasses import replace +from types import SimpleNamespace +from typing import Any + +import pytest +from openpilot.starpilot.flm.canonical_surface_adapter import NAMES, full_profile_sha256 +from openpilot.starpilot.flm.live_tracking_actor import LiveTrackingActor +from openpilot.starpilot.flm.live_tracking_ingress import LiveTrackingIngress +from openpilot.starpilot.flm.runtime_adapter_registry import SourceState +from openpilot.starpilot.flm.tests.test_live_tracking_episodes import requested_completed_builder, numerical_probe_from_saved + + +def pending_inline(monkeypatch): + builder, group, name, queued = requested_completed_builder(monkeypatch) + ingress = LiveTrackingIngress(builder.binding, builder.certificate, tuple((name, 0.) for name in NAMES), controller_domain='torque') + ingress.episodes = builder + polls = [] + probe = SimpleNamespace(poll=lambda **kwargs: polls.append(kwargs), poll_probe=lambda **kwargs: None, child=None) + actor: Any = LiveTrackingActor(probe_process=probe) + actor.ingress, actor.certificate, actor.seed = ingress, builder.certificate, builder.binding + actor.token = tuple(vars(actor.seed).values()) + token = (actor.token, actor.certificate.model_revision, full_profile_sha256(builder.binding.baseline_revision, ingress.optimizer.accepted)) + actor.prepared_probe_context = {'fixture': 'original numerical ownership only'} + actor.probe_preparation = SimpleNamespace(pid=123, token=token, canceled=False, started=queued) + source = SourceState(True, False, builder.binding.reset_generation, builder.binding.source_epoch, + queued, queued + 500_000_000, builder.binding.cp_sha256, + calibration_generation=builder.binding.calibration_generation, persistent_source_epoch=builder.binding.persistent_source_epoch) + return actor, source, SimpleNamespace(ready=[]), builder, group, name, queued, polls + + +def test_input_gap_retains_owned_inline_results_without_restoring_permission(monkeypatch): + actor, source, collector, builder, group, name, queued, polls = pending_inline(monkeypatch) + saved = next(iter(builder.completed.values())) + episode_id, deadline = saved['episode'].episode_id, saved['deadline'] + results = [numerical_probe_from_saved(builder, saved, point, group, name) for point in saved['points']] + assert actor.probe_token is None # Normal inline execution has no external child. + assert actor.tick(collector, source, now_ns=queued + 1, current_native=False, evidence_delay_only=True) is None + assert episode_id in builder.completed + assert builder.completed[episode_id]['deadline'] == deadline + assert not builder.ticks and not builder.endpoints and not actor.ingress.monitor.rows + assert actor.ingress.latest_offer is None and actor.ingress.optimizer.response_witness is None + assert polls == [{'token': None, 'now_ns': queued + 1}] + for index, result in enumerate(results): + assert actor.ingress.observe_probe(result, now_ns=queued + index + 2) + (episode,) = builder.drain(now_ns=queued + 8) + assert episode.episode_id == episode_id and episode.sensitivity[0].native_endpoints == 4 + assert actor.ingress.latest_offer is None and actor.ingress.optimizer.response_witness is None + assert actor.application is None # Retained numbers never grant an actuator target. + + +@pytest.mark.parametrize('failure', ['missing_context', 'reaped', 'canceled', 'expired', 'future', + 'owner_changed', 'model_changed', 'accepted_changed', 'disabled', 'manual', 'source_expired', 'source_changed']) +def test_unowned_inline_result_cannot_keep_completed_window(monkeypatch, failure): + actor, source, collector, builder, _group, _name, queued, _polls = pending_inline(monkeypatch) + if failure == 'missing_context': + actor.prepared_probe_context = None + elif failure == 'reaped': + actor.probe_preparation.pid = None + elif failure == 'canceled': + actor.probe_preparation.canceled = True + elif failure == 'expired': + actor.probe_preparation.started = queued - 30_000_000_001 + elif failure == 'future': + actor.probe_preparation.started = queued + 2 + elif failure == 'owner_changed': + actor.probe_preparation.token = ('other', *actor.probe_preparation.token[1:]) + elif failure == 'model_changed': + actor.probe_preparation.token = (actor.token, 'other', actor.probe_preparation.token[2]) + elif failure == 'accepted_changed': + actor.ingress.optimizer.accepted[NAMES[0]] = .01 + elif failure == 'disabled': + source = replace(source, enabled=False) + elif failure == 'manual': + source = replace(source, manual=True) + elif failure == 'source_expired': + source = replace(source, expires_ns=queued) + elif failure == 'source_changed': + source = replace(source, source_epoch='9' * 64) + assert actor.tick(collector, source, now_ns=queued + 1, current_native=False) is None + assert not builder.completed and not builder.probe_requests and not builder.ready + assert actor.ingress.latest_offer is None and actor.application is None diff --git a/prebuilt.json b/prebuilt.json index 888e011254..a6e47740a9 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "87c8754c25e75e1625e2bc3be6156686a0e75555551d75c5c2b54e8087a0724a", "schema_version": 1} +{"protected_inventory_sha256": "efd11c6fa09ef1c09fd6c394aec7b482b63101c1009824054635ef9bac18cf4b", "schema_version": 1}