fix: preflight gates so keeper Crashlanded failures are not model noise

src/rle/harness/brief.py

@@ -14,7 +14,7 @@
 
 from pydantic import BaseModel, ConfigDict
 
-from rle.rimapi.api_catalog import WRITE_CATALOG
+from rle.rimapi.api_catalog import visible_write_catalog
 from rle.rimapi.schemas import GameState
 from rle.rimapi.sse_client import RimAPIEvent
 from rle.scenarios.schema import ScenarioConfig
@@ -110,7 +110,7 @@ def action_catalog() -> list[dict[str, Any]]:
     out: list[dict[str, Any]] = [
         {"action_type": "no_action", "description": "Do nothing this tick.", "params": {}},
     ]
-    for name, raw in sorted(WRITE_CATALOG.items()):
+    for name, raw in sorted(visible_write_catalog().items()):
         entry = cast(dict[str, Any], raw)
         out.append({
             "action_type": name,

src/rle/harness/felix/agents/base_role.py

@@ -56,7 +56,8 @@
     "- blueprint: Place a building. params: {\"def_name\": \"Wall\", \"x\": int, "
     "\"z\": int, \"stuff_def\": \"WoodLog\", \"rotation\": 0}.\n"
     "- growing_zone: Create growing zone. params: {\"plant_def\": \"Plant_Potato\", "
-    "\"x1\": int, \"z1\": int, \"x2\": int, \"z2\": int}.\n"
+    "\"x1\": int, \"z1\": int, \"x2\": int, \"z2\": int}. "
+    "If a growing zone already covers the cells, do not recreate it.\n"
     "- stockpile_zone: Create stockpile. params: {\"x1\": int, \"z1\": int, "
     "\"x2\": int, \"z2\": int, \"name\": \"Supply\"}.\n"
     "- designate_area: Mine/harvest/deconstruct. params: {\"type\": \"Mine\", "
@@ -72,8 +73,9 @@
     "- time_assignment: Set schedule. params: {\"hours\": [18,19,20], "
     "\"assignment\": \"Joy\"}. target_colonist_id required.\n"
     "- bed_rest: Assign bed rest. target_colonist_id required.\n"
-    "- tend: Have doctor tend patient. params: {\"doctor_pawn_id\": int}. "
-    "target_colonist_id = patient.\n"
+    "- tend: Have a living doctor tend a living patient. "
+    "params: {\"doctor_pawn_id\": int}. target_colonist_id = patient. "
+    "Both must be distinct living colonists.\n"
     "- toggle_power: Toggle building power. params: {\"building_id\": int, "
     "\"power_on\": true}.\n"
     "- no_action: Do nothing (use when colonists are already productive).\n\n"
@@ -117,7 +119,7 @@
     "best Construction→Construction=1, best Intellectual→Research=1\n"
     "- growing_zone: use FARM SITE coordinates from MAP_SUMMARY, "
     "plant_def=Plant_Rice (fastest food). Create this zone ONCE — if a "
-    "growing zone already exists, do NOT create another (it will fail).\n\n"
+    "growing zone already exists, do NOT create another.\n\n"
     "TICK 2 (SHELTER — most critical):\n"
     "- blueprint Wall: place 5x5 rectangle using SHELTER SITE from MAP_SUMMARY. "
     "Use stuff_def=WoodLog. Leave one gap for a Door.\n"

src/rle/mcp/server.py

@@ -1,6 +1,6 @@
 """RimAPI as an MCP tool server (requires the ``mcp`` extra).
 
-One tool per ``WRITE_CATALOG`` entry (executed immediately through
+One tool per visible ``WRITE_CATALOG`` entry (executed immediately through
 ``ActionExecutor`` and recorded in the tick ledger), a generic read tool over
 ``READ_CATALOG``, the harness-neutral brief, and ``end_turn``. Any MCP-capable
 coding agent can attach and play; the harness packages that do so live in
@@ -15,7 +15,7 @@
 from mcp.server.mcpserver import MCPServer
 
 from rle.mcp.session import McpSession
-from rle.rimapi.api_catalog import READ_CATALOG, WRITE_CATALOG
+from rle.rimapi.api_catalog import READ_CATALOG, visible_write_catalog
 
 SERVER_NAME = "rle"
 INSTRUCTIONS = (
@@ -80,7 +80,7 @@ def end_turn(summary: str = "") -> str:
         n = len(session.ledger.actions)
         return f"Turn ended after {n} action(s)."
 
-    for name, raw in sorted(WRITE_CATALOG.items()):
+    for name, raw in sorted(visible_write_catalog().items()):
         entry = cast(dict[str, Any], raw)
         server.add_tool(
             _make_action_tool(session, name),

src/rle/mcp/session.py

@@ -51,13 +51,19 @@ async def act(
     ) -> dict[str, Any]:
         """Execute one write immediately, record it, and return the outcome."""
         self.ledger.require_active()
+        endpoint = resolve_endpoint(action_type)
+        if endpoint == "tend" and not target_colonist_id:
+            target_colonist_id = str(
+                (parameters or {}).get("patient_pawn_id")
+                or (parameters or {}).get("patient_id")
+                or ""
+            ) or None
         action = Action(
             action_type=action_type,
             target_colonist_id=target_colonist_id,
             parameters=dict(parameters or {}),
             reason=reason,
         )
-        endpoint = resolve_endpoint(action_type)
         if endpoint == "no_action":
             self.ledger.record(action, None)
             return {"ok": True, "action_type": action_type, "note": "no-op recorded"}
@@ -74,6 +80,7 @@ async def act(
                     "error": outcome.error}
         result = await self.executor.execute(
             ActionPlan(role=self.ledger.harness_name, tick=self.ledger.game_tick, actions=[action]),
+            state=self.state,
         )
         if result.outcomes:
             outcome = result.outcomes[0]

src/rle/orchestration/action_executor.py

@@ -13,8 +13,18 @@
     ExecutionResult,
     resolve_endpoint,
 )
+from rle.orchestration.preflight import (
+    VANILLA_WORK_TYPES,
+    growing_zone_already_covers,
+    is_already_covered_error,
+    is_quarantined_write,
+    normalize_work_priorities,
+    research_target_status,
+    resolve_tend_pair,
+)
 from rle.rimapi.api_catalog import WRITE_CATALOG
-from rle.rimapi.client import RimAPIClient, RimAPIResponseError
+from rle.rimapi.client import RimAPIClient, RimAPIConnectionError, RimAPIResponseError
+from rle.rimapi.schemas import GameState
 
 __all__ = ["NEEDS_PAWN", "ActionExecutor", "ActionOutcome", "ExecutionResult"]
 
@@ -47,31 +57,6 @@ def _extract_rimapi_error(detail: str) -> str:
     return detail
 
 
-def _normalize_work_priorities(params: dict[str, Any]) -> dict[str, int]:
-    """Accept the parameter shapes models actually emit for work_priority.
-
-    Documented shape is flat ``{"<WorkType>": <1-4>}``, but frontier models
-    also emit ``{"work_type": "Research", "priority": 2}`` and
-    ``{"work_priorities": {"Growing": 1, ...}}`` (issue #27). Passing those
-    through verbatim posted garbage like work="work_type" to RIMAPI.
-    """
-    nested = params.get("work_priorities")
-    if isinstance(nested, dict):
-        return {str(work): int(pri) for work, pri in nested.items()}
-    if "work_type" in params:
-        return {str(params["work_type"]): int(params.get("priority", 1))}
-    flat = {
-        str(work): int(pri) for work, pri in params.items()
-        if isinstance(pri, int) and not isinstance(pri, bool)
-    }
-    if not flat:
-        raise ValueError(
-            'work_priority requires {"<WorkType>": <1-4>} parameters '
-            "(e.g. {\"Growing\": 1})"
-        )
-    return flat
-
-
 class ActionExecutor:
     """Dispatches agent actions to RIMAPI write endpoints.
 
@@ -85,15 +70,15 @@ def __init__(self, client: RimAPIClient) -> None:
         # Agents perseverate — they re-issue the same growing zone every tick.
         # Each repeat targets cells already owned by the first zone, which RIMAPI
         # rejects (and historically mislabelled as "Invalid plant definition").
-        # We short-circuit overlapping repeats with an explicit error so the
-        # agent learns the zone exists instead of burning ticks on doomed calls
-        # (issue #33).
+        # Overlapping repeats are success-by-state (the farm already exists).
         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]] = []
+        self._tick_state: GameState | None = None
+        self._work_types: frozenset[str] | None = None
 
     @staticmethod
     def _rects_overlap(
@@ -103,8 +88,16 @@ def _rects_overlap(
         bx1, bz1, bx2, bz2 = b
         return not (ax2 < bx1 or ax1 > bx2 or az2 < bz1 or az1 > bz2)
 
-    async def execute(self, plan: ActionPlan) -> ExecutionResult:
-        """Execute all actions in a plan, return summary + per-action outcomes."""
+    async def execute(
+        self, plan: ActionPlan, state: GameState | None = None,
+    ) -> ExecutionResult:
+        """Execute all actions in a plan, return summary + per-action outcomes.
+
+        ``state`` is the current colony snapshot used by preflight gates
+        (research availability, living doctor/patient). Harnesses that omit
+        it keep the write path; gates that need state are skipped.
+        """
+        self._tick_state = state
         executed = 0
         failed = 0
         no_action_count = 0
@@ -158,6 +151,15 @@ async def _dispatch(self, action: Action, endpoint: str) -> None:
         cid = action.target_colonist_id or ""
         params = action.parameters
 
+        if is_quarantined_write(endpoint):
+            raise ValueError(
+                f"{endpoint} is quarantined — the endpoint is missing on "
+                "this RIMAPI build. Do not advertise or call it."
+            )
+
+        if endpoint == "tend" and (not cid or cid == "0"):
+            cid = str(params.get("patient_pawn_id") or params.get("patient_id") or "")
+
         # Skip pawn-targeting actions with no valid colonist ID
         if endpoint in _NEEDS_PAWN and (not cid or cid == "0"):
             logger.info("Skipping %s: no valid colonist ID", endpoint)
@@ -180,9 +182,22 @@ async def _dispatch(self, action: Action, endpoint: str) -> None:
 
     # -- Specialized handlers (parameter mapping for complex DTOs) -----------
 
+    async def _allowed_work_types(self) -> frozenset[str]:
+        if self._work_types is not None:
+            return self._work_types
+        try:
+            names = await self._client.get_work_list()
+        except (RimAPIResponseError, RimAPIConnectionError, TypeError, AttributeError):
+            names = None
+        if isinstance(names, list) and names:
+            self._work_types = frozenset(str(item) for item in names)
+        else:
+            self._work_types = VANILLA_WORK_TYPES
+        return self._work_types
+
     async def _h_work_priority(self, cid: str, params: dict[str, Any]) -> None:
         await self._client.set_work_priorities(
-            cid, _normalize_work_priorities(params),
+            cid, normalize_work_priorities(params, await self._allowed_work_types()),
         )
 
     async def _h_draft(self, cid: str, params: dict[str, Any]) -> None:
@@ -219,7 +234,8 @@ async def _h_bed_rest(self, cid: str, params: dict[str, Any]) -> None:
         await self._client.assign_bed_rest(cid, bed_building_id=params.get("bed_building_id"))
 
     async def _h_tend(self, cid: str, params: dict[str, Any]) -> None:
-        await self._client.administer_medicine(cid, doctor_id=params.get("doctor_id"))
+        doctor_id, patient_id = resolve_tend_pair(cid, params, self._tick_state)
+        await self._client.administer_medicine(patient_id, doctor_id=doctor_id)
 
     async def _h_blueprint(self, cid: str, params: dict[str, Any]) -> None:
         if "x" not in params or "z" not in params:
@@ -240,22 +256,39 @@ async def _h_growing_zone(self, cid: str, params: dict[str, Any]) -> None:
         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))
+        map_id = int(params.get("map_id", 0))
         if any(self._rects_overlap(rect, prev) for prev in self._created_growing_zones):
-            raise ValueError(
-                "A growing 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."
+            # Already satisfied this run — do not recreate or score as failure.
+            return
+        if await self._growing_zone_covered_on_map(map_id, x1, z1, x2, z2):
+            self._created_growing_zones.append(rect)
+            return
+        try:
+            await self._client.create_growing_zone(
+                map_id=map_id,
+                plant_def=params.get("plant_def", "Plant_Potato"),
+                x1=x1,
+                z1=z1,
+                x2=x2,
+                z2=z2,
             )
-        await self._client.create_growing_zone(
-            map_id=int(params.get("map_id", 0)),
-            plant_def=params.get("plant_def", "Plant_Potato"),
-            x1=x1,
-            z1=z1,
-            x2=x2,
-            z2=z2,
-        )
+        except RimAPIResponseError as exc:
+            if is_already_covered_error(_extract_rimapi_error(exc.detail)):
+                self._created_growing_zones.append(rect)
+                return
+            raise
         self._created_growing_zones.append(rect)
 
+    async def _growing_zone_covered_on_map(
+        self, map_id: int, x1: int, z1: int, x2: int, z2: int,
+    ) -> bool:
+        """Ask ``POST /builder/check-zone`` whether a zone already owns these cells."""
+        try:
+            payload = await self._client.check_zone(map_id, x1, z1, x2, z2)
+        except (RimAPIResponseError, RimAPIConnectionError, AttributeError, TypeError):
+            return False
+        return growing_zone_already_covers(payload)
+
     # 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").
@@ -324,8 +357,22 @@ async def _h_toggle_power(self, cid: str, params: dict[str, Any]) -> None:
         )
 
     async def _h_research_target(self, cid: str, params: dict[str, Any]) -> None:
-        project = params.get("project", params.get("name", ""))
-        await self._client.set_research_target(project, force=params.get("force", False))
+        project = str(params.get("project", params.get("name", "")))
+        force = bool(params.get("force", False))
+        research = None if self._tick_state is None else self._tick_state.research
+        if research is not None and not force:
+            status = research_target_status(project, research)
+            if status == "finished":
+                raise ValueError(
+                    f"Research project '{project}' is already finished."
+                )
+            if status == "locked":
+                raise ValueError(
+                    f"Research project '{project}' is not currently available "
+                    "(prerequisites or research bench missing). Only queue "
+                    "projects listed in research.available."
+                )
+        await self._client.set_research_target(project, force=force)
 
     async def _h_research_stop(self, cid: str, params: dict[str, Any]) -> None:
         await self._client.stop_research()

src/rle/orchestration/action_resolver.py

@@ -30,6 +30,7 @@
 from pydantic import BaseModel, ConfigDict
 
 from rle.agents.actions import Action, ActionPlan, resolve_endpoint
+from rle.orchestration.preflight import extract_work_priorities
 from rle.rimapi.schemas import GameState
 
 logger = logging.getLogger(__name__)
@@ -89,17 +90,8 @@ def _winner_key(ta: _TaggedAction) -> tuple[int, int, float, int]:
 
 
 def _work_priorities(params: dict[str, Any]) -> dict[str, int]:
-    """Accepted work_priority shapes, tolerant of garbage."""
-    nested = params.get("work_priorities")
-    if isinstance(nested, dict):
-        return {str(work): int(pri) for work, pri in nested.items() if isinstance(pri, int)}
-    if "work_type" in params:
-        pri = params.get("priority", 1)
-        return {str(params["work_type"]): int(pri)} if isinstance(pri, int) else {}
-    return {
-        str(work): pri for work, pri in params.items()
-        if isinstance(pri, int) and not isinstance(pri, bool)
-    }
+    """Accepted work_priority shapes; reserved DTO keys are not work types."""
+    return extract_work_priorities(params)
 
 
 def _merge_work_priority(candidates: list[_TaggedAction]) -> Action:

src/rle/orchestration/game_loop.py

@@ -429,7 +429,7 @@ async def run_tick(self) -> TickResult:
 
         # 4. Execute — unless the harness already applied its writes
         if step.execution is None:
-            exec_result = await self._executor.execute(step.plan)
+            exec_result = await self._executor.execute(step.plan, state=state)
             for outcome in exec_result.outcomes:
                 self._emit(
                     EventType.ACTION_EXEC, tick_num,