fix(eval_optimize_loop): address AI review round 14 — 1 critical + 6 warnings

Critical:
- run_pipeline.py: replace remove-then-create lock pattern with atomic
  os.replace() from temp file, followed by ownership verification.
  Eliminates the TOCTOU window where a concurrent process could steal
  the lock between os.remove() and O_CREAT|O_EXCL.

Warning:
- baseline.py: simplify sys.path finally block — unconditional
  try/except ValueError remove, no flag variable needed
- call_agent.py: add sys.path.remove() cleanup for plate_agent_root
  insert (was permanent pollution)
- auditor.py: total_cost now directly from validation.summary instead
  of sum(e.cost_candidate) which multiplied total by candidate count
- gate.py: unknown strategy now raises ValueError instead of silently
  falling back to all_must_pass behavior
- validator.py: merge duplicate warning logic for unknown
  failure_category into single if-block
- baseline.py: remove dead image_key variable in real mode fallback;
  keep case_id traceable via image_id rather than generating garbage

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

Author: woshidage77

examples/optimization/eval_optimize_loop/run_pipeline.py

@@ -120,13 +120,22 @@ def _pid_alive(pid):
                 sys.exit(75)
             if not args.quiet:
                 print(f"Cleaning stale lock from dead PID {old_pid}", file=sys.stderr)
-            _os.remove(LOCK_FILE)
-            fd = _os.open(LOCK_FILE, _os.O_CREAT | _os.O_EXCL | _os.O_WRONLY, 0o644)
-            with _os.fdopen(fd, "w", encoding="utf-8") as lf:
-                lf.write(f"{my_pid} {started_at}")
-                lf.flush()
-                _os.fsync(lf.fileno())
-            acquired = True
+            # Atomic takeover: write PID to temp file, then os.replace()
+            # is atomic (rename) on both POSIX and Windows -- no TOCTOU window.
+            tmp = LOCK_FILE + ".tmp"
+            with open(tmp, "w", encoding="utf-8") as tf:
+                tf.write(f"{my_pid} {started_at}")
+                tf.flush()
+                _os.fsync(tf.fileno())
+            _os.replace(tmp, LOCK_FILE)
+            # Verify ownership: if another process raced us, the file contains
+            # their PID, not ours.  In that case, back off.
+            with open(LOCK_FILE, "r", encoding="utf-8") as vf:
+                lock_owner = int(vf.read().strip().split()[0])
+            if lock_owner == my_pid:
+                acquired = True
+            # else: someone else replaced the lock between our replace and read;
+            # fall through to acquired=False -> exit(75)
         except (FileNotFoundError, ValueError, FileExistsError):
             pass
 

examples/optimization/eval_optimize_loop/src/auditor.py

@@ -88,7 +88,7 @@ def build_trail(self, pipeline_name, mode, random_seed, optimization, baseline_v
             mode=mode,
             random_seed=random_seed,
             entries=entries,
-            total_cost=sum(e.cost_candidate for e in entries),
+            total_cost=validation.summary.total_cost_candidate if validation else 0.0,
             avg_latency_ms=baseline_val.summary.avg_latency_ms if baseline_val else 0.0,
         )
 

examples/optimization/eval_optimize_loop/src/baseline.py

@@ -324,20 +324,19 @@ async def _run_real_split(
         import sys
         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.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
+            try:
+                sys.path.remove(plate_root_str)
+            except ValueError:
+                pass  # already removed or never inserted
         # 构建 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
@@ -374,7 +373,9 @@ async def _run_real_split(
         for r in report.details:
             case_id = id_to_case.get(r.image_id)
             if case_id is None:
-                image_key = Path(r.image_path).name if r.image_path else ""
+                # Defence-in-depth: should not happen with sequential IDs.
+                # Use image_id (which IS the sequential id from gt_items) so
+                # the case_id remains traceable rather than generating garbage.
                 case_id = f"case_{r.image_id}"
             case_result = BaselineCaseResult(
                 case_id=case_id,

examples/optimization/eval_optimize_loop/src/call_agent.py

@@ -37,9 +37,10 @@ def create_plate_call_agent(
     to ensure evaluation isolation.
     """
     import sys
-    sys.path.insert(0, str(Path(plate_agent_root)))
-
-    async def _call_agent(query: str) -> str:
+    _plate_root = str(Path(plate_agent_root))
+    sys.path.insert(0, _plate_root)
+    try:
+        async def _call_agent(query: str) -> str:
         try:
             from trpc_agent_sdk.runners import Runner
             from trpc_agent_sdk.sessions import InMemorySessionService
@@ -107,6 +108,11 @@ async def _call_agent(query: str) -> str:
 
         return final_text.strip() or "recognition failed"
 
+    finally:
+        try:
+            sys.path.remove(_plate_root)
+        except ValueError:
+            pass
     return _call_agent
 
 

examples/optimization/eval_optimize_loop/src/gate.py

@@ -116,7 +116,7 @@ def decide(
             # Strict majority: more than half must pass. Ties (exactly half) = reject.
             accepted = sum(1 for c in checks if c.passed) > len(checks) / 2
         else:
-            accepted = all(c.passed for c in checks)
+            raise ValueError(f"Unknown gate strategy: {self.strategy!r}. Expected 'all_must_pass' or 'majority'.")
 
         reason = self._build_reason(accepted, checks)
         return GateDecision(

examples/optimization/eval_optimize_loop/src/validator.py

@@ -68,9 +68,6 @@ def run(self, val_baseline, optimization_result, simulate_regression=False):
         return self._run_real(val_baseline, candidate)
 
     def _run_fake(self, val_baseline, candidate, simulate_regression=False):
-        if candidate.failure_category not in CANDIDATE_PREDICTIONS:
-            import warnings
-            warnings.warn(f"Unknown failure_category '{candidate.failure_category}', falling back to final_answer_mismatch")
         pred_map = REGRESSION_PREDICTIONS if simulate_regression else CANDIDATE_PREDICTIONS.get(
             candidate.failure_category)
         if pred_map is None: