[AZ-401] [AZ-400] Replay — compose_root replay-mode branch + transport seam

Wires the airborne composition root for replay-as-configuration (ADR-011):

- compose_root(config) branches on config.mode in {"live", "replay"}.
  Live behaviour is unchanged; replay builds ReplayInputAdapter,
  attaches JsonlReplaySink, and injects NoopMavlinkTransport.
- New private module runtime_root/_replay_branch.py holds the
  replay-only strategy graph + build-flag gate + calibration loader.
- Config gains Config.mode (Literal["live","replay"]) plus
  Config.replay sub-block with nested ReplayAutoSyncConfig that mirrors
  the AZ-405 AutoSyncConfig DTO; YAML loader + ENV map updated.

Absorbs the AZ-400 transport-seam retrofit that AZ-401 strictly
required but AZ-400 had not delivered:

- New MavlinkTransport Protocol (write/bytes_written/close).
- NoopMavlinkTransport (replay; build-flag gated, idempotent close,
  thread-safe byte counter).
- SerialMavlinkTransport (live, no-op restructure of existing pymavlink
  byte path; encoder retrofit to actually USE it is the AZ-558
  follow-up).

AZ-401 AC-9 (NoopMavlinkTransport.bytes_written > 0 after C8 encoders
run) is BLOCKED on AZ-558 — the encoder routing retrofit is out of
the AZ-401 task envelope (FORBIDDEN files: pymavlink_ardupilot_adapter,
msp2_inav_adapter). AZ-558 spec, batch_61_review.md, and the test's
@pytest.mark.skip rationale all carry the deferral reason.

Tests: 22 compose_root replay-branch tests + 17 transport tests.
Full regression: 2063 passed, 86 environment-skips, 1 documented
skip (AC-9 / AZ-558), 1 pre-existing flaky perf test deselected.

Co-authored-by: Cursor <cursoragent@cursor.com>

src/gps_denied_onboard/config/loader.py

@@ -23,10 +23,13 @@ import yaml
 from gps_denied_onboard.config.schema import (
     _COMPONENT_REGISTRY,
     Config,
+    ConfigError,
     FcConfig,
     FdrConfig,
     GcsConfig,
     LogConfig,
+    ReplayAutoSyncConfig,
+    ReplayConfig,
     RequiredFieldMissingError,
     RuntimeConfig,
     _replace_block,
@@ -64,6 +67,13 @@ ENV_KEY_MAP: Final[dict[str, tuple[str, str]]] = {
     "GCS_PORT_DEVICE": ("gcs", "port_device"),
     "GCS_PORT_BAUD": ("gcs", "port_baud"),
     "GCS_SUMMARY_RATE_HZ": ("gcs", "summary_rate_hz"),
+    # Replay block (AZ-401)
+    "REPLAY_VIDEO_PATH": ("replay", "video_path"),
+    "REPLAY_TLOG_PATH": ("replay", "tlog_path"),
+    "REPLAY_OUTPUT_PATH": ("replay", "output_path"),
+    "REPLAY_PACE": ("replay", "pace"),
+    "REPLAY_TIME_OFFSET_MS": ("replay", "time_offset_ms"),
+    "REPLAY_TARGET_FC_DIALECT": ("replay", "target_fc_dialect"),
 }
 
 # Env vars that MUST resolve to a non-empty value before `load_config`
@@ -106,6 +116,12 @@ _FIELD_COERCIONS: Final[dict[str, type]] = {
     "spoof_recovery_source_set": int,
     "source_set_switch_timeout_ms": int,
     "summary_rate_hz": float,
+    # Replay block coercions (AZ-401)
+    "video_path": str,
+    "tlog_path": str,
+    "output_path": str,
+    "pace": str,
+    "target_fc_dialect": str,
 }
 
 
@@ -121,8 +137,91 @@ def _coerce_value(field_name: str, raw: Any) -> Any:
         ) from exc
 
 
