install/restore: auto-fallback to host TFTP when pod PSRAM is too small (#95)

## Summary

- `defib install/restore --tftp-via=auto` (the default when
`power=rack`) now does a pre-flight `GET /tftp` on the pod and falls
back to host TFTP if `psram_largest_free_block < total_bytes + 256 KiB`
headroom — instead of crashing partway through the staging POST with a
503 OOM.
- Strict `--tftp-via=pod` is unchanged — still errors on OOM, since the
user explicitly opted out of the host path.
- New `RackController.psram_can_fit(total_bytes, headroom_bytes=256 KiB)
-> (bool, stats)` — any transport error returns `(False, {})` so an
unreachable pod is treated the same as "won't fit", and the CLI cleanly
falls back rather than crashing the pre-check.

## Test plan

- [x] `uv run pytest tests/test_power_rack.py -x -v` — 29 pass, 4 new in
`TestPsramCanFit`
- [x] `uv run pytest tests/ -x --ignore=tests/fuzz` — full suite clean
- [x] `uv run ruff check src/ tests/` — clean
- [x] `uv run mypy src/defib --ignore-missing-imports` — clean
- [x] Live against pod `10.216.128.69`:
- 6 MB filler staged → `largest_free=2 MB` → `auto` falls back: *"Pod
PSRAM has 2016 KB contiguous free, need 6535 KB ..."*
- same condition with `--tftp-via=pod` → 503 OOM (strict mode preserved)
  - cleared pod (`DELETE /tftp`) → `auto` uses pod path normally

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Dmitry Ilyin <widgetii@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: widgetii

src/defib/cli/app.py

@@ -2174,6 +2174,26 @@ async def _cmd(cmd: str, timeout: float = 60.0, **kw: object) -> str:
         rootfs_name: rootfs_data,
     }
 
+    # --tftp-via=auto pre-flight: if the pod doesn't have enough
+    # contiguous PSRAM for the firmware, fall back to host TFTP.
+    # Surfaces "too-big rootfs" cleanly instead of OOMing the staging
+    # POST mid-way.  --tftp-via=pod stays strict (error on OOM, no
+    # silent fallback).
+    if use_pod_tftp and tftp_via == "auto":
+        assert isinstance(power_controller, RackController)
+        total_bytes = sum(len(d) for d in tftp_files.values())
+        fits, pod_stats = await power_controller.psram_can_fit(total_bytes)
+        if not fits:
+            _raw = pod_stats.get("psram_largest_free_block", 0)
+            largest = int(_raw) if isinstance(_raw, (int, float)) else 0
+            if output == "human":
+                console.print(
+                    f"  [yellow]Pod PSRAM has {largest // 1024} KB contiguous free, "
+                    f"need {total_bytes // 1024} KB for this install — falling back "
+                    f"to host TFTP.[/yellow]"
+                )
+            use_pod_tftp = False
+
     if not use_pod_tftp:
         # Host TFTP needs a NIC + host_ip; pod path needs neither.
         if not nic:
@@ -2893,6 +2913,24 @@ async def _send(cmd: str, timeout: float = 60.0) -> str:
         await transport.close()
         raise typer.Exit(1)
 
+    # Auto-fallback: if the pod's PSRAM can't fit the dump, drop to
+    # host TFTP rather than OOMing mid-stage.  Explicit --tftp-via=pod
+    # stays strict.
+    if use_pod_tftp and tftp_via == "auto":
+        assert isinstance(power_controller, RackController)
+        total_bytes = sum(len(d) for _, d in partitions)
+        fits, pod_stats = await power_controller.psram_can_fit(total_bytes)
+        if not fits:
+            _raw = pod_stats.get("psram_largest_free_block", 0)
+            largest = int(_raw) if isinstance(_raw, (int, float)) else 0
+            if output == "human":
+                console.print(
+                    f"  [yellow]Pod PSRAM has {largest // 1024} KB contiguous free, "
+                    f"need {total_bytes // 1024} KB for this dump — falling back "
+                    f"to host TFTP.[/yellow]"
+                )
+            use_pod_tftp = False
+
     if use_pod_tftp:
         # Override host_ip + device_ip so they live on the pod's camera-
         # side subnet — the host_ip auto-detect picks something on the

src/defib/power/rack.py

