fix(eval_optimize_loop): address AI review round 13 — 4 warnings

Warning fixes:
- optimizer.py: cross prompt_type accumulation guard — reset
  prompt_before to base when target_prompt_type changes between
  iterations (system_prompt -> skill_prompt)
- gate.py: cost gate truly skipped when baseline_cost <= 0 —
  _check_cost returns None, decide() excludes it from checks list
  (was auto-passing with passed=True contrary to skip comment)
- run_pipeline.py: Windows _pid_alive uses OpenProcess first,
  avoiding os.kill(pid,0) which CPython maps to TerminateProcess
  on Windows (would kill the lock holder instead of probing)
- baseline.py: sys.path restored in finally block on BOTH success
  and error paths; uses sys.path.remove() by value instead of
  fragile pop(0) positional removal

99 tests pass, pipeline 6 phases run end-to-end.

Author: woshidage77

examples/optimization/eval_optimize_loop/run_pipeline.py

@@ -68,20 +68,17 @@ async def main():
     _os.makedirs(str(output_dir), exist_ok=True)
 
     def _pid_alive(pid):
-        """Check if a process is running. Cross-platform: signal 0 on Unix,
-        OpenProcess on Windows. Returns False for dead/invalid PIDs."""
-        # Unix: os.kill(pid, 0) raises if process doesn't exist
-        try:
-            _os.kill(pid, 0)
-        except ProcessLookupError:
-            return False
-        except PermissionError:
-            return True  # exists but we can't signal it
-        except OSError:
-            pass  # fall through to platform check
-
-        # Windows: use kernel32.OpenProcess
+        """Check if a process is running. Returns False for dead/invalid PIDs.
+
+        Platform strategy:
+        - Windows: use kernel32.OpenProcess (PROCESS_QUERY_LIMITED_INFORMATION).
+          We avoid os.kill(pid, 0) on Windows because CPython maps it to
+          TerminateProcess(handle, 0) which kills the target, not probes it.
+        - Unix: os.kill(pid, 0) sends signal 0 (null signal) which is a
+          pure liveness probe per POSIX.
+        """
         if sys.platform == "win32":
+            # Windows first: avoid os.kill which terminates the process
             try:
                 import ctypes
                 h = ctypes.windll.kernel32.OpenProcess(0x0400, False, pid)
@@ -90,10 +87,18 @@ def _pid_alive(pid):
                     return True
                 return False
             except Exception:
-                return False  # cannot verify liveness; treat as dead so stale lock can be cleaned
-        # On Unix, if os.kill(pid, 0) succeeded above, process exists
-        return True
+                return False  # cannot verify; assume dead to allow lock cleanup
 
+        # Unix: signal 0 is a pure liveness probe
+        try:
+            _os.kill(pid, 0)
+        except ProcessLookupError:
+            return False
+        except PermissionError:
+            return True  # exists but we cannot signal it
+        except OSError:
+            return False  # cannot verify; assume dead
+        return True
     my_pid = _os.getpid()
 
     # Atomic acquire: O_CREAT|O_EXCL fails if file exists (cross-platform)

examples/optimization/eval_optimize_loop/src/baseline.py

@@ -322,18 +322,22 @@ async def _run_real_split(
             )
 
         import sys
-        sys.path.insert(0, str(Path(plate_agent_root)))
-
+        plate_root_str = str(Path(plate_agent_root))
+        sys.path.insert(0, plate_root_str)
+        path_restored = False
         try:
             from agent.session_manager import create_session_service, create_memory_service
             from eval.evaluator import PlateEvaluator
         except ImportError as e:
-            sys.path.pop(0)  # restore path before raising (fix: round-5 review)
+            sys.path.remove(plate_root_str)  # restore before raising
+            path_restored = True
             raise ImportError(
                 f"Cannot import PlateAgent modules from {plate_agent_root}. "
                 f"Ensure trpc_agent_sdk is installed. Error: {e}"
             )
