import json import re from pathlib import Path from typing import List, Optional, Union def _extract_recommendation_items(text: str) -> List[str]: """Extract compact recommendation items from numbered or bulleted text.""" if not isinstance(text, str): return [] items: List[str] = [] for raw_line in text.splitlines(): line = raw_line.strip() if not line: continue if re.match(r"^\d+\.\s+", line) or line.startswith("- ") or line.startswith("•"): items.append(line) return items def build_meta_recommendation_context_lines( results_dir: Union[str, Path], max_current_items: int = 5, max_delta_items: int = 3, ) -> Optional[List[str]]: """ Build prompt lines describing current evolution-agent strategy from meta memory. Reads `meta_memory.json` under results_dir and returns a compact section with: - current recommendations used by the evolution loop - best-effort delta vs the previous recommendation refresh """ base_path = Path(results_dir) if base_path.name == "meta_memory.json": meta_path = base_path else: meta_path = base_path / "meta_memory.json" if not meta_path.exists(): return None try: with open(meta_path, "r", encoding="utf-8") as f: meta_data = json.load(f) except Exception: return None current_recs = meta_data.get("meta_recommendations") history = meta_data.get("meta_recommendations_history", []) if not isinstance(current_recs, str) or not current_recs.strip(): return None latest_items = _extract_recommendation_items(current_recs) previous_recs = history[-2] if isinstance(history, list) and len(history) >= 2 else "" previous_items = set(_extract_recommendation_items(previous_recs)) improvement_items = [item for item in latest_items if item not in previous_items] context_lines: List[str] = [ "", "🧭 Evolution Agent Strategy Context (from meta memory):", f"- Source: {meta_path}", "- Current recommendations used by evolution agent:", ] if latest_items: for item in latest_items[:max_current_items]: context_lines.append(f" {item}") else: trimmed = current_recs.strip().replace("\n", " ") if len(trimmed) > 1200: trimmed = trimmed[:1200] + "..." context_lines.append(f" {trimmed}") context_lines.append("- Latest recommendation updates since previous refresh:") if improvement_items: for item in improvement_items[:max_delta_items]: context_lines.append(f" {item}") else: context_lines.append( " - No explicit delta detected (recommendations likely unchanged or reformatted)." ) return context_lines