pureboot: PI lint, tightened gates, and the self-update test suite

check_pi.py asserts the two link-time facts position independence rests on
(no absolute jmp/call; the info block within the image's first 256 bytes);
the size gates drop to 510 on the tinies for the trampoline word.

New per-chip tests beside the reworked protocol test: the planner units
(programming orders and their recovery properties, the surgery, staging
composition, boot-fuse decode, and the update preflight's error/warning
matrix over synthetic fuse bytes), the relocated-copy sweep (the identical
image installed one slot lower serves the full command set — the PI
acceptance test, and the one that caught the temporary-buffer trap), and
the self-update end-to-end: --update-loader to a re-timed build
(pureboot9, byte-different by PUREBOOT_TIMEOUT alone), then every
power-fail phase killed mid-write, restarted from the runner's flash dump,
and completed by a re-run with the application intact throughout. The mega
rounds run the BOOTRST-unprogrammed profile: the fixture application's 'L'
jump is the application-owned loader entry that profile relies on.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

CMakeLists.txt

@@ -118,6 +118,15 @@ endif()
 # the trampoline word just below the loader on the tinies (host-side vector
 # surgery points it at the application). --pmem-wrap-around models AVR's
 # modulo-flash PC where the flash is big enough to need it.
+#
+# The image is position-independent (check_pi.py asserts the two link-time
+# facts that make it so), and on the tinies its budget is 510, not 512: the
+# slot's last word is the trampoline the host composes — the resident slot's
+# holds the application entry, and a staging copy's holds the jump through
+# which it reaches the loader it installed. The activation window is a
+# compile-time constant; a different PUREBOOT_TIMEOUT builds the re-timed
+# binary a self-update then installs.
+set(PUREBOOT_TIMEOUT 8 CACHE STRING "pureboot activation window, seconds")
 if(LIBAVR_MCU STREQUAL "attiny13a")
 	set(_pb_flash 1024)
 	set(_pb_wrap "")
@@ -125,6 +134,7 @@ if(LIBAVR_MCU STREQUAL "attiny13a")
 	set(_pb_hz 9600000)
 	set(_pb_baud 57600)
 	set(_pb_eeprom 64)
+	set(_pb_limit 510)
 elseif(LIBAVR_MCU STREQUAL "attiny85")
 	set(_pb_flash 8192)
 	set(_pb_wrap -Wl,--pmem-wrap-around=8k)
@@ -132,6 +142,7 @@ elseif(LIBAVR_MCU STREQUAL "attiny85")
 	set(_pb_hz 8000000)
 	set(_pb_baud 57600)
 	set(_pb_eeprom 512)
+	set(_pb_limit 510)
 else()
 	set(_pb_flash 32768)
 	set(_pb_wrap -Wl,--pmem-wrap-around=32k)
@@ -139,6 +150,7 @@ else()
 	set(_pb_hz 16000000)
 	set(_pb_baud 115200)
 	set(_pb_eeprom 1024)
+	set(_pb_limit 512)
 endif()
 math(EXPR _pb_base "${_pb_flash} - 512")
 math(EXPR _pb_base_hex "${_pb_base}" OUTPUT_FORMAT HEXADECIMAL)
@@ -150,13 +162,22 @@ endif()
 
 add_executable(pureboot pureboot/pureboot.cpp)
 target_link_libraries(pureboot PRIVATE libavr)
+target_compile_definitions(pureboot PRIVATE PUREBOOT_TIMEOUT=${PUREBOOT_TIMEOUT})
 target_link_options(pureboot PRIVATE -nostartfiles -Wl,--section-start=.text=${_pb_base_hex}
                     -Wl,--defsym=pureboot_app=${_pb_app} ${_pb_wrap})
 add_custom_command(TARGET pureboot POST_BUILD COMMAND ${CMAKE_SIZE} $<TARGET_FILE:pureboot>)
 if(PROJECT_IS_TOP_LEVEL)
 	add_test(NAME pureboot.size
 	         COMMAND ${CMAKE_COMMAND} -DSIZE_TOOL=${CMAKE_SIZE} -DELF=$<TARGET_FILE:pureboot>
-	                 -DLIMIT=512 -P ${CMAKE_CURRENT_SOURCE_DIR}/test/check_size.cmake)
+	                 -DLIMIT=${_pb_limit} -P ${CMAKE_CURRENT_SOURCE_DIR}/test/check_size.cmake)
+	if(Python3_FOUND)
+		add_test(NAME pureboot.pi
+		         COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/check_pi.py
+		                 ${CMAKE_OBJDUMP} ${CMAKE_NM} $<TARGET_FILE:pureboot> ${_pb_base_hex})
+		add_test(NAME pureboot.planner
+		         COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/test_planner.py
+		                 ${CMAKE_CURRENT_SOURCE_DIR}/pureboot/pureboot.py)
+	endif()
 
 	# The protocol test flashes this fixture through the loader with the real
 	# host tool and expects its banner after the hand-over; a normally linked
