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

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 == {}