[AZ-422] Add FT-P-17 + FT-N-06 mid-flight tile blackbox tests

Implement the AC-8.4 and AC-NEW-6 blackbox scenarios for mid-flight
tile generation, dedup, landing-time upload, and freshness gating.

Helpers:
- runner/helpers/mid_flight_tile_evaluator.py — pure-logic evaluators
  for tile generation rate, Mode B Fact #105 schema check, footprint+
  GSD dedup (via geo.distance_m), upload-audit reconciliation, and
  the AC-5/AC-6 capture_utc + freshness-gate checks.
- runner/helpers/mock_suite_sat_audit.py — httpx wrapper for the
  mock-suite-sat-service /tiles/audit endpoint with strict response-
  shape validation.

Scenarios:
- tests/positive/test_ft_p_17_mid_flight_tiles.py
- tests/negative/test_ft_n_06_mid_flight_freshness.py

Both skip when sitl_replay_ready is false and fail loudly when fixture
records are missing (tests-as-gates discipline). 52 new unit tests
(41 evaluator + 11 audit client) cover every helper branch.

Review: PASS_WITH_WARNINGS (2 Low — duplicate haversine carry-over,
upstream production dependency surface).

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

e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py

@@ -0,0 +1,486 @@
+"""Unit tests for ``runner.helpers.mid_flight_tile_evaluator`` (AZ-422).
+
+Pure-logic AC-8.4 / AC-NEW-6 coverage for FT-P-17 / FT-N-06.
+
+The scenarios in ``e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py``
+and ``e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py`` exercise
+the same helpers end-to-end when the SITL fixture is prepared; this
+file covers them in isolation.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from runner.helpers import mid_flight_tile_evaluator as mfe
+
+
+def _full_quality(**overrides: object) -> dict[str, object]:
+    base: dict[str, object] = {
+        "capture_utc": "2026-05-17T11:30:00Z",
+        "source_provider": "operator-supplied",
+        "resolution_m_per_px": 0.4,
+        "cloud_coverage_pct": 5.0,
+        "geo_accuracy_m": 2.0,
+    }
+    base.update(overrides)
+    return base
+
+
+def _tile(
+    tile_id: str = "tile_001",
+    *,
+    bbox: tuple[float, float, float, float] = (36.20, 49.95, 36.21, 49.96),
+    zoom: int = 18,
+    sha: str = "a" * 64,
+    payload_size: int = 1024,
+    quality: dict[str, object] | None = None,
+    generated_at_ms: int = 1_700_000_000_000,
+    capture_utc: str | None = "2026-05-17T11:30:00Z",
+) -> mfe.TileSpec:
+    return mfe.TileSpec(
+        tile_id=tile_id,
+        bbox_wgs84=bbox,
+        zoom_level=zoom,
+        descriptor_sha256=sha,
+        payload_size_bytes=payload_size,
+        quality=quality if quality is not None else _full_quality(capture_utc=capture_utc or "2026-05-17T11:30:00Z"),
+        generated_at_monotonic_ms=generated_at_ms,
+        capture_utc_iso=capture_utc,
+    )
+
+
+# ─────────────────────── bbox_centre ───────────────────────
+
+
+def test_bbox_centre_returns_midpoint() -> None:
+    # Act
+    lat, lon = mfe.bbox_centre((36.0, 50.0, 36.2, 50.2))
+    # Assert
+    assert lat == pytest.approx(50.1)
+    assert lon == pytest.approx(36.1)
+
+
+# ─────────────────────── evaluate_tile_generation_rate ───────────────────────
+
+
+def test_evaluate_tile_generation_rate_one_per_3s_exact_pass() -> None:
+    # Arrange — 10 tiles over 30s = 1 tile / 3s
+    tiles = [_tile(f"t_{i}") for i in range(10)]
+    # Act
+    report = mfe.evaluate_tile_generation_rate(tiles, high_quality_window_s=30.0)
+    # Assert
+    assert report.passes
+    assert report.observed_rate_per_3s == pytest.approx(1.0)
+
+
+def test_evaluate_tile_generation_rate_under_min_fails() -> None:
+    # Arrange — 1 tile over 30s = 0.1 tile / 3s
+    tiles = [_tile("t_0")]
+    # Act
+    report = mfe.evaluate_tile_generation_rate(tiles, high_quality_window_s=30.0)
+    # Assert
+    assert not report.passes
+
+
+def test_evaluate_tile_generation_rate_zero_window_fails() -> None:
+    # Act
+    report = mfe.evaluate_tile_generation_rate([_tile()], high_quality_window_s=0)
+    # Assert
+    assert not report.passes
+
+
+def test_evaluate_tile_generation_rate_invalid_window_per_tile_raises() -> None:
+    with pytest.raises(ValueError, match="window_s_per_tile"):
+        mfe.evaluate_tile_generation_rate([_tile()], 30.0, window_s_per_tile=0)
+
+
+def test_evaluate_tile_generation_rate_empty_tiles_fails() -> None:
+    # Act
+    report = mfe.evaluate_tile_generation_rate([], high_quality_window_s=30.0)
+    # Assert
+    assert not report.passes
+
+
+# ─────────────────────── evaluate_tile_quality_metadata ───────────────────────
+
+
+def test_evaluate_tile_quality_metadata_all_fields_present_passes() -> None:
+    # Act
+    report = mfe.evaluate_tile_quality_metadata([_tile()])
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_tile_quality_metadata_missing_quality_field_fails() -> None:
+    # Arrange
+    q = _full_quality()
+    del q["resolution_m_per_px"]
+    # Act
+    report = mfe.evaluate_tile_quality_metadata([_tile(quality=q)])
+    # Assert
+    assert not report.passes
+    assert report.failing_entries[0].missing_quality_fields == ("resolution_m_per_px",)
+
+
+def test_evaluate_tile_quality_metadata_partial_quality_field_drop_fails() -> None:
+    # Arrange — drop one of the AC-2 Mode B Fact #105 quality fields
+    q = _full_quality()
+    del q["cloud_coverage_pct"]
+    # Act
+    report = mfe.evaluate_tile_quality_metadata([_tile(quality=q)])
+    # Assert
+    assert not report.passes
+    assert "cloud_coverage_pct" in report.failing_entries[0].missing_quality_fields
+
+
+def test_evaluate_tile_quality_metadata_quality_not_dict_fails() -> None:
+    # Arrange
+    tile = mfe.TileSpec(
+        tile_id="bad",
+        bbox_wgs84=(0, 0, 1, 1),
+        zoom_level=18,
+        descriptor_sha256="a" * 64,
+        payload_size_bytes=1,
+        quality={},  # ensure the dataclass holds a dict; we mutate via object.__setattr__ below
+        generated_at_monotonic_ms=0,
+    )
+    object.__setattr__(tile, "quality", None)
+    # Act
+    report = mfe.evaluate_tile_quality_metadata([tile])
+    # Assert
+    assert not report.passes
+    assert set(report.failing_entries[0].missing_quality_fields) == set(mfe.TILE_REQUIRED_QUALITY_FIELDS)
+
+
+def test_evaluate_tile_quality_metadata_empty_list_fails() -> None:
+    # Act
+    report = mfe.evaluate_tile_quality_metadata([])
+    # Assert
+    assert not report.passes
+
+
+def test_evaluate_tile_quality_metadata_null_quality_field_value_fails() -> None:
+    # Arrange
+    q = _full_quality(cloud_coverage_pct=None)
+    # Act
+    report = mfe.evaluate_tile_quality_metadata([_tile(quality=q)])
+    # Assert
+    assert not report.passes
+
+
+# ─────────────────────── evaluate_dedup ───────────────────────
+
+
+def test_evaluate_dedup_two_tiles_same_centre_same_gsd_dupes() -> None:
+    # Arrange — same bbox + identical GSD
+    bbox = (36.20, 49.95, 36.21, 49.96)
+    tiles = [
+        _tile("a", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.5)),
+        _tile("b", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.5)),
+    ]
+    # Act
+    report = mfe.evaluate_dedup(tiles)
+    # Assert
+    assert not report.passes
+    assert report.duplicate_pairs == (("a", "b"),)
+
+
+def test_evaluate_dedup_far_apart_bboxes_pass() -> None:
+    # Arrange — bboxes 1 km apart
+    tiles = [
+        _tile("a", bbox=(36.20, 49.95, 36.21, 49.96)),
+        _tile("b", bbox=(36.30, 49.95, 36.31, 49.96)),
+    ]
+    # Act
+    report = mfe.evaluate_dedup(tiles)
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_dedup_close_centres_different_gsd_pass() -> None:
+    # Arrange — same bbox but very different GSD (0.5 vs 1.0 = 50% delta > 5%)
+    bbox = (36.20, 49.95, 36.21, 49.96)
+    tiles = [
+        _tile("a", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.5)),
+        _tile("b", bbox=bbox, quality=_full_quality(resolution_m_per_px=1.0)),
+    ]
+    # Act
+    report = mfe.evaluate_dedup(tiles)
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_dedup_close_centres_close_gsd_dupes() -> None:
+    # Arrange — same bbox + GSD 0.50 vs 0.51 = 2% delta ≤ 5%
+    bbox = (36.20, 49.95, 36.21, 49.96)
+    tiles = [
+        _tile("a", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.50)),
+        _tile("b", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.51)),
+    ]
+    # Act
+    report = mfe.evaluate_dedup(tiles)
+    # Assert
+    assert not report.passes
+
+
+def test_evaluate_dedup_missing_gsd_skips_pair() -> None:
+    # Arrange — one tile missing resolution_m_per_px → cannot be a duplicate
+    bbox = (36.20, 49.95, 36.21, 49.96)
+    q_no_gsd = _full_quality()
+    del q_no_gsd["resolution_m_per_px"]
+    tiles = [
+        _tile("a", bbox=bbox, quality=q_no_gsd),
+        _tile("b", bbox=bbox),
+    ]
+    # Act
+    report = mfe.evaluate_dedup(tiles)
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_dedup_empty_list_passes() -> None:
+    # Act
+    report = mfe.evaluate_dedup([])
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_dedup_invalid_tolerances_raise() -> None:
+    with pytest.raises(ValueError, match="footprint_tolerance_m"):
+        mfe.evaluate_dedup([_tile()], footprint_tolerance_m=-1)
+    with pytest.raises(ValueError, match="gsd_tolerance_fraction"):
+        mfe.evaluate_dedup([_tile()], gsd_tolerance_fraction=-1)
+
+
+def test_evaluate_dedup_three_tiles_two_pairs() -> None:
+    # Arrange — a, b are dupes; c is far away
+    bbox_close = (36.20, 49.95, 36.21, 49.96)
+    bbox_far = (36.40, 49.95, 36.41, 49.96)
+    tiles = [
+        _tile("a", bbox=bbox_close),
+        _tile("b", bbox=bbox_close),
+        _tile("c", bbox=bbox_far),
+    ]
+    # Act
+    report = mfe.evaluate_dedup(tiles)
+    # Assert
+    assert report.duplicate_pairs == (("a", "b"),)
+
+
+# ─────────────────────── evaluate_upload_acks ───────────────────────
+
+
+def test_evaluate_upload_acks_all_acked_passes() -> None:
+    # Arrange
+    tiles = [_tile("t_a"), _tile("t_b")]
+    audit = [{"tile_id": "t_a"}, {"tile_id": "t_b"}]
+    # Act
+    report = mfe.evaluate_upload_acks(tiles, audit)
+    # Assert
+    assert report.passes
+    assert report.missing_from_audit == ()
+
+
+def test_evaluate_upload_acks_missing_tile_fails() -> None:
+    # Arrange
+    tiles = [_tile("t_a"), _tile("t_b")]
+    audit = [{"tile_id": "t_a"}]
+    # Act
+    report = mfe.evaluate_upload_acks(tiles, audit)
+    # Assert
+    assert not report.passes
+    assert report.missing_from_audit == ("t_b",)
+
+
+def test_evaluate_upload_acks_audit_extra_tiles_ok() -> None:
+    # Arrange — audit may contain stale entries from earlier runs
+    tiles = [_tile("t_a")]
+    audit = [{"tile_id": "t_a"}, {"tile_id": "old_run_tile"}]
+    # Act
+    report = mfe.evaluate_upload_acks(tiles, audit)
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_upload_acks_empty_generated_fails() -> None:
+    # Act
+    report = mfe.evaluate_upload_acks([], [{"tile_id": "x"}])
+    # Assert
+    assert not report.passes
+
+
+def test_evaluate_upload_acks_audit_entry_missing_tile_id_skipped() -> None:
+    # Arrange
+    tiles = [_tile("t_a")]
+    audit = [{"not_tile_id": "garbage"}, {"tile_id": "t_a"}]
+    # Act
+    report = mfe.evaluate_upload_acks(tiles, audit)
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_upload_acks_non_dict_audit_entries_skipped() -> None:
+    # Arrange
+    tiles = [_tile("t_a")]
+    audit = ["not a dict", {"tile_id": "t_a"}]  # type: ignore[list-item]
+    # Act
+    report = mfe.evaluate_upload_acks(tiles, audit)
+    # Assert
+    assert report.passes
+
+
+# ─────────────────────── evaluate_capture_date_freshness ───────────────────────
+
+
+def test_evaluate_capture_date_freshness_zero_drift_passes() -> None:
+    # Arrange — generated_at == 1_700_000_000_000 ms == 1_700_000_000 s == 2023-11-14T22:13:20Z
+    capture = "2023-11-14T22:13:20Z"
+    tile = _tile(
+        capture_utc=capture, generated_at_ms=1_700_000_000_000
+    )
+    # Act
+    report = mfe.evaluate_capture_date_freshness([tile])
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_capture_date_freshness_within_tolerance_passes() -> None:
+    # Arrange — capture 30s before generation
+    tile = _tile(
+        capture_utc="2023-11-14T22:12:50Z", generated_at_ms=1_700_000_000_000
+    )
+    # Act
+    report = mfe.evaluate_capture_date_freshness([tile])
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_capture_date_freshness_over_tolerance_fails() -> None:
+    # Arrange — capture 120s before generation
+    tile = _tile(
+        capture_utc="2023-11-14T22:11:20Z", generated_at_ms=1_700_000_000_000
+    )
+    # Act
+    report = mfe.evaluate_capture_date_freshness([tile])
+    # Assert
+    assert not report.passes
+
+
+def test_evaluate_capture_date_freshness_unparseable_capture_fails() -> None:
+    # Arrange
+    tile = _tile(capture_utc="not-a-timestamp")
+    # Act
+    report = mfe.evaluate_capture_date_freshness([tile])
+    # Assert
+    assert not report.passes
+    assert report.entries[0].drift_s is None
+
+
+def test_evaluate_capture_date_freshness_missing_capture_falls_back_to_quality_dict() -> None:
+    # Arrange — capture_utc_iso None but quality dict carries the field
+    tile = _tile(capture_utc=None)
+    # Act
+    report = mfe.evaluate_capture_date_freshness([tile])
+    # Assert
+    # The quality dict's "capture_utc" is 2026-05-17T11:30:00Z; generated_at is 2023-11-14
+    # so drift is huge — should fail
+    assert not report.passes
+    assert report.entries[0].drift_s is not None
+
+
+def test_evaluate_capture_date_freshness_custom_tolerance() -> None:
+    # Arrange — capture 120s before; widen tolerance to 200s
+    tile = _tile(
+        capture_utc="2023-11-14T22:11:20Z", generated_at_ms=1_700_000_000_000
+    )
+    # Act
+    report = mfe.evaluate_capture_date_freshness([tile], tolerance_s=200.0)
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_capture_date_freshness_invalid_tolerance_raises() -> None:
+    with pytest.raises(ValueError, match="tolerance_s"):
+        mfe.evaluate_capture_date_freshness([_tile()], tolerance_s=0)
+
+
+def test_evaluate_capture_date_freshness_empty_list_fails() -> None:
+    # Act
+    report = mfe.evaluate_capture_date_freshness([])
+    # Assert
+    assert not report.passes
+
+
+# ─────────────────────── evaluate_freshness_gate ───────────────────────
+
+
+def test_evaluate_freshness_gate_no_rejections_passes() -> None:
+    # Arrange
+    tiles = [_tile("t_a"), _tile("t_b")]
+    # Act
+    report = mfe.evaluate_freshness_gate(tiles, [])
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_freshness_gate_unrelated_rejection_passes() -> None:
+    # Arrange — rejection for some other tile
+    tiles = [_tile("t_a")]
+    rejections = [{"id": "old_tile", "reason": "stale"}]
+    # Act
+    report = mfe.evaluate_freshness_gate(tiles, rejections)
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_freshness_gate_fresh_tile_rejected_stale_fails() -> None:
+    # Arrange
+    tiles = [_tile("t_a")]
+    rejections = [{"id": "t_a", "reason": "stale"}]
+    # Act
+    report = mfe.evaluate_freshness_gate(tiles, rejections)
+    # Assert
+    assert not report.passes
+    assert report.stale_rejections == ("t_a",)
+
+
+def test_evaluate_freshness_gate_non_stale_reason_ignored() -> None:
+    # Arrange
+    tiles = [_tile("t_a")]
+    rejections = [{"id": "t_a", "reason": "below_floor"}]
+    # Act
+    report = mfe.evaluate_freshness_gate(tiles, rejections)
+    # Assert
+    assert report.passes
+
+
+def test_evaluate_freshness_gate_tile_id_key_variant() -> None:
+    # Arrange — some rejection records use "tile_id" instead of "id"
+    tiles = [_tile("t_a")]
+    rejections = [{"tile_id": "t_a", "reason": "stale"}]
+    # Act
+    report = mfe.evaluate_freshness_gate(tiles, rejections)
+    # Assert
+    assert not report.passes
+
+
+def test_evaluate_freshness_gate_non_dict_payload_skipped() -> None:
+    # Arrange
+    tiles = [_tile("t_a")]
+    rejections = ["not a dict", {"id": "t_a", "reason": "stale"}]  # type: ignore[list-item]
+    # Act
+    report = mfe.evaluate_freshness_gate(tiles, rejections)
+    # Assert
+    assert not report.passes
+    assert report.stale_rejections == ("t_a",)
+
+
+def test_evaluate_freshness_gate_custom_stale_reason() -> None:
+    # Arrange
+    tiles = [_tile("t_a")]
+    rejections = [{"id": "t_a", "reason": "expired_freshness"}]
+    # Act
+    report = mfe.evaluate_freshness_gate(tiles, rejections, stale_reason="expired_freshness")
+    # Assert
+    assert not report.passes

e2e/_unit_tests/helpers/test_mock_suite_sat_audit.py

@@ -0,0 +1,185 @@
+"""Unit tests for ``runner.helpers.mock_suite_sat_audit`` (AZ-422)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+import pytest
+
+from runner.helpers import mock_suite_sat_audit
+
+
+def _transport(handler) -> httpx.MockTransport:  # type: ignore[no-untyped-def]
+    return httpx.MockTransport(handler)
+
+
+# ─────────────────────── happy path ───────────────────────
+
+
+def test_fetch_audit_returns_entries_list() -> None:
+    # Arrange
+    captured: dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["url"] = str(request.url)
+        return httpx.Response(
+            200,
+            json={
+                "run_id": "run_xyz",
+                "entries": [
+                    {"tile_id": "t_a", "received_at": 1.0},
+                    {"tile_id": "t_b", "received_at": 2.0},
+                ],
+            },
+        )
+
+    # Act
+    entries = mock_suite_sat_audit.fetch_audit(
+        "http://mock-suite-sat-service:8080",
+        run_id="run_xyz",
+        transport=_transport(handler),
+    )
+    # Assert
+    assert entries == [
+        {"tile_id": "t_a", "received_at": 1.0},
+        {"tile_id": "t_b", "received_at": 2.0},
+    ]
+    assert "run_id=run_xyz" in captured["url"]
+    assert "/tiles/audit" in captured["url"]
+
+
+def test_fetch_audit_empty_entries_list_returned_verbatim() -> None:
+    # Arrange
+    def handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"run_id": "run_xyz", "entries": []})
+
+    # Act
+    entries = mock_suite_sat_audit.fetch_audit(
+        "http://service",
+        run_id="run_xyz",
+        transport=_transport(handler),
+    )
+    # Assert
+    assert entries == []
+
+
+def test_fetch_audit_strips_trailing_slash_in_base_url() -> None:
+    # Arrange
+    captured: dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["url"] = str(request.url)
+        return httpx.Response(200, json={"run_id": "run_xyz", "entries": []})
+
+    # Act
+    mock_suite_sat_audit.fetch_audit(
+        "http://service/",
+        run_id="run_xyz",
+        transport=_transport(handler),
+    )
+    # Assert
+    assert "//tiles/audit" not in captured["url"]
+    assert "/tiles/audit?" in captured["url"]
+
+
+def test_fetch_audit_custom_audit_path() -> None:
+    # Arrange
+    captured: dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["url"] = str(request.url)
+        return httpx.Response(200, json={"run_id": "run_xyz", "entries": []})
+
+    # Act
+    mock_suite_sat_audit.fetch_audit(
+        "http://service",
+        run_id="run_xyz",
+        audit_path="/mock/audit",
+        transport=_transport(handler),
+    )
+    # Assert
+    assert "/mock/audit?" in captured["url"]
+
+
+# ─────────────────────── error paths ───────────────────────
+
+
+def test_fetch_audit_empty_base_url_raises() -> None:
+    with pytest.raises(RuntimeError, match="base_url"):
+        mock_suite_sat_audit.fetch_audit("", run_id="run_xyz")
+
+
+def test_fetch_audit_empty_run_id_raises() -> None:
+    with pytest.raises(RuntimeError, match="run_id"):
+        mock_suite_sat_audit.fetch_audit("http://service", run_id="")
+
+
+def test_fetch_audit_non_2xx_raises() -> None:
+    # Arrange
+    def handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, text="boom")
+
+    # Act / Assert
+    with pytest.raises(RuntimeError, match="HTTP 500"):
+        mock_suite_sat_audit.fetch_audit(
+            "http://service",
+            run_id="run_xyz",
+            transport=_transport(handler),
+        )
+
+
+def test_fetch_audit_non_json_body_raises() -> None:
+    # Arrange
+    def handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, text="<<<not json>>>")
+
+    # Act / Assert
+    with pytest.raises(RuntimeError, match="not valid JSON"):
+        mock_suite_sat_audit.fetch_audit(
+            "http://service",
+            run_id="run_xyz",
+            transport=_transport(handler),
+        )
+
+
+def test_fetch_audit_body_not_object_raises() -> None:
+    # Arrange
+    def handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json=["not", "an", "object"])
+
+    # Act / Assert
+    with pytest.raises(RuntimeError, match="not a JSON object"):
+        mock_suite_sat_audit.fetch_audit(
+            "http://service",
+            run_id="run_xyz",
+            transport=_transport(handler),
+        )
+
+
+def test_fetch_audit_missing_entries_raises() -> None:
+    # Arrange
+    def handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"run_id": "run_xyz"})
+
+    # Act / Assert
+    with pytest.raises(RuntimeError, match="entries"):
+        mock_suite_sat_audit.fetch_audit(
+            "http://service",
+            run_id="run_xyz",
+            transport=_transport(handler),
+        )
+
+
+def test_fetch_audit_entries_not_list_raises() -> None:
+    # Arrange
+    def handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"run_id": "run_xyz", "entries": "stringly"})
+
+    # Act / Assert
+    with pytest.raises(RuntimeError, match="entries"):
+        mock_suite_sat_audit.fetch_audit(
+            "http://service",
+            run_id="run_xyz",
+            transport=_transport(handler),
+        )

e2e/_unit_tests/test_directory_layout.py

@@ -53,6 +53,8 @@ E2E_ROOT = Path(__file__).resolve().parents[1]
         "runner/helpers/ap_contract_evaluator.py",
         "runner/helpers/gcs_telemetry_evaluator.py",
         "runner/helpers/tile_cache_inspector.py",
+        "runner/helpers/mid_flight_tile_evaluator.py",
+        "runner/helpers/mock_suite_sat_audit.py",
         "runner/helpers/cold_start_evaluator.py",
         "runner/helpers/outlier_tolerance_evaluator.py",
         "runner/helpers/outage_request_evaluator.py",
@@ -112,11 +114,13 @@ E2E_ROOT = Path(__file__).resolve().parents[1]
         "tests/positive/test_ft_p_13_gcs_command.py",
         "tests/positive/test_ft_p_15_cache_schema.py",
         "tests/positive/test_ft_p_16_offline_only.py",
+        "tests/positive/test_ft_p_17_mid_flight_tiles.py",
         "tests/positive/test_ft_p_18_no_raw_retention.py",
         "tests/negative/test_ft_n_01_outlier_tolerance.py",
         "tests/negative/test_ft_n_02_sharp_turn_failure.py",
         "tests/negative/test_ft_n_03_outage_reloc.py",
         "tests/negative/test_ft_n_04_blackout_spoof.py",