@@ -173,5 +194,33 @@ if(PROJECT_IS_TOP_LEVEL)
 		                 ${CMAKE_CURRENT_SOURCE_DIR}/pureboot/pureboot.py
 		                 ${CMAKE_BINARY_DIR}/pbtest-work)
 		set_tests_properties(pureboot.protocol PROPERTIES TIMEOUT 180)
+
+		# The position-independence acceptance test: the identical image,
+		# installed one slot lower, must serve the full command set.
+		add_test(NAME pureboot.reloc
+		         COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/pbreloc.py
+		                 ${PB_DEVICE} $<TARGET_FILE:pureboot> ${LIBAVR_MCU} ${_pb_hz} ${_pb_base_hex}
+		                 ${_pb_page} ${_pb_baud} ${CMAKE_CURRENT_SOURCE_DIR}/pureboot/pureboot.py
+		                 ${CMAKE_BINARY_DIR}/pbreloc-work)
+		set_tests_properties(pureboot.reloc PROPERTIES TIMEOUT 180
+		                     ENVIRONMENT "PB_OBJCOPY=${CMAKE_OBJCOPY}")
+
+		# The self-update end-to-end: the re-timed build (same source, only
+		# PUREBOOT_TIMEOUT differs — a byte-different image) replaces the
+		# resident through --update-loader, with every power-fail phase
+		# rehearsed from the runner's flash dumps.
+		add_executable(pureboot9 pureboot/pureboot.cpp)
+		target_link_libraries(pureboot9 PRIVATE libavr)
+		target_compile_definitions(pureboot9 PRIVATE PUREBOOT_TIMEOUT=9)
+		target_link_options(pureboot9 PRIVATE -nostartfiles -Wl,--section-start=.text=${_pb_base_hex}
+		                    -Wl,--defsym=pureboot_app=${_pb_app} ${_pb_wrap})
+		add_test(NAME pureboot.update
+		         COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/pbupdate.py
+		                 ${PB_DEVICE} $<TARGET_FILE:pureboot> $<TARGET_FILE:pureboot9> ${LIBAVR_MCU}
+		                 ${_pb_hz} ${_pb_base_hex} ${_pb_page} ${_pb_baud} $<TARGET_FILE:pbapp>.bin
+		                 ${CMAKE_CURRENT_SOURCE_DIR}/pureboot/pureboot.py
+		                 ${CMAKE_BINARY_DIR}/pbupdate-work)
+		set_tests_properties(pureboot.update PROPERTIES TIMEOUT 600
+		                     ENVIRONMENT "PB_OBJCOPY=${CMAKE_OBJCOPY}")
 	endif()
 endif()

