Add exact LoRa OTA route cost objective

tools/lora_ota/rak3401_route_search.py

@@ -4,6 +4,12 @@
 Cache entries are keyed by source SHA, target SHA, and workspace size rather
 than inventory indices.  A cache generated for an older inventory can therefore
 be migrated safely when new candidate images are inserted.
+
+The default secondary objective remains container size. ``--objective transport``
+instead measures the exact live motatool encoder and minimizes clean serialized
+MeshCore bytes. Inventory boolean ``ota_transport_deflate`` and integer
+``ota_fetch_pipeline`` belong to the currently running/source image; the pinned
+first bootstrap is always legacy raw.
 """
 from __future__ import annotations
 
@@ -16,6 +22,7 @@ import os
 from pathlib import Path
 import re
 import shutil
+import subprocess
 import sys
 import tempfile
 
@@ -34,10 +41,25 @@ FIXED_MEMORY = 0x98000
 AVAILABLE_PAGES = (STAGE_CEILING - APP_BASE) // PAGE
 EXPECTED_TARGET_ID = 0x2FA509C1
 EXPECTED_HARDWARE = "RAK_3401"
+TRANSPORT_CAPABILITY = "ota_transport_deflate"
+TRANSPORT_PIPELINE_FIELD = "ota_fetch_pipeline"
+TRANSPORT_CAPABILITY_ALIAS = "transport_inflate"
+TRANSPORT_PIPELINE_ALIAS = "fetch_pipeline"
+BLOCK_SIZE = 1024
+LEGACY_DATA_BYTES = 160
+V2_DATA_BYTES = 171
+MANIFEST_BYTES = 197
+MANIFEST_FRAGMENT_BYTES = 176
+RAK3401_PIPELINE_MAX = 4
+OTA_MAX_HOPS = 8
+TRANSPORT_FIELDS = [
+    "payload_sha256", "transport_encoder_sha256", "v2_wire_bytes",
+    "v2_deflate_bytes", "v2_deflate_blocks", "v2_data_packets",
+]
 FIELDS = [
     "source", "target", "source_sha256", "target_sha256", "memory",
     "payload", "container", "stage_start", "margin", "feasible", "error",
-]
+] + TRANSPORT_FIELDS
 
 
 class RouteSearchError(RuntimeError):
@@ -56,6 +78,189 @@ def container_size(payload_size: int) -> int:
     return payload_size + 210 + 4 * ((payload_size + 1023) // 1024)
 
 
+def payload_size_from_container(total: int) -> int:
+    """Invert the fixed mOTA geometry used by the pinned bootstrap package."""
+    if total <= 210:
+        raise RouteSearchError("container is too small to contain a payload")
+    max_blocks = (total - 210 + BLOCK_SIZE - 1) // BLOCK_SIZE
+    for blocks in range(1, max_blocks + 1):
+        payload = total - 210 - 4 * blocks
+        if payload > 0 and (payload + BLOCK_SIZE - 1) // BLOCK_SIZE == blocks:
+            return payload
+    raise RouteSearchError(f"container size {total} has no valid payload geometry")
+
+
+def _block_lengths(payload_size: int) -> list[int]:
+    if payload_size <= 0:
+        raise RouteSearchError("transport payload must be positive")
+    full, tail = divmod(payload_size, BLOCK_SIZE)
+    result = [BLOCK_SIZE] * full
+    if tail:
+        result.append(tail)
+    return result
+
+
+def proof_sibling_count(block_count: int, block_index: int) -> int:
+    """Match the promote-odd Merkle proof emitted by MeshCore."""
+    if block_count <= 0 or not 0 <= block_index < block_count:
+        raise RouteSearchError("invalid block for Merkle proof sizing")
+    siblings = 0
+    count = block_count
+    index = block_index
+    while count > 1:
+        if not (count % 2 == 1 and index == count - 1):
+            siblings += 1
+        index //= 2
+        count = (count + 1) // 2
+    return siblings
+
+
+def _common_transport_cost(block_count: int) -> tuple[int, int, int]:
+    """Return origin-serialized Mesh bytes, packets, and proof-response bytes."""
+    manifest_fragments = (
+        MANIFEST_BYTES + MANIFEST_FRAGMENT_BYTES - 1
+    ) // MANIFEST_FRAGMENT_BYTES
+    # GET_MANIFEST is 7 OTA bytes; each MANIFEST has a 7-byte OTA header.
+    # Every source frame also has MeshCore's header + path-length bytes.
+    manifest_source_bytes = 7 + 2 + MANIFEST_BYTES + 9 * manifest_fragments
+    proof_source_bytes = sum(
+        10 + 4 * proof_sibling_count(block_count, block)
+        for block in range(block_count)
+    )
+    return (
+        manifest_source_bytes + proof_source_bytes,
+        1 + manifest_fragments + block_count,
+        proof_source_bytes,
+    )
+
+
+def legacy_transport_cost(payload_size: int) -> dict[str, int | str]:
+    """Ideal clean legacy transfer, including requests, DATA, and proofs."""
+    blocks = _block_lengths(payload_size)
+    block_count = len(blocks)
+    data_packets = sum(
+        (length + LEGACY_DATA_BYTES - 1) // LEGACY_DATA_BYTES
+        for length in blocks
+    )
+    common_bytes, common_packets, proof_bytes = _common_transport_cost(block_count)
+    # DATA is 9 OTA header + 2 Mesh bytes. Each legacy block also needs one
+    # 9-byte OTA_REQ frame and one 7-byte OTA_REQ_PROOF frame.
+    data_bytes = payload_size + 11 * data_packets
+    request_bytes = 11 * block_count
+    proof_request_bytes = 9 * block_count
+    manifest_bytes = common_bytes - proof_bytes
+    return {
+        "profile": "legacy-160-raw",
+        "payload_bytes": payload_size,
+        "wire_bytes": payload_size,
+        "deflate_bytes": 0,
+        "deflate_blocks": 0,
+        "data_packets": data_packets,
+        "manifest_packets": common_packets - block_count,
+        "request_pipeline": 1,
+        "request_packets": block_count,
+        "proof_request_packets": block_count,
+        "proof_packets": block_count,
+        "origin_mesh_bytes": (
+            common_bytes + data_bytes + request_bytes + proof_request_bytes
+        ),
+        "packets": common_packets + data_packets + 2 * block_count,
+        "manifest_bytes": manifest_bytes,
+        "block_request_bytes": request_bytes,
+        "proof_request_bytes": proof_request_bytes,
+        "proof_response_bytes": proof_bytes,
+        "data_bytes": data_bytes,
+        "request_bytes": request_bytes + proof_request_bytes,
+    }
+
+
+def _v2_request_flights(
+    block_count: int, pipeline_capacity: int = RAK3401_PIPELINE_MAX
+) -> list[int]:
+    if block_count <= 0 or not 1 <= pipeline_capacity <= RAK3401_PIPELINE_MAX:
+        raise RouteSearchError("invalid v2 request-flight geometry")
+    flights: list[int] = []
+    remaining = block_count
+    width = 1
+    while remaining:
+        active = min(width, remaining)
+        flights.append(active)
+        remaining -= active
+        width = min(width + 1, pipeline_capacity)
+    return flights
+
+
+def v2_transport_cost(
+    payload_size: int,
+    wire_bytes: int,
+    deflate_bytes: int,
+    deflate_blocks: int,
+    data_packets: int,
+    pipeline_capacity: int = RAK3401_PIPELINE_MAX,
+) -> dict[str, int | str]:
+    """Ideal clean negotiated v2 transfer using measured seeder output."""
+    blocks = _block_lengths(payload_size)
+    block_count = len(blocks)
+    raw_data_packets = sum(
+        (length + V2_DATA_BYTES - 1) // V2_DATA_BYTES for length in blocks
+    )
+    if not (0 < wire_bytes <= payload_size):
+        raise RouteSearchError("v2 wire-byte count is inconsistent")
+    if not (0 <= deflate_bytes <= wire_bytes):
+        raise RouteSearchError("v2 DEFLATE-byte count is inconsistent")
+    if not (0 <= deflate_blocks <= block_count):
+        raise RouteSearchError("v2 DEFLATE-block count is inconsistent")
+    if not (block_count <= data_packets <= raw_data_packets):
+        raise RouteSearchError("v2 DATA-packet count is inconsistent")
+    if (deflate_blocks == 0) != (deflate_bytes == 0):
+        raise RouteSearchError("v2 DEFLATE totals disagree")
+    if not 1 <= pipeline_capacity <= RAK3401_PIPELINE_MAX:
+        raise RouteSearchError("v2 request pipeline is out of range")
+
+    common_bytes, common_packets, proof_bytes = _common_transport_cost(block_count)
+    flights = _v2_request_flights(block_count, pipeline_capacity)
+    # DATA: 9-byte existing header + repeated 4-byte stream id + 2 Mesh bytes.
+    data_bytes = wire_bytes + 15 * data_packets
+    # One request frame per adaptive flight: 5-byte prefix + 4 bytes/block + 2 Mesh bytes.
+    request_bytes = sum(7 + 4 * width for width in flights)
+    manifest_bytes = common_bytes - proof_bytes
+    return {
+        "profile": "v2-171-deflate",
+        "payload_bytes": payload_size,
+        "wire_bytes": wire_bytes,
+        "deflate_bytes": deflate_bytes,
+        "deflate_blocks": deflate_blocks,
+        "data_packets": data_packets,
+        "manifest_packets": common_packets - block_count,
+        "request_pipeline": pipeline_capacity,
+        "request_packets": len(flights),
+        "proof_request_packets": 0,
+        "proof_packets": block_count,
+        "origin_mesh_bytes": common_bytes + data_bytes + request_bytes,
+        "packets": common_packets + data_packets + len(flights),
+        "manifest_bytes": manifest_bytes,
+        "block_request_bytes": request_bytes,
+        "proof_request_bytes": 0,
+        "proof_response_bytes": proof_bytes,
+        "data_bytes": data_bytes,
+        "request_bytes": request_bytes,
+    }
+
+
+def linear_path_bytes(origin_mesh_bytes: int, packets: int, relay_hops: int) -> int:
+    """Serialized bytes for an ideal single path; real flood fan-out is external."""
+    if origin_mesh_bytes < 0 or packets < 0:
+        raise RouteSearchError("transport byte inputs cannot be negative")
+    if not 0 <= relay_hops <= OTA_MAX_HOPS:
+        raise RouteSearchError(f"relay hops must be between 0 and {OTA_MAX_HOPS}")
+    transmissions = relay_hops + 1
+    # Each relay retransmits the frame after appending one one-byte path hash.
+    return (
+        transmissions * origin_mesh_bytes
+        + packets * relay_hops * transmissions // 2
+    )
+
+
 def truth(value: object) -> bool:
     return value is True or str(value).lower() == "true"
 
@@ -89,8 +294,29 @@ def load_inventory(path: Path) -> list[dict[str, object]]:
     if not isinstance(images, list) or len(images) < 2:
         raise RouteSearchError(f"invalid image inventory: {path}")
 
+    capability_contract = document.get("capability_contract", {})
+    strict_transport_contract = (
+        isinstance(capability_contract, dict)
+        and TRANSPORT_CAPABILITY in capability_contract
+    ) or any(
+        isinstance(record, dict) and TRANSPORT_CAPABILITY in record
+        for record in images
+    )
+    edge_policy = document.get("edge_policy", {})
+    strict_version_ranks = (
+        isinstance(edge_policy, dict)
+        and edge_policy.get("require_target_version_rank_gt_source_version_rank")
+        is True
+    ) or any(
+        isinstance(record, dict) and "version_rank" in record
+        for record in images
+    )
+
     result: list[dict[str, object]] = []
     seen: set[str] = set()
+    previous_version_rank = -1
+    rank_by_version: dict[str, int] = {}
+    version_by_rank: dict[int, str] = {}
     for index, raw in enumerate(images):
         if not isinstance(raw, dict):
             raise RouteSearchError(f"inventory image {index} is not an object")
@@ -153,6 +379,76 @@ def load_inventory(path: Path) -> list[dict[str, object]]:
             or not re.fullmatch(r"[0-9a-f]{40}", source_commit)
         ):
             raise RouteSearchError(f"image {index} source commit is invalid")
+        if TRANSPORT_CAPABILITY in record:
+            capability = record[TRANSPORT_CAPABILITY]
+        elif strict_transport_contract:
+            raise RouteSearchError(
+                f"image {index} {TRANSPORT_CAPABILITY} is required by the "
+                "inventory capability contract"
+            )
+        else:
+            # Compatibility for pre-contract inventories. New inventories must
+            # carry the canonical field explicitly so transport is fail-closed.
+            capability = record.get(TRANSPORT_CAPABILITY_ALIAS, False)
+        if not isinstance(capability, bool):
+            raise RouteSearchError(
+                f"image {index} {TRANSPORT_CAPABILITY} must be boolean"
+            )
+        alias_capability = record.get(TRANSPORT_CAPABILITY_ALIAS, capability)
+        if not isinstance(alias_capability, bool) or alias_capability != capability:
+            raise RouteSearchError(
+                f"image {index} {TRANSPORT_CAPABILITY_ALIAS} disagrees with "
+                f"{TRANSPORT_CAPABILITY}"
+            )
+        record[TRANSPORT_CAPABILITY] = capability
+        raw_pipeline = record.get(TRANSPORT_PIPELINE_FIELD)
+        if (
+            raw_pipeline is None
+            and not strict_transport_contract
+            and TRANSPORT_PIPELINE_ALIAS in record
+        ):
+            raw_pipeline = record[TRANSPORT_PIPELINE_ALIAS]
+        if capability and raw_pipeline is None:
+            raise RouteSearchError(
+                f"image {index} {TRANSPORT_PIPELINE_FIELD} is required when "
+                f"{TRANSPORT_CAPABILITY} is true"
+            )
+        if capability:
+            pipeline = _integer(
+                raw_pipeline, f"image {index} {TRANSPORT_PIPELINE_FIELD}"
+            )
+            if not 1 <= pipeline <= RAK3401_PIPELINE_MAX:
+                raise RouteSearchError(
+                    f"image {index} {TRANSPORT_PIPELINE_FIELD} must be between 1 "
+                    f"and {RAK3401_PIPELINE_MAX}"
+                )
+            record[TRANSPORT_PIPELINE_FIELD] = pipeline
+        elif raw_pipeline is not None and strict_transport_contract:
+            raise RouteSearchError(
+                f"image {index} {TRANSPORT_PIPELINE_FIELD} must be null when "
+                f"{TRANSPORT_CAPABILITY} is false"
+            )
+        else:
+            record[TRANSPORT_PIPELINE_FIELD] = None
+
+        if "version_rank" not in record and strict_version_ranks:
+            raise RouteSearchError(
+                f"image {index} version_rank is required by the inventory edge policy"
+            )
+        version_rank = _integer(
+            record.get("version_rank", index), f"image {index} version_rank"
+        )
+        if version_rank < previous_version_rank:
+            raise RouteSearchError("inventory version_rank values must be nondecreasing")
+        if strict_version_ranks:
+            known_rank = rank_by_version.setdefault(version, version_rank)
+            known_version = version_by_rank.setdefault(version_rank, version)
+            if known_rank != version_rank or known_version != version:
+                raise RouteSearchError(
+                    "inventory version_rank must map one-to-one with firmware version"
+                )
+        previous_version_rank = version_rank
+        record["version_rank"] = version_rank
         if index == 1 and _integer(
             record.get("baseline_container_size"),
             "node 1 baseline_container_size",
@@ -231,11 +527,88 @@ def cache_key(source_sha: str, target_sha: str, memory: int) -> tuple[str, str,
     return source_sha.lower(), target_sha.lower(), memory
 
 
+def transport_stats_complete(row: dict[str, object]) -> bool:
+    return all(row.get(field) not in (None, "") for field in TRANSPORT_FIELDS)
+
+
+def normalize_transport_stats(
+    row: dict[str, object], payload_size: int, label: str
+) -> dict[str, object]:
+    present = [row.get(field) not in (None, "") for field in TRANSPORT_FIELDS]
+    if any(present) and not all(present):
+        raise RouteSearchError(f"partial transport statistics at {label}")
+    if not any(present):
+        return {field: "" for field in TRANSPORT_FIELDS}
+
+    normalized: dict[str, object] = {}
+    for field in TRANSPORT_FIELDS[:2]:
+        digest = row.get(field)
+        if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest):
+            raise RouteSearchError(f"invalid {field} at {label}")
+        normalized[field] = digest
+    for field in TRANSPORT_FIELDS[2:]:
+        normalized[field] = _integer(row.get(field), f"{field} at {label}")
+    v2_transport_cost(
+        payload_size,
+        int(normalized["v2_wire_bytes"]),
+        int(normalized["v2_deflate_bytes"]),
+        int(normalized["v2_deflate_blocks"]),
+        int(normalized["v2_data_packets"]),
+    )
+    return normalized
+
+
+def measure_transport_size(tool: Path, payload: Path) -> dict[str, object]:
+    """Ask motatool's live-seeder encoder for exact per-block byte counts."""
+    try:
+        completed = subprocess.run(
+            [str(tool), "transport-size", "--payload", str(payload),
+             "--block-size", str(BLOCK_SIZE)],
+            check=False, capture_output=True, text=True, encoding="utf-8",
+            timeout=60,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise RouteSearchError(f"transport-size tool timed out for {payload}") from exc
+    except OSError as exc:
+        raise RouteSearchError(f"cannot run transport-size tool {tool}: {exc}") from exc
+    if completed.returncode != 0:
+        detail = (completed.stderr or completed.stdout).strip()[-500:]
+        raise RouteSearchError(
+            f"transport-size tool failed for {payload}: {detail or completed.returncode}"
+        )
+    try:
+        measured = json.loads(completed.stdout)
+    except json.JSONDecodeError as exc:
+        raise RouteSearchError("transport-size tool returned invalid JSON") from exc
+    if not isinstance(measured, dict) or measured.get("schema") != 1:
+        raise RouteSearchError("transport-size tool returned an unsupported result")
+    expected_payload = payload.stat().st_size
+    if measured.get("payload_bytes") != expected_payload:
+        raise RouteSearchError("transport-size tool measured the wrong payload")
+    if measured.get("block_size") != BLOCK_SIZE:
+        raise RouteSearchError("transport-size tool used the wrong block size")
+    expected_blocks = (expected_payload + BLOCK_SIZE - 1) // BLOCK_SIZE
+    if measured.get("block_count") != expected_blocks:
+        raise RouteSearchError("transport-size tool returned the wrong block count")
+    row = {
+        "payload_sha256": sha256_file(payload),
+        "transport_encoder_sha256": sha256_file(tool),
+        "v2_wire_bytes": measured.get("wire_bytes"),
+        "v2_deflate_bytes": measured.get("deflate_bytes"),
+        "v2_deflate_blocks": measured.get("deflate_blocks"),
+        "v2_data_packets": measured.get("data_packets"),
+    }
+    return normalize_transport_stats(row, expected_payload, str(payload))
+
+
 def migrate_csv(
     inventory_path: Path, csv_path: Path
 ) -> dict[tuple[str, str, int], dict[str, object]]:
-    """Migrate numeric CSV indices through the inventory that created them."""
+    """Migrate cached edges by hash, with numeric-index fallback for old CSVs."""
     images = load_inventory(inventory_path)
+    nodes_by_sha = {
+        str(image["sha256"]): int(image["node"]) for image in images
+    }
     migrated: dict[tuple[str, str, int], dict[str, object]] = {}
     try:
         source = csv_path.open(newline="", encoding="ascii")
@@ -247,15 +620,18 @@ def migrate_csv(
         if reader.fieldnames is None or not required_fields.issubset(reader.fieldnames):
             raise RouteSearchError(f"geometry cache has missing columns: {csv_path}")
         for line, row in enumerate(reader, 2):
-            source_node = _integer(row["source"], f"source at {csv_path}:{line}")
-            target_node = _integer(row["target"], f"target at {csv_path}:{line}")
-            if not (0 <= source_node < target_node < len(images)):
+            recorded_source = _integer(
+                row["source"], f"source at {csv_path}:{line}"
+            )
+            recorded_target = _integer(
+                row["target"], f"target at {csv_path}:{line}"
+            )
+            if recorded_source >= recorded_target:
                 raise RouteSearchError(f"bad nodes at {csv_path}:{line}")
-            source_sha = str(images[source_node]["sha256"])
-            target_sha = str(images[target_node]["sha256"])
-            memory = _integer(row["memory"], f"memory at {csv_path}:{line}")
-            for column, expected in (
-                ("source_sha256", source_sha), ("target_sha256", target_sha)
+            resolved: list[tuple[int, str]] = []
+            for column, recorded_node in (
+                ("source_sha256", recorded_source),
+                ("target_sha256", recorded_target),
             ):
                 raw_recorded = row.get(column, "")
                 if raw_recorded is None:
@@ -264,11 +640,28 @@ def migrate_csv(
                     raise RouteSearchError(
                         f"{column} is invalid at {csv_path}:{line}"
                     )
-                recorded = raw_recorded.lower()
-                if recorded and recorded != expected:
-                    raise RouteSearchError(
-                        f"{column} disagrees with inventory at {csv_path}:{line}"
+                recorded_sha = raw_recorded.lower()
+                if recorded_sha:
+                    if not re.fullmatch(r"[0-9a-f]{64}", recorded_sha):
+                        raise RouteSearchError(
+                            f"{column} is invalid at {csv_path}:{line}"
+                        )
+                    resolved_node = nodes_by_sha.get(recorded_sha)
+                    if resolved_node is None:
+                        raise RouteSearchError(
+                            f"{column} is absent from inventory at {csv_path}:{line}"
+                        )
+                    resolved.append((resolved_node, recorded_sha))
+                else:
+                    if recorded_node >= len(images):
+                        raise RouteSearchError(f"bad nodes at {csv_path}:{line}")
+                    resolved.append(
+                        (recorded_node, str(images[recorded_node]["sha256"]))
                     )
+            (source_node, source_sha), (target_node, target_sha) = resolved
+            if source_node >= target_node:
+                raise RouteSearchError(f"bad nodes at {csv_path}:{line}")
+            memory = _integer(row["memory"], f"memory at {csv_path}:{line}")
             try:
                 payload = int(row.get("payload", ""))
                 total = int(row.get("container", ""))
@@ -306,6 +699,9 @@ def migrate_csv(
             normalized.update(source=source_node, target=target_node,
                               source_sha256=source_sha, target_sha256=target_sha,
                               memory=memory)
+            normalized.update(normalize_transport_stats(
+                row, payload, f"{csv_path}:{line}"
+            ))
             previous = migrated.get(key)
             comparable = ("payload", "container", "stage_start", "margin", "feasible", "error")
             if previous is not None and any(
@@ -313,6 +709,16 @@ def migrate_csv(
                 for item in comparable
             ):
                 raise RouteSearchError(f"conflicting cached geometry for {key}")
+            if previous is not None and transport_stats_complete(previous):
+                if transport_stats_complete(normalized) and any(
+                    str(previous.get(item, "")) != str(normalized.get(item, ""))
+                    for item in TRANSPORT_FIELDS
+                ):
+                    raise RouteSearchError(f"conflicting transport statistics for {key}")
+                if not transport_stats_complete(normalized):
+                    normalized.update(
+                        {field: previous[field] for field in TRANSPORT_FIELDS}
+                    )
             migrated[key] = normalized
     return migrated
 
@@ -320,6 +726,13 @@ def migrate_csv(
 Job = tuple[int, int, int, str, str, str, str]
 
 
+def is_forward_progress(
+    images: list[dict[str, object]], source: int, target: int
+) -> bool:
+    """Only a strictly newer firmware version may be a route edge."""
+    return int(images[target]["version_rank"]) > int(images[source]["version_rank"])
+
+
 def all_jobs(images: list[dict[str, object]]) -> list[Job]:
     """Enumerate every valid page-aligned workspace for every forward edge."""
     jobs: list[Job] = []
@@ -328,12 +741,16 @@ def all_jobs(images: list[dict[str, object]]) -> list[Job]:
     # Node 1 still runs the fixed-workspace receiver, so measure each of its
     # possible forward transitions at exactly that legacy workspace.
     for target in range(2, endpoint + 1):
+        if not is_forward_progress(images, 1, target):
+            continue
         jobs.append((1, target, FIXED_MEMORY, str(images[1]["path"]),
                      str(images[target]["path"]), str(images[1]["sha256"]),
                      str(images[target]["sha256"])))
     for source in range(2, endpoint):
         source_size = int(images[source]["size"])
         for target in range(source + 1, endpoint + 1):
+            if not is_forward_progress(images, source, target):
+                continue
             target_size = int(images[target]["size"])
             first_page = align_up(max(source_size + 2 * PAGE, target_size)) // PAGE
             for memory_page in range(first_page, AVAILABLE_PAGES):
@@ -343,7 +760,30 @@ def all_jobs(images: list[dict[str, object]]) -> list[Job]:
     return jobs
 
 
-def geometry_job(args: tuple[int, int, int, str, str, str, str, str]) -> dict[str, object]:
+def job_needs_transport_stats(job: Job, images: list[dict[str, object]],
+                              objective: str) -> bool:
+    return objective == "transport" and source_uses_v2(images, job[0])
+
+
+def cache_satisfies_job(
+    row: dict[str, object] | None,
+    needs_transport: bool,
+    expected_encoder_sha256: str | None = None,
+) -> bool:
+    return row is not None and (
+        not needs_transport
+        or not truth(row.get("feasible", False))
+        or (
+            transport_stats_complete(row)
+            and (
+                expected_encoder_sha256 is None
+                or row.get("transport_encoder_sha256") == expected_encoder_sha256
+            )
+        )
+    )
+
+
+def geometry_job(args: tuple[object, ...]) -> dict[str, object]:
     try:
         import detools
     except ImportError as exc:
@@ -351,7 +791,18 @@ def geometry_job(args: tuple[int, int, int, str, str, str, str, str]) -> dict[st
             "detools is required to generate geometry; run this tool in the "
             "detools pipx environment"
         ) from exc
-    source, target, memory, from_raw, to_raw, source_sha, target_sha, patches_raw = args
+    if len(args) not in (8, 9):
+        raise RouteSearchError("invalid geometry job")
+    source, target, memory, from_raw, to_raw, source_sha, target_sha, patches_raw = args[:8]
+    transport_tool = str(args[8]) if len(args) == 9 and args[8] else ""
+    source = int(source)
+    target = int(target)
+    memory = int(memory)
+    from_raw = str(from_raw)
+    to_raw = str(to_raw)
+    source_sha = str(source_sha)
+    target_sha = str(target_sha)
+    patches_raw = str(patches_raw)
     patch = Path(patches_raw) / f"{source_sha}-{target_sha}-{memory // PAGE:03d}.patch"
     try:
         detools.create_patch_filenames(
@@ -364,6 +815,11 @@ def geometry_job(args: tuple[int, int, int, str, str, str, str, str]) -> dict[st
         stage_start = align_down(STAGE_CEILING - total)
         margin = stage_start - (APP_BASE + memory)
         error = ""
+        transport = (
+            measure_transport_size(Path(transport_tool), patch)
+            if transport_tool and margin >= 0
+            else {field: "" for field in TRANSPORT_FIELDS}
+        )
     except Exception as exc:  # detools reports several backend exception types
         detail = f"{type(exc).__name__}: {exc}"[-500:].replace("\n", " ")
         raise RouteSearchError(
@@ -374,7 +830,8 @@ def geometry_job(args: tuple[int, int, int, str, str, str, str, str]) -> dict[st
     return {"source": source, "target": target, "source_sha256": source_sha,
             "target_sha256": target_sha, "memory": memory, "payload": payload,
             "container": total, "stage_start": stage_start, "margin": margin,
-            "feasible": payload >= 0 and margin >= 0, "error": error}
+            "feasible": payload >= 0 and margin >= 0, "error": error,
+            **transport}
 
 
 def project_cache(cache: dict[tuple[str, str, int], dict[str, object]],
@@ -457,31 +914,127 @@ def write_csv(path: Path, rows: list[dict[str, object]]) -> None:
         raise
 
 
+def source_uses_v2(images: list[dict[str, object]], source: int) -> bool:
+    """Capability is the currently running/source image; bootstrap is raw."""
+    return source != 0 and bool(images[source][TRANSPORT_CAPABILITY])
+
+
+def source_transport_pipeline(images: list[dict[str, object]], source: int) -> int:
+    """Return the request-flight capacity compiled into the source image."""
+    pipeline = images[source][TRANSPORT_PIPELINE_FIELD]
+    if pipeline is None:
+        raise RouteSearchError(f"source image {source} has no transport pipeline")
+    return int(pipeline)
+
+
+def transport_encoder_for_rows(
+    rows: list[dict[str, object]], images: list[dict[str, object]]
+) -> str | None:
+    """Require one exact live-seeder encoder across comparable measurements."""
+    encoders = {
+        str(row["transport_encoder_sha256"])
+        for row in rows
+        if truth(row.get("feasible", False))
+        and source_uses_v2(images, int(row["source"]))
+        and transport_stats_complete(row)
+    }
+    if len(encoders) > 1:
+        raise RouteSearchError(
+            "transport statistics mix different motatool encoder binaries"
+        )
+    return next(iter(encoders)) if encoders else None
+
+
+def edge_transport_cost(
+    row: dict[str, object], images: list[dict[str, object]], relay_hops: int
+) -> dict[str, int | str]:
+    source = int(row["source"])
+    payload = int(row.get("payload", 0))
+    if payload <= 0:
+        payload = payload_size_from_container(int(row["container"]))
+    if source_uses_v2(images, source):
+        if not transport_stats_complete(row):
+            raise RouteSearchError(
+                f"missing transport statistics for capable source edge "
+                f"{source}->{int(row['target'])} memory=0x{int(row['memory']):X}"
+            )
+        cost = v2_transport_cost(
+            payload,
+            int(row["v2_wire_bytes"]),
+            int(row["v2_deflate_bytes"]),
+            int(row["v2_deflate_blocks"]),
+            int(row["v2_data_packets"]),
+            source_transport_pipeline(images, source),
+        )
+    else:
+        cost = legacy_transport_cost(payload)
+    result = dict(cost)
+    result["linear_path_bytes"] = linear_path_bytes(
+        int(cost["origin_mesh_bytes"]), int(cost["packets"]), relay_hops
+    )
+    return result
+
+
 def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
-                 baseline_size: int, output: Path, complete: bool) -> dict[str, object]:
+                 baseline_size: int, output: Path, complete: bool,
+                 objective: str = "container", relay_hops: int = 0) -> dict[str, object]:
+    if objective not in ("container", "transport"):
+        raise RouteSearchError(f"unknown route objective: {objective}")
+    if not 0 <= relay_hops <= OTA_MAX_HOPS:
+        raise RouteSearchError(f"relay hops must be between 0 and {OTA_MAX_HOPS}")
     endpoint = len(images) - 1
     baseline_stage = align_down(STAGE_CEILING - baseline_size)
     baseline_margin = baseline_stage - (APP_BASE + FIXED_MEMORY)
     if baseline_margin < 0:
         raise RouteSearchError("pinned first package does not fit")
-    edges: dict[tuple[int, int], dict[str, object]] = {
-        (0, 1): {"source": 0, "target": 1, "memory": FIXED_MEMORY,
-                 "container": baseline_size, "stage_start": baseline_stage,
-                 "margin": baseline_margin, "feasible": True}
+    baseline_row: dict[str, object] = {
+        "source": 0, "target": 1, "memory": FIXED_MEMORY,
+        "container": baseline_size, "stage_start": baseline_stage,
+        "margin": baseline_margin, "feasible": True,
     }
+    if objective == "transport":
+        baseline_row["payload"] = payload_size_from_container(baseline_size)
+    edges: dict[tuple[int, int], dict[str, object]] = {(0, 1): baseline_row}
     for row in rows:
-        if int(row["source"]) == 0:
+        source = int(row["source"])
+        target = int(row["target"])
+        if not (0 <= source < len(images) and 0 <= target < len(images)):
+            raise RouteSearchError(f"geometry edge {source}->{target} is out of range")
+        if source == 0:
             # Older evidence tables measured these irrelevant shortcuts. Keep
             # them countable for archive verification, but never admit them to
             # the route graph; node 0 is pinned unconditionally to node 1.
             continue
+        if not is_forward_progress(images, source, target):
+            continue
         if not truth(row.get("feasible", False)):
             continue
-        key = int(row["source"]), int(row["target"])
-        rank = (int(row["container"]), -int(row["margin"]), int(row["memory"]))
+        key = source, target
+        if objective == "transport":
+            transport_bytes = int(
+                edge_transport_cost(row, images, relay_hops)["linear_path_bytes"]
+            )
+            rank = (
+                transport_bytes, int(row["container"]),
+                -int(row["margin"]), int(row["memory"]),
+            )
+        else:
+            rank = (int(row["container"]), -int(row["margin"]), int(row["memory"]))
         old = edges.get(key)
-        if old is None or rank < (int(old["container"]), -int(old["margin"]), int(old["memory"])):
+        if old is None:
             edges[key] = row
+        else:
+            if objective == "transport":
+                old_rank = (
+                    int(edge_transport_cost(old, images, relay_hops)["linear_path_bytes"]),
+                    int(old["container"]), -int(old["margin"]), int(old["memory"]),
+                )
+            else:
+                old_rank = (
+                    int(old["container"]), -int(old["margin"]), int(old["memory"]),
+                )
+            if rank < old_rank:
+                edges[key] = row
 
     best: dict[int, tuple[int, int, list[int]]] = {0: (0, 0, [0])}
     hop_distance = [sys.maxsize] * len(images)
@@ -494,8 +1047,12 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
         if source not in best:
             continue
         for target, row in outgoing.get(source, []):
+            edge_bytes = (
+                int(edge_transport_cost(row, images, relay_hops)["linear_path_bytes"])
+                if objective == "transport" else int(row["container"])
+            )
             candidate = (best[source][0] + 1,
-                         best[source][1] + int(row["container"]),
+                         best[source][1] + edge_bytes,
                          best[source][2] + [target])
             if target not in best or candidate < best[target]:
                 best[target] = candidate
@@ -505,11 +1062,33 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
             elif distance == hop_distance[target]:
                 route_count[target] += route_count[source]
 
+    objective_text = (
+        "minimum packages, then minimum ideal linear-path serialized MeshCore bytes"
+        if objective == "transport"
+        else "minimum packages, then minimum total container bytes"
+    )
     common = {"schema": 2, "app_base": f"0x{APP_BASE:X}",
               "stage_ceiling": f"0x{STAGE_CEILING:X}", "node_count": len(images),
               "search_complete": complete, "candidate_geometries": len(rows),
               "feasible_edges": len(edges),
-              "objective": "minimum packages, then minimum total container bytes"}
+              "objective": objective_text}
+    if objective == "transport":
+        transport_encoder = transport_encoder_for_rows(rows, images)
+        common["transport_accounting"] = {
+            "relay_hops": relay_hops,
+            "source_capability_field": TRANSPORT_CAPABILITY,
+            "source_pipeline_field": TRANSPORT_PIPELINE_FIELD,
+            "first_bootstrap_profile": "legacy-160-raw",
+            "included": (
+                "OTA messages, repeated 4-byte v2 stream IDs, 171-byte DATA slicing, "
+                "adaptive requests, manifest, exact per-block proofs, and MeshCore framing"
+            ),
+            "excluded": (
+                "discovery, retries, flood fan-out, and radio-dependent LoRa PHY coding/preamble"
+            ),
+        }
+        if transport_encoder is not None:
+            common["transport_accounting"]["encoder_sha256"] = transport_encoder
     if endpoint not in best:
         result = {**common, "status": "unreachable", "reachable_nodes": sorted(best),
                   "endpoint_node": endpoint,
@@ -521,17 +1100,41 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
         steps = []
         for source, target in zip(nodes, nodes[1:]):
             row = edges[source, target]
-            steps.append({"source_node": source, "target_node": target,
-                          "inplace_memory": f"0x{int(row['memory']):X}",
-                          "reuse_baseline_package": source == 0 and target == 1,
-                          "expected_container_size": int(row["container"]),
-                          "expected_staging_margin": int(row["margin"]),
-                          "expected_target_sha256": images[target]["sha256"],
-                          "expected_target_version": images[target].get("version", "unknown")})
+            step = {"source_node": source, "target_node": target,
+                    "inplace_memory": f"0x{int(row['memory']):X}",
+                    "reuse_baseline_package": source == 0 and target == 1,
+                    "expected_container_size": int(row["container"]),
+                    "expected_staging_margin": int(row["margin"]),
+                    "expected_target_sha256": images[target]["sha256"],
+                    "expected_target_version": images[target].get("version", "unknown")}
+            if objective == "transport":
+                cost = edge_transport_cost(row, images, relay_hops)
+                step["transport"] = {
+                    key: cost[key] for key in (
+                        "profile", "payload_bytes", "wire_bytes", "deflate_bytes",
+                        "deflate_blocks", "data_packets", "request_packets",
+                        "manifest_packets", "request_pipeline", "proof_request_packets",
+                        "proof_packets", "packets", "manifest_bytes", "data_bytes",
+                        "block_request_bytes", "proof_request_bytes",
+                        "proof_response_bytes", "origin_mesh_bytes", "linear_path_bytes",
+                    )
+                }
+                if source_uses_v2(images, source):
+                    step["transport"]["payload_sha256"] = row["payload_sha256"]
+                    step["transport"]["encoder_sha256"] = row[
+                        "transport_encoder_sha256"
+                    ]
+            steps.append(step)
+        selected_container_bytes = sum(
+            int(edges[source, target]["container"])
+            for source, target in zip(nodes, nodes[1:])
+        )
         result = {**common, "status": "reachable", "nodes": nodes,
                   "shortest_package_count": hop_distance[endpoint],
                   "shortest_route_count": route_count[endpoint],
-                  "selected_total_bytes": best[endpoint][1], "steps": steps}
+                  "selected_total_bytes": selected_container_bytes, "steps": steps}
+        if objective == "transport":
+            result["selected_total_transport_bytes"] = best[endpoint][1]
     write_atomic_text(
         output, json.dumps(result, indent=2, sort_keys=True) + "\n"
     )
@@ -545,14 +1148,47 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--work-dir", type=Path, required=True)
     parser.add_argument("--workers", type=int, default=min(8, os.cpu_count() or 1))
     parser.add_argument("--no-generate", action="store_true")
+    parser.add_argument(
+        "--objective", choices=("container", "transport"), default="container",
+        help="secondary objective after minimum package count",
+    )
+    parser.add_argument(
+        "--motatool", metavar="PATH",
+        help="motatool with `transport-size`; required to measure capable-source edges",
+    )
+    parser.add_argument(
+        "--relay-hops", type=int, default=0,
+        help="ideal linear-path relays for transport accounting (default: direct)",
+    )
     args = parser.parse_args(argv)
     if args.workers < 1:
         parser.error("--workers must be positive")
+    if not 0 <= args.relay_hops <= OTA_MAX_HOPS:
+        parser.error(f"--relay-hops must be between 0 and {OTA_MAX_HOPS}")
+    transport_tool: Path | None = None
+    if args.motatool:
+        discovered = shutil.which(args.motatool)
+        transport_tool = Path(discovered or args.motatool).resolve()
+        if not transport_tool.is_file():
+            parser.error(f"--motatool is not a regular file: {transport_tool}")
     args.work_dir.mkdir(parents=True, exist_ok=True)
     with tempfile.TemporaryDirectory(
         prefix=".frozen-inventory-", dir=args.work_dir
     ) as frozen_raw:
         frozen = Path(frozen_raw)
+        frozen_transport_tool: Path | None = None
+        frozen_transport_tool_sha256: str | None = None
+        if transport_tool is not None:
+            frozen_transport_tool = frozen / (
+                "motatool.exe" if transport_tool.suffix.lower() == ".exe" else "motatool"
+            )
+            try:
+                shutil.copy2(transport_tool, frozen_transport_tool, follow_symlinks=True)
+            except OSError as exc:
+                raise RouteSearchError(
+                    f"cannot snapshot transport-size tool {transport_tool}: {exc}"
+                ) from exc
+            frozen_transport_tool_sha256 = sha256_file(frozen_transport_tool)
         inventory_path = snapshot_inventory(
             args.inventory, frozen / "current" / "images.json"
         )
@@ -587,19 +1223,40 @@ def main(argv: list[str] | None = None) -> int:
                     for field in comparable
                 ):
                     raise RouteSearchError(f"conflicting reuse inputs for {key}")
+                if old is not None and transport_stats_complete(old):
+                    if transport_stats_complete(row) and any(
+                        str(old.get(field, "")) != str(row.get(field, ""))
+                        for field in TRANSPORT_FIELDS
+                    ):
+                        raise RouteSearchError(
+                            f"conflicting reuse transport statistics for {key}"
+                        )
+                    if not transport_stats_complete(row):
+                        row.update({field: old[field] for field in TRANSPORT_FIELDS})
                 cache[key] = row
 
         jobs = all_jobs(images)
         required = {cache_key(job[5], job[6], job[2]) for job in jobs}
         remaining = [
             job for job in jobs
-            if cache_key(job[5], job[6], job[2]) not in cache
+            if not cache_satisfies_job(
+                cache.get(cache_key(job[5], job[6], job[2])),
+                job_needs_transport_stats(job, images, args.objective),
+                frozen_transport_tool_sha256,
+            )
         ]
         print(
             f"nodes={len(images)} geometries={len(jobs)} "
             f"reused={len(required & cache.keys())} remaining={len(remaining)}"
         )
         if remaining and not args.no_generate:
+            if frozen_transport_tool is None and any(
+                job_needs_transport_stats(job, images, args.objective)
+                for job in remaining
+            ):
+                parser.error(
+                    "--motatool is required to measure transport costs for capable sources"
+                )
             # Patch payloads are disposable worker scratch. Keep them in a
             # uniquely owned directory so cleanup can never remove a caller's
             # pre-existing `work-dir/patches` tree.
@@ -609,7 +1266,13 @@ def main(argv: list[str] | None = None) -> int:
                 patches = Path(patches_raw)
                 with ProcessPoolExecutor(max_workers=args.workers) as pool:
                     futures = [
-                        pool.submit(geometry_job, (*job, str(patches)))
+                        pool.submit(
+                            geometry_job,
+                            (*job, str(patches),
+                             str(frozen_transport_tool) if job_needs_transport_stats(
+                                 job, images, args.objective
+                             ) else ""),
+                        )
                         for job in remaining
                     ]
                     for done, future in enumerate(as_completed(futures), 1):
@@ -622,13 +1285,21 @@ def main(argv: list[str] | None = None) -> int:
                         cache[key] = row
                         if done % 100 == 0 or done == len(futures):
                             print(f"measured={done}/{len(remaining)}", flush=True)
-        completed = required & cache.keys()
+        completed = {
+            cache_key(job[5], job[6], job[2]) for job in jobs
+            if cache_satisfies_job(
+                cache.get(cache_key(job[5], job[6], job[2])),
+                job_needs_transport_stats(job, images, args.objective),
+                frozen_transport_tool_sha256,
+            )
+        }
         rows = project_cache(cache, images, completed)
         write_csv(args.work_dir / "geometry.csv", rows)
         complete = completed == required
         result = select_route(
             rows, images, baseline_size, args.work_dir / "route.json",
             complete=complete,
+            objective=args.objective, relay_hops=args.relay_hops,
         )
         print(json.dumps(result, indent=2, sort_keys=True))
         return 0 if complete else 3

tools/lora_ota/test_rak3401_route_search.py

@@ -32,6 +32,9 @@ class RouteSearchTests(unittest.TestCase):
     def write_inventory(
         self, root: Path, payloads: list[bytes], *, baseline_size: int = 100,
         versions: list[str] | None = None,
+        transport_capabilities: list[bool] | None = None,
+        transport_pipelines: list[int | None] | None = None,
+        version_ranks: list[int] | None = None,
     ) -> tuple[Path, list[dict[str, object]]]:
         records = []
         for node, payload in enumerate(payloads):
@@ -46,12 +49,32 @@ class RouteSearchTests(unittest.TestCase):
                 "sha256": hashlib.sha256(encoded).hexdigest(),
                 "body_hash": hashlib.sha256(payload).digest()[:8].hex(),
                 "version": version,
+                "version_rank": (
+                    version_ranks[node] if version_ranks is not None else node
+                ),
             }
             if node == 1:
                 record["baseline_container_size"] = baseline_size
+            capability = (
+                transport_capabilities[node]
+                if transport_capabilities is not None else False
+            )
+            record[search.TRANSPORT_CAPABILITY] = capability
+            record[search.TRANSPORT_PIPELINE_FIELD] = (
+                transport_pipelines[node]
+                if transport_pipelines is not None else None
+            )
             records.append(record)
         manifest = root / "images.json"
-        manifest.write_text(json.dumps({"images": records}), encoding="ascii")
+        manifest.write_text(json.dumps({
+            "capability_contract": {
+                search.TRANSPORT_CAPABILITY: "required boolean on every record",
+            },
+            "edge_policy": {
+                "require_target_version_rank_gt_source_version_rank": True,
+            },
+            "images": records,
+        }), encoding="ascii")
         return manifest, records
 
     def test_cache_migration_survives_inserted_node_and_remaps_indices(self) -> None:
@@ -70,6 +93,8 @@ class RouteSearchTests(unittest.TestCase):
                 writer.writeheader()
                 writer.writerow({
                     "source": 1, "target": 2, "memory": search.FIXED_MEMORY,
+                    "source_sha256": old[1]["sha256"],
+                    "target_sha256": old[2]["sha256"],
                     "payload": 7,
                     "container": search.container_size(7),
                     "stage_start": search.align_down(
@@ -84,7 +109,9 @@ class RouteSearchTests(unittest.TestCase):
                 new_dir, [b"A", b"X", b"B", b"C"],
                 versions=["1.0.0.0", "1.0.0.99", "1.0.0.1", "1.0.0.2"],
             )
-            migrated = search.migrate_csv(old_manifest, cache)
+            # The recorded numeric nodes no longer identify these images in
+            # the expanded inventory; authenticated hashes still do.
+            migrated = search.migrate_csv(new_manifest, cache)
             key = search.cache_key(str(old[1]["sha256"]), str(old[2]["sha256"]),
                                    search.FIXED_MEMORY)
             self.assertIn(key, migrated)
@@ -309,6 +336,287 @@ class RouteSearchTests(unittest.TestCase):
             self.assertEqual(result["nodes"], [0, 1, 4])
             self.assertEqual(result["selected_total_bytes"], 1100)
 
+    def test_same_version_variants_are_never_progress_edges(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            manifest, _ = self.write_inventory(
+                root, [b"0", b"1", b"2-raw", b"2-inflate", b"3"],
+                versions=[
+                    "1.0.0.0", "1.0.0.1", "1.0.0.2", "1.0.0.2", "1.0.0.3",
+                ],
+                version_ranks=[0, 1, 2, 2, 3],
+            )
+            images = search.load_inventory(manifest)
+            jobs = search.all_jobs(images)
+            self.assertFalse(any(job[0] == 2 and job[1] == 3 for job in jobs))
+
+            def edge(source: int, target: int) -> dict[str, object]:
+                return {
+                    "source": source, "target": target, "memory": 0x1000,
+                    "container": 100, "stage_start": 0xC0000,
+                    "margin": 100, "feasible": True,
+                }
+
+            # The only apparent path depends on changing variants without
+            # advancing the running version. The route must remain unreachable.
+            result = search.select_route(
+                [edge(1, 2), edge(2, 3), edge(3, 4)], images, 100,
+                root / "route.json", True,
+            )
+            self.assertEqual(result["status"], "unreachable")
+            self.assertNotIn(3, result["reachable_nodes"])
+
+    def test_transport_cost_accounts_for_every_clean_wire_message(self) -> None:
+        legacy = search.legacy_transport_cost(1024)
+        self.assertEqual(legacy["data_packets"], 7)
+        self.assertEqual(legacy["packets"], 13)
+        self.assertEqual(legacy["origin_mesh_bytes"], 1355)
+        self.assertEqual(legacy["manifest_bytes"], 224)
+        self.assertEqual(
+            legacy["origin_mesh_bytes"],
+            sum(
+                int(legacy[field])
+                for field in (
+                    "manifest_bytes", "data_bytes", "block_request_bytes",
+                    "proof_request_bytes", "proof_response_bytes",
+                )
+            ),
+        )
+
+        raw_v2 = search.v2_transport_cost(1024, 1024, 0, 0, 6)
+        self.assertEqual(raw_v2["data_packets"], 6)
+        self.assertEqual(raw_v2["packets"], 11)
+        self.assertEqual(raw_v2["origin_mesh_bytes"], 1359)
+        self.assertEqual(raw_v2["manifest_bytes"], 224)
+
+        compressed = search.v2_transport_cost(1024, 800, 800, 1, 5)
+        self.assertEqual(compressed["data_bytes"], 875)
+        self.assertEqual(compressed["origin_mesh_bytes"], 1120)
+        self.assertEqual(
+            search.linear_path_bytes(1120, int(compressed["packets"]), 2),
+            3 * 1120 + 3 * int(compressed["packets"]),
+        )
+
+        # The receive-inflate compatibility bridges are compiled with a single
+        # block in flight. Account for their five individual request frames,
+        # rather than the full RAK3401 receiver's clean [1, 2, 2] flights.
+        full = search.v2_transport_cost(5 * 1024, 5 * 1024, 0, 0, 30, 4)
+        bridge = search.v2_transport_cost(5 * 1024, 5 * 1024, 0, 0, 30, 1)
+        self.assertEqual(full["request_packets"], 3)
+        self.assertEqual(full["request_bytes"], 41)
+        self.assertEqual(bridge["request_packets"], 5)
+        self.assertEqual(bridge["request_bytes"], 55)
+        self.assertEqual(
+            int(bridge["origin_mesh_bytes"]) - int(full["origin_mesh_bytes"]), 14
+        )
+
+    def test_published_legacy_chain_and_new_bootstrap_cost_regressions(self) -> None:
+        old_payloads = [
+            89282, 45974, 105407, 149133, 110666,
+            119942, 138883, 173312, 270303, 173854,
+        ]
+        old = [search.legacy_transport_cost(size) for size in old_payloads]
+        self.assertEqual(sum(old_payloads), 1_376_756)
+        self.assertEqual(sum(int(cost["packets"]) for cost in old), 13_490)
+        self.assertEqual(
+            sum(int(cost["origin_mesh_bytes"]) for cost in old), 1_563_957
+        )
+
+        new_bootstrap = search.legacy_transport_cost(64_088)
+        self.assertEqual(new_bootstrap["data_packets"], 438)
+        self.assertEqual(new_bootstrap["packets"], 630)
+        self.assertEqual(new_bootstrap["origin_mesh_bytes"], 72_528)
+        with self.assertRaisesRegex(search.RouteSearchError, "between 0 and 8"):
+            search.linear_path_bytes(100, 1, 9)
+
+    def test_proof_sizing_matches_promote_odd_merkle_implementation(self) -> None:
+        for count in range(1, 33):
+            leaves = [bytes([index]) * 4 for index in range(count)]
+            for index in range(count):
+                self.assertEqual(
+                    search.proof_sibling_count(count, index),
+                    len(search.motalib.merkle_proof(leaves, index)),
+                )
+
+    def test_transport_objective_uses_running_source_capability(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            baseline = search.container_size(64)
+            manifest, _ = self.write_inventory(
+                root, [b"0", b"1", b"2", b"3", b"4"],
+                baseline_size=baseline,
+                transport_capabilities=[True, False, True, False, True],
+                transport_pipelines=[4, None, 1, None, 4],
+            )
+            images = search.load_inventory(manifest)
+
+            def edge(source: int, target: int, payload: int,
+                     *, compressed: int | None = None) -> dict[str, object]:
+                row: dict[str, object] = {
+                    "source": source, "target": target, "memory": 0x1000,
+                    "payload": payload, "container": search.container_size(payload),
+                    "stage_start": 0xC0000, "margin": 100, "feasible": True,
+                }
+                if compressed is not None:
+                    row.update({
+                        "payload_sha256": "a" * 64,
+                        "transport_encoder_sha256": "b" * 64,
+                        "v2_wire_bytes": compressed,
+                        "v2_deflate_bytes": compressed,
+                        "v2_deflate_blocks": 1,
+                        "v2_data_packets": (compressed + 170) // 171,
+                    })
+                return row
+
+            rows = [
+                edge(1, 2, 100), edge(1, 3, 100),
+                edge(2, 4, 1000, compressed=50), edge(3, 4, 100),
+            ]
+            transport = search.select_route(
+                rows, images, baseline, root / "transport.json", True,
+                objective="transport",
+            )
+            self.assertEqual(transport["nodes"], [0, 1, 2, 4])
+            # Node 0 is marked capable on purpose: bootstrap is nevertheless raw.
+            self.assertEqual(
+                transport["steps"][0]["transport"]["profile"], "legacy-160-raw"
+            )
+            # Target node 2's capability does not apply until node 2 is running.
+            self.assertEqual(
+                transport["steps"][1]["transport"]["profile"], "legacy-160-raw"
+            )
+            self.assertEqual(
+                transport["steps"][2]["transport"]["profile"], "v2-171-deflate"
+            )
+            self.assertEqual(
+                transport["steps"][2]["transport"]["request_pipeline"], 1
+            )
+            self.assertEqual(
+                transport["selected_total_bytes"],
+                sum(step["expected_container_size"] for step in transport["steps"]),
+            )
+            self.assertEqual(
+                transport["selected_total_transport_bytes"],
+                sum(
+                    step["transport"]["linear_path_bytes"]
+                    for step in transport["steps"]
+                ),
+            )
+
+            container = search.select_route(
+                rows, images, baseline, root / "container.json", True,
+                objective="container",
+            )
+            self.assertEqual(container["nodes"], [0, 1, 3, 4])
+
+    def test_transport_geometry_rank_includes_per_packet_stream_id_and_framing(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            baseline = search.container_size(64)
+            manifest, _ = self.write_inventory(
+                root, [b"0", b"1", b"2"], baseline_size=baseline,
+                transport_capabilities=[False, True, False],
+                transport_pipelines=[None, 1, None],
+            )
+            images = search.load_inventory(manifest)
+
+            def edge(memory: int, payload: int, wire: int, packets: int) -> dict[str, object]:
+                return {
+                    "source": 1, "target": 2, "memory": memory,
+                    "payload": payload, "container": search.container_size(payload),
+                    "stage_start": 0xC0000, "margin": 100, "feasible": True,
+                    "payload_sha256": f"{memory:064x}",
+                    "transport_encoder_sha256": "c" * 64,
+                    "v2_wire_bytes": wire, "v2_deflate_bytes": wire,
+                    "v2_deflate_blocks": 1, "v2_data_packets": packets,
+                }
+
+            # The 342-byte representation wins despite its larger patch because
+            # crossing 342 -> 343 adds a third repeated stream ID and frame.
+            rows = [edge(0x2000, 400, 343, 3), edge(0x3000, 500, 342, 2)]
+            result = search.select_route(
+                rows, images, baseline, root / "route.json", True,
+                objective="transport",
+            )
+            self.assertEqual(result["steps"][1]["inplace_memory"], "0x3000")
+            self.assertEqual(result["steps"][1]["transport"]["data_packets"], 2)
+
+    def test_transport_measurement_is_bound_to_payload_and_encoder_hashes(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            payload = root / "payload.patch"
+            payload.write_bytes(b"x" * 300)
+            tool = root / "motatool"
+            tool.write_bytes(b"exact encoder binary")
+            measured = {
+                "schema": 1, "payload_bytes": 300, "block_size": 1024,
+                "block_count": 1, "wire_bytes": 200, "deflate_bytes": 200,
+                "deflate_blocks": 1, "data_packets": 2,
+            }
+            completed = mock.Mock(returncode=0, stdout=json.dumps(measured), stderr="")
+            with mock.patch.object(search.subprocess, "run", return_value=completed) as run:
+                result = search.measure_transport_size(tool, payload)
+            self.assertEqual(result["payload_sha256"], search.sha256_file(payload))
+            self.assertEqual(result["transport_encoder_sha256"], search.sha256_file(tool))
+            self.assertEqual(result["v2_data_packets"], 2)
+            self.assertIn("transport-size", run.call_args.args[0])
+
+    def test_partial_or_missing_capable_transport_evidence_is_rejected(self) -> None:
+        with self.assertRaisesRegex(search.RouteSearchError, "partial transport"):
+            search.normalize_transport_stats(
+                {"payload_sha256": "a" * 64}, 100, "test row"
+            )
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            baseline = search.container_size(64)
+            manifest, _ = self.write_inventory(
+                root, [b"0", b"1", b"2"], baseline_size=baseline,
+                transport_capabilities=[False, True, False],
+                transport_pipelines=[None, 1, None],
+            )
+            images = search.load_inventory(manifest)
+            row = {
+                "source": 1, "target": 2, "memory": 0x1000,
+                "payload": 100, "container": search.container_size(100),
+                "stage_start": 0xC0000, "margin": 100, "feasible": True,
+            }
+            with self.assertRaisesRegex(search.RouteSearchError, "missing transport"):
+                search.select_route(
+                    [row], images, baseline, root / "route.json", True,
+                    objective="transport",
+                )
+
+    def test_transport_cache_cannot_mix_or_reuse_a_different_encoder(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            baseline = search.container_size(64)
+            manifest, _ = self.write_inventory(
+                root, [b"0", b"1", b"2"], baseline_size=baseline,
+                transport_capabilities=[False, True, False],
+                transport_pipelines=[None, 1, None],
+            )
+            images = search.load_inventory(manifest)
+
+            def row(memory: int, encoder: str) -> dict[str, object]:
+                return {
+                    "source": 1, "target": 2, "memory": memory,
+                    "payload": 100, "container": search.container_size(100),
+                    "stage_start": 0xC0000, "margin": 100, "feasible": True,
+                    "payload_sha256": "a" * 64,
+                    "transport_encoder_sha256": encoder,
+                    "v2_wire_bytes": 50, "v2_deflate_bytes": 50,
+                    "v2_deflate_blocks": 1, "v2_data_packets": 1,
+                }
+
+            first = row(0x1000, "b" * 64)
+            self.assertTrue(search.cache_satisfies_job(first, True, "b" * 64))
+            self.assertFalse(search.cache_satisfies_job(first, True, "c" * 64))
+            with self.assertRaisesRegex(search.RouteSearchError, "mix different"):
+                search.select_route(
+                    [first, row(0x2000, "c" * 64)], images, baseline,
+                    root / "route.json", True, objective="transport",
+                )
+
     def test_fixed_outputs_never_follow_preexisting_symlinks(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)
@@ -345,6 +653,28 @@ class RouteSearchTests(unittest.TestCase):
             bad = json.loads(manifest.read_text())
             bad["images"][1]["path"] = "missing.bin"
             cases.append((bad, "missing inventory image"))
+            bad = json.loads(manifest.read_text())
+            bad["images"][1][search.TRANSPORT_CAPABILITY] = "yes"
+            cases.append((bad, "must be boolean"))
+            bad = json.loads(manifest.read_text())
+            del bad["images"][1][search.TRANSPORT_CAPABILITY]
+            cases.append((bad, "required by the inventory capability contract"))
+            bad = json.loads(manifest.read_text())
+            bad["images"][1][search.TRANSPORT_CAPABILITY_ALIAS] = True
+            cases.append((bad, "disagrees"))
+            bad = json.loads(manifest.read_text())
+            bad["images"][1][search.TRANSPORT_CAPABILITY] = True
+            cases.append((bad, "is required"))
+            bad = json.loads(manifest.read_text())
+            bad["images"][1][search.TRANSPORT_CAPABILITY] = True
+            bad["images"][1][search.TRANSPORT_PIPELINE_FIELD] = 0
+            cases.append((bad, "must be between"))
+            bad = json.loads(manifest.read_text())
+            del bad["images"][1]["version_rank"]
+            cases.append((bad, "required by the inventory edge policy"))
+            bad = json.loads(manifest.read_text())
+            bad["images"][1]["version_rank"] = 0
+            cases.append((bad, "one-to-one"))
             for number, (document, message) in enumerate(cases):
                 candidate = root / f"bad-{number}.json"
                 candidate.write_text(json.dumps(document), encoding="ascii")