+        "tests/negative/test_ft_n_06_mid_flight_freshness.py",
     ],
 )
 def test_required_path_exists(relative_path: str) -> None:

e2e/runner/helpers/mid_flight_tile_evaluator.py

@@ -0,0 +1,500 @@
+"""Mid-flight tile generation + freshness evaluators (AZ-422 / FT-P-17 + FT-N-06).
+
+Pure-logic evaluators sourced from the FDR archive (per-tile generation
+records + freshness-gate events) and the mock-suite-sat-service audit
+log (landing-time upload acks).
+
+Sub-scenarios:
+
+* **FT-P-17 / AC-8.4** — five evaluators:
+  * generation cadence (≥ 1 tile / 3 s of high-quality nav frames);
+  * quality-metadata sufficiency (per-tile fields the Service voting
+    layer needs — Mode B Fact #105: capture_utc, source_provider,
+    resolution_m_per_px, cloud_coverage_pct, geo_accuracy_m, plus
+    publish-request fields: tile_id, bbox_wgs84, zoom_level,
+    descriptor_sha256, payload_size_bytes);
+  * dedup (no two tiles share footprint within ±1 m AND GSD within
+    ±5 %);
+  * landing-event upload (every generated tile has an audit entry
+    in the mock-suite-sat-service).
+* **FT-N-06 / AC-NEW-6** — two evaluators:
+  * capture-date freshness (|capture_utc − generated_at| ≤ 60 s);
+  * freshness-gate (no ``tile-load-rejected: stale`` FDR event for a
+    freshly generated tile).
+
+All evaluators consume Python dataclasses / dicts. The HTTP fetch
+and FDR walk live in scenario tests; this module only decides whether
+the parsed inputs satisfy the AC.
+
+Public-boundary discipline: NO imports from ``src/gps_denied_onboard``.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Iterable, Sequence
+
+from .geo import distance_m
+
+# ─────────────────────── FDR record kinds & schema ───────────────────────
+
+MID_FLIGHT_TILE_FDR_KIND = "mid-flight-tile-output"
+TILE_LOAD_REJECTED_FDR_KIND = "tile-load-rejected"
+TILE_LOAD_REJECTED_STALE_REASON = "stale"
+
+MIN_TILES_PER_HIGH_QUALITY_WINDOW_S = 3.0  # ≥ 1 tile per ~3 s of high-quality nav frames
+
+CAPTURE_DATE_FRESHNESS_TOLERANCE_S = 60.0
+
+DEDUP_FOOTPRINT_TOLERANCE_M = 1.0
+DEDUP_GSD_TOLERANCE_FRACTION = 0.05  # ±5 %
+
+# Schema mirror — must stay in sync with ``e2e/fixtures/mock-suite-sat/app.py``
+# ``TilePublishRequest`` + ``TileQualityMetadata``.
+TILE_REQUIRED_TOP_LEVEL_FIELDS: tuple[str, ...] = (
+    "tile_id",
+    "bbox_wgs84",
+    "zoom_level",
+    "descriptor_sha256",
+    "payload_size_bytes",
+    "quality",
+)
+
+TILE_REQUIRED_QUALITY_FIELDS: tuple[str, ...] = (
+    "capture_utc",
+    "source_provider",
+    "resolution_m_per_px",
+    "cloud_coverage_pct",
+    "geo_accuracy_m",
+)
+
+
+@dataclass(frozen=True)
+class TileSpec:
+    """Public-boundary projection of one mid-flight-tile-output record.
+
+    Sourced from the FDR ``mid-flight-tile-output`` record. Mirrors the
+    TilePublishRequest schema so the same dataclass feeds both the
+    landing-event upload comparison (AC-4) and the per-tile evaluators.
+
+    ``bbox_wgs84`` is ``(west_lon, south_lat, east_lon, north_lat)``
+    matching the mock-suite-sat-service contract.
+
+    ``generated_at_monotonic_ms`` is the SUT's emission timestamp from
+    the FDR envelope's ``ts`` (projected to monotonic ms by the FDR
+    reader). ``capture_utc_iso`` is the per-tile field — they should
+    agree within ``CAPTURE_DATE_FRESHNESS_TOLERANCE_S`` (FT-N-06).
+    """
+
+    tile_id: str
+    bbox_wgs84: tuple[float, float, float, float]
+    zoom_level: int
+    descriptor_sha256: str
+    payload_size_bytes: int
+    quality: dict[str, object]
+    generated_at_monotonic_ms: int
+    capture_utc_iso: str | None = None  # convenience accessor; same as quality["capture_utc"]
+
+
+def bbox_centre(bbox: tuple[float, float, float, float]) -> tuple[float, float]:
+    """Return ``(lat, lon)`` of a WGS84 bbox ``(west, south, east, north)``."""
+    west, south, east, north = bbox
+    return ((south + north) / 2.0, (west + east) / 2.0)
+
+
+# ─────────────────────────── FT-P-17 / AC-1 ───────────────────────────
+
+
+@dataclass(frozen=True)
+class TileGenerationRateReport:
+    """AC-1 of FT-P-17: ≥ 1 tile per ~3 s of high-quality nav frames."""
+
+    tile_count: int
+    high_quality_window_s: float
+    observed_rate_per_3s: float
+    min_required_rate_per_3s: float = 1.0
+
+    @property
+    def passes(self) -> bool:
+        if self.high_quality_window_s <= 0:
+            return False
+        return self.observed_rate_per_3s >= self.min_required_rate_per_3s
+
+
+def evaluate_tile_generation_rate(
+    tiles: Sequence[TileSpec],
+    high_quality_window_s: float,
+    *,
+    window_s_per_tile: float = MIN_TILES_PER_HIGH_QUALITY_WINDOW_S,
+) -> TileGenerationRateReport:
+    """AC-1: rate of generated tiles over the high-quality nav-frame window.
+
+    ``high_quality_window_s`` is the total wall-clock seconds during the
+    replay that produced "high-quality" nav frames (defined by AC-2.1a
+    normal-segment in `_docs/02_document/tests/blackbox-tests.md`).
+    The scenario test computes this from the FDR's segment-quality
+    records; the helper only divides.
+
+    The AC threshold is ≥ 1 tile per ``window_s_per_tile`` seconds.
+    Normalised to a "tiles per 3 s" rate so the report is unitless.
+    """
+    if window_s_per_tile <= 0:
+        raise ValueError(f"window_s_per_tile must be > 0, got {window_s_per_tile}")
+    if high_quality_window_s <= 0:
+        return TileGenerationRateReport(
+            tile_count=len(tiles),
+            high_quality_window_s=high_quality_window_s,
+            observed_rate_per_3s=0.0,
+        )
+    rate = (len(tiles) / high_quality_window_s) * window_s_per_tile
+    return TileGenerationRateReport(
+        tile_count=len(tiles),
+        high_quality_window_s=high_quality_window_s,
+        observed_rate_per_3s=rate,
+    )
+
+
+# ─────────────────────────── FT-P-17 / AC-2 ───────────────────────────
+
+
+@dataclass(frozen=True)
+class TileQualityEntryReport:
+    """Per-tile schema-completeness result."""
+
+    tile_id: str
+    missing_top_level_fields: tuple[str, ...]
+    missing_quality_fields: tuple[str, ...]
+
+    @property
+    def passes(self) -> bool:
+        return not self.missing_top_level_fields and not self.missing_quality_fields
+
+
+@dataclass(frozen=True)
+class TileQualityReport:
+    """AC-2 of FT-P-17: every tile carries the Mode B Fact #105 fields."""
+
+    entries: tuple[TileQualityEntryReport, ...]
+
+    @property
+    def failing_entries(self) -> tuple[TileQualityEntryReport, ...]:
+        return tuple(e for e in self.entries if not e.passes)
+
+    @property
+    def passes(self) -> bool:
+        if not self.entries:
+            return False
+        return not self.failing_entries
+
+
+def evaluate_tile_quality_metadata(
+    tiles: Sequence[TileSpec],
+    *,
+    required_top_level: Sequence[str] = TILE_REQUIRED_TOP_LEVEL_FIELDS,
+    required_quality: Sequence[str] = TILE_REQUIRED_QUALITY_FIELDS,
+) -> TileQualityReport:
+    """AC-2: every tile has all top-level + quality fields populated.
+
+    "Populated" means the key is present in the underlying dict
+    representation AND the value is not ``None``. A ``TileSpec``
+    constructed by the scenario test from the FDR record carries
+    these fields as dataclass attributes; this helper still re-checks
+    the quality dict for completeness because the dict mirror is the
+    actual contract with the Service voting layer.
+    """
+    entries: list[TileQualityEntryReport] = []
+    for tile in tiles:
+        missing_top: list[str] = []
+        for f in required_top_level:
+            if f == "quality":
+                continue
+            value = getattr(tile, _top_level_field_to_attr(f), None)
+            if value is None:
+                missing_top.append(f)
+        missing_quality: list[str] = []
+        if not isinstance(tile.quality, dict):
+            missing_quality = list(required_quality)
+        else:
+            for f in required_quality:
+                if f not in tile.quality or tile.quality[f] is None:
+                    missing_quality.append(f)
+        entries.append(
+            TileQualityEntryReport(
+                tile_id=tile.tile_id or "<unknown>",
+                missing_top_level_fields=tuple(missing_top),
+                missing_quality_fields=tuple(missing_quality),
+            )
+        )
+    return TileQualityReport(entries=tuple(entries))
+
+
+def _top_level_field_to_attr(field: str) -> str:
+    """Map TilePublishRequest field name to the TileSpec attribute."""
+    return field  # 1:1 mapping; documented for future drift handling
+
+
+# ─────────────────────────── FT-P-17 / AC-3 ───────────────────────────
+
+
+@dataclass(frozen=True)
+class TileDedupReport:
+    """AC-3 of FT-P-17: no two tiles share a (footprint, GSD) bin."""
+
+    duplicate_pairs: tuple[tuple[str, str], ...]
+    footprint_tolerance_m: float = DEDUP_FOOTPRINT_TOLERANCE_M
+    gsd_tolerance_fraction: float = DEDUP_GSD_TOLERANCE_FRACTION
+
+    @property
+    def duplicate_count(self) -> int:
+        return len(self.duplicate_pairs)
+
+    @property
+    def passes(self) -> bool:
+        return self.duplicate_count == 0
+
+
+def evaluate_dedup(
+    tiles: Sequence[TileSpec],
+    *,
+    footprint_tolerance_m: float = DEDUP_FOOTPRINT_TOLERANCE_M,
+    gsd_tolerance_fraction: float = DEDUP_GSD_TOLERANCE_FRACTION,
+) -> TileDedupReport:
+    """AC-3: pair-wise dedup check.
+
+    Two tiles are duplicates iff:
+      * Vincenty distance between their bbox centres ≤ ``footprint_tolerance_m`` AND
+      * ``|gsd_a − gsd_b| / max(gsd_a, gsd_b) ≤ gsd_tolerance_fraction``
+
+    O(N²) — fine for the < 100 tiles per 5 min replay scenarios produce.
+    Returns the offending ``(tile_id, tile_id)`` pairs.
+    """
+    if footprint_tolerance_m < 0:
+        raise ValueError(f"footprint_tolerance_m must be ≥0, got {footprint_tolerance_m}")
+    if gsd_tolerance_fraction < 0:
+        raise ValueError(
+            f"gsd_tolerance_fraction must be ≥0, got {gsd_tolerance_fraction}"
+        )
+    centres: list[tuple[float, float]] = [bbox_centre(t.bbox_wgs84) for t in tiles]
+    gsds: list[float | None] = [_extract_gsd(t) for t in tiles]
+    pairs: list[tuple[str, str]] = []
+    for i in range(len(tiles)):
+        gsd_i = gsds[i]
+        if gsd_i is None:
+            continue
+        for j in range(i + 1, len(tiles)):
+            gsd_j = gsds[j]
+            if gsd_j is None:
+                continue
+            denom = max(gsd_i, gsd_j)
+            if denom == 0:
+                continue
+            gsd_delta_fraction = abs(gsd_i - gsd_j) / denom
+            if gsd_delta_fraction > gsd_tolerance_fraction:
+                continue
+            d_m = distance_m(
+                centres[i][0], centres[i][1], centres[j][0], centres[j][1]
+            )
+            if d_m <= footprint_tolerance_m:
+                pairs.append((tiles[i].tile_id, tiles[j].tile_id))
+    return TileDedupReport(
+        duplicate_pairs=tuple(pairs),
+        footprint_tolerance_m=footprint_tolerance_m,
+        gsd_tolerance_fraction=gsd_tolerance_fraction,
+    )
+
+
+def _extract_gsd(tile: TileSpec) -> float | None:
+    """Pull GSD (resolution_m_per_px) from the tile's quality dict."""
+    if not isinstance(tile.quality, dict):
+        return None
+    raw = tile.quality.get("resolution_m_per_px")
+    if isinstance(raw, (int, float)):
+        return float(raw)
+    return None
+
+
+# ─────────────────────────── FT-P-17 / AC-4 ───────────────────────────
+
+
+@dataclass(frozen=True)
+class TileUploadAckReport:
+    """AC-4 of FT-P-17: every generated tile uploaded with HTTP 202."""
+
+    generated_tile_ids: tuple[str, ...]
+    audit_tile_ids: tuple[str, ...]
+    missing_from_audit: tuple[str, ...]
+
+    @property
+    def passes(self) -> bool:
+        if not self.generated_tile_ids:
+            return False
+        return not self.missing_from_audit
+
+
+def evaluate_upload_acks(
+    generated_tiles: Sequence[TileSpec],
+    audit_entries: Sequence[dict],
+) -> TileUploadAckReport:
+    """AC-4: every generated tile_id appears in the mock-suite-sat-service audit.
+
+    The mock-suite-sat-service ``POST /tiles`` endpoint records HTTP 202
+    responses to its run-scoped audit log; a tile that did not return
+    202 (i.e., was rejected with 400 or any forced-5xx) is NOT in the
+    audit. So a tile_id present in ``generated_tiles`` but absent from
+    ``audit_entries`` is by construction a missing ack.
+
+    ``audit_entries`` is the ``entries`` field of the JSON response from
+    ``GET /tiles/audit?run_id=<RUN_ID>``.
+    """
+    generated_ids = tuple(t.tile_id for t in generated_tiles)
+    audit_ids = tuple(
+        e["tile_id"] for e in audit_entries if isinstance(e, dict) and "tile_id" in e
+    )
+    audit_id_set = set(audit_ids)
+    missing = tuple(tid for tid in generated_ids if tid not in audit_id_set)
+    return TileUploadAckReport(
+        generated_tile_ids=generated_ids,
+        audit_tile_ids=audit_ids,
+        missing_from_audit=missing,
+    )
+
+
+# ─────────────────────────── FT-N-06 / AC-5 ───────────────────────────
+
+
+@dataclass(frozen=True)
+class CaptureDateFreshnessEntryReport:
+    """Per-tile drift between ``capture_utc`` and ``generated_at``.
+
+    Whether the drift passes the AC threshold is decided at the
+    ``CaptureDateFreshnessReport`` level because the tolerance is a
+    report-wide knob (AC-5 stipulates 60 s globally).
+    """
+
+    tile_id: str
+    drift_s: float | None  # None when capture_utc cannot be parsed
+
+
+@dataclass(frozen=True)
+class CaptureDateFreshnessReport:
+    """AC-5 of FT-N-06: |capture_utc - generated_at_wall_clock| ≤ 60 s."""
+
+    entries: tuple[CaptureDateFreshnessEntryReport, ...]
+    tolerance_s: float = CAPTURE_DATE_FRESHNESS_TOLERANCE_S
+
+    @property
+    def failing_entries(self) -> tuple[CaptureDateFreshnessEntryReport, ...]:
+        return tuple(
+            e for e in self.entries
+            if e.drift_s is None or abs(e.drift_s) > self.tolerance_s
+        )
+
+    @property
+    def passes(self) -> bool:
+        if not self.entries:
+            return False
+        return not self.failing_entries
+
+
+def evaluate_capture_date_freshness(
+    tiles: Sequence[TileSpec],
+    *,
+    tolerance_s: float = CAPTURE_DATE_FRESHNESS_TOLERANCE_S,
+) -> CaptureDateFreshnessReport:
+    """AC-5: per-tile capture_utc drift against generated_at_monotonic_ms.
+
+    Drift is signed: ``capture_utc − generated_at``. A drift of +0 is
+    "capture happened at generation"; negative drift means capture
+    happened BEFORE generation (the usual direction — capture is
+    instantaneous, generation is the orthorectification step that
+    follows).
+
+    A tile whose ``capture_utc`` cannot be parsed as ISO 8601 records
+    drift_s = None and fails the AC.
+    """
+    if tolerance_s <= 0:
+        raise ValueError(f"tolerance_s must be > 0, got {tolerance_s}")
+    entries: list[CaptureDateFreshnessEntryReport] = []
+    for tile in tiles:
+        capture_str = tile.capture_utc_iso
+        if capture_str is None and isinstance(tile.quality, dict):
+            raw = tile.quality.get("capture_utc")
+            if isinstance(raw, str):
+                capture_str = raw
+        drift: float | None
+        if capture_str is None:
+            drift = None
+        else:
+            parsed = _parse_iso8601_utc_seconds(capture_str)
+            if parsed is None:
+                drift = None
+            else:
+                drift = parsed - (tile.generated_at_monotonic_ms / 1000.0)
+        entries.append(
+            CaptureDateFreshnessEntryReport(tile_id=tile.tile_id, drift_s=drift)
+        )
+    return CaptureDateFreshnessReport(
+        entries=tuple(entries), tolerance_s=tolerance_s
+    )
+
+
+def _parse_iso8601_utc_seconds(ts: str) -> float | None:
+    """Parse ISO 8601 ``ts`` into seconds-since-epoch; ``None`` on failure.
+
+    Accepts the trailing ``Z`` shorthand that ``datetime.fromisoformat``
+    did not accept until 3.11.
+    """
+    try:
+        normalised = ts[:-1] + "+00:00" if ts.endswith("Z") else ts
+        dt = datetime.fromisoformat(normalised)
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=timezone.utc)
+        return dt.timestamp()
+    except (TypeError, ValueError):
+        return None
+
+
+# ─────────────────────────── FT-N-06 / AC-6 ───────────────────────────
+
+
+@dataclass(frozen=True)
+class FreshnessGateReport:
+    """AC-6 of FT-N-06: no `tile-load-rejected: stale` for freshly generated tiles."""
+
+    generated_tile_ids: tuple[str, ...]
+    stale_rejections: tuple[str, ...]
+
+    @property
+    def passes(self) -> bool:
+        return not self.stale_rejections
+
+
+def evaluate_freshness_gate(
+    generated_tiles: Sequence[TileSpec],
+    fdr_rejection_records: Iterable[dict],
+    *,
+    stale_reason: str = TILE_LOAD_REJECTED_STALE_REASON,
+) -> FreshnessGateReport:
+    """AC-6: any ``tile-load-rejected: stale`` for a freshly generated tile fails.
+
+    ``fdr_rejection_records`` is the payload dict of each FDR record whose
+    ``record_type == TILE_LOAD_REJECTED_FDR_KIND``. A "stale" rejection
+    sets ``reason == "stale"``. If the rejected tile_id matches a
+    generated tile_id, the freshness gate misclassified it.
+    """
+    generated_ids = tuple(t.tile_id for t in generated_tiles)
+    gen_id_set = set(generated_ids)
+    stale: list[str] = []
+    for payload in fdr_rejection_records:
+        if not isinstance(payload, dict):
+            continue
+        reason = payload.get("reason")
+        if reason != stale_reason:
+            continue
+        tile_id = payload.get("id") or payload.get("tile_id")
+        if isinstance(tile_id, str) and tile_id in gen_id_set:
+            stale.append(tile_id)
+    return FreshnessGateReport(
+        generated_tile_ids=generated_ids, stale_rejections=tuple(stale)
+    )