test/check_pi.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""Position-independence lint for the pureboot image.
+
+The self-staging design lets the identical binary run from any 512-byte
+slot, which holds only if nothing in the image addresses itself absolutely.
+Two link-time facts guarantee it, both asserted here from the built ELF:
+
+  1. No absolute jmp/call opcodes — all control flow is PC-relative
+     (rjmp/rcall/ijmp/icall). -mrelax normally guarantees this; a code
+     change that grows a branch out of relaxation range would break it
+     silently.
+  2. The info block sits within the image's first 256 bytes: the 'b'
+     command rebuilds its address as (running slot high byte : low byte of
+     the link address), which needs the offset to fit that low byte.
+
+Usage: check_pi.py <objdump> <nm> <elf> <text_start_hex>
+"""
+
+import re
+import subprocess
+import sys
+
+
+def main():
+    objdump, nm, elf, text_start = sys.argv[1:]
+    text_start = int(text_start, 0)
+
+    listing = subprocess.run([objdump, "-d", elf], capture_output=True, text=True, check=True).stdout
+    absolute = [
+        line
+        for line in listing.splitlines()
+        if re.search(r"\t(jmp|call)\t", line)
+    ]
+    if absolute:
+        print("FAIL: absolute control flow in the image:")
+        print("\n".join(absolute))
+        sys.exit(1)
+
+    symbols = subprocess.run([nm, "-C", elf], capture_output=True, text=True, check=True).stdout
+    info = [line for line in symbols.splitlines() if "flash_table" in line and "::storage" in line]
+    if len(info) != 1:
+        print(f"FAIL: expected one info-block storage symbol, found {len(info)}")
+        sys.exit(1)
+    offset = int(info[0].split()[0], 16) - text_start
+    if not 0 <= offset < 256:
+        print(f"FAIL: info block at image offset {offset:#x}, must sit in the first 256 bytes")
+        sys.exit(1)
+
+    print(f"PI lint: control flow PC-relative, info block at offset {offset:#x}")
+
+
+if __name__ == "__main__":
+    main()

test/pbapp.cpp

@@ -1,8 +1,14 @@
-// Test-fixture application for the pureboot protocol test: prints "APP" on
+// Test-fixture application for the pureboot protocol tests: prints "APP" on
-// the chip's serial link (the same link the loader uses) and idles — the
+// the chip's serial link (the same link the loader uses) — the proof that
-// proof that the loader's hand-over, and on the tinies the host's
+// the loader's hand-over, and on the tinies the host's reset-vector
-// reset-vector surgery, actually launched it. Linked normally (crt, vectors
+// surgery, actually launched it. Linked normally (crt, vectors at 0); on
-// at 0); on the tinies its reset vector is the rjmp the host re-homes.
+// the tinies its reset vector is the rjmp the host re-homes.
+//
+// On the mega it then listens, and an 'L' makes it jump into the resident
+// loader — the application-owned loader entry a BOOTRST-unprogrammed mega
+// relies on (reset always boots the application there), exercised by the
+// self-update tests. The tinies idle: reset reaches their loader through
+// the patched vector, so the application owes it nothing.
 #include <libavr/libavr.hpp>
 
 using namespace avr::literals;
@@ -27,6 +33,12 @@ struct link {
 	{
 		tx_t::write(static_cast<std::uint8_t>(c));
 	}
+	[[noreturn]] static void idle()
+	{
+		for (;;)
+			if (tx_t::read_blocking() == 'L')
+				reinterpret_cast<void (*)()>((avr::hw::db.mem.flash_size - 512) / 2)();
+	}
 };
 
 template <avr::hertz_t C>
@@ -36,6 +48,11 @@ struct link<C, false> {
 	{
 		tx_t::write(static_cast<std::uint8_t>(c));
 	}
+	[[noreturn]] static void idle()
+	{
+		while (true) {
+		}
+	}
 };
 
 } // namespace
@@ -46,6 +63,5 @@ int main()
 	link<dev::clock>::tx('A');
 	link<dev::clock>::tx('P');
 	link<dev::clock>::tx('P');
-	while (true) {
+	link<dev::clock>::idle();
-	}
 }

test/pbreloc.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+"""Position-independence acceptance test: the identical pureboot binary,
+flashed one slot below the resident loader, must serve the complete command
+set from there. The resident installs it (through-word composed by the host
+layer), 'J' transfers control, and every command is exercised against the
+staged copy — the info block must come back byte-identical, the write guard
+must protect the staged copy's own slot and permit the resident's, and the
+staged copy must be able to rewrite the resident slot verbatim.
+
+Usage: pbreloc.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
+                  <baud> <tool_py> <workdir>
+"""
+
+import os
+import subprocess
+import sys
+
+
+def fail(message):
+    print(f"FAIL: {message}")
+    sys.exit(1)
+
+
+def main():
+    device_bin, elf, mcu, hz, base_hex, page, baud, tool, workdir = sys.argv[1:]
+    base, page, baud = int(base_hex, 0), int(page), int(baud)
+    stage = base - 512
+    sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
+    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+    import pbsim
+    import pureboot as pb
+
+    os.makedirs(workdir, exist_ok=True)
+    objcopy = os.environ.get("PB_OBJCOPY", "avr-objcopy")
+    image_path = os.path.join(workdir, "pureboot.bin")
+    subprocess.run([objcopy, "-O", "binary", elf, image_path], check=True)
+    image = open(image_path, "rb").read()
+
+    device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, os.path.join(workdir, "dump.bin"))
+    try:
+        port = pb.Port(device.pty, baud)
+        loader = pb.Loader(port)
+        info = loader.connect(25)
+        if info.base != base:
+            fail(f"info reports base {info.base:#06x}")
+        resident_info = info.raw
+
+        # Install the staging copy exactly as the update flow would.
+        staged = pb.staging_content(image, info)
+        pb.write_differing(loader, stage, staged)
+
+        # Enter it; from here on, every command runs in the relocated copy.
+        staged_info = loader.enter_copy(stage, 25)
+        if staged_info.raw != resident_info:
+            fail(f"staged info {staged_info.raw.hex()} != resident info {resident_info.hex()}")
+
+        # 'R' from the staged copy already proved itself in the install
+        # verify; 'F' must answer 4 bytes (values are unmodeled in simavr).
+        if len(loader.read_fuses()) != 4:
+            fail("fuse read from the staged copy")
+
+        # EEPROM round-trip through the staged copy.
+        pattern = bytes(range(0x50, 0x60))
+        loader.write_eeprom(0, pattern)
+        if loader.read_eeprom(0, len(pattern)) != pattern:
+            fail("EEPROM round-trip through the staged copy")
+
+        # The guard, both ways: its own slot refused (drained, unchanged),
+        # the resident slot writable.
+        before = loader.read_flash(stage, page)
+        loader.write_page(stage, bytes(page))
+        if loader.read_flash(stage, page) != before:
+            fail("the staged copy's guard let its own slot change")
+        marker = bytes((i * 3) & 0xFF for i in range(page))
+        loader.write_page(base, marker)
+        if loader.read_flash(base, page) != marker:
+            fail("the staged copy could not write the resident slot")
+
+        # Restore the resident image through the staged copy, then 'J' back
+        # into it and prove it lives.
+        resident = image + b"\xff" * (512 - len(image))
+        pb.write_differing(loader, base, resident)
+        back_info = loader.enter_copy(base, 25)
+        if back_info.raw != resident_info:
+            fail("the restored resident does not serve its info block")
+        port.close()
+    finally:
+        device.stop()
+    print("pbreloc: the relocated copy serves the full command set")
+
+
+if __name__ == "__main__":
+    main()

test/pbsim.py

@@ -0,0 +1,61 @@
+"""Shared simavr harness for the pureboot tests: spawn the device runner,
+hand out its pty, restart it from a flash dump (the power-fail path), and
+keep its chatter out of undrained pipes."""
+
+import os
+import signal
+import subprocess
+
+
+class Device:
+    def __init__(self, binary, elf, mcu, hz, base_hex, page, baud, dump, reset_hex=None, resume=None):
+        cmd = [binary, elf, mcu, hz, base_hex, str(page), str(baud), dump]
+        if reset_hex is not None or resume is not None:
+            cmd.append(reset_hex if reset_hex is not None else ("0" if mcu != "atmega328p" else base_hex))
+        if resume is not None:
+            cmd.append(resume)
+        self.log = open(dump + ".log", "a")
+        self.proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=self.log, text=True)
+        self.dump = dump
+        self.pty = None
+        for _ in range(50):
+            line = self.proc.stdout.readline()
+            if not line:
+                break
+            if line.startswith("PB_PTY"):
+                self.pty = line.split()[1]
+                break
+        if not self.pty:
+            self.stop()
+            raise RuntimeError("device did not report a pty")
+
+    def reset(self):
+        """The external reset line: SIGUSR1 re-enters at the reset vector."""
+        self.proc.send_signal(signal.SIGUSR1)
+
+    def power_fail(self):
+        """SIGTERM: the runner dumps its flash and exits — the image a
+        restart resumes from."""
+        self.stop()
+        return self.dump
+
+    def stop(self):
+        self.proc.terminate()
+        try:
+            self.proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            self.proc.kill()
+        self.log.close()
+
+
+def run_tool(tool, pty, baud, *args, timeout=180):
+    result = subprocess.run(
+        [os.environ.get("PYTHON", "python3"), tool, "--port", pty, "--baud", str(baud), "--wait", "25", *args],
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+    )
+    print(result.stdout, end="")
+    if result.returncode != 0:
+        raise RuntimeError(f"tool exited {result.returncode}: {result.stderr.strip()}")
+    return result.stdout

test/pbupdate.py

@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+"""Self-update end-to-end: an application is flashed, then the loader
+replaces itself with a re-timed build through the host tool's
+--update-loader — and the power-fail phases of that update are rehearsed by
+killing the simulated device mid-write, restarting it from its flash dump,
+and letting a re-run complete the update.
+
+The mega runs the BOOTRST-unprogrammed profile (reset boots the application;
+the fixture application's 'L' jump is the application-owned loader entry),
+with --assume-fuses standing in for the fuse read simavr cannot model. The
+tinies reset into a loader at every phase by construction — the t13a because
+its staging slot carries the reset vector itself, the t85 through the word-0
+redirect the tool plants around the resident rewrite.
+
+Usage: pbupdate.py <device_bin> <pureboot_elf> <update_elf> <mcu> <hz>
+                   <base_hex> <page> <baud> <app_bin> <tool_py> <workdir>
+"""
+
+import os
+import subprocess
+import sys
+
+
+def fail(message):
+    print(f"FAIL: {message}")
+    sys.exit(1)
+
+
+def rjmp_decode(word, at, flash_words):
+    """Written against the instruction-set definition, not with the tool's
+    encoder, so an encoding bug cannot verify itself."""
+    if word & 0xF000 != 0xC000:
+        fail(f"word at {at * 2:#06x} is {word:#06x}, not an rjmp")
+    offset = word & 0x0FFF
+    if offset >= 0x800:
+        offset -= 0x1000
+    return (at + 1 + offset) % flash_words
+
+
+class PowerFail(Exception):
+    pass
+
+
+MEGA_FUSES = "ffffffdd"  # high 0xdd: BOOTSZ = 1 KB, BOOTRST unprogrammed
+
+
+def make_fault_loader(pb, base, kill_region, kill_hits, device):
+    """A Loader whose write_page kills the device (or, with device=None,
+    just the host) at the Nth write into a region; the sequence
+    stage->resident->stage distinguishes the install from the restore."""
+
+    class FaultLoader(pb.Loader):
+        def __init__(self, port):
+            super().__init__(port)
+            self.seen_resident = False
+            self.hits = 0
+
+        def write_page(self, address, data):
+            if address >= base:
+                phase = "resident"
+                self.seen_resident = True
+            elif address >= base - 512:
+                phase = "stage_restore" if self.seen_resident else "stage"
+            else:
+                phase = "app"
+            if phase == kill_region:
+                self.hits += 1
+                if self.hits == kill_hits:
+                    if device is not None:
+                        device.power_fail()
+                    raise PowerFail(f"{kill_region} write {kill_hits}")
+            super().write_page(address, data)
+
+    return FaultLoader
+
+
+def main():
+    (device_bin, elf, update_elf, mcu, hz, base_hex, page, baud, app_bin, tool, workdir) = sys.argv[1:]
+    base, page, baud = int(base_hex, 0), int(page), int(baud)
+    mega = mcu == "atmega328p"
+    reset_hex = "0" if mega else None  # the mega runs BOOTRST-unprogrammed here
+    sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
+    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+    import pbsim
+    import pureboot as pb
+
+    os.makedirs(workdir, exist_ok=True)
+    objcopy = os.environ.get("PB_OBJCOPY", "avr-objcopy")
+    images = {}
+    for name, source in (("v0", elf), ("v9", update_elf)):
+        path = os.path.join(workdir, name + ".bin")
+        subprocess.run([objcopy, "-O", "binary", source, path], check=True)
+        images[name] = open(path, "rb").read()
+    if images["v0"] == images["v9"]:
+        fail("the update image is byte-identical to the resident build")
+    dump = os.path.join(workdir, "dump.bin")
+    state = os.path.join(workdir, "update.pbstate")
+    fuses = bytes.fromhex(MEGA_FUSES) if mega else None
+
+    def connect(device):
+        port = pb.Port(device.pty, baud)
+        if mega:
+            # Reset boots the application here; its 'L' is the loader entry.
+            # To a live loader the same byte is an ignored command.
+            port.read_available(0.5)
+            port.write(b"L")
+        loader = pb.Loader(port)
+        loader.connect(25)
+        return port, loader
+
+    def padded(image):
+        return image + b"\xff" * (512 - len(image))
+
+    def resident_bytes(loader):
+        return loader.read_flash(base, 256) + loader.read_flash(base + 256, 256)
+
+    def assert_state(loader, image, app_pages):
+        if resident_bytes(loader) != padded(image):
+            fail("resident loader does not match the update image")
+        stage = base - 512
+        got = loader.read_flash(stage, 256) + loader.read_flash(stage + 256, 256)
+        for address, data in app_pages.items():
+            if stage <= address < base:
+                if got[address - stage : address - stage + page] != data:
+                    fail(f"staging region page {address:#06x} not restored")
+
+    device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump, reset_hex=reset_hex)
+    final = "v0"
+    try:
+        # The application first — its planner output is the restore truth.
+        pbsim.run_tool(tool, device.pty, baud, "--flash", app_bin, "--stay")
+        port, loader = connect(device)
+        app_pages = pb.plan_flash(open(app_bin, "rb").read(), loader.info)
+        port.close()
+
+        # A clean CLI update, resident -> v9.
+        args = ["--update-loader", os.path.join(workdir, "v9.bin"), "--state", state, "--stay"]
+        if mega:
+            args += ["--assume-fuses", MEGA_FUSES]
+        out = pbsim.run_tool(tool, device.pty, baud, *args)
+        if "loader updated" not in out:
+            fail("update did not report success")
+        if os.path.exists(state):
+            fail("state file survived a completed update")
+        port, loader = connect(device)
+        assert_state(loader, images["v9"], app_pages)
+        loader.run_application()
+        if port.read_exact(3, 5.0) != b"APP":
+            fail("application does not banner after the update")
+        port.close()
+        final = "v9"
+        print("clean update: resident replaced, staging restored, application intact")
+
+        # Power-fail rehearsal: kill mid-phase, restart from the dump,
+        # re-run, and the update must still complete. Each round flips the
+        # direction so the flash is never already at its target. The mega's
+        # mid-resident-rewrite loss is exercised as a host crash instead:
+        # with BOOTRST unprogrammed and the resident mid-erase, a power loss
+        # there has no reset path into the staging copy — the documented
+        # cost of that profile (README).
+        for kill_region, kill_hits, kill_device in (
+            ("stage", 2, True),
+            ("resident", 1, not mega),
+            ("stage_restore", 2, True),
+        ):
+            device.reset()  # the previous round left the application running
+            port, loader = connect(device)
+            target = "v9" if resident_bytes(loader) == padded(images["v0"]) else "v0"
+            image_path = os.path.join(workdir, target + ".bin")
+            injected = make_fault_loader(pb, base, kill_region, kill_hits, device if kill_device else None)(port)
+            injected.info = loader.info
+            try:
+                pb.op_update_loader(injected, 25, image_path, state, fuses)
+                fail(f"{kill_region}: fault never triggered")
+            except PowerFail as event:
+                print(f"power fail injected: {event}")
+            port.close()
+            if kill_device:
+                device = pbsim.Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump,
+                                      reset_hex=reset_hex, resume=dump)
+            port, loader = connect(device)
+            pb.op_update_loader(loader, 25, image_path, state, fuses)
+            assert_state(loader, images[target], app_pages)
+            loader.run_application()
+            if port.read_exact(3, 5.0) != b"APP":
+                fail(f"{kill_region}: application lost after the resumed update")
+            port.close()
+            final = target
+            print(f"resumed after {kill_region} loss: update completed, application intact")
+    finally:
+        device.stop()
+
+    # Ground truth: the simulator's own flash against the final state, and
+    # on the tinies an independent decode of the reset routing.
+    flash = open(dump, "rb").read()
+    if flash[base : base + 512] != padded(images[final]):
+        fail("ground-truth resident region does not match the final image")
+    if not mega:
+        flash_words = (base + 512) // 2
+        word0 = flash[0] | (flash[1] << 8)
+        if rjmp_decode(word0, 0, flash_words) != base // 2:
+            fail("ground-truth reset vector does not land on the loader")
+        app = open(app_bin, "rb").read()
+        trampoline = flash[base - 2] | (flash[base - 1] << 8)
+        if rjmp_decode(trampoline, (base - 2) // 2, flash_words) != rjmp_decode(app[0] | (app[1] << 8), 0, flash_words):
+            fail("ground-truth trampoline does not land on the application entry")
+    print("pbupdate: clean update + all power-fail phases recovered")
+
+
+if __name__ == "__main__":
+    main()

test/test_planner.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""Host-tool unit tests — the pure planning and policy logic, no simulator:
+the flash-programming orders and their recovery properties, the reset-vector
+surgery, the staging-slot composition, the mega boot-fuse decode, and the
+update preflight's error/warning matrix (fuse combinations simavr cannot
+model reach it here as synthetic bytes).
+
+Usage: test_planner.py <tool_py>
+"""
+
+import sys
+
+
+def fail(message):
+    print(f"FAIL: {message}")
+    sys.exit(1)
+
+
+def expect_error(what, fn, *needles):
+    try:
+        fn()
+    except Exception as error:
+        for needle in needles:
+            if needle not in str(error):
+                fail(f"{what}: error lacks {needle!r}: {error}")
+        return
+    fail(f"{what}: no error raised")
+
+
+def info_of(pb, base, page, patch, flash):
+    raw = bytes((0x50, 0x42, 1, 0x1E, 0x93, 0x0B, page, base & 0xFF, base >> 8,
+                 0, 2, 1 if patch else 0))
+    info = pb.Info(raw)
+    assert info.flash_size == flash
+    return info
+
+
+def rjmp_decode(word, at, flash_words):
+    if word & 0xF000 != 0xC000:
+        fail(f"not an rjmp: {word:#06x}")
+    offset = word & 0x0FFF
+    if offset >= 0x800:
+        offset -= 0x1000
+    return (at + 1 + offset) % flash_words
+
+
+def main():
+    import os
+    sys.path.insert(0, os.path.dirname(os.path.abspath(sys.argv[1])))
+    import pureboot as pb
+
+    tiny = info_of(pb, 0x1E00, 64, True, 0x2000)
+    mega = info_of(pb, 0x7E00, 128, False, 0x8000)
+
+    # mega_boot: BOOTSZ words and the BOOTRST sense, DS40002061B §27.
+    for bits, start in ((0b11, 0x7E00), (0b10, 0x7C00), (0b01, 0x7800), (0b00, 0x7000)):
+        prog, at = pb.mega_boot((0xF8 | (bits << 1)) & ~1)
+        if not prog or at != start:
+            fail(f"mega_boot BOOTSZ={bits:02b} programmed: {prog} {at:#06x}")
+        prog, at = pb.mega_boot(0xF8 | (bits << 1) | 1)
+        if prog or at != start:
+            fail(f"mega_boot BOOTSZ={bits:02b} unprogrammed: {prog} {at:#06x}")
+
+    # Surgery: word 0 lands on the loader, the trampoline on the original
+    # entry — checked with an independent decoder.
+    app = bytes((0xC0 | 0x00, 0xC0)) + bytes((0x12,)) * 300  # rjmp .+0x00C0... entry word 0xC0C0
+    entry = rjmp_decode(app[0] | (app[1] << 8), 0, tiny.flash_size // 2)
+    pages = pb.plan_flash(app, tiny)
+    word0 = pages[0][0] | (pages[0][1] << 8)
+    if rjmp_decode(word0, 0, tiny.flash_size // 2) != tiny.base // 2:
+        fail("surgery: patched word 0 misses the loader")
+    tp = pages[tiny.base - 64]
+    tramp = tp[62] | (tp[63] << 8)
+    if rjmp_decode(tramp, (tiny.base - 2) // 2, tiny.flash_size // 2) != entry:
+        fail("surgery: trampoline misses the original entry")
+    expect_error("non-rjmp vector", lambda: pb.plan_flash(bytes((0x0C, 0x94)) + app[2:], tiny), "not an rjmp")
+    looped = bytearray(app)
+    word = pb.rjmp_to(0, tiny.base // 2, tiny.flash_size // 2)
+    looped[0], looped[1] = word & 0xFF, word >> 8
+    expect_error("read-back image", lambda: pb.plan_flash(bytes(looped), tiny), "read-back")
+    expect_error("oversize image", lambda: pb.plan_flash(bytes(0x1DFF), tiny), "application flash ends")
+
+    # Ordering: patched vector puts page 0 first and the trampoline second;
+    # a boot section puts page 0 last. Blank pages drop only when erased.
+    order = pb.covered(pages, tiny, skip_blank=False)
+    if order[0] != 0 or order[1] != tiny.base - 64:
+        fail(f"tiny order starts {order[:2]}, want page 0 then trampoline page")
+    if sorted(order[2:]) != order[2:]:
+        fail("tiny order tail not ascending")
+    mega_pages = pb.plan_flash(bytes((0xFF,)) * 600, mega)
+    morder = pb.covered(mega_pages, mega, skip_blank=False)
+    if morder[-1] != 0 or sorted(morder[:-1]) != morder[:-1]:
+        fail(f"mega order {morder}, want ascending with page 0 last")
+    blanky = {0: pages[0], 64: bytes((0xFF,)) * 64, 128: pages[128], tiny.base - 64: tp}
+    slim = pb.covered(blanky, tiny, skip_blank=True)
+    if 64 in slim or 0 not in slim or tiny.base - 64 not in slim:
+        fail(f"skip_blank order wrong: {slim}")
+
+    # Staging content: the identical image plus the through-word on a
+    # patched-vector chip; hard size clamps either way.
+    image = bytes(range(256)) * 2  # 512 B — too big for a tiny slot
+    expect_error("tiny staging size", lambda: pb.staging_content(image, tiny), "510")
+    staged = pb.staging_content(image[:508], tiny)
+    through = staged[510] | (staged[511] << 8)
+    if rjmp_decode(through, (tiny.base - 2) // 2, tiny.flash_size // 2) != tiny.base // 2:
+        fail("through-word misses the resident base")
+    if pb.staging_content(image, mega) != image:
+        fail("mega staging content should be the bare image")
+    expect_error("mega staging size", lambda: pb.staging_content(image + b"!", mega), "512")
+
+    # The embedded info block: found in a synthetic binary, absent in noise.
+    binary = bytes((0xAA,)) * 10 + tiny.raw + bytes((0xBB,)) * 10
+    found = pb.image_info(binary)
+    if found is None or found.raw != tiny.raw:
+        fail("image_info misses the embedded block")
+    if pb.image_info(bytes((0xAA,)) * 40) is not None:
+        fail("image_info invents a block")
+
+    # Update preflight: the full fuse matrix, plus target mismatch.
+    other = info_of(pb, 0x1E00, 32, True, 0x2000)
+    expect_error("wrong-target image", lambda: pb.update_preflight(binary, other, None), "another target")
+    expect_error("mega needs fuses", lambda: pb.update_preflight(bytes((0xAA,)) * 8 + mega.raw, mega, None),
+                 "--assume-fuses")
+    mega_image = bytes((0xAA,)) * 8 + mega.raw
+
+    def fuses(high):
+        return bytes((0xFF, 0xFF, 0xFF, high))
+
+    expect_error("BOOTSZ 512 B", lambda: pb.update_preflight(mega_image, mega, fuses(0xFE)),
+                 "cannot self-update", "BOOTSZ")
+    notes = pb.update_preflight(mega_image, mega, fuses(0xFD))  # 1 KB, BOOTRST unprogrammed
+    if not any("BOOTRST unprogrammed" in n for n in notes):
+        fail(f"1K/unprogrammed notes: {notes}")
+    notes = pb.update_preflight(mega_image, mega, fuses(0xFC))  # 1 KB, BOOTRST programmed
+    if not any("staging slot" in n for n in notes):
+        fail(f"1K/programmed notes: {notes}")
+    notes = pb.update_preflight(mega_image, mega, fuses(0xFA))  # 2 KB, BOOTRST programmed
+    if not any("application flash" in n for n in notes):
+        fail(f"2K/programmed notes: {notes}")
+    if pb.update_preflight(bytes((0xAA,)) * 8 + tiny.raw, tiny, None) != []:
+        fail("tiny preflight should pass without fuses")
+
+    # The walk-region refusal: BOOTRST aimed below the loader plus app data
+    # in the walk span errors without --force; erased spans and unprogrammed
+    # BOOTRST pass.
+    deep = {0x7800: bytes((1,)) * 128}
+    expect_error("walk region", lambda: pb.check_walk_region(deep, mega, fuses(0xFA), False), "--force")
+    pb.check_walk_region(deep, mega, fuses(0xFA), True)
+    pb.check_walk_region(deep, mega, fuses(0xFB), False)  # BOOTRST unprogrammed
+    pb.check_walk_region({0x7800: bytes((0xFF,)) * 128}, mega, fuses(0xFA), False)
+    pb.check_walk_region(deep, mega, None, False)  # fuses unknown: no check
+
+    print("test_planner: all planner and policy checks pass")
+
+
+if __name__ == "__main__":
+    main()