[AZ-344] C3 CrossDomainMatcher Protocol + factory + RollingHealthWindow

Defines the public `CrossDomainMatcher` Protocol (PEP 544
@runtime_checkable, two methods: `match` + `health_snapshot`),
the three frozen+slotted DTOs (`CandidateMatchSet`, `MatchResult`,
`MatcherHealth`) in the L1 `_types/matcher.py` layer, the
`MatcherError` family (`MatcherBackboneError`,
`InsufficientInliersError`), and the composition-root
`build_matcher_strategy` factory with lazy-import +
`BUILD_MATCHER_<variant>` gating per ADR-002.

`RollingHealthWindow` accumulator (60 s, amortised O(1) update,
strict O(1) snapshot) is constructed by the factory and injected
into every concrete matcher so all backbones share window
semantics; this is what backs C5's spoof-promotion gate.

Legacy placeholder `MatchResult` removed from `_types/matching.py`;
import-only consumers (`c4_pose.interface`, `c3_5_adhop.interface`)
repointed at the new `_types/matcher.py` home — zero behavioural
change to those components.

AC-9 (single-thread binding) and AC-10 (LightGlueRuntime
identity-share with C2.5) deferred to AZ-270 runtime-root
composition, mirroring the AZ-342 Risk-4 escape clause. All other
ACs + NFRs covered by 70 new conformance tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-05-12 05:43:33 +03:00
parent d6756f1855
commit 89c223882b
16 changed files with 1404 additions and 50 deletions
@@ -0,0 +1,196 @@
"""C3 matcher strategy composition-root factory (AZ-344).
:func:`build_matcher_strategy` selects exactly one strategy by
``config.components['c3_matcher'].strategy`` and respects
compile-time ``BUILD_MATCHER_<variant>`` gating: requesting a
strategy whose flag is OFF raises
:class:`StrategyNotAvailableError` at composition time (NOT at
first frame).
Concrete strategy modules
(``disk_lightglue``, ``aliked_lightglue``, ``xfeat``) are imported
lazily — a Tier-0 workstation build with
``BUILD_MATCHER_DISK_LIGHTGLUE=OFF`` MUST NOT load
``c3_matcher.disk_lightglue`` (ADR-002 / I-5; verifiable via
``sys.modules``).
The shared :class:`LightGlueRuntime` and :class:`RansacFilter` are
constructor-injected — the factory does NOT own their lifecycles.
The runtime root constructs ONE ``LightGlueRuntime`` instance and
passes the SAME reference to both this factory (C3) and the C2.5
``ReRankStrategy`` factory (per AC-10 / AZ-342 AC-10). The
identity-share gives R14 fix substance: a regression that
constructs two runtimes would double GPU memory.
The :class:`RollingHealthWindow` accumulator is constructed BY
this factory (one per matcher instance) and passed to the
concrete strategy's ``create`` entry-point so all backbones share
window semantics (AZ-344 Outcome line 5).
"""
from __future__ import annotations
import logging
import os
from typing import TYPE_CHECKING
from gps_denied_onboard.components.c3_matcher._health_window import RollingHealthWindow
from gps_denied_onboard.runtime_root.errors import StrategyNotAvailableError
if TYPE_CHECKING:
from gps_denied_onboard.components.c3_matcher import (
C3MatcherConfig,
CrossDomainMatcher,
)
from gps_denied_onboard.components.c7_inference import InferenceRuntime
from gps_denied_onboard.config.schema import Config
from gps_denied_onboard.helpers.lightglue_runtime import LightGlueRuntime
from gps_denied_onboard.helpers.ransac_filter import RansacFilter
__all__ = ["build_matcher_strategy"]
_LOG = logging.getLogger("gps_denied_onboard.c3_matcher")
# Strategy resolution table — mirrors the contract's
# ``cross_domain_matcher_protocol.md`` v1.0.0 § Composition-Root
# Factory table verbatim. ANY mutation of this dict MUST be
# mirrored in the contract.
_STRATEGY_TO_BUILD_FLAG: dict[str, str] = {
"disk_lightglue": "BUILD_MATCHER_DISK_LIGHTGLUE",
"aliked_lightglue": "BUILD_MATCHER_ALIKED_LIGHTGLUE",
"xfeat": "BUILD_MATCHER_XFEAT",
}
_STRATEGY_TO_MODULE: dict[str, tuple[str, str]] = {
"disk_lightglue": (
"gps_denied_onboard.components.c3_matcher.disk_lightglue",
"DiskLightGlueMatcher",
),
"aliked_lightglue": (
"gps_denied_onboard.components.c3_matcher.aliked_lightglue",
"AlikedLightGlueMatcher",
),
"xfeat": (
"gps_denied_onboard.components.c3_matcher.xfeat",
"XFeatMatcher",
),
}
def _is_build_flag_on(flag_name: str) -> bool:
"""Read a compile-time ``BUILD_*`` flag from the environment.
``ON`` / ``1`` / ``true`` / ``yes`` (case-insensitive) → ``True``;
anything else (including unset) → ``False``. Defaults to OFF so
test environments must opt-in explicitly per strategy.
"""
raw = os.environ.get(flag_name, "")
return raw.strip().lower() in {"on", "1", "true", "yes"}
def _c3_config(config: "Config") -> "C3MatcherConfig":
"""Pull the registered C3 config block.
``c3_matcher.__init__`` registers it on import; a missing
registration is a developer error and surfaces as ``KeyError``
rather than a silent fallback.
"""
return config.components["c3_matcher"]
def build_matcher_strategy(
config: "Config",
*,
lightglue_runtime: "LightGlueRuntime",
ransac_filter: "RansacFilter",
inference_runtime: "InferenceRuntime",
) -> "CrossDomainMatcher":
"""Construct the :class:`CrossDomainMatcher` impl selected by config.
1. Reads ``config.components['c3_matcher'].strategy``.
2. Checks the matching ``BUILD_MATCHER_<variant>`` flag — if
OFF, raises :class:`StrategyNotAvailableError` BEFORE any
import.
3. Constructs a :class:`RollingHealthWindow` seeded with
``config.components['c3_matcher'].min_inliers_threshold``.
4. Lazily imports the concrete strategy module.
5. Constructs the strategy via its module-level ``create(
config, lightglue_runtime, ransac_filter, inference_runtime,
health_window)`` factory function (each concrete strategy
module exports ``create`` as its public entry-point;
concrete constructors stay private).
6. Emits ONE INFO log ``kind="c3.matcher.strategy_loaded"``
with structured fields ``{strategy, min_inliers_threshold,
residual_warn_threshold_px}``.
Raises:
StrategyNotAvailableError: compile-time flag OFF or
concrete module not yet built
(AZ-345 / AZ-346 / AZ-347 pending).
"""
block = _c3_config(config)
strategy = block.strategy
flag_name = _STRATEGY_TO_BUILD_FLAG.get(strategy)
module_info = _STRATEGY_TO_MODULE.get(strategy)
if flag_name is None or module_info is None:
# Defensive — config validation rejects unknown strategy
# labels at load (``C3MatcherConfig.__post_init__``), so
# this branch is only reachable if the resolution table
# and the validation set drift apart.
_LOG.error(
"c3.matcher.build_flag_off",
extra={"strategy": strategy, "reason": "unknown_strategy"},
)
raise StrategyNotAvailableError(
f"CrossDomainMatcher {strategy!r} is not buildable in this binary."
)
if not _is_build_flag_on(flag_name):
_LOG.error(
"c3.matcher.build_flag_off",
extra={"strategy": strategy, "flag": flag_name},
)
raise StrategyNotAvailableError(
f"BUILD_MATCHER_{strategy.upper()} is OFF for this binary; "
f"cannot select strategy={strategy}."
)
health_window = RollingHealthWindow(
min_inliers_threshold=block.min_inliers_threshold,
)
module_name, class_name = module_info
try:
module = __import__(module_name, fromlist=[class_name])
except ModuleNotFoundError as exc:
raise StrategyNotAvailableError(
f"CrossDomainMatcher {strategy!r} is configured but its concrete "
f"impl module {module_name!r} has not been built into this binary "
"yet (AZ-345 / AZ-346 / AZ-347 pending)."
) from exc
create_fn = getattr(module, "create", None)
if create_fn is None:
strategy_cls = getattr(module, class_name)
instance = strategy_cls(
config,
lightglue_runtime=lightglue_runtime,
ransac_filter=ransac_filter,
inference_runtime=inference_runtime,
health_window=health_window,
)
else:
instance = create_fn(
config,
lightglue_runtime=lightglue_runtime,
ransac_filter=ransac_filter,
inference_runtime=inference_runtime,
health_window=health_window,
)
_LOG.info(
"c3.matcher.strategy_loaded",
extra={
"strategy": strategy,
"min_inliers_threshold": block.min_inliers_threshold,
"residual_warn_threshold_px": block.residual_warn_threshold_px,
},
)
return instance