e2e/runner/helpers/mock_suite_sat_audit.py

@@ -0,0 +1,76 @@
+"""HTTP client for the mock-suite-sat-service audit log.
+
+Thin wrapper over ``GET /tiles/audit`` (and its alias ``GET /mock/audit``)
+that the FT-P-17 scenario uses to verify every generated tile was
+accepted with HTTP 202 at landing-time upload.
+
+Reading the audit log is a one-shot, end-of-run operation, so the
+helper is synchronous httpx — no streaming, no retries (the service is
+co-located in the compose harness and a failure to reach it is itself
+a test failure, not something to paper over).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+
+DEFAULT_TIMEOUT_S = 10.0
+DEFAULT_AUDIT_PATH = "/tiles/audit"
+
+
+def fetch_audit(
+    base_url: str,
+    run_id: str,
+    *,
+    timeout_s: float = DEFAULT_TIMEOUT_S,
+    audit_path: str = DEFAULT_AUDIT_PATH,
+    transport: httpx.BaseTransport | None = None,
+) -> list[dict[str, Any]]:
+    """Return the ``entries`` list from the mock-suite-sat-service audit log.
+
+    The endpoint returns ``{"run_id": str, "entries": [...]}``; this
+    helper unwraps the list. An empty list is a legal response (the
+    SUT may not have uploaded anything yet).
+
+    Raises ``RuntimeError`` on non-2xx HTTP status or a malformed
+    response shape — the scenario test wants those to fail loudly.
+
+    ``transport`` is a unit-test seam: pass an
+    ``httpx.MockTransport`` to feed canned responses without spinning
+    up the real service.
+    """
+    if not base_url:
+        raise RuntimeError("fetch_audit: base_url must be a non-empty string")
+    if not run_id:
+        raise RuntimeError("fetch_audit: run_id must be a non-empty string")
+    url = base_url.rstrip("/") + audit_path
+    client_kwargs: dict[str, Any] = {"timeout": timeout_s}
+    if transport is not None:
+        client_kwargs["transport"] = transport
+    with httpx.Client(**client_kwargs) as client:
+        resp = client.get(url, params={"run_id": run_id})
+    if resp.status_code >= 300:
+        raise RuntimeError(
+            f"fetch_audit: {url}?run_id={run_id} returned HTTP "
+            f"{resp.status_code}: body={resp.text[:200]!r}"
+        )
+    try:
+        body = resp.json()
+    except ValueError as exc:
+        raise RuntimeError(
+            f"fetch_audit: {url}?run_id={run_id} body is not valid JSON: {exc}"
+        ) from exc
+    if not isinstance(body, dict):
+        raise RuntimeError(
+            f"fetch_audit: {url}?run_id={run_id} body is not a JSON object: "
+            f"got {type(body).__name__}"
+        )
+    entries = body.get("entries")
+    if not isinstance(entries, list):
+        raise RuntimeError(
+            f"fetch_audit: {url}?run_id={run_id} body missing 'entries' list: "
+            f"keys={list(body.keys())}"
+        )
+    return entries

e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py

@@ -0,0 +1,126 @@
+"""FT-N-06 — Mid-flight tile current-timestamp + fresh-treatment (AZ-422 / AC-NEW-6).
+
+The full scenario:
+
+1. Same 5 min Derkachi replay as FT-P-17; the SUT generates one
+   FDR ``mid-flight-tile-output`` record per tile.
+2. Inspect each tile's manifest entry:
+   * AC-5: ``|capture_utc - generated_at_monotonic_ms| ≤ 60 s``.
+   * AC-6: no FDR ``tile-load-rejected`` record with
+     ``reason == "stale"`` carries any of the generated tile IDs
+     (a fresh tile must not be misclassified by the freshness gate).
+
+Gated on:
+
+* ``sitl_replay_ready`` — full replay requires the SITL fixture.
+* ``runner.helpers.mid_flight_tile_evaluator`` — pure-logic
+  evaluator covered by
+  ``e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py``.
+
+This is a "negative" test in the sense that it asserts a *non*-event:
+no stale rejection of a freshly generated tile. The test still skips
+cleanly when the SITL fixture is not prepared.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from runner.helpers import mid_flight_tile_evaluator as mfe
+
+
+@pytest.mark.traces_to("AC-NEW-6,AC-5,AC-6,AC-7")
+def test_ft_n_06_mid_flight_freshness(
+    fc_adapter: str,
+    vio_strategy: str,
+    evidence_dir,  # type: ignore[no-untyped-def]
+    run_id: str,
+    nfr_recorder,  # type: ignore[no-untyped-def]
+    sitl_replay_ready: bool,
+) -> None:
+    """Full FT-N-06 scenario (AC-NEW-6)."""
+    if not sitl_replay_ready:
+        pytest.skip(
+            "FT-N-06 requires `E2E_SITL_REPLAY_DIR` to point at a SITL replay "
+            "fixture exposing `mid-flight-tile-output` FDR records and any "
+            "`tile-load-rejected` events emitted by the freshness gate "
+            "(AZ-595 + AZ-422 fixture builder). Pure-logic AC-NEW-6 "
+            "coverage lives in "
+            "e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py."
+        )
+
+    from runner.helpers import fdr_reader
+
+    fdr_root = Path(evidence_dir).parent / f"run-{run_id}" / "fdr"
+
+    tiles: list[mfe.TileSpec] = []
+    rejection_payloads: list[dict] = []
+    for rec in fdr_reader.iter_records(fdr_root):
+        if rec.record_type == mfe.MID_FLIGHT_TILE_FDR_KIND:
+            tile = _project_tile(rec)
+            if tile is not None:
+                tiles.append(tile)
+        elif rec.record_type == mfe.TILE_LOAD_REJECTED_FDR_KIND:
+            rejection_payloads.append(dict(rec.payload))
+
+    if not tiles:
+        pytest.fail(
+            f"FT-N-06: no `{mfe.MID_FLIGHT_TILE_FDR_KIND}` FDR records at "
+            f"{fdr_root}. The fixture builder must produce at least one "
+            "generated tile for the freshness/stale check to be meaningful."
+        )
+
+    capture_report = mfe.evaluate_capture_date_freshness(tiles)
+    freshness_report = mfe.evaluate_freshness_gate(tiles, rejection_payloads)
+
+    nfr_recorder.record_metric(
+        "ft_n_06.tile_count", float(len(tiles)), ac_id="AC-NEW-6"
+    )
+    nfr_recorder.record_metric(
+        "ft_n_06.capture_drift_failures",
+        float(len(capture_report.failing_entries)),
+        ac_id="AC-5",
+    )
+    nfr_recorder.record_metric(
+        "ft_n_06.stale_rejection_count",
+        float(len(freshness_report.stale_rejections)),
+        ac_id="AC-6",
+    )
+
+    assert capture_report.passes, (
+        f"AC-5 (|capture_utc - generated_at| ≤ {capture_report.tolerance_s} s) failed: "
+        f"failures={[(e.tile_id, e.drift_s) for e in capture_report.failing_entries]}"
+    )
+    assert freshness_report.passes, (
+        "AC-6 (no `tile-load-rejected: stale` for freshly generated tile) failed: "
+        f"stale_rejected_tile_ids={freshness_report.stale_rejections}"
+    )
+
+
+def _project_tile(rec) -> mfe.TileSpec | None:  # type: ignore[no-untyped-def]
+    """Project an FDR record onto a ``TileSpec``; ``None`` if malformed."""
+    p = rec.payload
+    try:
+        bbox = tuple(p["bbox_wgs84"])  # type: ignore[index]
+    except (KeyError, TypeError):
+        return None
+    if len(bbox) != 4:
+        return None
+    quality = p.get("quality") if isinstance(p.get("quality"), dict) else {}
+    capture_utc: str | None = None
+    if isinstance(quality, dict):
+        raw_capture = quality.get("capture_utc")
+        if isinstance(raw_capture, str):
+            capture_utc = raw_capture
+    return mfe.TileSpec(
+        tile_id=str(p.get("tile_id") or ""),
+        bbox_wgs84=(float(bbox[0]), float(bbox[1]), float(bbox[2]), float(bbox[3])),
+        zoom_level=int(p.get("zoom_level") or 0),
+        descriptor_sha256=str(p.get("descriptor_sha256") or ""),
+        payload_size_bytes=int(p.get("payload_size_bytes") or 0),
+        quality=dict(quality) if isinstance(quality, dict) else {},
+        generated_at_monotonic_ms=int(rec.monotonic_ms),
+        capture_utc_iso=capture_utc,
+    )

