fix(eval_optimize_loop): address AI review round 15 — 1 critical + 5 warnings + 1 suggestion

Author: woshidage77

examples/optimization/eval_optimize_loop/run_pipeline.py

@@ -128,14 +128,16 @@ def _pid_alive(pid):
                 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.
+            # Lock is ours after atomic replace.  Set acquired=True immediately
+            # so that finally-block cleanup works even if we crash during the
+            # verification read below (prevents permanent lock residue).
+            acquired = True
+            # Verify ownership: if another process somehow 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)
+            if lock_owner != my_pid:
+                acquired = False  # not our lock; don't clean up in finally
         except (FileNotFoundError, ValueError, FileExistsError):
             pass
 

examples/optimization/eval_optimize_loop/src/attribution.py

@@ -86,7 +86,9 @@ class AttributionReport:
     def primary_failure_category(self) -> Optional[AttributionCluster]:
         if not self.clusters:
             return None
-        return max(self.clusters, key=lambda c: c.count)
+        # Tie-break: (-c.count, c.category) for consistency with
+        # optimization_priority ordering (same sort key used below).
+        return sorted(self.clusters, key=lambda c: (-c.count, c.category))[0]
 
     @property
     def cluster_map(self) -> dict[str, AttributionCluster]:
@@ -131,7 +133,7 @@ def run(
         for case in val_result.failed_cases:
             all_attrs.append(self._attribute_case(case, "val"))
         clusters = self._build_clusters(all_attrs)
-        opt_priority = [c.category for c in sorted(clusters, key=lambda x: -x.count)]
+        opt_priority = [c.category for c in sorted(clusters, key=lambda x: (-x.count, x.category))]
         attributed = [a for a in all_attrs if a.category != "unattributed"]
         return AttributionReport(
             total_failures=len(all_attrs),

examples/optimization/eval_optimize_loop/src/auditor.py

@@ -59,6 +59,11 @@ def build_trail(self, pipeline_name, mode, random_seed, optimization, baseline_v
         now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
         run_id = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + f"_{random_seed}"
         entries = []
+        # NOTE: per-candidate AuditEntry records all carry the same aggregate
+        # baseline_scores / candidate_scores / cost / latency from the single
+        # validation run.  This is intentional for fake mode where there is no
+        # per-iteration re-evaluation; in real mode each iteration would have
+        # its own scores.
         for cand in optimization.candidates:
             entry = AuditEntry(
                 timestamp=now,

examples/optimization/eval_optimize_loop/src/baseline.py

@@ -362,7 +362,7 @@ async def _run_real_split(
             memory_service=memory_service,
         )
         # ???? ground_truth ??
-        evaluator.ground_truth = gt_items
+        evaluator.ground_truth = gt_items  # NOTE: relies on PlateEvaluator internal attr; fragile if field renamed
 
         report = await evaluator.run(verbose=False)
 

examples/optimization/eval_optimize_loop/src/call_agent.py

@@ -32,6 +32,10 @@ def create_plate_call_agent(
 ) -> "callable":
     """Create a PlateAgent call_agent for AgentOptimizer.
 
+    Args:
+        plate_agent_root: Absolute path to PlateAgent project root.
+        prompt_dir: Reserved for future prompt-directory override (currently unused).
+
     Returns async (query: str) -> str.
     Each call re-reads prompt from disk and creates a fresh session
     to ensure evaluation isolation.

examples/optimization/eval_optimize_loop/tests/test_validator.py

@@ -255,7 +255,7 @@ async def test_four_phase_to_gate(self):
             baseline_scores=results["val"].score_map,
             candidate_scores=val_result.score_map,
             baseline_train_scores=results["train"].score_map,
-            candidate_train_scores=results["train"].score_map,
+            candidate_train_scores=val_result.score_map,
             baseline_cost=results["val"].summary.avg_cost * results["val"].summary.total,
             candidate_cost=val_result.summary.total_cost_candidate,
         )
@@ -276,6 +276,7 @@ async def test_four_phase_to_gate(self):
         }
         j = json.dumps(full_output, ensure_ascii=False, indent=2)
         assert len(j) > 2000
+        assert decision.accepted, f"Gate should accept: {decision.reason}"
 
         print(f"\n  Gate decision: accepted={decision.accepted} reason={decision.reason[:80]}")
         print(f"  Val delta: {val_result.summary.avg_score_delta:+.3f}")