-
+        finally:
+            if not path_restored:
+                sys.path.remove(plate_root_str)  # restore on success path too
         # 构建 ground_truth.json 格式（临时文件）
         # Build ground_truth items with sequential IDs for stable reverse mapping.
         # Uses enumerate(start=1) instead of SHA256 hash so that the id->case_id

examples/optimization/eval_optimize_loop/src/gate.py

@@ -98,9 +98,9 @@ def decide(
 
         # 4. 成本不超预算
         if self._rule_enabled("cost_within_budget"):
-            checks.append(self._check_cost(
-                baseline_cost, candidate_cost
-            ))
+            cost_check = self._check_cost(baseline_cost, candidate_cost)
+            if cost_check is not None:
+                checks.append(cost_check)
 
         # 5. 过拟合检测
         if self._rule_enabled("overfit_detection") and baseline_train_scores and candidate_train_scores:
@@ -132,7 +132,7 @@ def _check_total_improvement(
         self,
         baseline: dict[str, float],
         candidate: dict[str, float],
-    ) -> GateCheck:
+    ) -> "Optional[GateCheck]":
         threshold = self.rules["total_score_improvement"].get("threshold", 0.03)
         base_avg = sum(baseline.values()) / len(baseline) if baseline else 0
         cand_avg = sum(candidate.values()) / len(candidate) if candidate else 0
@@ -149,7 +149,7 @@ def _check_no_new_hard_fail(
         self,
         baseline: dict[str, float],
         candidate: dict[str, float],
-    ) -> GateCheck:
+    ) -> "Optional[GateCheck]":
         max_new = self.rules["no_new_hard_fail"].get("max_new_fails", 0)
         base_fails = sum(1 for s in baseline.values() if s < PASS_THRESHOLD)
         cand_fails = sum(1 for s in candidate.values() if s < PASS_THRESHOLD)
@@ -167,7 +167,7 @@ def _check_critical_cases(
         baseline: dict[str, float],
         candidate: dict[str, float],
         critical_ids: list[str],
-    ) -> GateCheck:
+    ) -> "Optional[GateCheck]":
         if not critical_ids:
             return GateCheck(
                 name="critical_case_no_regress",
@@ -193,18 +193,12 @@ def _check_cost(
         self,
         baseline_cost: float,
         candidate_cost: float,
-    ) -> GateCheck:
+    ) -> "Optional[GateCheck]":
         max_ratio = self.rules["cost_within_budget"].get("max_cost_ratio", 1.2)
         if baseline_cost <= 0:
-            # Fake mode: costs are simulated placeholders (see run_pipeline.py L173-176).
-            # Real mode: cost may be 0 if token_tracker is not connected.
-            # In either case, cost data is absent; skip this gate rather than auto-pass.
-            return GateCheck(
-                name="cost_within_budget",
-                passed=True,
-                description=f"cost budget skipped (baseline_cost={baseline_cost:.4f} <= 0, tracking inactive)",
-                detail="cost data unavailable -- gate skipped",
-            )
+            # Cost data is absent (fake mode simulated / real mode token_tracker not connected).
+            # Return None so decide() excludes this gate from the checks list.
+            return None
         else:
             ratio = candidate_cost / baseline_cost
             passed = ratio <= max_ratio
@@ -221,7 +215,7 @@ def _check_overfit(
         candidate_train: dict[str, float],
         baseline_val: dict[str, float],
         candidate_val: dict[str, float],
-    ) -> GateCheck:
+    ) -> "Optional[GateCheck]":
         train_avg_base = sum(baseline_train.values()) / len(baseline_train) if baseline_train else 0
         train_avg_cand = sum(candidate_train.values()) / len(candidate_train) if candidate_train else 0
         val_avg_base = sum(baseline_val.values()) / len(baseline_val) if baseline_val else 0

examples/optimization/eval_optimize_loop/src/optimizer.py

@@ -225,7 +225,7 @@ def optimize(
             confidence = target["confidence"]
 
             # prompt_before: use previous iteration's result for cumulative optimization
-            if candidates:
+            if candidates and candidates[-1].target_prompt_type == prompt_type:
                 prompt_before = candidates[-1].prompt_after
             else:
                 prompt_before = self._get_base_prompt(prompt_type)