e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py

@@ -0,0 +1,182 @@
+"""FT-P-17 — Mid-flight tile generation + landing-time upload (AZ-422 / AC-8.4).
+
+The full scenario:
+
+1. The SUT cold-starts against an empty ``mid-flight-tile-output/``
+   FDR directory + the bind-mounted Derkachi fixture.
+2. Replay 5 min of Derkachi at the SUT's runtime cadence. While the
+   SUT generates orthorectified tiles it writes one FDR record per
+   tile under ``mid-flight-tile-output`` carrying every field the
+   mock-suite-sat-service ingest schema requires (Mode B Fact #105).
+3. After replay, the test simulates a landing event (mechanism is
+   public-input — ``simulate_landing()`` MAVLink command, owned by
+   AZ-595 fixture builder); the SUT then uploads every generated
+   tile to ``mock-suite-sat-service``.
+4. The test parses the FDR archive for generated tiles, fetches the
+   mock-service audit log, and asserts:
+   * AC-1: ≥ 1 tile per ~3 s of high-quality nav frames.
+   * AC-2: every tile has all Mode B Fact #105 fields populated.
+   * AC-3: no two tiles share footprint within ±1 m AND GSD within ±5 %.
+   * AC-4: every generated tile_id is in the audit log (HTTP 202).
+   * AC-7: parameterised across ``(fc_adapter, vio_strategy)``.
+
+FT-N-06 (AC-5/AC-6) is a separate file: ``test_ft_n_06_mid_flight_freshness.py``.
+
+Gated on:
+
+* ``sitl_replay_ready`` — full replay requires the SITL fixture.
+* ``runner.helpers.mid_flight_tile_evaluator`` — pure-logic evaluator
+  covered by ``e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py``.
+* ``runner.helpers.mock_suite_sat_audit.fetch_audit`` — HTTP wrapper
+  covered by ``e2e/_unit_tests/helpers/test_mock_suite_sat_audit.py``.
+* ``FT_P_17_HIGH_QUALITY_WINDOW_S_ENV`` — the fixture builder records
+  the total wall-clock seconds of high-quality nav frames produced
+  by the replay (per AC-2.1a normal-segment criterion). Without this
+  env var the scenario can't compute the AC-1 denominator and skips.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from runner.helpers import mid_flight_tile_evaluator as mfe
+from runner.helpers import mock_suite_sat_audit
+
+FT_P_17_HIGH_QUALITY_WINDOW_S_ENV = "FT_P_17_HIGH_QUALITY_WINDOW_S"
+
+
+@pytest.mark.traces_to("AC-8.4,AC-1,AC-2,AC-3,AC-4,AC-7")
+def test_ft_p_17_mid_flight_tiles(
+    fc_adapter: str,
+    vio_strategy: str,
+    evidence_dir,  # type: ignore[no-untyped-def]
+    run_id: str,
+    nfr_recorder,  # type: ignore[no-untyped-def]
+    sitl_replay_ready: bool,
+    mock_suite_sat_url: str,
+) -> None:
+    """Full FT-P-17 scenario (AC-8.4)."""
+    if not sitl_replay_ready:
+        pytest.skip(
+            "FT-P-17 requires `E2E_SITL_REPLAY_DIR` to point at a SITL replay "
+            "fixture exposing `mid-flight-tile-output` FDR records and the "
+            "post-landing audit population on mock-suite-sat-service "
+            "(AZ-595 + AZ-422 fixture builder). Pure-logic AC-8.4 coverage "
+            "lives in e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py."
+        )
+
+    high_quality_window_s_str = os.environ.get(FT_P_17_HIGH_QUALITY_WINDOW_S_ENV)
+    if not high_quality_window_s_str:
+        pytest.skip(
+            f"FT-P-17 needs `{FT_P_17_HIGH_QUALITY_WINDOW_S_ENV}` env var "
+            "(total wall-clock seconds of high-quality nav frames per "
+            "AC-2.1a). The fixture builder records this from the replay's "
+            "segment-quality FDR records."
+        )
+    try:
+        high_quality_window_s = float(high_quality_window_s_str)
+    except ValueError as exc:
+        pytest.fail(
+            f"FT-P-17: `{FT_P_17_HIGH_QUALITY_WINDOW_S_ENV}` must parse as "
+            f"float; got {high_quality_window_s_str!r}: {exc}"
+        )
+
+    from runner.helpers import fdr_reader
+
+    fdr_root = Path(evidence_dir).parent / f"run-{run_id}" / "fdr"
+    tiles = list(_extract_tiles_from_fdr(fdr_reader, fdr_root))
+    if not tiles:
+        pytest.fail(
+            f"FT-P-17: no `{mfe.MID_FLIGHT_TILE_FDR_KIND}` FDR records under "
+            f"{fdr_root}. The SUT must generate at least one tile per AC-1."
+        )
+
+    audit_entries = mock_suite_sat_audit.fetch_audit(mock_suite_sat_url, run_id=run_id)
+
+    rate_report = mfe.evaluate_tile_generation_rate(tiles, high_quality_window_s)
+    quality_report = mfe.evaluate_tile_quality_metadata(tiles)
+    dedup_report = mfe.evaluate_dedup(tiles)
+    upload_report = mfe.evaluate_upload_acks(tiles, audit_entries)
+
+    nfr_recorder.record_metric(
+        "ft_p_17.tile_count", float(rate_report.tile_count), ac_id="AC-1"
+    )
+    nfr_recorder.record_metric(
+        "ft_p_17.observed_rate_per_3s", rate_report.observed_rate_per_3s, ac_id="AC-1"
+    )
+    nfr_recorder.record_metric(
+        "ft_p_17.high_quality_window_s", high_quality_window_s, ac_id="AC-1"
+    )
+    nfr_recorder.record_metric(
+        "ft_p_17.tile_quality_failures",
+        float(len(quality_report.failing_entries)),
+        ac_id="AC-2",
+    )
+    nfr_recorder.record_metric(
+        "ft_p_17.dedup_duplicate_pairs",
+        float(dedup_report.duplicate_count),
+        ac_id="AC-3",
+    )
+    nfr_recorder.record_metric(
+        "ft_p_17.audit_missing_count",
+        float(len(upload_report.missing_from_audit)),
+        ac_id="AC-4",
+    )
+
+    assert rate_report.passes, (
+        f"AC-1 (≥1 tile per {mfe.MIN_TILES_PER_HIGH_QUALITY_WINDOW_S} s) failed: "
+        f"{rate_report.tile_count} tiles over {high_quality_window_s} s "
+        f"high-quality window → rate={rate_report.observed_rate_per_3s:.3f}/3s"
+    )
+    assert quality_report.passes, (
+        "AC-2 (every tile has Mode B Fact #105 quality fields) failed: "
+        f"failures={[(e.tile_id, e.missing_top_level_fields, e.missing_quality_fields) for e in quality_report.failing_entries]}"
+    )
+    assert dedup_report.passes, (
+        "AC-3 (no duplicate footprint+GSD bins) failed: "
+        f"duplicate_pairs={dedup_report.duplicate_pairs}"
+    )
+    assert upload_report.passes, (
+        "AC-4 (landing-event upload accepted) failed: "
+        f"generated={len(upload_report.generated_tile_ids)}, "
+        f"audited={len(upload_report.audit_tile_ids)}, "
+        f"missing={upload_report.missing_from_audit}"
+    )
+
+
+def _extract_tiles_from_fdr(fdr_reader, fdr_root: Path):  # type: ignore[no-untyped-def]
+    """Yield ``TileSpec``s from every ``mid-flight-tile-output`` FDR record.
+
+    Each record's payload mirrors the mock-suite-sat-service TilePublishRequest
+    shape; the scenario only projects it onto a ``TileSpec`` and lets the
+    evaluators do the AC math.
+    """
+    for rec in fdr_reader.iter_records(fdr_root):
+        if rec.record_type != mfe.MID_FLIGHT_TILE_FDR_KIND:
+            continue
+        p = rec.payload
+        try:
+            bbox = tuple(p["bbox_wgs84"])  # type: ignore[index]
+        except (KeyError, TypeError):
+            continue
+        if len(bbox) != 4:
+            continue
+        quality = p.get("quality") if isinstance(p.get("quality"), dict) else {}
+        capture_utc: str | None = None
+        if isinstance(quality, dict):
+            raw_capture = quality.get("capture_utc")
+            if isinstance(raw_capture, str):
+                capture_utc = raw_capture
+        yield mfe.TileSpec(
+            tile_id=str(p.get("tile_id") or ""),
+            bbox_wgs84=(float(bbox[0]), float(bbox[1]), float(bbox[2]), float(bbox[3])),
+            zoom_level=int(p.get("zoom_level") or 0),
+            descriptor_sha256=str(p.get("descriptor_sha256") or ""),
+            payload_size_bytes=int(p.get("payload_size_bytes") or 0),
+            quality=dict(quality) if isinstance(quality, dict) else {},
+            generated_at_monotonic_ms=int(rec.monotonic_ms),
+            capture_utc_iso=capture_utc,
+        )