377 lines
16 KiB
Python
377 lines
16 KiB
Python
"""Temporal authority, source identity and fresh-data fences for initialization."""
|
|
|
|
from dataclasses import replace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from k1link.missions.stationary_bootstrap import StationaryBootstrap
|
|
from k1link.missions.stationary_entry import STATIONARY_POLICY
|
|
from k1link.sessions.live_planning import PlanningLiveEvent
|
|
|
|
PATH = np.array([[0.0, 0, 0], [40.0, 0, 0]])
|
|
POINTS = np.random.default_rng(36).uniform([-4, -4, -1], [4, 4, 3], (1800, 3))
|
|
|
|
|
|
def event(t, seq, *, kind="pose", **kw):
|
|
return PlanningLiveEvent(
|
|
"B",
|
|
1,
|
|
seq,
|
|
int((t + 1) * 1e9),
|
|
1,
|
|
kind,
|
|
points=POINTS if kind == "points" else None,
|
|
position=(0.0, 0, 0),
|
|
**kw,
|
|
)
|
|
|
|
|
|
def fit():
|
|
return dict(status="candidate", T_reference_query=np.eye(4).tolist())
|
|
|
|
|
|
def initialized():
|
|
boot = StationaryBootstrap(PATH)
|
|
for i in range(20):
|
|
boot.ingest(event(i * 0.5, i * 2 + 1), 0)
|
|
boot.ingest(event(i * 0.5 + 0.001, i * 2 + 2, kind="points"), 0)
|
|
assert boot.start_search(10_999_999_999) is None
|
|
assert boot.start_search(11_000_000_000) is not None
|
|
result = dict(
|
|
**fit(), initialization=dict(complete=True, policy=STATIONARY_POLICY, attempts=[{}] * 108)
|
|
)
|
|
return boot, result
|
|
|
|
|
|
def long_route_search():
|
|
from k1link.missions.route_relocalization import ROUTE_RELOCALIZATION_POLICY
|
|
boot, result = initialized()
|
|
boot.initialization_policy = ROUTE_RELOCALIZATION_POLICY
|
|
result["initialization"].update(policy=ROUTE_RELOCALIZATION_POLICY,
|
|
scope="selected-route", expected_attempts=108)
|
|
for i in range(20, 200):
|
|
boot.ingest(event(i * .5, i * 2 + 1), 0)
|
|
boot.ingest(event(i * .5 + .001, i * 2 + 2, kind="points"), 0)
|
|
boot.tick(int((i * .5 + 1.01) * 1e9), 0)
|
|
assert boot.phase == "searching"
|
|
return boot, result
|
|
|
|
|
|
def test_long_complete_search_is_only_a_hypothesis_until_three_new_windows():
|
|
boot, result = long_route_search()
|
|
ready = 101_000_000_000
|
|
decision = boot.offer_prior(result, ready, 0)
|
|
assert decision["age_s"] > 90
|
|
assert decision["provisional"] and decision["stationary_search_continuity"]
|
|
assert not decision["accepted"] and boot.gate.matrix is None
|
|
seen = set()
|
|
for i, start in enumerate((101., 106., 111.)):
|
|
sample, _ = fresh_window(boot, start, 500 + i * 20, segment=0)
|
|
ids = {e["sequence"] for e in sample["events"]}
|
|
assert not ids & seen
|
|
seen |= ids
|
|
assert all(e["monotonic_ns"] > ready for e in sample["events"])
|
|
decision = boot.accept_fresh(fit(), sample, int((start + 4.2) * 1e9), 0)
|
|
assert decision["accepted"]
|
|
assert boot.phase == ("tracking" if i == 2 else "validating")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"failure", ["motion", "gap", "pose-only", "cloud-only", "silent", "identity"]
|
|
)
|
|
def test_long_search_cannot_survive_motion_or_invalid_live_receipts(failure):
|
|
boot, result = long_route_search()
|
|
now = 111_000_000_000
|
|
if failure == "motion":
|
|
boot.ingest(replace(event(100., 500), position=(.11, 0., 0.)), 0)
|
|
elif failure == "gap":
|
|
boot.ingest(event(100., 500), 1)
|
|
elif failure == "identity":
|
|
with pytest.raises(ValueError):
|
|
boot.ingest(replace(event(100., 500), generation=2), 0)
|
|
elif failure in {"pose-only", "cloud-only"}:
|
|
boot.ingest(event(109., 500, kind="pose" if failure == "pose-only" else "points"), 0)
|
|
boot.tick(now, boot.segment)
|
|
decision = boot.offer_prior(result, now, boot.segment)
|
|
assert boot.phase == "lost" and not decision["provisional"]
|
|
assert boot.gate.matrix is None
|
|
|
|
|
|
def test_long_search_can_try_next_place_only_with_another_fresh_window():
|
|
boot, result = long_route_search()
|
|
alternative = np.eye(4)
|
|
alternative[0, 3] = 40
|
|
result["initialization"]["candidate_queue"] = [
|
|
dict(candidate_index=1, T_reference_query=np.eye(4).tolist(), ambiguous=False),
|
|
dict(candidate_index=2, T_reference_query=alternative.tolist(), ambiguous=False),
|
|
]
|
|
assert boot.offer_prior(result, 101_000_000_000, 0)["provisional"]
|
|
sample, _ = fresh_window(boot, 101., 500, segment=0)
|
|
decision = boot.accept_fresh(dict(status="rejected"), sample, 105_200_000_000, 0)
|
|
assert not decision["accepted"] and decision["next_candidate_trial"] == 2
|
|
assert boot.gate.matrix is None and boot.floor_ns == 105_200_000_000
|
|
assert np.allclose(boot.prior, alternative)
|
|
assert boot.validation(106_000_000_000, 0) is None
|
|
|
|
|
|
def fresh_window(boot, start, seq, segment=1):
|
|
for i in range(7):
|
|
boot.ingest(event(start + i * 0.5, seq + i * 2), segment)
|
|
boot.ingest(event(start + i * 0.5 + 0.001, seq + i * 2 + 1, kind="points"), segment)
|
|
return boot.validation(int((start + 4.1) * 1e9), 20.0)
|
|
|
|
|
|
def test_slow_prior_never_counts_and_three_disjoint_fresh_windows_are_required():
|
|
boot, result = initialized()
|
|
ready = 36_000_000_000
|
|
temporal = boot.offer_prior(result, ready, 1)
|
|
assert temporal["provisional"] and not temporal["accepted"]
|
|
assert temporal["age_s"] > 8 and temporal["source_segment"] == 0
|
|
assert boot.gate.matrix is None and boot.gate.streak == 0
|
|
seen = set()
|
|
for i, start in enumerate((36.0, 41.0, 46.0)):
|
|
sample, seed = fresh_window(boot, start, 100 + 20 * i)
|
|
assert np.allclose(seed, np.eye(4))
|
|
ids = {e["sequence"] for e in sample["events"]}
|
|
assert not ids & seen
|
|
seen |= ids
|
|
assert all(e["monotonic_ns"] > ready for e in sample["events"])
|
|
stamp = sample["monotonic_ns"]
|
|
temporal = boot.accept_fresh(fit(), sample, int((start + 4.2) * 1e9), 1)
|
|
assert temporal["accepted"] and boot.gate.sample_ns == stamp
|
|
assert boot.gate.streak == i + 1
|
|
assert boot.phase == ("tracking" if i == 2 else "validating")
|
|
assert boot.tracking_established
|
|
boot.tick(boot.gate.sample_ns + 8_000_000_001, 1)
|
|
assert boot.phase == "lost" and boot.gate.matrix is None and boot.prior is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"change,reason",
|
|
[
|
|
("rejected", "initialization-rejected"),
|
|
("partial", "initialization-incomplete"),
|
|
("wrong-policy", "initialization-incomplete"),
|
|
("late", "initialization-expired"),
|
|
("gaps", "too-many-receipt-gaps"),
|
|
],
|
|
)
|
|
def test_bad_initialization_never_becomes_prior(change, reason):
|
|
boot, result = initialized()
|
|
now, segment = 36_000_000_000, 1
|
|
if change == "rejected":
|
|
result["status"] = "rejected"
|
|
elif change == "partial":
|
|
result["initialization"]["attempts"] = [{}] * 107
|
|
elif change == "wrong-policy":
|
|
result["initialization"]["policy"] = {**STATIONARY_POLICY, "version": "other"}
|
|
elif change == "late":
|
|
now = 41_000_000_001
|
|
else:
|
|
segment = 2
|
|
temporal = boot.offer_prior(result, now, segment)
|
|
assert temporal["reason"] == reason and not temporal["provisional"]
|
|
assert boot.gate.matrix is None and boot.prior is None
|
|
|
|
|
|
def test_receipts_from_before_ready_cannot_validate_and_silence_expires_prior():
|
|
boot, result = initialized()
|
|
boot.offer_prior(result, 36_000_000_000, 0)
|
|
for i in range(5):
|
|
boot.ingest(event(34 + i * 0.1, 100 + i * 2), 0)
|
|
boot.ingest(event(34 + i * 0.1 + 0.001, 101 + i * 2, kind="points"), 0)
|
|
assert boot.validation(40_000_000_000, 0) is None
|
|
assert boot.fresh.events == [] or not boot.fresh.events
|
|
boot.tick(46_000_000_001, 0)
|
|
assert boot.reason == "prior-expired" and boot.gate.matrix is None
|
|
assert not boot.offer_prior(result, 47_000_000_000, 0)["provisional"]
|
|
|
|
|
|
def test_fresh_rejection_consumes_prior_without_automatic_retry():
|
|
boot, result = initialized()
|
|
boot.offer_prior(result, 36_000_000_000, 1)
|
|
sample, _ = fresh_window(boot, 36.0, 100)
|
|
assert boot.prior is None
|
|
temporal = boot.accept_fresh(dict(status="rejected"), sample, 40_200_000_000, 1)
|
|
assert not temporal["accepted"] and boot.phase == "lost"
|
|
assert boot.validation(41_000_000_000, 20) is None
|
|
assert boot.start_search(41_000_000_000) is None
|
|
|
|
|
|
def test_gap_while_fresh_fit_runs_or_input_end_cannot_restore_tracking():
|
|
for reason in ("receipt-gap", "input-ended"):
|
|
boot, result = initialized()
|
|
boot.offer_prior(result, 36_000_000_000, 1)
|
|
sample, _ = fresh_window(boot, 36.0, 100)
|
|
if reason == "receipt-gap":
|
|
boot.ingest(event(40.0, 120), 2)
|
|
else:
|
|
boot.stop(reason)
|
|
temporal = boot.accept_fresh(fit(), sample, 41_100_000_000, boot.segment)
|
|
assert not temporal["accepted"] and temporal["reason"] == reason
|
|
assert boot.gate.matrix is None
|
|
|
|
|
|
def test_identity_and_clock_regression_stop_the_chain():
|
|
boot, _ = initialized()
|
|
changed = replace(event(11.0, 100), generation=2)
|
|
with pytest.raises(ValueError, match="generation"):
|
|
boot.ingest(changed, 0)
|
|
assert boot.reason == "identity-changed"
|
|
boot, _ = initialized()
|
|
with pytest.raises(ValueError, match="regressed"):
|
|
boot.ingest(event(8.0, 100), 0)
|
|
assert boot.reason == "source-order-changed"
|
|
|
|
|
|
def test_fresh_data_fence_is_checked_again_at_completion():
|
|
boot, result = initialized()
|
|
boot.offer_prior(result, 36_000_000_000, 1)
|
|
sample, _ = fresh_window(boot, 36.0, 100)
|
|
sample["events"][0]["monotonic_ns"] = sample["fresh_floor_ns"]
|
|
assert not boot.accept_fresh(fit(), sample, 40_200_000_000, 1)["accepted"]
|
|
assert boot.reason == "pre-validation-data" and boot.gate.matrix is None
|
|
|
|
|
|
def test_initial_failure_and_later_loss_have_distinct_operator_messages():
|
|
from k1link.missions.stationary_live import phase_message
|
|
boot, result = initialized()
|
|
result.update(status="rejected", initialization={"reason": "incomplete-search"})
|
|
rejected = boot.offer_prior(result, 36_000_000_000, 0)
|
|
assert rejected["reason"] == "initialization-incomplete"
|
|
assert not boot.tracking_established
|
|
assert "Поиск не завершён" in phase_message(boot)
|
|
assert "Переинициализировать" in phase_message(boot)
|
|
assert "Остановите устройство" not in phase_message(boot)
|
|
assert "потеряна" not in phase_message(boot)
|
|
boot.tracking_established = True
|
|
boot.stop("stale")
|
|
assert "Привязка потеряна" in phase_message(boot)
|
|
|
|
|
|
def test_initial_geometric_rejection_tells_operator_that_route_did_not_sync():
|
|
from k1link.missions.stationary_live import phase_message
|
|
|
|
boot, result = initialized()
|
|
result.update(status="rejected", initialization={"reason": "no-admissible-entry"})
|
|
rejected = boot.offer_prior(result, 36_000_000_000, 0)
|
|
|
|
assert rejected["reason"] == "initialization-rejected"
|
|
message = phase_message(boot)
|
|
assert "Синхронизация маршрута не выполнена" in message
|
|
assert "Переинициализировать" in message
|
|
assert "исследованного участка" in message
|
|
|
|
|
|
def test_gap_straddling_ready_can_finish_before_first_fresh_cloud_only():
|
|
boot, result = initialized()
|
|
ready = 30_000_000_000
|
|
assert boot.offer_prior(result, ready, 0)["provisional"]
|
|
# First new receipt arrives after the pre-existing gap. Readiness has never
|
|
# been granted, no query points have entered validation, and the one-shot
|
|
# prior lifetime/fresh fence are not extended.
|
|
boot.ingest(event(30, 100), 1)
|
|
assert boot.phase == "refreshing" and boot.ready_ns == ready
|
|
assert boot.floor_ns == ready and boot.gate.matrix is None
|
|
sample, _ = fresh_window(boot, 31, 101, segment=1)
|
|
assert sample["segment"] == 1
|
|
assert all(e["monotonic_ns"] > ready for e in sample["events"])
|
|
assert boot.accept_fresh(fit(), sample, 35_200_000_000, 1)["accepted"]
|
|
assert boot.gate.streak == 1 and not boot.tracking_established
|
|
boot.ingest(event(37, 150), 2)
|
|
assert boot.phase == "lost" and boot.prior is None
|
|
|
|
|
|
@pytest.mark.parametrize("case", ["second-gap", "fresh-cloud", "expired"])
|
|
def test_pre_validation_gap_never_extends_or_reuses_a_prior(case):
|
|
boot, result = initialized()
|
|
boot.offer_prior(result, 30_000_000_000, 0)
|
|
first_segment = 0 if case == "fresh-cloud" else 1
|
|
boot.ingest(event(30, 100), first_segment)
|
|
if case == "expired":
|
|
boot.tick(40_000_000_001, 1)
|
|
assert boot.reason == "prior-expired"
|
|
else:
|
|
if case == "fresh-cloud":
|
|
boot.ingest(event(30.001, 101, kind="points"), first_segment)
|
|
boot.ingest(event(33, 102), first_segment + 1)
|
|
assert boot.reason == "receipt-gap"
|
|
assert boot.prior is None and boot.gate.matrix is None
|
|
|
|
|
|
def queued_prior():
|
|
boot, result = initialized()
|
|
matrix = np.eye(4)
|
|
matrix[0, 3] = 10
|
|
result["initialization"]["candidate_queue"] = [
|
|
dict(T_reference_query=np.eye(4).tolist(), candidate_index=0, ambiguous=False),
|
|
dict(T_reference_query=matrix.tolist(), candidate_index=2, ambiguous=False),
|
|
]
|
|
return boot, result, matrix
|
|
|
|
|
|
def test_next_distinct_candidate_requires_three_new_disjoint_windows():
|
|
boot, result, matrix = queued_prior()
|
|
boot.offer_prior(result, 25_000_000_000, 1)
|
|
sample, _ = fresh_window(boot, 25.0, 100)
|
|
rejected_ids = {e["sequence"] for e in sample["events"]}
|
|
rejected = boot.accept_fresh(dict(status="rejected"), sample, 29_200_000_000, 1)
|
|
assert not rejected["accepted"] and rejected["next_candidate_trial"] == 2
|
|
assert boot.phase == "refreshing" and boot.gate.matrix is None
|
|
assert boot.floor_ns == 29_200_000_000 and not boot.fresh.events
|
|
assert boot.candidate_trial == 2 and not boot.tracking_established
|
|
for i, start in enumerate((30.0, 35.0, 40.0)):
|
|
fresh, seed = fresh_window(boot, start, 200 + i * 20)
|
|
assert np.allclose(seed, matrix)
|
|
assert not rejected_ids.intersection(e["sequence"] for e in fresh["events"])
|
|
assert all(e["monotonic_ns"] > 29_200_000_000 for e in fresh["events"])
|
|
boot.accept_fresh(dict(status="candidate", T_reference_query=matrix.tolist()),
|
|
fresh, int((start + 4.2) * 1e9), 1)
|
|
assert boot.gate.streak == i + 1
|
|
assert boot.phase == "tracking" and not boot.candidate_queue
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["expired", "gap", "ended", "ambiguous", "stale-result"])
|
|
def test_candidate_queue_never_bypasses_freshness_identity_or_ambiguity(failure):
|
|
boot, result, _ = queued_prior()
|
|
ready = 45_000_000_000 if failure == "expired" else 25_000_000_000
|
|
if failure == "ambiguous":
|
|
result["initialization"]["candidate_queue"][1]["ambiguous"] = True
|
|
# The legacy local protocol still has its 30 s search / 40 s source fences.
|
|
if failure == "expired":
|
|
boot.search_started_ns += 5_000_000_000
|
|
assert boot.offer_prior(result, ready, 1)["provisional"]
|
|
start = ready / 1e9
|
|
sample, _ = fresh_window(boot, start, 100)
|
|
now = int((start + 4.2) * 1e9)
|
|
if failure == "expired":
|
|
now = 51_000_000_000
|
|
elif failure == "gap":
|
|
boot.ingest(event(start + 4, 120), 2)
|
|
elif failure == "ended":
|
|
boot.stop("input-ended")
|
|
elif failure == "stale-result":
|
|
now = sample["monotonic_ns"] + 9_000_000_000
|
|
rejected = boot.accept_fresh(dict(status="rejected"), sample, now, boot.segment)
|
|
assert not rejected["accepted"] and boot.phase == "lost"
|
|
assert boot.prior is None and boot.gate.matrix is None and not boot.candidate_queue
|
|
|
|
|
|
def test_dense_start_fresh_rejection_requests_new_prefix_then_route_search():
|
|
boot, result = initialized()
|
|
result["initialization"]["stages"] = [dict(name="dense-start")]
|
|
boot.offer_prior(result, 25_000_000_000, 1)
|
|
sample, _ = fresh_window(boot, 25.0, 100)
|
|
boot.accept_fresh(dict(status="rejected"), sample, 29_200_000_000, 1)
|
|
assert boot.phase == "lost" and boot.retry_route_search
|
|
boot.stop("input-ended")
|
|
assert not boot.retry_route_search
|
|
|
|
|
|
def test_ambiguous_first_hypothesis_never_becomes_a_provisional_prior():
|
|
boot, result, _ = queued_prior()
|
|
result["initialization"]["candidate_queue"][0]["ambiguous"] = True
|
|
assert not boot.offer_prior(result, 25_000_000_000, 1)["provisional"]
|
|
assert boot.reason == "initialization-ambiguous" and boot.prior is None
|