feat: add action validation and sanitization in AIPlanner to ensure only allowed actions are processed

ai_planner.py

@@ -24,6 +24,19 @@ import httpx
 
 OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "qwen3.5:9b")
 OLLAMA_HOST  = os.environ.get("OLLAMA_HOST",  "http://192.168.50.61:1234")
+ALLOWED_ACTIONS = {
+    "explore",
+    "move",
+    "mine_resource",
+    "craft_item",
+    "place_entity",
+    "build_smelting_line",
+    "place_belt_line",
+    "insert_to_entity",
+    "set_recipe",
+    "start_research",
+    "wait",
+}
 
 
 SYSTEM_PROMPT = """당신은 팩토리오 게임을 순수하게 플레이하는 AI 에이전트입니다.
@@ -167,6 +180,8 @@ class AIPlanner:
                 traceback.print_exc()
             plan = self._fallback_plan_from_summary(state_summary, last_error=str(e))
 
+        plan = self._sanitize_plan(plan)
+
         thinking = plan.get("thinking", "")
         if thinking:
             print(f"\n[AI] 판단:\n{thinking}\n")
@@ -244,13 +259,10 @@ class AIPlanner:
         print(f"[AI] 응답 수신 ({dt:.2f}s, {len(content)}자)")
         if _debug_enabled():
             print(f"[AI][디버그] raw={content[:300]}")
-        try:
-            return json.loads(content)
-        except json.JSONDecodeError:
-            repaired = self._extract_json_object_text(content)
-            if repaired is None:
-                raise
-            return json.loads(repaired)
+        plan = self._parse_plan_json(content)
+        if plan is None:
+            raise ValueError("유효한 계획 JSON을 찾지 못했습니다.")
+        return plan
 
     @staticmethod
     def _build_chat_payload(user_message: str) -> dict:
@@ -380,6 +392,96 @@ class AIPlanner:
                     return text[start:i + 1]
         return None
 
+    @staticmethod
+    def _iter_json_object_texts(text: str) -> list[str]:
+        out: list[str] = []
+        i = 0
+        while i < len(text):
+            start = text.find("{", i)
+            if start < 0:
+                break
+            depth = 0
+            in_str = False
+            esc = False
+            end = -1
+            for j in range(start, len(text)):
+                ch = text[j]
+                if in_str:
+                    if esc:
+                        esc = False
+                    elif ch == "\\":
+                        esc = True
+                    elif ch == '"':
+                        in_str = False
+                    continue
+                if ch == '"':
+                    in_str = True
+                elif ch == "{":
+                    depth += 1
+                elif ch == "}":
+                    depth -= 1
+                    if depth == 0:
+                        end = j
+                        break
+            if end >= 0:
+                out.append(text[start:end + 1])
+                i = end + 1
+            else:
+                break
+        return out
+
+    @classmethod
+    def _parse_plan_json(cls, content: str) -> dict | None:
+        try:
+            obj = json.loads(content)
+            if cls._looks_like_plan(obj):
+                return obj
+        except Exception:
+            pass
+        for obj_text in cls._iter_json_object_texts(content):
+            try:
+                obj = json.loads(obj_text)
+            except Exception:
+                continue
+            if cls._looks_like_plan(obj):
+                return obj
+        return None
+
+    @staticmethod
+    def _looks_like_plan(obj: object) -> bool:
+        if not isinstance(obj, dict):
+            return False
+        actions = obj.get("actions")
+        return isinstance(actions, list)
+
+    @staticmethod
+    def _sanitize_plan(plan: dict) -> dict:
+        if not isinstance(plan, dict):
+            return {"thinking": "", "current_goal": "", "actions": [], "after_this": ""}
+        actions = plan.get("actions")
+        if not isinstance(actions, list):
+            actions = []
+        valid_actions: list[dict] = []
+        for a in actions:
+            if not isinstance(a, dict):
+                continue
+            act = a.get("action")
+            if not isinstance(act, str) or act not in ALLOWED_ACTIONS:
+                continue
+            params = a.get("params")
+            if not isinstance(params, dict):
+                params = {}
+            reason = a.get("reason")
+            if not isinstance(reason, str):
+                reason = ""
+            valid_actions.append({"action": act, "params": params, "reason": reason})
+        return {
+            "thinking": str(plan.get("thinking", "")),
+            "current_goal": str(plan.get("current_goal", "")),
+            "actions": valid_actions,
+            "after_this": str(plan.get("after_this", "")),
+        }
+
     @staticmethod
     def _ensure_move_before_build_actions(actions: list[dict]) -> list[dict]:
         """