@@ -138,6 +138,30 @@ async def tftp_list(self, timeout: float = 10.0) -> dict[str, object]:
         url = f"http://{self._host}:{self._port}/tftp"
         return await asyncio.to_thread(self._http_send_sync, "GET", url, None, timeout)
 
+    async def psram_can_fit(
+        self,
+        total_bytes: int,
+        headroom_bytes: int = 256 * 1024,
+    ) -> tuple[bool, dict[str, object]]:
+        """Best-effort: does the pod have a contiguous PSRAM block large
+        enough to stage ``total_bytes`` (plus ``headroom_bytes`` slack)?
+
+        Queries the pod's ``GET /tftp`` for ``psram_largest_free_block``
+        and compares.  Returns ``(fits, stats)``; on any transport error
+        returns ``(False, {})`` so the caller can treat that as "fall
+        back to host TFTP".
+
+        Used by ``defib install --tftp-via=auto`` to pick between pod and
+        host TFTP without making the user predict file sizes vs PSRAM.
+        """
+        try:
+            stats = await self.tftp_list()
+        except Exception:
+            return False, {}
+        raw = stats.get("psram_largest_free_block", 0)
+        largest = int(raw) if isinstance(raw, (int, float)) else 0
+        return largest >= total_bytes + headroom_bytes, stats
+
     @staticmethod
     def _http_send_sync(
         method: str, url: str, body: bytes | None, timeout: float,

tests/test_power_rack.py

@@ -506,3 +506,70 @@ def http503(req: Any, timeout: float | None = None) -> None:
         ctrl = RackController(host="pod", port=8080)
         with pytest.raises(PowerControllerError, match="503"):
             await ctrl.tftp_put("rootfs", b"X" * 4096)
+
+
+class TestPsramCanFit:
+    """RackController.psram_can_fit — best-effort pre-flight check
+    used by --tftp-via=auto to pick pod vs host TFTP."""
+
+    @pytest.mark.asyncio
+    async def test_fits_with_headroom(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        body = (
+            b'{"files":[],"max_size_bytes":8388608,"max_slots":4,'
+            b'"psram_free_bytes":8000000,"psram_largest_free_block":7000000}'
+        )
+        ctrl = RackController(host="pod", port=8080)
+        with patched_urlopen(monkeypatch, body=body):
+            fits, stats = await ctrl.psram_can_fit(6 * 1024 * 1024)
+        assert fits is True
+        assert stats["psram_largest_free_block"] == 7000000
+
+    @pytest.mark.asyncio
+    async def test_does_not_fit_when_largest_block_too_small(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """OpenIPC nor-ultimate is ~10 MB; an N8R2 pod with 2 MB PSRAM
+        couldn't host it. Must say no clearly so auto-mode falls back."""
+        body = (
+            b'{"files":[],"max_size_bytes":1572864,"max_slots":4,'
+            b'"psram_free_bytes":1900000,"psram_largest_free_block":1700000}'
+        )
+        ctrl = RackController(host="pod", port=8080)
+        with patched_urlopen(monkeypatch, body=body):
+            fits, stats = await ctrl.psram_can_fit(10 * 1024 * 1024)
+        assert fits is False
+        assert stats["psram_largest_free_block"] == 1700000
+
+    @pytest.mark.asyncio
+    async def test_does_not_fit_when_headroom_eats_margin(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """Exactly-equal block size must NOT pass — leave headroom for
+        the HTTP handler's transient scratch + lwip buffers."""
+        body = (
+            b'{"psram_largest_free_block":1048576}'   # 1 MiB exactly
+        )
+        ctrl = RackController(host="pod", port=8080)
+        with patched_urlopen(monkeypatch, body=body):
+            # default headroom = 256 KiB; total 1 MiB → needs 1.25 MiB
+            fits, _ = await ctrl.psram_can_fit(1024 * 1024)
+        assert fits is False
+
+    @pytest.mark.asyncio
+    async def test_pod_unreachable_returns_false(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """Network error → treat as "doesn't fit" so the CLI cleanly
+        falls back to host instead of crashing on the pre-check."""
+        import urllib.error
+
+        def boom(req: Any, timeout: float | None = None) -> None:
+            raise urllib.error.URLError("connection refused")
+
+        monkeypatch.setattr(rack_mod.urllib.request, "urlopen", boom)
+        ctrl = RackController(host="pod", port=8080)
+        fits, stats = await ctrl.psram_can_fit(1024)
+        assert fits is False
+        assert stats == {}