Preserve qualified release artifacts across resumed builds

.github/workflows/run-unit-tests.yml

@@ -412,6 +412,7 @@ jobs:
           python3 -B test/test_cascade_release_package.py
           python3 -B test/test_release_publication.py
           python3 -B test/test_local_release_full_selection.py
+          python3 -B test/test_local_release_capacity_archive.py
           python3 -B test/test_t096_full_memory.py
           python3 -B test/test_nrf52_ble_startup.py
           python3 -B test/test_nrf52_ble_trace.py

build_legacy.sh

@@ -6174,7 +6174,7 @@ run_logged_build_targets() {
       fi
       preserved_log=0
       echo "Building ${env} (${profile}); log: ${log_path}"
-      build_firmware "$env" > "$log_tmp" 2>&1
+      PIO_BUILD_JOBS_OVERRIDE=$pio_job_limit build_firmware "$env" > "$log_tmp" 2>&1
       build_status=$?
       if [ "$build_status" -eq 0 ] \
           && grep -Eq "^Skipping ${env}; (existing artifacts found|matching recipe and qualified artifacts found)" "$log_tmp" \

scripts/build_local_release.py

@@ -51,6 +51,145 @@ def run_logged(command: list[str], log: Path, environment: dict[str, str]) -> No
             raise RuntimeError(f"build failed; see {log}")
 
 
+def capacity_rejected_attempt(manifest_path: Path, version: str, source: str) -> dict:
+    """Recognize only the successful qualification followed by portable-size rejection."""
+    work = manifest_path.parent
+    if manifest_path.is_symlink() or not manifest_path.is_file():
+        raise ValueError("manifest is not a regular non-symlink file")
+    raw = manifest_path.read_bytes()
+    manifest = json.loads(raw)
+    target = manifest.get("target")
+    if not isinstance(target, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.+-]*", target):
+        raise ValueError("invalid target identity")
+    artifact_version = f"{version}-{source[:8]}"
+    stems = (f"{target}-{artifact_version}", f"{target}-ota-{artifact_version}")
+    if manifest_path.name not in {stem + ".capabilities.json" for stem in stems}:
+        raise ValueError("manifest filename does not match target/version/source")
+    stem = manifest_path.name.removesuffix(".capabilities.json")
+    # Even an unfamiliar sidecar can contain firmware or other failed-build
+    # evidence. Never consume such a package merely because its bin is absent.
+    siblings = list(work.glob(stem + "*"))
+    if siblings != [manifest_path] or list(work.rglob(stem + "*")) != [manifest_path]:
+        raise ValueError("manifest is not the sole same-stem output")
+    recipe = manifest.get("build_recipe")
+    checks = manifest.get("verification")
+    if (type(manifest.get("schema_version")) is not int or manifest["schema_version"] != 2
+            or manifest.get("platform") != "ESP32_PLATFORM"
+            or manifest.get("build_profile") != "standard"
+            or manifest.get("artifact_target") != target
+            or manifest.get("verified") is not True
+            or manifest.get("ota_update_verified") is not True
+            or manifest.get("ota_update_evidence") !=
+               "firmware fits both OTA application slots; otadata present"
+            or not isinstance(checks, list) or not checks
+            or any(not isinstance(check, dict) or check.get("present") is not True
+                   for check in checks)
+            or not isinstance(recipe, dict) or set(recipe) != {"schema_version", "sha256"}
+            or type(recipe.get("schema_version")) is not int or recipe["schema_version"] != 1
+            or not isinstance(recipe.get("sha256"), str)
+            or not re.fullmatch(r"[0-9a-f]{64}", recipe["sha256"])):
+        raise ValueError("qualification or original recipe binding is incomplete")
+    log = work / "build-logs" / f"{target}-standard.log"
+    if log.parent.is_symlink() or log.is_symlink() or not log.is_file():
+        raise ValueError("original standard log is missing or unsafe")
+    log_raw = log.read_bytes()
+    lines = log_raw.decode("utf-8").splitlines()
+    overflow = re.fullmatch(
+        r"ESP32 app image is (\d+) bytes, exceeding portable LoRa-OTA "
+        r"slot 0x10000\.\.0x150000 \((\d+) bytes\) by (\d+) bytes",
+        lines[-2] if len(lines) >= 2 else "")
+    marker = (f"DEFERRED: {target} (standard) exceeds the portable OTA slot; "
+              "the expanded FULL pass is required.")
+    version_line = f'    -DFIRMWARE_VERSION="{artifact_version}"'
+    qualification = re.compile(
+        r"Verified \d+ capability marker\(s\); manifest: " + re.escape(str(manifest_path)))
+    if (not overflow or lines[-1] != marker or version_line not in lines
+            or not any(qualification.fullmatch(line) for line in lines[:-2])):
+        raise ValueError("original log does not prove this exact portable-size rejection")
+    size, limit, excess = map(int, overflow.groups())
+    if limit != 0x140000 or size <= limit or excess != size - limit:
+        raise ValueError("portable overflow measurements are inconsistent")
+    return {"target": target, "manifest": manifest_path.name,
+            "manifest_sha256": hashlib.sha256(raw).hexdigest(),
+            "original_log": str(log.relative_to(work)),
+            "original_log_sha256": hashlib.sha256(log_raw).hexdigest(),
+            "build_recipe": recipe, "image_bytes": size,
+            "portable_limit_bytes": limit, "excess_bytes": excess,
+            "reason": "portable_slot_overflow"}
+
+
+def archive_capacity_rejected_attempts(work: Path, version: str, source: str,
+                                      *, project_root: Path = ROOT,
+                                      archive_parent: Path | None = None) -> dict:
+    """Preserve proven manifest-only rejections outside recursive package discovery."""
+    if not re.fullmatch(r"[0-9a-f]{40}", source):
+        raise ValueError("invalid release source commit")
+    if not re.fullmatch(r"v[0-9]+(?:\.[0-9]+){3}-halo-keymind-cascade-dev", version):
+        raise ValueError("invalid release firmware version")
+    work = Path(os.path.abspath(work))
+    parent = Path(os.path.abspath(archive_parent or work.parent))
+    if (work.resolve() != work or not work.is_dir() or parent.resolve() != parent
+            or parent == work or work in parent.parents):
+        raise ValueError("unsafe work/archive location")
+    lock_path = project_root / ".pio" / "build-sh.lock"
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    if lock_path.parent.resolve() != lock_path.parent or lock_path.is_symlink():
+        raise ValueError("unsafe build lock location")
+    with lock_path.open("a+") as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as error:
+            raise RuntimeError("active build owns the checkout; no attempts were archived") from error
+        candidates = []
+        preserved = []
+        for manifest in sorted(work.glob("*.capabilities.json")):
+            try:
+                candidates.append(capacity_rejected_attempt(manifest, version, source))
+            except (ValueError, OSError, TypeError, AttributeError) as error:
+                preserved.append({"manifest": manifest.name, "reason": str(error)})
+        if not candidates:
+            return {"archived": [], "preserved": preserved, "archive": None}
+        # Complete the scan and recheck every candidate before any file moves.
+        for record in candidates:
+            if capacity_rejected_attempt(work / record["manifest"], version, source) != record:
+                raise ValueError("capacity attempt changed during archival scan")
+        archive = Path(tempfile.mkdtemp(
+            prefix=f".capacity-rejected-{version}-{source[:8]}-", dir=parent))
+        (archive / "manifests").mkdir()
+        (archive / "logs").mkdir()
+        for record in candidates:
+            original = work / record["original_log"]
+            copy = archive / "logs" / original.name
+            shutil.copy2(original, copy)
+            if digest(copy) != record["original_log_sha256"]:
+                raise ValueError("original log changed while preserving rejection evidence")
+        index = {"format": "meshcore-capacity-rejected-attempts-v1",
+                 "source_commit": source, "firmware_version": version,
+                 "work_directory": str(work), "state": "prepared",
+                 "attempts": candidates, "moved_manifests": []}
+        def save_index() -> None:
+            temporary = archive / ".index.json.tmp"
+            temporary.write_text(json.dumps(index, indent=2, sort_keys=True,
+                                             ensure_ascii=True) + "\n", encoding="ascii")
+            temporary.replace(archive / "index.json")
+        save_index()
+        try:
+            for record in candidates:
+                manifest = work / record["manifest"]
+                if capacity_rejected_attempt(manifest, version, source) != record:
+                    raise ValueError("capacity attempt changed before archival move")
+                manifest.rename(archive / "manifests" / manifest.name)
+                index["moved_manifests"].append(manifest.name)
+                save_index()
+        except (ValueError, OSError):
+            index["state"] = "interrupted"
+            save_index()
+            raise
+        index["state"] = "archived"
+        save_index()
+        return {"archived": candidates, "preserved": preserved, "archive": str(archive)}
+
+
 def full_image_pattern(target: str, version: str, short_source: str) -> str:
     return f"{target}-full-*-{version}-{short_source}.bin"
 
@@ -400,6 +539,15 @@ def main() -> None:
     environment["PLATFORMIO_BUILD_DIR"] = str(utility_build_dir)
     work.mkdir(parents=True, exist_ok=True)
     environment["OUTPUT_DIR"] = str(work)
+    if args.resume:
+        archived = archive_capacity_rejected_attempts(work, args.firmware_version, source)
+        if archived["archived"]:
+            print(f'Archived {len(archived["archived"])} proven portable-capacity rejections: '
+                  f'{archived["archive"]}', flush=True)
+        print(f'Preserved {len(archived["preserved"])} other package manifests; '
+              'ordinary resume qualification still applies.', flush=True)
+        # All other packages retain the existing recipe/qualification checks
+        # in build_legacy.sh; this helper never makes them resumable.
     run_logged([
         "bash", "build_legacy.sh", "build-firmwares-logging-matrix",
         "--firmware-version", args.firmware_version,

scripts/generate_mesh_america_release.py

@@ -81,10 +81,9 @@ def read_stage(stage):
         if len(manifests) != group["target_count"]:
             raise ValueError("release target count differs from manifest")
         for row in manifests:
-            key = row["artifact_target"]
-            if key in seen or row.get("verified") is not True:
-                raise ValueError("duplicate or unqualified release profile")
-            seen.add(key)
+            logical_target = row["artifact_target"]
+            if row.get("verified") is not True or row.get("source_commit", source) != source:
+                raise ValueError("unqualified or wrong-source release profile")
             files = []
             for name in row["files"]:
                 if Path(name).name != name or name not in checksums:
@@ -93,12 +92,35 @@ def read_stage(stage):
                 if hashlib.sha256(path.read_bytes()).hexdigest() != checksums[name]:
                     raise ValueError(f"release asset checksum failed: {name}")
                 if path.suffix in SUFFIXES:
-                    if identity(name) != key or family not in name:
-                        raise ValueError("firmware filename differs from qualified profile")
+                    if family not in name:
+                        raise ValueError("firmware filename belongs to a different family")
                     files.append(path)
             if not files:
                 raise ValueError("qualified profile contains no firmware")
-            records.append((row, group["tag"], files))
+            identities = {identity(path.name) for path in files}
+            if len(identities) != 1:
+                raise ValueError("firmware filenames disagree on publication profile")
+            published_target = identities.pop()
+            # Full images keep their OTA identity while the filename adds its
+            # qualified build profile. Do not relabel the logical target.
+            infix = published_target.removeprefix(logical_target)
+            valid_infix = (published_target == logical_target
+                or (published_target.startswith(logical_target + "-") and (
+                    (infix == "-ota" and "lora" in row.get("ota_update_methods", []))
+                    or (row["build_profile"] == "full" and
+                        re.fullmatch(r"-full(?:-usb-wifi|-logging)?(?:-ota)?", infix))
+                    or (row["build_profile"] == "logging" and
+                        re.fullmatch(r"-logging(?:-ota)?", infix)))))
+            qualified_stem = published_target + "-" + family
+            if any(path.stem.removesuffix("-merged") != qualified_stem for path in files):
+                raise ValueError("firmware filename differs from exact release source/version")
+            if (not valid_infix or any(qualified_stem + suffix not in row["files"]
+                                      for suffix in (".capabilities.json", ".memory.json"))):
+                raise ValueError("firmware filename differs from qualified profile")
+            if published_target in seen:
+                raise ValueError("duplicate release publication profile")
+            seen.add(published_target)
+            records.append(({**row, "publication_target": published_target}, group["tag"], files))
     return plan, family, records
 
 
@@ -116,7 +138,7 @@ def generate(catalog, plan, family, records, profiles, repo):
                     by_hardware.setdefault((device["type"], hardware), set()).add(index)
         result["device"][index]["firmware"] = []
     for row, tag, files in records:
-        target = row["artifact_target"]
+        target = row.get("publication_target", row["artifact_target"])
         parsed = profiles[target]
         platform = PLATFORMS[row["platform"]]
         indexes = by_identity.get(target) or by_identity.get(row["target"])
@@ -201,7 +223,7 @@ def main():
     args = parser.parse_args()
     catalog = json.loads(args.catalog.read_text())
     plan, family, records = read_stage(args.stage)
-    targets = {row["artifact_target"] for row, _, _ in records}
+    targets = {row.get("publication_target", row["artifact_target"]) for row, _, _ in records}
     targets.update(identity(file["name"]) for device in catalog["device"]
                    for firmware in device["firmware"] for version in firmware["version"].values()
                    for file in version["files"])

scripts/package_cascade_release.py

@@ -290,18 +290,24 @@ def collect_local_release(directory, version, source):
         if any(category(record) != group.name for record in group_records):
             raise ValueError("local release firmware group disagrees with qualification")
         records.extend(group_records)
-    indexed = {record["manifest"]["artifact_target"]: record for record in records}
+    # Standard and expanded Full images may retain the same OTA identity.
+    # Match each independently qualified package by its exact file inventory;
+    # the logical target alone cannot distinguish those publication profiles.
+    indexed = {frozenset(str(p.relative_to(directory)) for p in record["files"]): record
+               for record in records}
     entries = manifest["firmware"]
     if (len(indexed) != len(records) or len(entries) != len(records)
             or len(records) != manifest["firmware_target_count"]
-            or len({entry["artifact_target"] for entry in entries}) != len(entries)):
+            or any(len(entry["files"]) != len(set(entry["files"])) for entry in entries)
+            or len({frozenset(entry["files"]) for entry in entries}) != len(entries)):
         raise ValueError("local release target inventory disagrees with qualification")
     for entry in entries:
-        record = indexed.get(entry["artifact_target"])
+        record = indexed.get(frozenset(entry["files"]))
         if record is None:
             raise ValueError("local release contains an unqualified target")
         qualified = record["manifest"]
         if (qualified["target"] != entry["target"]
+                or qualified["artifact_target"] != entry["artifact_target"]
                 or qualified.get("source_commit", source) != source
                 or qualified["platform"] != entry["platform"]
                 or qualified["build_profile"] != entry["profile"]
@@ -404,6 +410,8 @@ def main():
         summaries = []
         for record in group["records"]:
             manifest = record["manifest"]
+            if manifest.get("source_commit", args.commit) != args.commit:
+                raise ValueError("qualified firmware belongs to another source")
             file_links = []
             for path in record["files"]:
                 shutil.copy2(path, destination / path.name)
@@ -412,7 +420,10 @@ def main():
                     label = "merged.bin (USB)" if path.name.endswith("-merged.bin") else path.suffix[1:]
                     file_links.append(f'<a href="{html.escape(url)}">{html.escape(label)}</a>')
             methods = ", ".join(manifest.get("ota_update_methods", [])) or "USB"
-            summaries.append({**manifest, "sensor_profile": nrf52_sensor_profile(manifest),
+            # Older capability manifests inherit source provenance from the
+            # validated completed release inventory. Preserve their raw bytes.
+            summaries.append({**manifest, "source_commit": args.commit,
+                              "sensor_profile": nrf52_sensor_profile(manifest),
                               "files": [path.name for path in record["files"]]})
             rows.append(f"<tr><td>{html.escape(manifest['artifact_target'])}</td><td>{html.escape(release_profile_label(manifest))}</td><td>{html.escape(methods)}</td><td>{' | '.join(file_links)}</td></tr>")
         if group['key'] == 'utility':

test/test_firmware_build_recipe.py

@@ -680,5 +680,31 @@ printf 'RESULT:%s BATCH:%s\n' "$status" "$BATCH_BUILD_MODE"
             self.assertEqual(before, (original.read_bytes(), original.stat().st_ino, original.stat().st_mtime_ns))
 
 
+class CompilerJobLimitTests(unittest.TestCase):
+    def test_serial_matrix_passes_validated_job_limit_and_restores_caller_state(self):
+        for requested, expected in (("4", "4"), ("2", "2"), ("0", "8"), ("invalid", "8")):
+            with self.subTest(requested=requested), tempfile.TemporaryDirectory(prefix="mesh-compiler-jobs-") as temporary:
+                script = r'''
+source "$1/build_legacy.sh"
+OUTPUT_DIR=$2
+OPTION3_PIO_JOBS=$3
+PIO_BUILD_JOBS_OVERRIDE=3
+pio() { printf 'FORBIDDEN_PLATFORMIO\n' >&2; return 99; }
+build_firmware() {
+  printf '%s:%s:%s\n' "$1" "$PIO_BUILD_JOBS_OVERRIDE" "$BATCH_BUILD_MODE" >> "$OUTPUT_DIR/jobs-observed.txt"
+}
+run_logged_build_targets first second
+status=$?
+printf 'RESULT:%s RESTORED_JOBS:%s RESTORED_BATCH:%s\n' "$status" "$PIO_BUILD_JOBS_OVERRIDE" "$BATCH_BUILD_MODE"
+'''
+                result = subprocess.run(["bash", "-c", script, "jobs-test", str(ROOT), temporary, requested],
+                                        cwd=temporary, text=True, capture_output=True, timeout=10)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                self.assertNotIn("FORBIDDEN_PLATFORMIO", result.stdout + result.stderr)
+                self.assertIn("RESULT:0 RESTORED_JOBS:3 RESTORED_BATCH:0", result.stdout)
+                self.assertEqual((Path(temporary) / "jobs-observed.txt").read_text().splitlines(),
+                                 ["first:" + expected + ":1", "second:" + expected + ":1"])
+
+
 if __name__ == "__main__":
     unittest.main()

test/test_local_release_capacity_archive.py

@@ -0,0 +1,282 @@
+#!/usr/bin/env python3
+"""Exercise production archival and resume wiring without invoking PlatformIO."""
+
+import fcntl
+import hashlib
+import importlib.util
+import io
+import json
+import os
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+REPO = Path(os.environ.get("CAPACITY_ARCHIVE_REPO", str(Path(__file__).resolve().parents[1])))
+sys.path.insert(0, str(REPO / "scripts"))
+from firmware_build_recipe import package_occupied
+MODULE = Path(os.environ.get("CAPACITY_ARCHIVE_MODULE", str(REPO / "scripts/build_local_release.py")))
+SPEC = importlib.util.spec_from_file_location("capacity_archive_release", MODULE)
+BUILD = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(BUILD)
+VERSION = "v1.17.1.9-halo-keymind-cascade-dev"
+SOURCE = "8d31b3932de614ac38d238f1a01010bc5f362fe8"
+TARGET = "heltec_v4_repeater_lora_ota_no_external_sensors"
+
+
+class CapacityArchiveTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory(prefix="mesh-capacity-archive-test-")
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        self.work = self.root / ".releases" / (".build-" + VERSION + "-" + SOURCE[:8])
+        (self.work / "build-logs").mkdir(parents=True)
+
+    def attempt(self, target=TARGET, *, infix="-ota"):
+        stem = f"{target}{infix}-{VERSION}-{SOURCE[:8]}"
+        manifest = self.work / (stem + ".capabilities.json")
+        value = {"schema_version": 2, "target": target, "artifact_target": target,
+                 "platformio_env": "heltec_v4_repeater", "platform": "ESP32_PLATFORM",
+                 "build_profile": "standard", "verified": True,
+                 "ota_update_verified": True,
+                 "ota_update_evidence": "firmware fits both OTA application slots; otadata present",
+                 "verification": [{"capability": "ota.update.lora", "present": True,
+                                   "source": "packaged application", "evidence": "OTA: status"}],
+                 "build_recipe": {"schema_version": 1, "sha256": "a" * 64}}
+        manifest.write_text(json.dumps(value, indent=2) + "\n", encoding="ascii")
+        log = self.work / "build-logs" / (target + "-standard.log")
+        log.write_text(
+            f'    -DFIRMWARE_VERSION="{VERSION}-{SOURCE[:8]}"\n'
+            f"Verified 1 capability marker(s); manifest: {manifest}\n"
+            "ESP32 app image is 1334184 bytes, exceeding portable LoRa-OTA "
+            "slot 0x10000..0x150000 (1310720 bytes) by 23464 bytes\n"
+            f"DEFERRED: {target} (standard) exceeds the portable OTA slot; "
+            "the expanded FULL pass is required.\n", encoding="ascii")
+        return manifest, log
+
+    def archive(self, **kwargs):
+        return BUILD.archive_capacity_rejected_attempts(
+            self.work, VERSION, SOURCE, project_root=self.root, **kwargs)
+
+    def mutate(self, manifest, **updates):
+        value = json.loads(manifest.read_text()); value.update(updates)
+        manifest.write_text(json.dumps(value) + "\n", encoding="ascii")
+
+    def assert_preserved(self, manifest, log):
+        raw_manifest, raw_log = manifest.read_bytes(), log.read_bytes()
+        before = sorted(path.name for path in self.work.parent.iterdir())
+        result = self.archive()
+        self.assertEqual(result["archived"], [])
+        self.assertEqual(manifest.read_bytes(), raw_manifest)
+        self.assertEqual(log.read_bytes(), raw_log)
+        self.assertEqual(sorted(path.name for path in self.work.parent.iterdir()), before)
+        self.assertEqual(len(result["preserved"]), 1)
+
+    def test_real_log_shape_preserves_raw_bytes_and_leaves_log_in_place(self):
+        manifest, log = self.attempt()
+        raw_manifest, raw_log = manifest.read_bytes(), log.read_bytes()
+        self.assertTrue(package_occupied(manifest.with_name(
+            manifest.name.removesuffix(".capabilities.json"))))
+        result = self.archive(); archive = Path(result["archive"])
+        self.assertEqual(len(result["archived"]), 1)
+        self.assertFalse(manifest.exists())
+        self.assertFalse(package_occupied(manifest.with_name(
+            manifest.name.removesuffix(".capabilities.json"))))
+        self.assertEqual(log.read_bytes(), raw_log)
+        self.assertEqual((archive / "manifests" / manifest.name).read_bytes(), raw_manifest)
+        self.assertEqual((archive / "logs" / log.name).read_bytes(), raw_log)
+        self.assertEqual(archive.parent, self.work.parent)
+        self.assertNotIn(archive, self.work.rglob("*"))
+        index = json.loads((archive / "index.json").read_text(encoding="ascii"))
+        self.assertEqual(index["state"], "archived")
+        self.assertEqual(index["source_commit"], SOURCE)
+        record = index["attempts"][0]
+        self.assertEqual(record["manifest_sha256"], hashlib.sha256(raw_manifest).hexdigest())
+        self.assertEqual(record["original_log_sha256"], hashlib.sha256(raw_log).hexdigest())
+        self.assertEqual(record["image_bytes"] - record["portable_limit_bytes"], 23464)
+        self.assertEqual(self.archive()["archived"], [])
+
+    def test_plain_c6_filename_and_multiple_attempts_receive_unique_archive(self):
+        self.attempt("Xiao_C6_repeater_", infix="")
+        self.attempt()
+        first = self.archive()
+        self.assertEqual(len(first["archived"]), 2)
+        self.attempt()
+        second = self.archive()
+        self.assertNotEqual(first["archive"], second["archive"])
+        self.assertTrue(Path(first["archive"]).is_dir())
+
+    def test_log_only_precollection_size_deferral_needs_no_archive(self):
+        manifest, log = self.attempt("Station_G2_repeater_lora_ota_no_external_sensors")
+        manifest.unlink(); raw_log = log.read_bytes()
+        result = self.archive()
+        self.assertEqual(result["archived"], [])
+        self.assertIsNone(result["archive"])
+        self.assertEqual(log.read_bytes(), raw_log)
+
+    def test_any_firmware_memory_license_unknown_sidecar_or_directory_is_preserved(self):
+        for suffix in (".bin", "-merged.bin", ".uf2", ".zip", ".hex", ".memory.json",
+                       ".font-license.txt", ".unexpected", ".directory", ".bin-link"):
+            with self.subTest(suffix=suffix):
+                manifest, log = self.attempt()
+                sibling = self.work / (manifest.name.removesuffix(".capabilities.json") + suffix)
+                if suffix == ".directory": sibling.mkdir()
+                elif suffix == ".bin-link": sibling.symlink_to(self.root / "missing.bin")
+                else: sibling.write_bytes(b"original evidence or qualified firmware")
+                self.assert_preserved(manifest, log)
+                if sibling.is_dir(): sibling.rmdir()
+                else: sibling.unlink()
+
+    def test_unknown_failures_and_incomplete_or_nonstandard_contracts_are_preserved(self):
+        faults = [{"verified": False}, {"ota_update_verified": False},
+                  {"platform": "NRF52_PLATFORM"}, {"build_profile": "full"},
+                  {"artifact_target": "different"}, {"schema_version": 1},
+                  {"schema_version": 2.0},
+                  {"verification": []}, {"verification": [{"present": False}]},
+                  {"ota_update_evidence": "firmware does not fit both OTA application slots"},
+                  {"build_recipe": None}, {"build_recipe": {"schema_version": 1, "sha256": "x" * 64}},
+                  {"build_recipe": {"schema_version": True, "sha256": "a" * 64}},
+                  {"build_recipe": {"schema_version": 1, "sha256": "a" * 64, "unknown": 1}}]
+        for fault in faults:
+            with self.subTest(fault=fault):
+                manifest, log = self.attempt(); self.mutate(manifest, **fault)
+                self.assert_preserved(manifest, log)
+        manifest, log = self.attempt(); manifest.write_bytes(b"{malformed")
+        self.assert_preserved(manifest, log)
+
+    def test_nested_firmware_is_not_hidden_by_flat_manifest_scan(self):
+        manifest, log = self.attempt()
+        nested = self.work / "nested"; nested.mkdir()
+        (nested / (manifest.name.removesuffix(".capabilities.json") + ".bin")).write_bytes(b"firmware")
+        self.assert_preserved(manifest, log)
+
+    def test_numeric_source_identity_and_final_failure_consistency_are_required(self):
+        replacements = [("23464 bytes", "23465 bytes"), ("1310720 bytes", "1310721 bytes"),
+                        ("1334184 bytes", "1310719 bytes"),
+                        ("portable LoRa-OTA slot", "ota_0"),
+                        (SOURCE[:8], "deadbeef"), ("(standard)", "(full)"),
+                        ("Verified 1 capability", "Failed 1 capability"),
+                        ("; the expanded FULL pass is required.", "; unknown failure.")]
+        for old, new in replacements:
+            with self.subTest(old=old):
+                manifest, log = self.attempt()
+                log.write_text(log.read_text().replace(old, new), encoding="ascii")
+                self.assert_preserved(manifest, log)
+        manifest, log = self.attempt()
+        with log.open("a") as output: output.write("FAILED: unrelated compiler error\n")
+        self.assert_preserved(manifest, log)
+
+    def test_wrong_source_version_or_target_filename_is_not_consumed(self):
+        for token, replacement in ((SOURCE[:8], "deadbeef"), (VERSION, VERSION.replace("1.9", "1.8")),
+                                   (TARGET, "different_target")):
+            with self.subTest(token=token):
+                manifest, log = self.attempt()
+                renamed = manifest.with_name(manifest.name.replace(token, replacement))
+                manifest.rename(renamed)
+                self.assert_preserved(renamed, log); renamed.unlink()
+
+    def test_symlink_manifest_log_log_directory_or_work_is_never_consumed(self):
+        manifest, log = self.attempt()
+        original = self.root / "original.capabilities.json"
+        manifest.rename(original); manifest.symlink_to(original)
+        self.assert_preserved(manifest, log)
+        manifest.unlink(); original.rename(manifest)
+        real_log = self.root / "original.log"; log.rename(real_log); log.symlink_to(real_log)
+        self.assert_preserved(manifest, log)
+        log.unlink(); real_log.rename(log)
+        real_logs = self.root / "original-logs"
+        log.parent.rename(real_logs); log.parent.symlink_to(real_logs, target_is_directory=True)
+        self.assert_preserved(manifest, log)
+        alias = self.root / "work-alias"; alias.symlink_to(self.work, target_is_directory=True)
+        with self.assertRaisesRegex(ValueError, "unsafe work/archive"):
+            BUILD.archive_capacity_rejected_attempts(alias, VERSION, SOURCE, project_root=self.root)
+
+    def test_archive_overlap_and_held_build_lock_leave_every_package_byte_untouched(self):
+        manifest, log = self.attempt(); raw_manifest, raw_log = manifest.read_bytes(), log.read_bytes()
+        for parent in (self.work, self.work / "nested"):
+            with self.subTest(parent=parent), self.assertRaisesRegex(ValueError, "unsafe work/archive"):
+                self.archive(archive_parent=parent)
+        lock_path = self.root / ".pio" / "build-sh.lock"; lock_path.parent.mkdir()
+        lock_path.write_bytes(b"original active build lock owner\n")
+        with lock_path.open("a+") as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with self.assertRaisesRegex(RuntimeError, "active build owns"):
+                self.archive()
+        self.assertEqual(lock_path.read_bytes(), b"original active build lock owner\n")
+        self.assertEqual(manifest.read_bytes(), raw_manifest)
+        self.assertEqual(log.read_bytes(), raw_log)
+        self.assertEqual(list(self.work.parent.glob(".capacity-rejected-*")), [])
+
+    def test_complete_scan_recheck_blocks_changes_before_any_move(self):
+        first, _ = self.attempt(); second, _ = self.attempt("second_repeater")
+        original = BUILD.capacity_rejected_attempt; calls = 0
+        def racing(*args):
+            nonlocal calls
+            calls += 1
+            record = original(*args)
+            if calls == 2: self.mutate(first, verified=False)
+            return record
+        with mock.patch.object(BUILD, "capacity_rejected_attempt", side_effect=racing):
+            with self.assertRaises(ValueError): self.archive()
+        self.assertTrue(first.exists()); self.assertTrue(second.exists())
+        self.assertEqual(list(self.work.parent.glob(".capacity-rejected-*")), [])
+
+    def test_late_failure_keeps_moved_raw_evidence_and_marks_partial_archive(self):
+        first, _ = self.attempt(); second, _ = self.attempt("second_repeater")
+        raw_first, raw_second = first.read_bytes(), second.read_bytes()
+        original = BUILD.capacity_rejected_attempt; calls = 0
+        def racing(*args):
+            nonlocal calls
+            calls += 1
+            if calls == 6: raise ValueError("injected late race")
+            return original(*args)
+        with mock.patch.object(BUILD, "capacity_rejected_attempt", side_effect=racing):
+            with self.assertRaisesRegex(ValueError, "injected late race"): self.archive()
+        archive, = self.work.parent.glob(".capacity-rejected-*")
+        self.assertFalse(first.exists()); self.assertTrue(second.exists())
+        self.assertEqual((archive / "manifests" / first.name).read_bytes(), raw_first)
+        self.assertEqual(second.read_bytes(), raw_second)
+        index = json.loads((archive / "index.json").read_text())
+        self.assertEqual(index["state"], "interrupted")
+        self.assertEqual(index["moved_manifests"], [first.name])
+
+    def test_resume_main_archives_before_matrix_and_nonresume_dryrun_do_not_archive(self):
+        (self.work / "build-logs").rmdir()
+        for arguments, expected in ((["--resume"], True), ([], False), (["--resume", "--dry-run"], False)):
+            with self.subTest(arguments=arguments):
+                order = []
+                def archive(*args):
+                    order.append("archive")
+                    return {"archived": [], "preserved": [], "archive": None}
+                def matrix(*args):
+                    order.append("matrix")
+                    raise RuntimeError("stop before any external build")
+                with mock.patch.object(BUILD, "ROOT", self.root), \
+                     mock.patch.object(BUILD, "git", side_effect=[SOURCE, ""]), \
+                     mock.patch.object(BUILD, "archive_capacity_rejected_attempts", side_effect=archive), \
+                     mock.patch.object(BUILD, "run_logged", side_effect=matrix), \
+                     mock.patch.object(BUILD, "resolve_pio_build_dir", return_value=self.root / ".pio/build"), \
+                     mock.patch.object(sys, "argv", ["build_local_release.py"] + arguments), \
+                     mock.patch("sys.stdout", new=io.StringIO()), \
+                     mock.patch("sys.stderr", new=io.StringIO()):
+                    if "--dry-run" in arguments: BUILD.main()
+                    else:
+                        with self.assertRaisesRegex(RuntimeError, "stop before"): BUILD.main()
+                self.assertEqual(order, ["archive", "matrix"] if expected else
+                                 ([] if "--dry-run" in arguments else ["matrix"]))
+
+    def test_explicit_release_ci_step_runs_capacity_archive_regressions(self):
+        workflow = Path(os.environ.get("CAPACITY_ARCHIVE_WORKFLOW", str(
+            REPO / ".github/workflows/run-unit-tests.yml"))).read_text(encoding="utf-8")
+        command = "python3 -B test/test_local_release_capacity_archive.py"
+        self.assertEqual(workflow.count(command), 1)
+        # This workflow lists host suites explicitly, rather than discovering
+        # test files. Keep the new resume boundary in the existing release step.
+        self.assertIn("python3 -B test/test_release_publication.py\n"
+                      "          python3 -B test/test_local_release_full_selection.py\n"
+                      "          " + command, workflow)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_release_publication.py

@@ -21,17 +21,19 @@ FAMILY = VERSION + "-" + SOURCE[:8]
 
 
 class ReleasePublicationTest(unittest.TestCase):
-    def record(self, directory, target, platform="ESP32_PLATFORM", sensor=""):
+    def record(self, directory, target, platform="ESP32_PLATFORM", sensor="", *,
+               publication_target=None, build_profile="auto"):
         directory.mkdir(parents=True, exist_ok=True)
-        stem = target + "-" + FAMILY
+        stem = (publication_target or target) + "-" + FAMILY
         names = [stem + suffix for suffix in (
             [".uf2", ".zip"] if platform == "NRF52_PLATFORM" else [".bin", "-merged.bin"])]
         names += [stem + ".capabilities.json", stem + ".memory.json"]
         for name in names:
             (directory / name).write_bytes((name + "\n").encode("ascii"))
         row = {"artifact_target": target, "target": target.removesuffix("-" + sensor + "-ota") if sensor else target,
-               "platform": platform, "build_profile": "auto", "verified": True,
-               "sensor_profile": sensor, "files": names, "reductions": []}
+               "platform": platform, "build_profile": build_profile, "verified": True,
+               "sensor_profile": sensor, "files": names, "reductions": [],
+               "source_commit": SOURCE, "ota_update_methods": ["lora"]}
         return row, [directory / name for name in names]
 
     def stage(self, directory):
@@ -140,10 +142,114 @@ class ReleasePublicationTest(unittest.TestCase):
                 catalogs.generate(template, plan, family, records,
                                   catalogs.target_profiles(targets), "mikecarper/MeshCore")
 
+    def test_catalog_accepts_standard_and_full_with_the_same_logical_identity(self):
+        with tempfile.TemporaryDirectory() as temp:
+            directory = Path(temp)
+            plan = self.stage(directory)
+            target = "Station_G2_repeater"
+            published = target + "-full-usb-wifi-ota"
+            row, _ = self.record(directory / "full-profiles", target,
+                                 publication_target=published, build_profile="full")
+            folder = directory / "full-profiles"
+            (folder / "TARGET-MANIFEST.json").write_text(json.dumps([row]))
+            (folder / "SHA256SUMS.txt").write_text("".join(
+                hashlib.sha256(p.read_bytes()).hexdigest() + "  " + p.name + "\n"
+                for p in sorted(folder.iterdir())))
+            plan["groups"].append({"key": "full-profiles", "tag": "full-profiles-" + FAMILY,
+                                   "target_count": 1})
+            (directory / "release-plan.json").write_text(json.dumps(plan))
+            plan, family, records = catalogs.read_stage(directory)
+            self.assertEqual([row["artifact_target"] for row, _, _ in records].count(target), 2)
+            targets = {row["publication_target"] for row, _, _ in records} | {
+                "Station_G2_companion_radio_full", target, "RAK_4631_repeater"}
+            result = catalogs.generate(self.template(), plan, family, records,
+                                       catalogs.target_profiles(targets), "mikecarper/MeshCore")
+            entries = result["device"][0]["firmware"]
+            self.assertEqual(len(entries), 3)
+            full = next(entry for entry in entries if entry["subTitle"].startswith(published))
+            self.assertIn("Full profile", full["subTitle"])
+            self.assertTrue(all("/full-profiles-" + FAMILY + "/" in file["url"]
+                                for file in full["version"][FAMILY]["files"]))
+            # Matching filenames do not authorize a profile for another board
+            # or a fabricated Full infix on a standard qualified build.
+            for change in ({"artifact_target": "Other_repeater"}, {"build_profile": "auto"},
+                           {"source_commit": "b" * 40}):
+                (folder / "TARGET-MANIFEST.json").write_text(json.dumps([{**row, **change}]))
+                (folder / "SHA256SUMS.txt").write_text("".join(
+                    hashlib.sha256(p.read_bytes()).hexdigest() + "  " + p.name + "\n"
+                    for p in sorted(folder.iterdir()) if p.name != "SHA256SUMS.txt"))
+                with self.subTest(change=change), self.assertRaises(ValueError):
+                    catalogs.read_stage(directory)
+            # A second source suffix cannot hide inside an otherwise familiar
+            # family filename while keeping the original cap/memory stems.
+            wrong = {**row, "files": list(row["files"])}
+            for index, name in enumerate(wrong["files"]):
+                if name.endswith(".bin"):
+                    extra = name.replace(FAMILY, FAMILY + "-bbbbbbbb")
+                    (folder / name).rename(folder / extra)
+                    wrong["files"][index] = extra
+            (folder / "TARGET-MANIFEST.json").write_text(json.dumps([wrong]))
+            (folder / "SHA256SUMS.txt").write_text("".join(
+                hashlib.sha256(p.read_bytes()).hexdigest() + "  " + p.name + "\n"
+                for p in sorted(folder.iterdir()) if p.name != "SHA256SUMS.txt"))
+            with self.assertRaisesRegex(ValueError, "exact release source/version"):
+                catalogs.read_stage(directory)
+
+    def test_local_release_distinguishes_standard_and_full_by_exact_qualified_files(self):
+        with tempfile.TemporaryDirectory() as temp:
+            directory = Path(temp)
+            target = "Station_G2_companion_radio_usb"
+            folder = directory / "firmware/companion"
+            standard, standard_files = self.record(folder, target)
+            full, full_files = self.record(folder, target,
+                publication_target=target + "-full-usb-wifi-ota", build_profile="full")
+            records = [{"manifest": row, "files": files} for row, files in
+                       ((standard, standard_files), (full, full_files))]
+            manifest = {"format": "meshcore-local-release-v1", "source_commit": SOURCE,
+                        "firmware_version": VERSION, "profile": "cascade", "publication": "local-only",
+                        "radio": {"frequency_mhz": 910.525, "bandwidth_khz": 62.5,
+                                  "spreading_factor": 7, "coding_rate": 5},
+                        "firmware_target_count": 2, "migration_board_role_count": 1,
+                        "firmware": [{"target": row["target"], "artifact_target": row["artifact_target"],
+                            "profile": row["build_profile"], "platform": row["platform"],
+                            "files": [str(p.relative_to(directory)) for p in files]}
+                            for row, files in ((standard, standard_files), (full, full_files))]}
+            migrations = directory / "esp32-partition-migration"
+            migrations.mkdir()
+            archive = migrations / ("example-" + FAMILY + "-migration.zip")
+            archive.write_bytes(b"qualified migration fixture")
+            bundle = directory / ("esp32-partition-migration-" + FAMILY + "-release.zip")
+            with zipfile.ZipFile(bundle, "w") as z:
+                z.writestr("release-manifest.json", "{}")
+                z.writestr(archive.name, archive.read_bytes())
+            def write_inventory():
+                (directory / "manifest.json").write_text(json.dumps(manifest))
+                (directory / "SHA256SUMS.txt").write_text("".join(
+                    hashlib.sha256(p.read_bytes()).hexdigest() + "  " + str(p.relative_to(directory)) + "\n"
+                    for p in sorted(directory.rglob("*")) if p.is_file() and p.name != "SHA256SUMS.txt"))
+            with patch.object(package, "collect_artifacts", return_value=records), \
+                 patch("package_esp32_partition_migration.BOARDS", {"example": {}}), \
+                 patch("build_esp32_partition_migration.verify_archive"):
+                write_inventory()
+                self.assertEqual(len(package.collect_local_release(directory, VERSION, SOURCE)[0]), 2)
+                for field, value in (("artifact_target", "Other_companion_radio_usb"),
+                                     ("profile", "auto")):
+                    original = manifest["firmware"][1][field]
+                    manifest["firmware"][1][field] = value
+                    write_inventory()
+                    with self.subTest(field=field), self.assertRaisesRegex(ValueError, "manifest disagrees"):
+                        package.collect_local_release(directory, VERSION, SOURCE)
+                    manifest["firmware"][1][field] = original
+                manifest["firmware"][1]["files"] = manifest["firmware"][0]["files"]
+                write_inventory()
+                with self.assertRaisesRegex(ValueError, "inventory disagrees"):
+                    package.collect_local_release(directory, VERSION, SOURCE)
+
     def test_stable_publication_notes_and_assets_are_ascii(self):
         with tempfile.TemporaryDirectory() as temp:
             base = Path(temp)
             row, files = self.record(base / "input", "Station_G2_companion_radio_usb")
+            row.pop("source_commit")  # Actual older cap manifests inherit validated release provenance.
             output = base / "stage"
             argv = ["package", "--input", str(base / "input"), "--output", str(output),
                     "--local-release", "--stable", "--version", "1.17.1.9", "--commit", SOURCE]
@@ -153,12 +259,23 @@ class ReleasePublicationTest(unittest.TestCase):
                 package.main()
             plan = json.loads((output / "release-plan.json").read_text())
             self.assertFalse(plan["groups"][0]["prerelease"])
+            staged_rows = json.loads((output / "companion/TARGET-MANIFEST.json").read_text())
+            self.assertEqual(staged_rows[0]["source_commit"], SOURCE)
+            self.assertNotIn("source_commit", row, "raw capability record was relabeled")
             notes = (output / "companion-notes.md").read_text(encoding="ascii")
             self.assertIn("Stable release", notes)
             self.assertNotIn("Development prerelease", notes)
             for path in output.rglob("*"):
                 if path.is_file():
                     path.read_text(encoding="ascii")
+            bad = {**row, "source_commit": "b" * 40}
+            bad_argv = list(argv)
+            bad_argv[bad_argv.index(str(output))] = str(base / "wrong-source-stage")
+            with patch.object(sys, "argv", bad_argv), patch.object(package, "collect_local_release",
+                    return_value=([{"manifest": bad, "files": files}], [],
+                                  {"frequency": "910.525", "bandwidth": "62.5", "sf": "7", "cr": "5"})), \
+                    self.assertRaisesRegex(ValueError, "another source"):
+                package.main()
 
     def test_local_release_preserves_selected_profiles_and_verifies_complete_inventory(self):
         with tempfile.TemporaryDirectory() as temp: