feat: spread capture polish + post-spread harness fixes

scripts/mirror_latest_tick.py

@@ -0,0 +1,46 @@
+"""Mirror the most-recently-updated per-model latest_tick.json up to the watch
+root, so a single dashboard tick server (serve_dashboard.py on :9000) follows
+whichever model is currently running in a spread.
+
+    python scripts/mirror_latest_tick.py results/spread
+
+Runs until killed. Pairs with run_spread_n1.sh (per-model output dirs) and the
+OBS score ticker (obs_studio.py overlay), which use the same discovery logic.
+"""
+
+from __future__ import annotations
+
+import shutil
+import sys
+import time
+from pathlib import Path
+
+
+def main() -> None:
+    root = Path(sys.argv[1] if len(sys.argv) > 1 else "results/spread")
+    root.mkdir(parents=True, exist_ok=True)
+    dest = root / "latest_tick.json"
+    print(f"mirroring newest {root}/*/latest_tick.json -> {dest} (ctrl-c to stop)")
+    last_src: Path | None = None
+    last_mtime = 0.0
+    while True:
+        candidates = list(root.glob("*/latest_tick.json"))
+        if candidates:
+            newest = max(candidates, key=lambda p: p.stat().st_mtime)
+            mtime = newest.stat().st_mtime
+            if newest != last_src or mtime != last_mtime:
+                try:
+                    shutil.copyfile(newest, dest)
+                    if newest != last_src:
+                        print(f"now following {newest.parent.name}")
+                    last_src, last_mtime = newest, mtime
+                except OSError:
+                    pass  # mid-write; retry next cycle
+        time.sleep(1.0)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        pass

scripts/obs_studio.py

@@ -46,14 +46,9 @@
 # ?api skips the dashboard's first-run setup screen; ?preset selects the
 # built-in capture layouts (both ship in rimapi-dashboard for fresh browser
 # profiles like OBS's embedded CEF).
-DASHBOARD_URL = (
-    "http://localhost:3000/rimapi-dashboard"
-    "?api=http%3A%2F%2Flocalhost%3A8765%2Fapi%2Fv1&preset=RLE%20Capture"
-)
-DASHBOARD_VERTICAL_URL = (
-    "http://localhost:3000/rimapi-dashboard"
-    "?api=http%3A%2F%2Flocalhost%3A8765%2Fapi%2Fv1&preset=RLE%20Vertical"
-)
+_DASH_BASE = "http://localhost:3000/rimapi-dashboard?api=http%3A%2F%2Flocalhost%3A8765%2Fapi%2Fv1&norefresh=1"
+DASHBOARD_URL = f"{_DASH_BASE}&preset=RLE%20Capture"
+DASHBOARD_VERTICAL_URL = f"{_DASH_BASE}&preset=RLE%20Vertical"
 CANVAS_W, CANVAS_H = 1920, 1080
 VERTICAL_W = 608  # 608x1080 is exactly 9:16 on a 1080p canvas
 
@@ -215,10 +210,15 @@ def cmd_setup(c: obs.ReqClient) -> None:
 
 
 def _latest_tick_file(watch: Path) -> Path | None:
+    # Prefer per-model subdirs (their parent name is the model label). Only
+    # fall back to a root-level latest_tick.json (single-scenario --output, or
+    # the spread mirror file) when no subdir candidates exist — otherwise the
+    # mirror file, written last, always wins and the label reads "spread".
     candidates = list(watch.glob("*/latest_tick.json"))
-    direct = watch / "latest_tick.json"
-    if direct.exists():
-        candidates.append(direct)
+    if not candidates:
+        direct = watch / "latest_tick.json"
+        if direct.exists():
+            candidates = [direct]
     if not candidates:
         return None
     return max(candidates, key=lambda p: p.stat().st_mtime)

src/rle/orchestration/action_executor.py

@@ -107,6 +107,11 @@ def __init__(self, client: RimAPIClient) -> None:
         # agent learns the zone exists instead of burning ticks on doomed calls
         # (issue #33).
         self._created_growing_zones: list[tuple[int, int, int, int]] = []
+        # Same guard for stockpiles: RimWorld logs one "overwriting slot group
+        # square" error PER CELL when a stockpile overlaps an existing one,
+        # which spams the dev log (and pops it over the game when auto-open
+        # is enabled).
+        self._created_stockpile_zones: list[tuple[int, int, int, int]] = []
 
     @staticmethod
     def _rects_overlap(
@@ -269,22 +274,49 @@ async def _h_growing_zone(self, cid: str, params: dict[str, Any]) -> None:
         )
         self._created_growing_zones.append(rect)
 
+    # RimWorld's StoragePriority is semantic; models reasonably emit the
+    # words. Map them instead of crashing in int() (found in the 2026-06-11
+    # spread: fable5 sent priority="important").
+    _STOCKPILE_PRIORITIES = {
+        "unstored": 0, "low": 1, "normal": 2,
+        "preferred": 3, "important": 4, "critical": 5,
+    }
+
+    @classmethod
+    def _parse_stockpile_priority(cls, value: Any) -> int:
+        if isinstance(value, str):
+            named = cls._STOCKPILE_PRIORITIES.get(value.strip().lower())
+            if named is not None:
+                return named
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return 3
+
     async def _h_stockpile_zone(self, cid: str, params: dict[str, Any]) -> None:
-        x1 = params.get("x1", params.get("x", 0))
-        z1 = params.get("z1", params.get("z", 0))
-        x2 = params.get("x2", x1 + 5)
-        z2 = params.get("z2", z1 + 5)
+        x1 = int(params.get("x1", params.get("x", 0)))
+        z1 = int(params.get("z1", params.get("z", 0)))
+        x2 = int(params.get("x2", x1 + 5))
+        z2 = int(params.get("z2", z1 + 5))
+        rect = (min(x1, x2), min(z1, z2), max(x1, x2), max(z1, z2))
+        if any(self._rects_overlap(rect, prev) for prev in self._created_stockpile_zones):
+            raise ValueError(
+                "A stockpile zone already covers these cells — it was created "
+                "earlier this run. Do NOT recreate it; pick a different, "
+                "non-overlapping rectangle or move on to another task."
+            )
         await self._client.create_stockpile_zone(
             map_id=int(params.get("map_id", 0)),
             x1=x1,
             z1=z1,
             x2=x2,
             z2=z2,
             name=params.get("name", ""),
-            priority=int(params.get("priority", 3)),
+            priority=self._parse_stockpile_priority(params.get("priority", 3)),
             allowed_item_defs=params.get("allowed_item_defs"),
             allowed_item_categories=params.get("allowed_item_categories"),
         )
+        self._created_stockpile_zones.append(rect)
 
     async def _h_designate_area(self, cid: str, params: dict[str, Any]) -> None:
         x1 = params.get("x1", params.get("x", 0))

src/rle/orchestration/game_loop.py

@@ -748,6 +748,14 @@ async def run_tick(self) -> TickResult:
     async def run(self, max_ticks: int | None = None) -> list[TickResult]:
         """Run the game loop for N ticks or until stopped."""
         self._running = True
+        if not await self._client.window_endpoints_available():
+            logger.warning(
+                "RIMAPI build lacks /api/v1/ui/window endpoints (404) — "
+                "auto-dismiss of force-pause dialogs is INERT this run. "
+                "Rebuild and deploy the rle-testing DLL (PR #77), or expect "
+                "the colony-name dialog and dev debug log to need manual "
+                "dismissal.",
+            )
         if self._no_pause:
             await self._client.unpause_game()
         tick_count = 0

src/rle/rimapi/client.py

@@ -519,6 +519,22 @@ async def list_windows(self) -> list[dict[str, Any]]:
             return []
         return data if isinstance(data, list) else []
 
+    async def window_endpoints_available(self) -> bool:
+        """Probe whether the deployed RIMAPI build has the /ui/window endpoints.
+
+        The 2026-06-11 spread ran an entire 11-model batch with auto-dismiss
+        silently inert because the Workshop DLL predated PR #77 and every
+        window call 404'd inside the swallow-everything error handling. A
+        connection error returns True (can't tell — don't warn spuriously).
+        """
+        try:
+            await self._get("/api/v1/ui/windows")
+        except RimAPIResponseError:
+            return False
+        except RimAPIConnectionError:
+            return True
+        return True
+
     async def get_resources(
         self, power_info: PowerData | None = None,
     ) -> ResourceData:

tests/unit/test_action_executor.py

@@ -183,6 +183,98 @@ async def test_non_overlapping_zone_allowed(self) -> None:
         assert client.create_growing_zone.await_count == 2
 
 