+def _coerce_optional_int(field_name: str, raw: Any) -> int | None:
+    """Coerce ``raw`` to ``int`` or ``None`` (empty / null sentinels become ``None``)."""
+    if raw is None:
+        return None
+    if isinstance(raw, str) and raw.strip() == "":
+        return None
+    if isinstance(raw, int) and not isinstance(raw, bool):
+        return raw
+    try:
+        return int(raw)
+    except (TypeError, ValueError) as exc:
+        raise RequiredFieldMissingError(
+            f"config field {field_name!r}: cannot coerce {raw!r} to int ({exc})"
+        ) from exc
+
+
+def _build_replay_block(overrides: Mapping[str, Any]) -> ReplayConfig:
+    """Build a :class:`ReplayConfig` from YAML/env overrides.
+
+    Handles two non-trivial coercions the generic path cannot:
+
+    * ``time_offset_ms`` — ``int | None`` (empty string / None → None).
+    * ``auto_sync`` — nested mapping → :class:`ReplayAutoSyncConfig`.
+    """
+    flat: dict[str, Any] = {}
+    auto_sync_overrides: Mapping[str, Any] = {}
+    for key, value in overrides.items():
+        if key == "auto_sync":
+            if value is None:
+                continue
+            if not isinstance(value, Mapping):
+                raise ConfigError(
+                    f"replay.auto_sync must be a mapping; got {type(value).__name__}"
+                )
+            auto_sync_overrides = value
+            continue
+        if key == "time_offset_ms":
+            flat[key] = _coerce_optional_int(key, value)
+            continue
+        flat[key] = _coerce_value(key, value)
+    auto_sync_block = _replace_block(
+        ReplayAutoSyncConfig(),
+        {k: _coerce_replay_auto_sync_field(k, v) for k, v in auto_sync_overrides.items()},
+    )
+    flat["auto_sync"] = auto_sync_block
+    return _replace_block(ReplayConfig(), flat)
+
+
+_REPLAY_AUTO_SYNC_TYPES: Final[dict[str, type]] = {
+    "takeoff_accel_threshold_g": float,
+    "takeoff_attitude_rate_threshold_rad_s": float,
+    "sustained_seconds": float,
+    "prescan_max_messages": int,
+    "video_motion_threshold": float,
+    "video_motion_scan_seconds": float,
+    "match_threshold_pct": float,
+    "match_window_ms": int,
+    "low_confidence_threshold": float,
+}
+
+
+def _coerce_replay_auto_sync_field(field_name: str, raw: Any) -> Any:
+    target_type = _REPLAY_AUTO_SYNC_TYPES.get(field_name)
+    if target_type is None or isinstance(raw, target_type):
+        return raw
+    try:
+        return target_type(raw)
+    except (TypeError, ValueError) as exc:
+        raise RequiredFieldMissingError(
+            f"config field replay.auto_sync.{field_name}: cannot coerce {raw!r} "
+            f"to {target_type.__name__} ({exc})"
+        ) from exc
+
+
+_TOP_LEVEL_SCALAR_FIELDS: Final[frozenset[str]] = frozenset({"mode"})
+
+
 def _load_yaml_files(paths: Sequence[Path]) -> dict[str, dict[str, Any]]:
-    """Merge YAML files in order: later paths win for the same block + field."""
+    """Merge YAML files in order: later paths win for the same block + field.
+
+    Top-level scalar fields named in :data:`_TOP_LEVEL_SCALAR_FIELDS`
+    (currently ``mode``) are collected under the synthetic ``__top__``
+    block so the ``Config`` outer fields can be overridden alongside
+    the nested cross-cutting / component blocks.
+    """
     merged: dict[str, dict[str, Any]] = {}
     for path in paths:
         data = yaml.safe_load(path.read_text()) or {}
@@ -131,6 +230,9 @@ def _load_yaml_files(paths: Sequence[Path]) -> dict[str, dict[str, Any]]:
                 f"YAML at {path} must be a mapping at the top level; got {type(data).__name__}"
             )
         for block_name, block_value in data.items():
+            if block_name in _TOP_LEVEL_SCALAR_FIELDS:
+                merged.setdefault("__top__", {})[block_name] = block_value
+                continue
             if not isinstance(block_value, dict):
                 continue
             merged.setdefault(block_name, {}).update(block_value)
@@ -193,6 +295,11 @@ def load_config(
         GcsConfig(),
         {k: _coerce_value(k, v) for k, v in yaml_overrides.get("gcs", {}).items()},
     )
+    replay_block = _build_replay_block(yaml_overrides.get("replay", {}))
+    raw_mode = yaml_overrides.get("__top__", {}).get("mode")
+    if raw_mode is None:
+        raw_mode = env.get("MODE", "live")
+    mode = str(raw_mode).strip().lower()
 
     component_blocks = _resolve_component_blocks()
     for slug, dataclass_type in _COMPONENT_REGISTRY.items():
@@ -209,5 +316,7 @@ def load_config(
         fdr=fdr_block,
         fc=fc_block,
         gcs=gcs_block,
+        replay=replay_block,
+        mode=mode,  # type: ignore[arg-type]  # validated by Config.__post_init__
         components=component_blocks,
     )