+class TestStockpileZoneIdempotency:
+    """Stockpile overlaps log one RimWorld error PER CELL ('overwriting slot
+    group square'), spamming the dev log. Same guard as growing zones."""
+
+    def _zone(self, x1: int, z1: int, x2: int, z2: int) -> Action:
+        return Action(
+            action_type="stockpile_zone",
+            parameters={"map_id": 0, "x1": x1, "z1": z1, "x2": x2, "z2": z2},
+        )
+
+    async def test_first_stockpile_succeeds(self) -> None:
+        client = AsyncMock()
+        executor = ActionExecutor(client)
+        result = await executor.execute(_make_plan(self._zone(132, 134, 136, 140)))
+        assert result.executed == 1
+        client.create_stockpile_zone.assert_awaited_once()
+
+    async def test_overlapping_repeat_is_rejected(self) -> None:
+        client = AsyncMock()
+        executor = ActionExecutor(client)
+        await executor.execute(_make_plan(self._zone(132, 134, 136, 140)))
+        result = await executor.execute(_make_plan(self._zone(132, 134, 136, 140)))
+        assert result.failed == 1
+        assert result.executed == 0
+        assert result.outcomes[0].error is not None
+        assert "already" in result.outcomes[0].error.lower()
+        client.create_stockpile_zone.assert_awaited_once()
+
+    async def test_non_overlapping_stockpile_allowed(self) -> None:
+        client = AsyncMock()
+        executor = ActionExecutor(client)
+        await executor.execute(_make_plan(self._zone(132, 134, 136, 140)))
+        result = await executor.execute(_make_plan(self._zone(150, 150, 154, 154)))
+        assert result.executed == 1
+        assert client.create_stockpile_zone.await_count == 2
+
+    async def test_growing_and_stockpile_guards_are_independent(self) -> None:
+        """A stockpile may legitimately sit where no growing zone is — the two
+        guards must not share state."""
+        client = AsyncMock()
+        executor = ActionExecutor(client)
+        grow = Action(
+            action_type="growing_zone",
+            parameters={"x1": 10, "z1": 10, "x2": 15, "z2": 15},
+        )
+        await executor.execute(_make_plan(grow))
+        result = await executor.execute(_make_plan(self._zone(10, 10, 15, 15)))
+        assert result.executed == 1
+
+
+class TestStockpilePriorityParsing:
+    """Models emit RimWorld's semantic priority words ('important'); the
+    handler crashed in int() during the 2026-06-11 spread."""
+
+    def _zone_with_priority(self, priority: object) -> Action:
+        return Action(
+            action_type="stockpile_zone",
+            parameters={"x1": 1, "z1": 1, "x2": 4, "z2": 4, "priority": priority},
+        )
+
+    async def test_semantic_priority_word_mapped(self) -> None:
+        client = AsyncMock()
+        executor = ActionExecutor(client)
+        result = await executor.execute(_make_plan(self._zone_with_priority("important")))
+        assert result.executed == 1
+        assert client.create_stockpile_zone.await_args.kwargs["priority"] == 4
+
+    async def test_priority_word_case_insensitive(self) -> None:
+        client = AsyncMock()
+        executor = ActionExecutor(client)
+        await executor.execute(_make_plan(self._zone_with_priority(" Critical ")))
+        assert client.create_stockpile_zone.await_args.kwargs["priority"] == 5
+
+    async def test_numeric_priority_passthrough(self) -> None:
+        client = AsyncMock()
+        executor = ActionExecutor(client)
+        await executor.execute(_make_plan(self._zone_with_priority(2)))
+        assert client.create_stockpile_zone.await_args.kwargs["priority"] == 2
+
+    async def test_numeric_string_priority(self) -> None:
+        client = AsyncMock()
+        executor = ActionExecutor(client)
+        await executor.execute(_make_plan(self._zone_with_priority("4")))
+        assert client.create_stockpile_zone.await_args.kwargs["priority"] == 4
+
+    async def test_garbage_priority_falls_back_to_default(self) -> None:
+        client = AsyncMock()
+        executor = ActionExecutor(client)
+        await executor.execute(_make_plan(self._zone_with_priority("highest")))
+        assert client.create_stockpile_zone.await_args.kwargs["priority"] == 3
+
+
 class TestDispatch:
     async def test_set_work_priority(self) -> None:
         client = AsyncMock()